feat: complete phase 2 multi-strategy deliberation and scheduling (#2062)

Co-authored-by: zhulinsen <42829555+ZhuLinsen@users.noreply.github.com>
2026-10-06 14:33:11 +08:00 · 2026-07-22 23:44:59 +08:00
parent b275b4ebea
commit de8aa6a0ba
20 changed files with 2603 additions and 18 deletions
@@ -502,6 +502,9 @@ AGENT_SKILLS=
 # AGENT_PORTFOLIO_AGENT_TIMEOUT_S=0
 # AGENT_SKILL_AGENT_TIMEOUT_S=0

+# 策略专家并发数（仅 specialist 模式生效；范围 1-4，默认 3）
+# AGENT_SKILL_CONCURRENCY=3
+
 # 风控 Agent 是否可以否决买入信号（默认开启）
 # AGENT_RISK_OVERRIDE=true

@@ -911,6 +911,17 @@ const settingsHelpZhCN: SettingsHelpMap = {
    impact: ['影响 Agent 分析的最大等待时间。'],
    notes: ['超时不影响其他股票的分析流程。'],
  },
+  'settings.agent.AGENT_SKILL_CONCURRENCY': {
+    title: '策略专家并发数',
+    summary: '控制 specialist 模式下最多同时运行多少个策略专家 Agent。',
+    usage: '默认 3，允许范围 1 到 4。调低可减少瞬时模型调用压力，调高可缩短多策略批处理等待时间。',
+    valueNotes: [
+      '该值只限制策略专家 batch 的并发，不改变最终参与综合的策略选择。',
+      '整体 Agent 超时仍是共享预算；并发数低于策略数时，单个策略会按批次数量分摊剩余预算。',
+    ],
+    impact: ['影响 specialist 多策略分析的并发度、耗时和模型调用峰值。'],
+    notes: ['单个策略失败或超时会进入诊断信息，不阻塞其它策略和最终决策。'],
+  },
  'settings.agent.AGENT_RISK_OVERRIDE': {
    title: '风险 Agent 否决权',
    summary: '允许风险 Agent 在检测到关键风险信号时否决买入信号。',
@@ -2068,6 +2079,17 @@ const settingsHelpEnUS: SettingsHelpMap = {
    impact: ['Affects the maximum wait time for Agent analysis.'],
    notes: ['Timeout does not affect other stocks in the analysis pipeline.'],
  },
+  'settings.agent.AGENT_SKILL_CONCURRENCY': {
+    title: 'Strategy Skill Concurrency',
+    summary: 'Controls how many specialist strategy agents can run at the same time in specialist mode.',
+    usage: 'Default is 3, allowed range is 1 to 4. Lower values reduce peak model pressure; higher values can shorten multi-strategy batch latency.',
+    valueNotes: [
+      'This only limits specialist batch concurrency and does not change which strategies participate in synthesis.',
+      'The overall Agent timeout remains a shared budget; when strategy count exceeds concurrency, each skill receives a budget slice based on the number of waves.',
+    ],
+    impact: ['Affects specialist multi-strategy concurrency, latency, and peak model calls.'],
+    notes: ['A single strategy failure or timeout enters diagnostics and does not block other strategies or the final decision.'],
+  },
  'settings.agent.AGENT_RISK_OVERRIDE': {
    title: 'Risk Agent Veto',
    summary: 'Allows the risk agent to veto buy signals when critical risk flags are detected.',
@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [新功能] Multi-Agent specialist 运行在分析历史保存成功后，按独立 skill 持久化版本化、低敏且幂等的有效 opinion 样本，为后续后验评估提供真实数据；本阶段不计算 outcome、不统计表现、不调整权重。
 - [新功能] Multi-Agent 报告按八态用户 action 追踪 Pipeline 最终调整，排除非法 Agent 意见；仅在 canonical action 可唯一解析时生成 explanation 与 DecisionSignal，并以同一个 `final_action` 统一最终动作契约。
 - [新功能] 新增 `--portfolio futu`，只读导入 Futu OpenD 真实账户的沪深 A 股、港股、美股 LONG 正股持仓作为分析列表。
+- [新功能] 多策略综合新增受控 deliberation v0、可注入 mediator/self-review v1-v2、只读 revision projection v3 与 multi-round v4；所有增强层相对上一层 baseline 只能保持或继续 softened，不覆盖权威最终信号。
+- [新功能] `specialist` 模式最多选择 4 个策略专家，并通过 `AGENT_SKILL_CONCURRENCY` 控制 1–4 个 worker 并发；worker 继承主管线冻结的 target date 等 `ContextVar` 状态，失败 skill 进入权威 Diagnostics 计数且不阻断其它策略或最终决策。
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
 - [修复] #2026 外股代码映射到中文显示名时英文新闻相关性判定漏判：新增同源 STOCK_ENGLISH_NAME_MAP 单一真源、canonicalize_foreign_stock_code 规范化入口与 _foreign_english_query_terms 别名解析，使 AAPL/00700/BABA 等 ticker 即使 stock_name 为中文也能在查询构建、相关性打分与多维度情报路径上复用 canonical 英文名，并补齐 .US/.HK suffix / HK 前缀全形式的归类与回归用例；同时在 _score_news_relevance 对 alias 展开 term 做去重，避免 legal alias 展开短名与显式 short alias 重复计分。
@@ -242,6 +242,7 @@ daily_stock_analysis/
 | `LOCAL_CLI_BACKEND_MAX_CONCURRENCY` | 本地 CLI backend 并发上限；范围 `1-4`，有效并发取它与 `GENERATION_BACKEND_MAX_CONCURRENCY` 的较小值 | `1` | 否 |
 | `AGENT_BACKEND` | 现有问股 Chat 的运行方式：`auto`（推荐，保持默认模型）、`litellm` 或 `codex_app_server`（实验，仅 single-agent Chat） | `auto` | 否 |
 | `AGENT_GENERATION_BACKEND` | Agent Chat 生成后端；Web 设置页仅暴露 `auto|litellm`，手写 local CLI backend 会返回 unsupported tool-calling 诊断 | `auto` | 否 |
+| `AGENT_SKILL_CONCURRENCY` | `specialist` 模式策略专家 worker 并发上限，范围 `1-4`；最多选择 4 个策略，默认 3 个并发，第 4 个进入下一批次并共享整体超时预算 | `3` | 否 |
 | `LITELLM_MODEL` | 主模型，格式 `provider/model`（如 `gemini/gemini-3.1-pro-preview`），推荐优先使用 | - | 否 |
 | `AGENT_LITELLM_MODEL` | 「默认模型」问股的主模型（可选）；留空继承主模型，无 provider 前缀按 `openai/<model>` 解析；Codex 不使用此项 | - | 否 |
 | `AGENT_CONTEXT_COMPRESSION_ENABLED` | 「默认模型」问股可见历史的 LLM 压缩开关；Codex 使用最近 20 条可见对话且保留该配置 | `false` | 否 |
@@ -210,6 +210,7 @@ Default schedule: Every weekday at **18:00 (Beijing Time)** automatic execution.
 | `LOCAL_CLI_BACKEND_MAX_CONCURRENCY` | Local CLI backend concurrency cap; range `1-4`, effective concurrency is the lower of this value and `GENERATION_BACKEND_MAX_CONCURRENCY` | `1` | No |
 | `AGENT_BACKEND` | Runtime for the existing ask-stock Chat: `auto` (recommended, preserves the default model), `litellm`, or `codex_app_server` (experimental, single-agent Chat only) | `auto` | No |
 | `AGENT_GENERATION_BACKEND` | Agent Chat generation backend. Web settings only expose `auto|litellm`; hand-written local CLI backends return an unsupported tool-calling diagnostic | `auto` | No |
+| `AGENT_SKILL_CONCURRENCY` | Specialist-mode strategy worker concurrency cap, range `1-4`. Up to four strategies are selected; the default runs three concurrently and queues the fourth under the shared pipeline budget | `3` | No |
 | `LITELLM_MODEL` | Primary model, format `provider/model` (e.g. `gemini/gemini-3.1-pro-preview`), recommended | - | No |
 | `AGENT_LITELLM_MODEL` | Optional primary model for **Default model** ask-stock; empty inherits the primary model and bare names become `openai/<model>`; Codex does not use this setting | - | No |
 | `AGENT_CONTEXT_COMPRESSION_ENABLED` | LLM compression for visible **Default model** ask-stock history; Codex uses the 20 most recent visible messages and retains this setting | `false` | No |
@@ -1,6 +1,6 @@
 # 多策略投资建议契约：Baseline 语义、Phase 1 收敛、Phase 2/3/4 边界

-本页是 Issue #1964「多策略投资建议」的专题文档，用于记录 2 个及以上策略/技能（skill）观点在系统内的**语义收敛边界**：有效证据集合、无效观点隔离、阵营分组、共识度、跨消费面一致性。Baseline 负责契约边界和现状盘点；Phase 1 只在 Baseline 契约内完成有效证据集合分拣、`strategy_synthesis` 确定性合成、DecisionAgent prompt 收敛、四条 renderer 一致性以及 E2E 反例覆盖；Phase 2 只在 Phase 1 契约下新增 2–4 策略并发调度与阶段调度；Phase 3 只在 Phase 2 之上补前端多语言完整展示；Phase 4 只在同一 `CONTRACT_VERSION = "1.0"` 内补权重回测反馈闭环。Baseline 的所有约束对后续 Phase 均永久生效，Phase N 不得静默降级 Baseline 中已经写死的边界。
+本页是 Issue #1964「多策略投资建议」的专题文档，用于记录 2 个及以上策略/技能（skill）观点在系统内的**语义收敛边界**：有效证据集合、无效观点隔离、阵营分组、共识度、跨消费面一致性。Baseline 负责契约边界和现状盘点；Phase 1 只在 Baseline 契约内完成有效证据集合分拣、`strategy_synthesis` 确定性合成、DecisionAgent prompt 收敛、四条 renderer 一致性以及 E2E 反例覆盖；Phase 1.5 在 Phase 1 契约上新增受控协同推理 v0（mediator_v0），只记录冲突议题、策略回应、softened 修正和置信度折减原因；Phase 1.6 新增可注入 LLM mediator v1（llm_mediator_v1），只允许 schema 合法的结构化修订，并在缺失、异常或越界时回退 v0；Phase 1.7 新增可注入 strategy self-review v2（self_review_v2），只允许冲突参与策略按固定 schema 自审，并在任一参与方越界时整轮回退 baseline；Phase 1.8 新增修订投影 v3（revision_projection），只预览采纳 softened 修订后的综合信号、置信度和冲突状态，不覆盖权威 `final_signal`；Phase 1.9 新增可配置多轮协同推理 v4（multi_round_v4），按 `max_rounds` 继续结构化修订并保留 `round_history`，任一轮越界时回到上一轮已验证结果；Phase 2 只在 Phase 1/1.5/1.6/1.7/1.8/1.9 契约下新增 2–4 策略并发调度与阶段调度；Phase 3 只在 Phase 2 之上补前端多语言完整展示；Phase 4 只在同一 `CONTRACT_VERSION = "1.0"` 内补权重回测反馈闭环。Baseline 的所有约束对后续 Phase 均永久生效，Phase N 不得静默降级 Baseline 中已经写死的边界。

 ## Skill opinion 样本边界（Issue #1904 P2 PR1）

@@ -134,6 +134,52 @@ Baseline 禁止使用 `sum(...) or 1.0` 之类的兜底把零权重掩盖成分
    "consensus_level": "high",
    "conflict_severity": "none",
    "conflict_count": 0
+  },
+  "deliberation": {                       // 可选；仅 material conflicts 触发
+    "status": "completed",
+    "mode": "multi_round_v4",
+    "rounds": 2,
+    "agenda": [ /* conflict agenda item */ ],
+    "responses": [ /* per-agenda participant response */ ],
+    "summary": {
+      "resolution_status": "partially_resolved",
+      "resolved_conflict_count": 0,
+      "unresolved_conflict_count": 1,
+      "minority_view_preserved": true,
+      "confidence_adjustment": -0.06,
+      "confidence_adjustment_reason_key": "deliberation.confidence.high_partially_resolved"
+    },
+    "round_history": [
+      {
+        "round": 1,
+        "source_mode": "mediator_v0",
+        "status": "baseline",
+        "changed_response_count": 2,
+        "confidence_adjustment": -0.06
+      },
+      {
+        "round": 2,
+        "source_mode": "multi_round_v4",
+        "status": "accepted",
+        "changed_response_count": 1,
+        "confidence_adjustment": -0.09
+      }
+    ]
+  },
+  "revision_projection": {                // 可选；仅 deliberation 存在时生成的 preview
+    "status": "computed",
+    "mode": "preview_only",
+    "source_mode": "mediator_v0",
+    "projected_signal": "hold",
+    "projected_weighted_score": 3.0,
+    "projected_confidence": 0.6696,
+    "projected_original_confidence": 0.72,
+    "projected_conflict_count": 1,
+    "projected_conflict_severity": "medium",
+    "projected_consensus_level": "low",
+    "changed_skill_count": 2,
+    "changed_skills": ["trend_v1", "theme_v1"],
+    "final_signal_overridden": false
  }
 }
 ```
@@ -154,9 +200,111 @@ Opinion Item 结构（`supporting_skills` / `opposing_skills` 每个元素）：

 Baseline 明确 `strategy_synthesis` 是**由 SkillAggregator 确定性算法产出的唯一权威合成结果**。Orchestrator 的 `_collect_strategy_synthesis()` 必须优先使用 `ctx.get_data("skill_consensus")` 中的 synthesis，只有在 SkillAggregator 未产出时才允许回退到 `ctx.opinions` 中的 `skill_consensus` opinion。**LLM 返回的 dashboard 不得覆盖或修改 `dashboard.strategy_synthesis`**；`normalize_dashboard_payload` 收到 LLM 输出时应剥离 LLM 侧的 `strategy_synthesis` 字段，避免 LLM 幻觉污染权威合成结果。

+### Strategy Deliberation v0（Phase 1.5）
+
+`strategy_synthesis.deliberation` 是可选协同推理块，只在中高强度冲突或明确关键冲突类型出现时生成。v0 使用确定性 `mediator_v0`，不调用 LLM、不让策略自由聊天、不修改原始 opinion、不重新计算 `final_signal`。它的职责是把冲突转成可审计议题，并记录策略回应、轻量修正与综合置信度折减原因。
+
+触发条件：
+
+- `len(valid_opinions) >= 2`
+- 且存在 `severity in {"medium", "high"}` 的 conflict，或 conflict type 属于 `directional_opposition` / `high_confidence_dissent`
+
+v0 revision 只允许：
+
+- `unchanged`：坚持原观点。
+- `softened`：仅降低 confidence，或将 `strong_buy -> buy`、`strong_sell -> sell`；`buy` / `sell` / `hold` 不反转，只可降低 confidence。
+
+v0 明确禁止：
+
+- `reversed`：不得反转观点。
+- 重新计算 `final_signal`。
+- 引入多轮 debate、并发调度、前端展示或新配置项。
+
+`deliberation.summary.confidence_adjustment` 只作为 `StrategySynthesizer` 在原 conflict severity 折减后的额外保守折减。高冲突部分缓解时默认约 `-0.06`，未缓解时约 `-0.08`；中冲突部分缓解时默认约 `-0.04`，未缓解时约 `-0.05`。该字段必须保留在 payload 中，方便后续 renderer 或 Web UI 展示“为什么置信度被继续下调”。
+
+### LLM Mediator v1（Phase 1.6）
+
+`llm_mediator_v1` 是 `StrategyDeliberation` 的可注入增强模式，不是默认运行时行为。调用方可以向 `StrategySynthesizer(deliberation_mediator=...)` 注入 `LLMDeliberationMediator`，由它先生成 v0 baseline agenda，再把低敏结构化 opinions/conflicts/baseline payload 发送给 LLM callable。LLM 只能返回同 schema 的 JSON 对象；返回文本、坏 JSON、缺字段、ID 漂移或越界 revision 时，必须无条件回退 v0。
+
+v1 schema guard：
+
+- `agenda` 必须保留 v0 的 `agenda_id` 集合；不得新增、删除或替换参与方。
+- `responses` 必须覆盖 v0 的 `(agenda_id, skill_id)` 集合；不得新增未参与策略。
+- `revision` 只允许 `unchanged` / `softened`；`reversed` 继续禁止。
+- v0 baseline 已经 `softened` 的 response 必须继续保持 `softened`，不得恢复 original signal，且 `revised_confidence` 不得高于 baseline 的已验证值。
+- v0 baseline 为 `unchanged` 的 response 可以保持不变，也可以按原规则继续 `softened`；不得反转 signal 或提高 confidence。
+- `summary.confidence_adjustment` 不得比 v0 baseline 更乐观，且单次额外折减下限为 `-0.10`，避免 LLM 撤销确定性折减。
+
+v1 输出通过校验时 `deliberation.mode="llm_mediator_v1"`；否则保持 `mediator_v0` 输出。v1 仍不调用策略 agent 自审、不多轮 debate、不重算 `final_signal`，也不新增配置项。
+
+### Strategy Self-Review v2（Phase 1.7）
+
+`self_review_v2` 是 `StrategyDeliberation` 的可注入自审模式，不是默认运行时行为。调用方可以向 `StrategySynthesizer(deliberation_mediator=...)` 注入 `StrategySelfReviewMediator`，由它先获取 baseline deliberation（可以是 `mediator_v0` 或通过校验的 `llm_mediator_v1`），再按每个 baseline response 的 `(agenda_id, skill_id)` 调用自审 callable。未来该 callable 可以由真实冲突参与 strategy agent 执行；当前契约只规定输入/输出与降级行为。
+
+v2 self-review guard：
+
+- 每个 baseline response 必须返回且只返回自己的 response JSON；不得修改其它策略回应。
+- 返回的 `agenda_id` / `skill_id` 必须与 baseline response 完全一致。
+- `revision` 仍只允许 `unchanged` / `softened`；`reversed` 继续禁止。
+- baseline 已经 `softened` 时不得改回 `unchanged`、恢复 original signal 或提高 baseline `revised_confidence`。
+- baseline 为 `unchanged` 时可以保持不变，也可以按原规则继续 `softened`；不得反转 signal 或提高 confidence。
+- v2 根据通过校验的 responses 重算 summary 时，最终 `confidence_adjustment` 不得比输入 baseline 更乐观。
+- 任一参与方缺失、坏 JSON、ID 漂移、越权修改或试图 `reversed`，整轮 self-review 回退到 baseline deliberation，禁止混合部分有效自审。
+
+v2 输出通过校验时 `deliberation.mode="self_review_v2"`。v2 仍只做一轮、不新增并发调度、不重算 `final_signal`、不改变原始 opinion，也不新增配置项。
+
+### Revision Projection v3（Phase 1.8）
+
+`strategy_synthesis.revision_projection` 是可选预览块，只在 `deliberation` 存在时由 `StrategySynthesizer` 计算。它读取已经通过 v0/v1/v2 schema guard 的 `responses`，把 `revision="softened"` 的回应应用到临时 `StrategyOpinion` 副本上，再用 confidence-weighted score 预览新的综合结果。
+
+v3 输出边界：
+
+- `revision_projection.mode` 固定为 `preview_only`。
+- `source_mode` 记录投影来源：`mediator_v0` / `llm_mediator_v1` / `self_review_v2`。
+- `projected_signal` / `projected_weighted_score` / `projected_confidence` 只描述采纳 softened 修订后的预览结果。
+- `projected_conflict_count` / `projected_conflict_severity` / `projected_consensus_level` 基于临时修订副本重新检测，不改写原始 conflicts。
+- `changed_skill_count` / `changed_skills` 只统计实际 softened 的策略。
+- `final_signal_overridden` 必须固定为 `false`，用于明确 v3 不覆盖权威最终信号。
+
+v3 明确禁止：
+
+- 把 `projected_signal` 回写到顶层 `final_signal`。
+- 把 `projected_weighted_score` 回写到顶层 `weighted_score`。
+- 把 `projected_confidence` 回写到顶层 `confidence`。
+- 在没有 `deliberation` 的场景输出空 projection。
+- 接受未经 v0/v1/v2 guard 的自由文本、反转信号或新增策略回应。
+
+v3 在投影入口还会重新核对 `original_signal`、允许的 softened signal 与 `revised_confidence` 上界；即使调用方注入了未使用内置 mediator guard 的自定义结果，也不会把更激进的 response 应用到临时 opinion 副本。
+
+### Configurable Multi-Round Deliberation v4（Phase 1.9）
+
+`multi_round_v4` 是 `StrategyDeliberation` 的可注入多轮增强模式，不是默认运行时行为。调用方可以向 `StrategySynthesizer(deliberation_mediator=...)` 注入 `MultiRoundDeliberationMediator`，并通过构造参数配置：
+
+- `fallback`：第一轮 baseline mediator，可为 `mediator_v0`、`llm_mediator_v1` 或 `self_review_v2`。
+- `max_rounds`：总轮数上限，范围 `1–4`；`1` 等价只保留 fallback baseline。
+- `stop_when_stable`：当某轮没有任何 response 变化时是否提前停止，默认开启。
+- `round_completion(round_index, messages)`：下一轮结构化修订 callable，只能返回同 schema JSON。
+
+v4 round guard：
+
+- 每轮必须保留上一轮的 `agenda_id` 集合和 `(agenda_id, skill_id)` response 集合；不得新增、删除或替换参与方。
+- `revision` 仍只允许 `unchanged` / `softened`；`reversed` 继续禁止。
+- 上一轮已经 `softened` 的 response 不能回到 `unchanged`。
+- 上一轮已经 `softened` 的 response 不能更换 `revised_signal`，也不能提高 `revised_confidence`。
+- 上一轮 `unchanged` 的 response 可以继续 `unchanged`，也可以按原规则 `softened`。
+- `summary.confidence_adjustment` 不能为正数，也不能比上一轮更乐观；单轮下限仍按 v1 guard 钳制到 `-0.10`。
+- 任一轮坏 JSON、ID 漂移、越界 revision、撤销 softened 或提高 confidence 时，停止后续轮次并返回上一轮已验证结果；如果第 2 轮即失败，则保持 fallback baseline。
+
+v4 输出：
+
+- 至少接受一轮额外修订时，`deliberation.mode="multi_round_v4"`。
+- `deliberation.rounds` 记录实际接受到的总轮数。
+- `deliberation.round_history` 记录 baseline 与每个已接受轮次的 `round`、`source_mode`、`status`、`changed_response_count` 和 `confidence_adjustment`。
+- v4 仍不重算顶层 `final_signal`，不改变原始 opinion，不直接覆盖顶层 `weighted_score` 或 `confidence`；顶层 confidence 只继续读取最终 `deliberation.summary.confidence_adjustment` 做保守折减。
+
 ### 关键不变量

-Baseline 的语义边界收敛为八条不变量。所有 Phase N 的实现必须同时满足这八条，任一违反视为契约破坏。
+Baseline 的语义边界收敛为九条不变量。所有 Phase N 的实现必须同时满足这九条，任一违反视为契约破坏。

 | ID | 不变量 | 场景 | 期望 |
 | --- | --- | --- | --- |
@@ -168,6 +316,7 @@ Baseline 的语义边界收敛为八条不变量。所有 Phase N 的实现必
 | I-6 | Payload 与 renderer 语义一致 | `dashboard.strategy_synthesis` 值 | 四条 renderer（Markdown / WeChat / Notification / History）实际文本必须与 payload 完全一致，不得出现"共识度：高 + 支持策略：无"等自相矛盾组合 |
 | I-7 | Canonical-First 评分 | Aggregator / ConflictDetector / Synthesizer 内部的评分、加权、冲突判定、分组 | 必须使用 `normalize_strategy_signal()` 返回的 canonical 小写值；禁止用大写 `"BUY"`、别名等原始字符串直接查 `strategy_signal_score` |
 | I-8 | 多语言空占位符 | `supporting_skills` / `opposing_skills` 为空时的展示 | 必须通过 `labels.none_label` 按 `report_language` 查表；禁止在代码或模板中硬编码中文 `"无"` / 英文 `"None"` / 韩文 `"없음"` 字面量 |
+| I-9 | Deliberation 单调保守 | v1/v2/v4 基于上一层已验证 baseline 修订，v3 应用 projection | 不得撤销已有 `softened`、恢复 original signal、提高 baseline revised confidence 或提高 baseline confidence adjustment；越界结果回退上一层 |

 ## Phase 1 语义收敛（本 PR 交付范围）

@@ -188,14 +337,17 @@ Phase 1 涉及的入口：

 Phase 1 不改变 `AgentOpinion` 字段、不改变 API 返回结构、不改变数据库 schema、不新增配置项、不改变现有 skill 的执行方式。

-## Phase 2 并发调度（本 PR 不做）
+## Phase 2 并发调度

-Phase 2 只在 Phase 1 契约下新增 2–4 策略并发调度与阶段调度：
+Phase 2 只在 Phase 1/1.5/1.6/1.7/1.8/1.9 契约下新增 2–4 策略并发调度与阶段调度：

- 策略执行从串行改为并发（`asyncio.gather` 或 thread pool），阶段调度中按 `SKILL_CONCURRENCY` / `SKILL_TIMEOUT_PER_SKILL` 控制。
- 单个 skill 超时或异常，走 Baseline Invalid 处理路径（`reason="skill_timeout"` / `skill_error"`），进入 Diagnostics，不阻塞其他 skill 与主流程。
- Phase 2 不改变 Baseline Evidence Chain / Diagnostics 分离原则、不改变阵营语义、不改变共识门槛、不改变 payload schema。
- Phase 2 不改变 renderer 展示逻辑；`invalid_opinion_count` 计数天然覆盖超时/异常 skill。
+- `src/agent/skills/scheduler.py::AgentSkillScheduler` 使用 thread pool 并发执行 specialist skill agents；每个 skill 使用 `AgentContext` 副本运行，并通过独立的 `copy_context()` 把主管线冻结的 target date 等 `ContextVar` 状态传播到 worker，主线程按路由顺序合并结构化 opinion，避免多个 skill 同时写共享 `ctx.opinions`。
+- specialist 最终入口最多选择 4 个策略；`AGENT_SKILL_CONCURRENCY` 控制同时运行的 worker 数，默认 `3`，范围 `1–4`。默认值下第 4 个策略进入下一 concurrency wave，不会被路由层静默丢弃。
+- `AGENT_SKILL_AGENT_TIMEOUT_S` 继续作为单个 skill 的独立超时上限；Pipeline 总预算开启时，`_run_stage_agent()` 仍取 Pipeline 剩余预算与 skill 独立上限的较小值。
+- 单个 skill 超时或异常，走 Diagnostics 路径（`reason="skill_timeout"` / `"skill_error"`），进入 `ctx.meta["invalid_opinions"]`，不阻塞其他 skill 与主流程。
+- Phase 2 不改变 Baseline Evidence Chain / Diagnostics 分离原则、不改变阵营语义、不改变共识门槛、不改变 `strategy_synthesis` payload schema。
+- Phase 2 不改变 renderer 展示逻辑；scheduler timeout/error/no-opinion 与 signal 校验失败统一进入 StrategyEngine 的 authoritative Diagnostics，`invalid_opinion_count` / `total_opinion_count` 覆盖这些失败 skill。
+- `ctx.meta["skill_scheduler"]` 仅作为运行时诊断，记录调度模式、并发数、单 skill timeout、调度数量、完成数量和 invalid 数量；不得参与综合评分。

 ## Phase 3 前端多语言完整展示（本 PR 不做）

@@ -193,9 +193,26 @@ should sum to 100; all-zero means no effective signal and must not be faked.

        invalid_opinions = ctx.meta.get("invalid_opinions") or []
        if invalid_opinions:
+            reason_labels = {
+                "skill_timeout": "执行超时",
+                "skill_error": "执行异常或未产出结构化观点",
+                "missing_signal": "signal 缺失",
+                "unrecognized_signal": "signal 无法识别",
+            }
+            reason_counts = {}
+            for item in invalid_opinions:
+                if not isinstance(item, dict):
+                    continue
+                reason = str(item.get("reason") or "unrecognized_signal")
+                reason_counts[reason] = reason_counts.get(reason, 0) + 1
+            reason_summary = "、".join(
+                f"{reason_labels.get(reason, reason)} {count} 个"
+                for reason, count in reason_counts.items()
+            )
            parts.append("## Invalid Skill Opinions (Diagnostics only — not in evidence chain)")
            parts.append(
-                f"共 {len(invalid_opinions)} 个 skill 观点因 signal 缺失或无法识别，已从证据链移除；"
+                f"共 {len(invalid_opinions)} 个 skill 观点未进入证据链"
+                f"（{reason_summary or '原因未分类'}）；"
                f"仅供你在 data_limitations 中标注，不得作为决策依据。"
            )
            parts.append("")
@@ -30,6 +30,7 @@ import logging
 import re
 import time
 from dataclasses import dataclass, field
+from math import ceil
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional

 from src.agent.chat_context import build_visible_chat_history
@@ -48,6 +49,7 @@ from src.agent.protocols import (
 )
 from src.agent.skills.defaults import is_skill_agent_name
 from src.agent.skills.engine import EvidencePartition, StrategyEngine, StrategyResult, StrategyResultStatus
+from src.agent.skills.scheduler import AgentSkillScheduler, SkillBatchResult
 from src.agent.risk_override import (
    RiskOverrideApplication,
    build_risk_override_application,
@@ -163,6 +165,14 @@ class AgentOrchestrator:
            if (val := getattr(config, attr, None)) is not None and val > 0
        }

+    def _get_skill_concurrency(self) -> int:
+        raw_value = getattr(self.config, "agent_skill_concurrency", 3)
+        try:
+            parsed = int(raw_value or 3)
+        except (TypeError, ValueError):
+            parsed = 3
+        return max(1, min(4, parsed))
+
    def _build_timeout_result(
        self,
        stats: AgentRunStats,
@@ -579,7 +589,34 @@ class AgentOrchestrator:
                self._skill_agent_names = {a.agent_name for a in specialist_agents}
                specialist_agents_inserted = True
                if specialist_agents:
-                    agents[index:index] = specialist_agents
+                    batch = self._run_specialist_agent_batch(
+                        specialist_agents,
+                        ctx,
+                        progress_callback=progress_callback,
+                        timeout_seconds=remaining_budget,
+                    )
+                    for stage_result in batch.stage_results:
+                        stats.record_stage(stage_result)
+                        all_tool_calls.extend(
+                            tc for tc in (stage_result.meta.get("tool_calls_log") or [])
+                        )
+                        models_used.extend(stage_result.meta.get("models_used", []))
+                        if stage_result.status == StageStatus.FAILED:
+                            self._record_degraded_stage(ctx, stage_result.stage_name, stage_result)
+                    ctx.opinions.extend(batch.opinions)
+                    invalid_bucket = ctx.meta.get("invalid_opinions")
+                    if not isinstance(invalid_bucket, list):
+                        invalid_bucket = []
+                    invalid_bucket.extend(batch.invalid_records)
+                    ctx.meta["invalid_opinions"] = invalid_bucket
+                    ctx.meta["skill_scheduler"] = {
+                        "mode": "thread_pool",
+                        "max_concurrency": batch.max_concurrency,
+                        "timeout_per_skill": batch.timeout_per_skill,
+                        "scheduled_skill_count": len(specialist_agents),
+                        "completed_skill_count": sum(1 for item in batch.stage_results if item.success),
+                        "invalid_skill_count": len(batch.invalid_records),
+                    }
                    continue

            if agent.agent_name == "decision":
@@ -774,13 +811,13 @@ class AgentOrchestrator:
                technical_skill_policy=self.technical_skill_policy,
            )
            router = SkillRouter()
-            selected = router.select_skills(ctx)
+            selected = router.select_skills(ctx, max_count=4)
            if not selected:
                return []

            from src.agent.skills.skill_agent import SkillAgent
            agents = []
-            for skill_id in selected[:3]:  # cap at 3 concurrent skills
+            for skill_id in selected:
                agent = self._prepare_agent(SkillAgent(
                    skill_id=skill_id,
                    **common_kwargs,
@@ -799,6 +836,77 @@ class AgentOrchestrator:
        """Compatibility wrapper for legacy tests/imports."""
        return self._build_specialist_agents(ctx)

+    def _run_specialist_agent_batch(
+        self,
+        agents: list,
+        ctx: AgentContext,
+        *,
+        progress_callback: Optional[Callable] = None,
+        timeout_seconds: Optional[float] = None,
+    ) -> SkillBatchResult:
+        sub_agent_timeout_map = self._get_sub_agent_timeout_map()
+        configured_skill_timeout = sub_agent_timeout_map.get("skill", 0.0)
+        budget_per_skill = self._skill_batch_timeout_slice(
+            len(agents),
+            timeout_seconds=timeout_seconds,
+        )
+        if configured_skill_timeout and budget_per_skill is not None:
+            timeout_per_skill = min(configured_skill_timeout, budget_per_skill)
+        elif configured_skill_timeout:
+            timeout_per_skill = configured_skill_timeout
+        elif budget_per_skill is not None:
+            timeout_per_skill = budget_per_skill
+        else:
+            timeout_per_skill = 0.0
+        scheduler = AgentSkillScheduler(
+            max_concurrency=self._get_skill_concurrency(),
+            timeout_per_skill=timeout_per_skill,
+        )
+        if progress_callback:
+            for agent in agents:
+                progress_callback(stream_event(
+                    "stage_start",
+                    stage=agent.agent_name,
+                    message=f"Starting {agent.agent_name} analysis...",
+                ))
+
+        batch = scheduler.run(
+            agents,
+            ctx,
+            self._run_stage_agent,
+            progress_callback=progress_callback,
+        )
+        if progress_callback:
+            for result in batch.stage_results:
+                progress_callback(stream_event(
+                    "stage_done",
+                    stage=result.stage_name,
+                    status=result.status.value,
+                    duration=result.duration_s,
+                ))
+        return batch
+
+    def _skill_batch_timeout_slice(
+        self,
+        agent_count: int,
+        *,
+        timeout_seconds: Optional[float],
+    ) -> Optional[float]:
+        """Split remaining specialist budget across queued concurrency waves."""
+        if timeout_seconds is None:
+            return None
+        try:
+            remaining = float(timeout_seconds)
+        except (TypeError, ValueError):
+            return None
+        if remaining <= 0:
+            return 0.0
+
+        count = max(1, int(agent_count or 1))
+        worker_count = min(self._get_skill_concurrency(), count)
+        wave_count = max(1, ceil(count / worker_count))
+        return remaining / wave_count
+
    # -----------------------------------------------------------------
    # Skill aggregation
    # -----------------------------------------------------------------
@@ -894,7 +1002,13 @@ class AgentOrchestrator:
        Replaces the old two-step _partition_skill_opinions + _aggregate_skill_opinions
        calls. The engine is the single authoritative owner of strategy_synthesis.
        """
-        result = self.strategy_engine.process(ctx.opinions)
+        existing_invalid = ctx.meta.get("invalid_opinions")
+        if not isinstance(existing_invalid, list):
+            existing_invalid = []
+        result = self.strategy_engine.process(
+            ctx.opinions,
+            diagnostic_records=existing_invalid,
+        )

        ctx.meta["invalid_opinions"] = list(result.invalid_records)
        ctx.opinions = list(result.non_skill_opinions) + list(result.valid_skill_opinions)
@@ -31,6 +31,10 @@ __all__ = [
    "SkillAgent",
    "SkillRouter",
    "SkillAggregator",
+    "DeliberationMediator",
+    "LLMDeliberationMediator",
+    "StrategySelfReviewMediator",
+    "MultiRoundDeliberationMediator",
    "DEFAULT_ACTIVE_SKILL_IDS",
    "DEFAULT_ROUTER_SKILL_IDS",
    "PRIMARY_DEFAULT_SKILL_ID",
@@ -59,4 +63,20 @@ def __getattr__(name):
        from src.agent.skills.aggregator import SkillAggregator

        return SkillAggregator
+    if name == "DeliberationMediator":
+        from src.agent.skills.deliberation import DeliberationMediator
+
+        return DeliberationMediator
+    if name == "LLMDeliberationMediator":
+        from src.agent.skills.deliberation import LLMDeliberationMediator
+
+        return LLMDeliberationMediator
+    if name == "StrategySelfReviewMediator":
+        from src.agent.skills.deliberation import StrategySelfReviewMediator
+
+        return StrategySelfReviewMediator
+    if name == "MultiRoundDeliberationMediator":
+        from src.agent.skills.deliberation import MultiRoundDeliberationMediator
+
+        return MultiRoundDeliberationMediator
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -131,8 +131,22 @@ class StrategyEngine:
            evidence_opinions=evidence_opinions,
        )

-    def process(self, opinions: List[AgentOpinion]) -> StrategyResult:
-        return self.process_partition(self.partition_only(opinions))
+    def process(
+        self,
+        opinions: List[AgentOpinion],
+        *,
+        diagnostic_records: Optional[List[Dict[str, Any]]] = None,
+    ) -> StrategyResult:
+        partition = self.partition_only(opinions)
+        existing_diagnostics = [
+            dict(record)
+            for record in (diagnostic_records or [])
+            if isinstance(record, dict)
+        ]
+        if existing_diagnostics:
+            partition.invalid_records = existing_diagnostics + partition.invalid_records
+            partition.invalid_count = len(partition.invalid_records)
+        return self.process_partition(partition)

    def process_partition(self, partition: EvidencePartition) -> StrategyResult:
        if not partition.valid_skill_opinions:
@@ -0,0 +1,176 @@
+# -*- coding: utf-8 -*-
+"""
+Concurrent scheduler for specialist strategy-skill agents.
+"""
+
+from __future__ import annotations
+
+import logging
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from contextvars import copy_context
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+from src.agent.protocols import AgentContext, AgentOpinion, StageResult, StageStatus
+logger = logging.getLogger(__name__)
+
+RunStageCallable = Callable[[Any, AgentContext, Optional[Callable], Optional[float]], StageResult]
+
+
+@dataclass
+class SkillBatchResult:
+    stage_results: List[StageResult] = field(default_factory=list)
+    opinions: List[AgentOpinion] = field(default_factory=list)
+    invalid_records: List[Dict[str, Any]] = field(default_factory=list)
+    max_concurrency: int = 1
+    timeout_per_skill: float = 0.0
+
+
+class AgentSkillScheduler:
+    """Run selected skill agents concurrently without sharing mutable context writes."""
+
+    def __init__(self, *, max_concurrency: int = 3, timeout_per_skill: float = 0.0) -> None:
+        self.max_concurrency = _clamp_int(max_concurrency, minimum=1, maximum=4)
+        self.timeout_per_skill = max(0.0, float(timeout_per_skill or 0.0))
+
+    def run(
+        self,
+        agents: List[Any],
+        ctx: AgentContext,
+        run_stage: RunStageCallable,
+        *,
+        progress_callback: Optional[Callable] = None,
+    ) -> SkillBatchResult:
+        selected_agents = list(agents)
+        if not selected_agents:
+            return SkillBatchResult(
+                max_concurrency=self.max_concurrency,
+                timeout_per_skill=self.timeout_per_skill,
+            )
+
+        stage_results_by_index: Dict[int, StageResult] = {}
+        opinions_by_index: Dict[int, List[AgentOpinion]] = {}
+        invalid_by_index: Dict[int, Dict[str, Any]] = {}
+        worker_count = min(self.max_concurrency, len(selected_agents))
+        executor = ThreadPoolExecutor(max_workers=worker_count, thread_name_prefix="agent-skill")
+        futures = {
+            executor.submit(
+                copy_context().run,
+                self._run_one,
+                agent,
+                ctx,
+                run_stage,
+                progress_callback,
+            ): (index, agent)
+            for index, agent in enumerate(selected_agents)
+        }
+
+        try:
+            for future in as_completed(futures):
+                index, agent = futures[future]
+                try:
+                    result, opinions = future.result()
+                except Exception as exc:
+                    logger.warning("[AgentSkillScheduler] skill '%s' failed: %s", agent.agent_name, exc)
+                    result = StageResult(
+                        stage_name=getattr(agent, "agent_name", ""),
+                        status=StageStatus.FAILED,
+                        error=str(exc),
+                    )
+                    opinions = []
+
+                stage_results_by_index[index] = result
+                if opinions:
+                    opinions_by_index[index] = opinions
+                invalid = self._invalid_record_for(agent, result, opinions)
+                if invalid is not None:
+                    invalid_by_index[index] = invalid
+        finally:
+            executor.shutdown(wait=True, cancel_futures=True)
+
+        ordered_results = [
+            stage_results_by_index[index]
+            for index in range(len(selected_agents))
+            if index in stage_results_by_index
+        ]
+        ordered_opinions = [
+            opinion
+            for index in range(len(selected_agents))
+            for opinion in opinions_by_index.get(index, [])
+        ]
+        ordered_invalid = [
+            invalid_by_index[index]
+            for index in range(len(selected_agents))
+            if index in invalid_by_index
+        ]
+        return SkillBatchResult(
+            stage_results=ordered_results,
+            opinions=ordered_opinions,
+            invalid_records=ordered_invalid,
+            max_concurrency=self.max_concurrency,
+            timeout_per_skill=self.timeout_per_skill,
+        )
+
+    def _run_one(
+        self,
+        agent: Any,
+        ctx: AgentContext,
+        run_stage: RunStageCallable,
+        progress_callback: Optional[Callable],
+    ) -> tuple[StageResult, List[AgentOpinion]]:
+        isolated_ctx = _clone_context_for_skill(ctx)
+        opinion_count_before = len(isolated_ctx.opinions)
+        timeout = self.timeout_per_skill if self.timeout_per_skill > 0 else None
+        result = run_stage(agent, isolated_ctx, progress_callback, timeout)
+
+        opinions: List[AgentOpinion] = []
+        if result.opinion is not None:
+            opinions.append(result.opinion)
+        else:
+            opinions.extend(isolated_ctx.opinions[opinion_count_before:])
+        return result, opinions
+
+    @staticmethod
+    def _invalid_record_for(
+        agent: Any,
+        result: StageResult,
+        opinions: List[AgentOpinion],
+    ) -> Optional[Dict[str, Any]]:
+        if result.success and opinions:
+            return None
+
+        error = result.error or ""
+        reason = "skill_timeout" if "timeout" in error.lower() or "timed out" in error.lower() else "skill_error"
+        if result.success and not opinions:
+            reason = "skill_error"
+            error = error or "skill completed without a structured opinion"
+
+        return {
+            "agent_name": getattr(agent, "agent_name", result.stage_name),
+            "raw_signal": None,
+            "confidence": 0.0,
+            "reason": reason,
+            "error": error,
+        }
+
+
+def _clone_context_for_skill(ctx: AgentContext) -> AgentContext:
+    return AgentContext(
+        query=ctx.query,
+        stock_code=ctx.stock_code,
+        stock_name=ctx.stock_name,
+        session_id=ctx.session_id,
+        data=dict(ctx.data or {}),
+        opinions=list(ctx.opinions or []),
+        risk_flags=[dict(flag) for flag in (ctx.risk_flags or []) if isinstance(flag, dict)],
+        meta=dict(ctx.meta or {}),
+        created_at=ctx.created_at,
+    )
+
+
+def _clamp_int(value: Any, *, minimum: int, maximum: int) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        parsed = minimum
+    return max(minimum, min(maximum, parsed))
@@ -5,7 +5,7 @@ Strategy synthesis helpers for skill-agent consensus.

 from __future__ import annotations

-from dataclasses import asdict
+from dataclasses import asdict, replace
 from typing import Any, Dict, Iterable, List, Optional

 from src.agent.protocols import (
@@ -15,9 +15,17 @@ from src.agent.protocols import (
    normalize_strategy_signal,
    strategy_signal_score,
 )
+from src.agent.skills.deliberation import DeliberationMediator
 from src.agent.skills.defaults import extract_skill_id

 _SEVERITY_RANK = {"none": 0, "low": 1, "medium": 2, "high": 3}
+_SCORE_TO_SIGNAL = [
+    (4.5, "strong_buy"),
+    (3.5, "buy"),
+    (2.5, "hold"),
+    (1.5, "sell"),
+    (0.0, "strong_sell"),
+]


 def strategy_opinion_from_agent_opinion(opinion: AgentOpinion) -> StrategyOpinion:
@@ -195,6 +203,12 @@ class ConflictDetector:
 class StrategySynthesizer:
    """Build an explainable synthesis payload for strategy consensus."""

+    def __init__(
+        self,
+        deliberation_mediator: Optional[Any] = None,
+    ) -> None:
+        self.deliberation_mediator = deliberation_mediator or DeliberationMediator()
+
    def synthesize(
        self,
        opinions: List[StrategyOpinion],
@@ -208,6 +222,17 @@ class StrategySynthesizer:
    ) -> Dict[str, Any]:
        conflict_severity = _highest_severity(conflicts)
        adjusted_confidence = self.adjust_confidence(weighted_confidence, conflict_severity)
+        deliberation = self.deliberation_mediator.deliberate(
+            opinions,
+            conflicts,
+            final_signal=final_signal,
+        )
+        if deliberation is not None:
+            adjusted_confidence = max(
+                0.0,
+                min(1.0, adjusted_confidence + deliberation.summary.confidence_adjustment),
+            )
+        revision_projection = self._build_revision_projection(opinions, deliberation)
        final_score = strategy_signal_score(final_signal)
        supporting, opposing = self._group_opinions(opinions, final_score)
        consensus_level = self._consensus_level(
@@ -222,7 +247,7 @@ class StrategySynthesizer:
        # list itself. In the E2E path the explicit partition value takes precedence.
        invalid_count = max(invalid_count, sum(1 for op in opinions if op.invalid_signal))

-        return {
+        payload = {
            "final_signal": final_signal,
            "weighted_score": round(weighted_score, 4),
            "confidence": round(adjusted_confidence, 4),
@@ -244,6 +269,11 @@ class StrategySynthesizer:
                "conflict_count": len(conflicts),
            },
        }
+        if deliberation is not None:
+            payload["deliberation"] = deliberation.to_dict()
+        if revision_projection is not None:
+            payload["revision_projection"] = revision_projection
+        return payload

    @staticmethod
    def adjust_confidence(confidence: float, conflict_severity: str) -> float:
@@ -346,6 +376,69 @@ class StrategySynthesizer:
            return "low"
        return "medium"

+    def _build_revision_projection(
+        self,
+        opinions: List[StrategyOpinion],
+        deliberation: Optional[Any],
+    ) -> Optional[Dict[str, Any]]:
+        """Preview synthesis if softened deliberation responses were accepted.
+
+        This is intentionally preview-only: it does not mutate opinions and does
+        not override the authoritative final_signal already produced upstream.
+        """
+        if deliberation is None:
+            return None
+
+        valid_opinions = [op for op in opinions if not op.invalid_signal]
+        if not valid_opinions:
+            return None
+
+        softened_by_skill = _softened_response_by_skill(deliberation.responses)
+        projected_opinions: List[StrategyOpinion] = []
+        changed_skills: List[str] = []
+        for opinion in valid_opinions:
+            response = softened_by_skill.get(opinion.skill_id)
+            if response is None or not _is_safe_projection_response(response, opinion):
+                projected_opinions.append(opinion)
+                continue
+
+            projected_opinions.append(
+                replace(
+                    opinion,
+                    signal=response.revised_signal,
+                    confidence=max(0.0, min(1.0, response.revised_confidence)),
+                )
+            )
+            if opinion.skill_id not in changed_skills:
+                changed_skills.append(opinion.skill_id)
+
+        weighted_score, weighted_confidence = _confidence_weighted_projection(projected_opinions)
+        projected_signal = _signal_from_score(weighted_score)
+        projected_conflicts = ConflictDetector().detect(projected_opinions, final_signal=projected_signal)
+        projected_conflict_severity = _highest_severity(projected_conflicts)
+        projected_confidence = self.adjust_confidence(weighted_confidence, projected_conflict_severity)
+
+        return {
+            "status": "computed",
+            "mode": "preview_only",
+            "source_mode": str(getattr(deliberation, "mode", "")),
+            "projected_signal": projected_signal,
+            "projected_weighted_score": round(weighted_score, 4),
+            "projected_confidence": round(projected_confidence, 4),
+            "projected_original_confidence": round(weighted_confidence, 4),
+            "projected_conflict_count": len(projected_conflicts),
+            "projected_conflict_severity": projected_conflict_severity,
+            "projected_consensus_level": self._consensus_level(
+                projected_opinions,
+                projected_conflicts,
+                projected_signal,
+                insufficient_evidence=False,
+            ),
+            "changed_skill_count": len(changed_skills),
+            "changed_skills": changed_skills,
+            "final_signal_overridden": False,
+        }
+

 def _as_float(value: Any, default: float) -> float:
    try:
@@ -378,6 +471,73 @@ def _highest_severity(conflicts: List[StrategyConflict]) -> str:
    return max((conflict.severity for conflict in conflicts), key=lambda severity: _SEVERITY_RANK.get(severity, 0))


+def _confidence_weighted_projection(opinions: List[StrategyOpinion]) -> tuple[float, float]:
+    weighted: List[tuple[StrategyOpinion, float]] = [
+        (opinion, max(0.0, min(1.0, opinion.confidence)))
+        for opinion in opinions
+        if not opinion.invalid_signal
+    ]
+    weight_sum = sum(weight for _, weight in weighted)
+    if weight_sum <= 0:
+        return 3.0, 0.0
+
+    weighted_score = sum(
+        strategy_signal_score(opinion.signal) * weight
+        for opinion, weight in weighted
+    ) / weight_sum
+    weighted_confidence = sum(
+        max(0.0, min(1.0, opinion.confidence)) * weight
+        for opinion, weight in weighted
+    ) / weight_sum
+    return weighted_score, weighted_confidence
+
+
+def _signal_from_score(score: float) -> str:
+    for threshold, signal in _SCORE_TO_SIGNAL:
+        if score >= threshold:
+            return signal
+    return "hold"
+
+
+def _softened_response_by_skill(responses: Any) -> Dict[str, Any]:
+    result: Dict[str, Any] = {}
+    if not isinstance(responses, list):
+        return result
+
+    for response in responses:
+        if getattr(response, "revision", "") != "softened":
+            continue
+        skill_id = str(getattr(response, "skill_id", "") or "")
+        revised_signal = str(getattr(response, "revised_signal", "") or "")
+        if not skill_id or not revised_signal:
+            continue
+        _, invalid, _ = normalize_strategy_signal(revised_signal)
+        if invalid:
+            continue
+
+        current = result.get(skill_id)
+        if current is None or _projection_preference(response) < _projection_preference(current):
+            result[skill_id] = response
+    return result
+
+
+def _is_safe_projection_response(response: Any, opinion: StrategyOpinion) -> bool:
+    """Reject projection inputs that are more aggressive than the source opinion."""
+    if str(getattr(response, "original_signal", "") or "") != opinion.signal:
+        return False
+    revised_signal = str(getattr(response, "revised_signal", "") or "")
+    if revised_signal != DeliberationMediator._softened_signal(opinion.signal):
+        return False
+    revised_confidence = _as_float(getattr(response, "revised_confidence", None), -1.0)
+    return 0.0 <= revised_confidence <= opinion.confidence + 0.0001
+
+
+def _projection_preference(response: Any) -> tuple[float, float]:
+    signal = str(getattr(response, "revised_signal", "") or "hold")
+    confidence = _as_float(getattr(response, "revised_confidence", 0.0), 0.0)
+    return abs(strategy_signal_score(signal) - strategy_signal_score("hold")), confidence
+
+
 def _opinion_to_item(opinion: StrategyOpinion) -> Dict[str, Any]:
    return {
        "skill_id": opinion.skill_id,
@@ -856,6 +856,7 @@ class Config:
    agent_decision_agent_timeout_s: float = 0
    agent_portfolio_agent_timeout_s: float = 0
    agent_skill_agent_timeout_s: float = 0
+    agent_skill_concurrency: int = 3
    agent_risk_override: bool = True  # Allow risk agent to veto buy signals
    agent_deep_research_budget: int = 30000  # Max token budget for deep research
    agent_deep_research_timeout: int = 180  # Max seconds for /research command before returning timeout
@@ -1787,6 +1788,13 @@ class Config:
                os.getenv('AGENT_SKILL_AGENT_TIMEOUT_S'), 0,
                field_name='AGENT_SKILL_AGENT_TIMEOUT_S', minimum=0,
            ),
+            agent_skill_concurrency=parse_env_int(
+                os.getenv('AGENT_SKILL_CONCURRENCY'),
+                3,
+                field_name='AGENT_SKILL_CONCURRENCY',
+                minimum=1,
+                maximum=4,
+            ),
            agent_risk_override=os.getenv('AGENT_RISK_OVERRIDE', 'true').lower() == 'true',
            agent_deep_research_budget=parse_env_int(
                os.getenv('AGENT_DEEP_RESEARCH_BUDGET'),
@@ -4014,6 +4014,32 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
        ],
        "warning_codes": [],
    },
+    "AGENT_SKILL_CONCURRENCY": {
+        "title": "Strategy Skill Concurrency",
+        "description": "Maximum number of specialist strategy-skill agents to run concurrently in specialist mode.",
+        "category": "agent",
+        "data_type": "integer",
+        "ui_control": "number",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": "3",
+        "options": [],
+        "validation": {"min": 1, "max": 4},
+        "display_order": 64,
+        "help_key": "settings.agent.AGENT_SKILL_CONCURRENCY",
+        "examples": [
+            "AGENT_SKILL_CONCURRENCY=3",
+            "AGENT_SKILL_CONCURRENCY=4",
+        ],
+        "docs": [
+            {
+                "label": "完整指南：Agent 配置",
+                "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#环境变量完整列表",
+            },
+        ],
+        "warning_codes": [],
+    },
    "AGENT_DEEP_RESEARCH_BUDGET": {
        "title": "Deep Research Token Budget",
        "description": "Maximum token budget for Deep Research planning, follow-up research, and final synthesis.",
@@ -51,11 +51,13 @@ class TestAgentConfig(unittest.TestCase):
        self.assertFalse(config.agent_mode)
        self.assertEqual(config.agent_max_steps, AGENT_MAX_STEPS_DEFAULT)
        self.assertEqual(config.agent_skills, [])
+        self.assertEqual(config.agent_skill_concurrency, 3)

    @patch.dict(os.environ, {
        'AGENT_MODE': 'true',
        'AGENT_MAX_STEPS': '15',
        'AGENT_SKILLS': 'dragon_head,shrink_pullback,volume_breakout',
+        'AGENT_SKILL_CONCURRENCY': '4',
    }, clear=True)
    def test_agent_config_from_env(self):
        """Agent config should be loaded from environment."""
@@ -65,6 +67,15 @@ class TestAgentConfig(unittest.TestCase):
        self.assertTrue(config.agent_mode)
        self.assertEqual(config.agent_max_steps, 15)
        self.assertEqual(config.agent_skills, ['dragon_head', 'shrink_pullback', 'volume_breakout'])
+        self.assertEqual(config.agent_skill_concurrency, 4)
+
+    @patch.dict(os.environ, {'AGENT_SKILL_CONCURRENCY': '9'}, clear=True)
+    def test_agent_skill_concurrency_is_clamped(self):
+        """Agent skill concurrency should stay within the supported 1-4 range."""
+        from src.config import Config
+        Config._instance = None
+        config = Config._load_from_env()
+        self.assertEqual(config.agent_skill_concurrency, 4)

    @patch.dict(os.environ, {'AGENT_MODE': 'false'}, clear=True)
    def test_agent_mode_disabled(self):
@@ -362,6 +362,7 @@ class TestSettingsHelpMetadata(unittest.TestCase):
        "AGENT_ARCH",
        "AGENT_ORCHESTRATOR_MODE",
        "AGENT_ORCHESTRATOR_TIMEOUT_S",
+        "AGENT_SKILL_CONCURRENCY",
        "AGENT_RISK_OVERRIDE",
        "AGENT_DEEP_RESEARCH_BUDGET",
        "AGENT_DEEP_RESEARCH_TIMEOUT",
@@ -733,8 +733,12 @@ class TestStrategyAggregator(unittest.TestCase):

        self.assertEqual(synthesis["final_signal"], "hold")
        self.assertEqual(synthesis["conflict_severity"], "high")
-        self.assertAlmostEqual(synthesis["confidence"], 0.68)
+        # high conflict severity first reduces 0.8 -> 0.68; mediator_v0 then
+        # applies a partially-resolved deliberation adjustment of -0.06.
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
        self.assertEqual(synthesis["summary_key"], "strategy_synthesis.with_conflicts")
+        self.assertEqual(synthesis["deliberation"]["status"], "completed")
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
        self.assertNotIn("summary", synthesis)
        self.assertEqual(synthesis["summary_params"]["final_signal"], "hold")
        self.assertNotIn("综合信号", json.dumps(synthesis, ensure_ascii=False))
@@ -1009,6 +1013,23 @@ class TestOrchestratorModes(unittest.TestCase):
        self.assertEqual(ctx.stock_name, "贵州茅台")
        self.assertEqual(ctx.meta["skills_requested"], ["bull_trend"])

+    def test_specialist_builder_allows_four_requested_skills(self):
+        orch = self._make_orchestrator("specialist")
+        ctx = AgentContext(query="test", stock_code="600519")
+        ctx.meta["skills_requested"] = [
+            "bull_trend",
+            "hot_theme",
+            "fund_flow",
+            "chan_theory",
+        ]
+
+        agents = orch._build_specialist_agents(ctx)
+
+        self.assertEqual(
+            [agent.skill_id for agent in agents],
+            ["bull_trend", "hot_theme", "fund_flow", "chan_theory"],
+        )
+
    def test_build_context_keeps_market_phase_context_in_meta_not_data(self):
        orch = self._make_orchestrator()
        phase_context = {"phase": "intraday", "is_partial_bar": True}
@@ -1450,6 +1471,72 @@ class TestOrchestratorExecution(unittest.TestCase):
        self.assertIn('"non_critical_stage_present": true', combined)
        self.assertIn('"non_critical": true', combined)

+    def test_strategy_engine_preserves_scheduler_diagnostics(self):
+        orch = self._make_orchestrator()
+        ctx = AgentContext(
+            query="test",
+            stock_code="600519",
+            meta={
+                "invalid_opinions": [
+                    {
+                        "agent_name": "skill_fund_flow",
+                        "reason": "skill_timeout",
+                    }
+                ]
+            },
+        )
+        ctx.add_opinion(AgentOpinion(
+            agent_name="skill_hot_theme",
+            signal="moon",
+            confidence=0.9,
+            reasoning="bad fixture",
+        ))
+        ctx.add_opinion(AgentOpinion(
+            agent_name="skill_bull_trend",
+            signal="buy",
+            confidence=0.8,
+            reasoning="valid fixture",
+            raw_data={"skill_id": "bull_trend"},
+        ))
+
+        orch._run_strategy_engine(ctx)
+
+        invalid_bucket = ctx.meta["invalid_opinions"]
+        self.assertEqual([item["agent_name"] for item in invalid_bucket], [
+            "skill_fund_flow",
+            "skill_hot_theme",
+        ])
+        self.assertEqual(invalid_bucket[0]["reason"], "skill_timeout")
+        self.assertEqual(invalid_bucket[1]["reason"], "unrecognized_signal")
+        synthesis = ctx.get_data("skill_consensus")["strategy_synthesis"]
+        self.assertEqual(synthesis["summary_params"]["opinion_count"], 1)
+        self.assertEqual(synthesis["summary_params"]["invalid_opinion_count"], 2)
+        self.assertEqual(synthesis["summary_params"]["total_opinion_count"], 3)
+
+        from src.agent.agents.decision_agent import DecisionAgent
+        prompt = DecisionAgent(
+            tool_registry=MagicMock(),
+            llm_adapter=MagicMock(),
+        ).build_user_message(ctx)
+        self.assertIn("执行超时 1 个", prompt)
+        self.assertIn("signal 无法识别 1 个", prompt)
+
+    def test_specialist_batch_timeout_is_split_across_concurrency_waves(self):
+        orch = self._make_orchestrator(config=SimpleNamespace(agent_skill_concurrency=2))
+
+        self.assertEqual(
+            orch._skill_batch_timeout_slice(3, timeout_seconds=30),
+            15,
+        )
+        self.assertEqual(
+            orch._skill_batch_timeout_slice(4, timeout_seconds=30),
+            15,
+        )
+        self.assertEqual(
+            orch._skill_batch_timeout_slice(2, timeout_seconds=30),
+            30,
+        )
+
    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="贵州茅台")
@@ -3512,6 +3599,41 @@ class TestStrategyEngineE2E(unittest.TestCase):
        self.assertEqual(result.invalid_count, 3)
        self.assertEqual(len(result.invalid_records), 3)

+    def test_engine_counts_scheduler_diagnostics_with_valid_opinions(self):
+        from src.agent.skills.engine import StrategyEngine, StrategyResultStatus
+
+        result = StrategyEngine().process(
+            [AgentOpinion(agent_name="skill_bull_trend", signal="buy", confidence=0.8)],
+            diagnostic_records=[{
+                "agent_name": "skill_hot_theme",
+                "reason": "skill_timeout",
+            }],
+        )
+
+        self.assertEqual(result.status, StrategyResultStatus.CONSENSUS)
+        params = result.synthesis_dict["summary_params"]
+        self.assertEqual(params["opinion_count"], 1)
+        self.assertEqual(params["invalid_opinion_count"], 1)
+        self.assertEqual(params["total_opinion_count"], 2)
+        self.assertEqual(result.invalid_records[0]["reason"], "skill_timeout")
+
+    def test_engine_builds_no_consensus_stub_for_scheduler_only_failure(self):
+        from src.agent.skills.engine import StrategyEngine, StrategyResultStatus
+
+        result = StrategyEngine().process(
+            [],
+            diagnostic_records=[{
+                "agent_name": "skill_hot_theme",
+                "reason": "skill_error",
+            }],
+        )
+
+        self.assertEqual(result.status, StrategyResultStatus.NO_CONSENSUS)
+        params = result.synthesis_dict["summary_params"]
+        self.assertEqual(params["opinion_count"], 0)
+        self.assertEqual(params["invalid_opinion_count"], 1)
+        self.assertEqual(params["total_opinion_count"], 1)
+
    # ------------------------------------------------------------------
    # 2. All invalid → NO_CONSENSUS stub
    # ------------------------------------------------------------------
@@ -0,0 +1,504 @@
+# -*- coding: utf-8 -*-
+"""Focused tests for Strategy Deliberation mediators."""
+
+import json
+import unittest
+
+from src.agent.protocols import StrategyConflict, StrategyOpinion
+from src.agent.skills.deliberation import (
+    DeliberationMediator,
+    LLMDeliberationMediator,
+    MultiRoundDeliberationMediator,
+    StrategySelfReviewMediator,
+)
+from src.agent.skills.synthesis import (
+    ConflictDetector,
+    StrategySynthesizer,
+)
+
+
+class TestStrategyDeliberationV0(unittest.TestCase):
+    def test_omits_deliberation_without_conflicts(self):
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="buy", confidence=0.8),
+            StrategyOpinion(skill_id="hot_theme", signal="buy", confidence=0.7),
+        ]
+
+        synthesis = StrategySynthesizer().synthesize(
+            opinions,
+            weighted_score=4.0,
+            final_signal="buy",
+            weighted_confidence=0.75,
+            conflicts=[],
+        )
+
+        self.assertEqual(synthesis["final_signal"], "buy")
+        self.assertAlmostEqual(synthesis["confidence"], 0.75)
+        self.assertNotIn("deliberation", synthesis)
+        self.assertNotIn("revision_projection", synthesis)
+
+    def test_softens_high_conflict_without_reversing_signal(self):
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="strong_buy", confidence=0.82),
+            StrategyOpinion(skill_id="hot_theme", signal="strong_sell", confidence=0.78),
+        ]
+        conflicts = ConflictDetector().detect(opinions, final_signal="hold")
+
+        synthesis = StrategySynthesizer().synthesize(
+            opinions,
+            weighted_score=3.0,
+            final_signal="hold",
+            weighted_confidence=0.8,
+            conflicts=conflicts,
+        )
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        deliberation = synthesis["deliberation"]
+        self.assertEqual(deliberation["status"], "completed")
+        self.assertEqual(deliberation["mode"], "mediator_v0")
+        self.assertEqual(deliberation["summary"]["resolution_status"], "partially_resolved")
+        self.assertEqual(deliberation["summary"]["confidence_adjustment"], -0.06)
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
+
+        responses = deliberation["responses"]
+        self.assertNotIn("reversed", {response["revision"] for response in responses})
+        bull_softened = [
+            response for response in responses
+            if response["skill_id"] == "bull_trend" and response["revision"] == "softened"
+        ]
+        bear_softened = [
+            response for response in responses
+            if response["skill_id"] == "hot_theme" and response["revision"] == "softened"
+        ]
+        self.assertTrue(bull_softened)
+        self.assertEqual(bull_softened[0]["original_signal"], "strong_buy")
+        self.assertEqual(bull_softened[0]["revised_signal"], "buy")
+        self.assertTrue(bear_softened)
+        self.assertEqual(bear_softened[0]["original_signal"], "strong_sell")
+        self.assertEqual(bear_softened[0]["revised_signal"], "sell")
+        projection = synthesis["revision_projection"]
+        self.assertEqual(projection["status"], "computed")
+        self.assertEqual(projection["mode"], "preview_only")
+        self.assertEqual(projection["source_mode"], "mediator_v0")
+        self.assertEqual(projection["projected_signal"], "hold")
+        self.assertFalse(projection["final_signal_overridden"])
+
+    def test_preserves_high_confidence_minority_view(self):
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="buy", confidence=0.82),
+            StrategyOpinion(skill_id="fund_flow", signal="sell", confidence=0.8),
+        ]
+        conflicts = [
+            StrategyConflict(
+                conflict_type="high_confidence_dissent",
+                severity="medium",
+                participants=["fund_flow"],
+                description_key="strategy_conflict.high_confidence_dissent",
+                metadata={"final_signal": "buy"},
+            )
+        ]
+
+        synthesis = StrategySynthesizer().synthesize(
+            opinions,
+            weighted_score=4.0,
+            final_signal="buy",
+            weighted_confidence=0.81,
+            conflicts=conflicts,
+        )
+
+        self.assertEqual(synthesis["final_signal"], "buy")
+        summary = synthesis["deliberation"]["summary"]
+        self.assertTrue(summary["minority_view_preserved"])
+        self.assertEqual(summary["resolution_status"], "unresolved")
+        self.assertEqual(summary["confidence_adjustment"], -0.05)
+        responses = synthesis["deliberation"]["responses"]
+        self.assertEqual(len(responses), 1)
+        self.assertEqual(responses[0]["skill_id"], "fund_flow")
+        self.assertEqual(responses[0]["revision"], "unchanged")
+        self.assertEqual(responses[0]["revised_signal"], "sell")
+
+    def test_revision_projection_does_not_override_final_signal(self):
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="strong_buy", confidence=0.82),
+            StrategyOpinion(skill_id="hot_theme", signal="strong_sell", confidence=0.3),
+        ]
+        conflicts = ConflictDetector().detect(opinions, final_signal="strong_buy")
+
+        synthesis = StrategySynthesizer().synthesize(
+            opinions,
+            weighted_score=4.5,
+            final_signal="strong_buy",
+            weighted_confidence=0.68,
+            conflicts=conflicts,
+        )
+
+        self.assertEqual(synthesis["final_signal"], "strong_buy")
+        self.assertEqual(synthesis["weighted_score"], 4.5)
+        projection = synthesis["revision_projection"]
+        self.assertEqual(projection["projected_signal"], "hold")
+        self.assertEqual(projection["changed_skill_count"], 2)
+        self.assertEqual(projection["changed_skills"], ["bull_trend", "hot_theme"])
+        self.assertFalse(projection["final_signal_overridden"])
+
+    def test_revision_projection_ignores_unguarded_aggressive_response(self):
+        class UnsafeProjectionMediator:
+            def deliberate(self, opinions, conflicts, *, final_signal):
+                baseline = DeliberationMediator().deliberate(
+                    opinions,
+                    conflicts,
+                    final_signal=final_signal,
+                )
+                for response in baseline.responses:
+                    response.revised_signal = response.original_signal
+                    response.revised_confidence = response.original_confidence
+                baseline.mode = "unsafe_test_mediator"
+                return baseline
+
+        synthesis = _synthesize_high_conflict(UnsafeProjectionMediator())
+
+        projection = synthesis["revision_projection"]
+        self.assertEqual(projection["source_mode"], "unsafe_test_mediator")
+        self.assertEqual(projection["changed_skill_count"], 0)
+        self.assertEqual(projection["projected_signal"], "hold")
+
+
+class TestStrategyDeliberationV1(unittest.TestCase):
+    def test_llm_mediator_accepts_schema_valid_payload(self):
+        def fake_completion(messages):
+            request = _request_payload(messages)
+            payload = request["baseline_deliberation"]
+            payload["summary"]["confidence_adjustment"] = -0.09
+            payload["summary"]["confidence_adjustment_reason_key"] = (
+                "deliberation.confidence.llm_v1_more_conservative"
+            )
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(LLMDeliberationMediator(fake_completion))
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        self.assertAlmostEqual(synthesis["confidence"], 0.59)
+        self.assertEqual(synthesis["deliberation"]["mode"], "llm_mediator_v1")
+        self.assertEqual(
+            synthesis["deliberation"]["summary"]["confidence_adjustment_reason_key"],
+            "deliberation.confidence.llm_v1_more_conservative",
+        )
+
+    def test_llm_mediator_rejects_reversed_revision_and_falls_back(self):
+        def fake_completion(messages):
+            request = _request_payload(messages)
+            payload = request["baseline_deliberation"]
+            payload["responses"][0]["revision"] = "reversed"
+            payload["responses"][0]["revised_signal"] = "sell"
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(LLMDeliberationMediator(fake_completion))
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertNotIn(
+            "reversed",
+            {response["revision"] for response in synthesis["deliberation"]["responses"]},
+        )
+
+    def test_llm_mediator_cannot_undo_baseline_softening(self):
+        def fake_completion(messages):
+            request = _request_payload(messages)
+            payload = request["baseline_deliberation"]
+            for response in payload["responses"]:
+                response["revision"] = "unchanged"
+                response["revised_signal"] = response["original_signal"]
+                response["revised_confidence"] = response["original_confidence"]
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(LLMDeliberationMediator(fake_completion))
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
+        self.assertEqual(synthesis["revision_projection"]["changed_skill_count"], 2)
+
+    def test_llm_mediator_cannot_raise_baseline_adjustment(self):
+        def fake_completion(messages):
+            request = _request_payload(messages)
+            payload = request["baseline_deliberation"]
+            payload["summary"]["confidence_adjustment"] = 0
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(LLMDeliberationMediator(fake_completion))
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertEqual(
+            synthesis["deliberation"]["summary"]["confidence_adjustment"],
+            -0.06,
+        )
+
+    def test_llm_mediator_cannot_raise_softened_baseline_confidence(self):
+        def fake_completion(messages):
+            request = _request_payload(messages)
+            payload = request["baseline_deliberation"]
+            response = payload["responses"][0]
+            self.assertEqual(response["revision"], "softened")
+            response["revised_confidence"] = response["original_confidence"]
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(LLMDeliberationMediator(fake_completion))
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertEqual(synthesis["revision_projection"]["changed_skill_count"], 2)
+
+
+class TestStrategyDeliberationV2(unittest.TestCase):
+    def test_self_review_mediator_accepts_participant_reviews(self):
+        def fake_self_review(skill_id, messages):
+            request = _request_payload(messages)
+            response = request["baseline_response"]
+            if skill_id == "bull_trend" and response["revision"] == "softened":
+                response["revised_confidence"] = 0.7
+                response["critique_key"] = "deliberation.self_review.bull_trend.softened"
+            else:
+                response["critique_key"] = "deliberation.self_review.hot_theme.unchanged"
+            return json.dumps(response)
+
+        synthesis = _synthesize_high_conflict(StrategySelfReviewMediator(fake_self_review))
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        self.assertEqual(synthesis["deliberation"]["mode"], "self_review_v2")
+        self.assertEqual(synthesis["deliberation"]["summary"]["resolution_status"], "partially_resolved")
+        self.assertEqual(synthesis["deliberation"]["summary"]["confidence_adjustment"], -0.06)
+        responses = synthesis["deliberation"]["responses"]
+        self.assertTrue(any(
+            response["skill_id"] == "bull_trend"
+            and response["revision"] == "softened"
+            and response["revised_signal"] == "buy"
+            for response in responses
+        ))
+        self.assertTrue(any(
+            response["skill_id"] == "hot_theme"
+            and response["revision"] == "softened"
+            and response["revised_signal"] == "sell"
+            for response in responses
+        ))
+
+    def test_self_review_projection_uses_accepted_reviews(self):
+        def fake_self_review(skill_id, messages):
+            request = _request_payload(messages)
+            response = request["baseline_response"]
+            if skill_id == "bull_trend" and response["revision"] == "softened":
+                response["revised_confidence"] = 0.4
+            return json.dumps(response)
+
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="strong_buy", confidence=0.9),
+            StrategyOpinion(skill_id="hot_theme", signal="strong_sell", confidence=0.7),
+        ]
+        conflicts = ConflictDetector().detect(opinions, final_signal="hold")
+        mediator = StrategySelfReviewMediator(fake_self_review)
+
+        synthesis = StrategySynthesizer(deliberation_mediator=mediator).synthesize(
+            opinions,
+            weighted_score=3.0,
+            final_signal="hold",
+            weighted_confidence=0.8,
+            conflicts=conflicts,
+        )
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        projection = synthesis["revision_projection"]
+        self.assertEqual(projection["source_mode"], "self_review_v2")
+        self.assertEqual(projection["projected_signal"], "hold")
+        self.assertEqual(projection["changed_skill_count"], 2)
+        self.assertEqual(projection["changed_skills"], ["bull_trend", "hot_theme"])
+
+    def test_self_review_mediator_rejects_any_reversed_review_and_falls_back(self):
+        def fake_self_review(skill_id, messages):
+            request = _request_payload(messages)
+            response = request["baseline_response"]
+            if skill_id == "bull_trend":
+                response["revision"] = "reversed"
+                response["revised_signal"] = "sell"
+            return json.dumps(response)
+
+        synthesis = _synthesize_high_conflict(StrategySelfReviewMediator(fake_self_review))
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
+        self.assertNotIn(
+            "reversed",
+            {response["revision"] for response in synthesis["deliberation"]["responses"]},
+        )
+
+    def test_self_review_mediator_cannot_undo_baseline_softening(self):
+        def fake_self_review(skill_id, messages):
+            request = _request_payload(messages)
+            response = request["baseline_response"]
+            response["revision"] = "unchanged"
+            response["revised_signal"] = response["original_signal"]
+            response["revised_confidence"] = response["original_confidence"]
+            return json.dumps(response)
+
+        synthesis = _synthesize_high_conflict(StrategySelfReviewMediator(fake_self_review))
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
+        self.assertEqual(synthesis["revision_projection"]["changed_skill_count"], 2)
+
+    def test_self_review_mediator_cannot_raise_softened_baseline_confidence(self):
+        def fake_self_review(skill_id, messages):
+            request = _request_payload(messages)
+            response = request["baseline_response"]
+            if response["revision"] == "softened":
+                response["revised_confidence"] = response["original_confidence"]
+            return json.dumps(response)
+
+        synthesis = _synthesize_high_conflict(StrategySelfReviewMediator(fake_self_review))
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertEqual(synthesis["revision_projection"]["changed_skill_count"], 2)
+
+    def test_self_review_keeps_baseline_adjustment_when_more_resolved(self):
+        def fake_self_review(skill_id, messages):
+            response = _request_payload(messages)["baseline_response"]
+            response["revision"] = "softened"
+            response["revised_signal"] = (
+                "buy" if response["original_signal"] == "strong_buy" else "sell"
+            )
+            response["revised_confidence"] = 0.7
+            return json.dumps(response)
+
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="strong_buy", confidence=0.82),
+            StrategyOpinion(skill_id="hot_theme", signal="strong_sell", confidence=0.78),
+        ]
+        conflicts = [StrategyConflict(
+            conflict_type="directional_opposition",
+            severity="medium",
+            participants=["bull_trend", "hot_theme"],
+        )]
+        synthesis = StrategySynthesizer(
+            deliberation_mediator=StrategySelfReviewMediator(fake_self_review),
+        ).synthesize(
+            opinions,
+            weighted_score=3.0,
+            final_signal="hold",
+            weighted_confidence=0.8,
+            conflicts=conflicts,
+        )
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "self_review_v2")
+        self.assertEqual(synthesis["deliberation"]["summary"]["resolution_status"], "partially_resolved")
+        self.assertEqual(synthesis["deliberation"]["summary"]["confidence_adjustment"], -0.05)
+
+
+class TestStrategyDeliberationV4(unittest.TestCase):
+    def test_multi_round_mediator_accepts_configured_second_round(self):
+        def fake_round(round_index, messages):
+            request = _request_payload(messages)
+            payload = request["current_deliberation"]
+            self.assertEqual(round_index, 2)
+            self.assertEqual(request["round_index"], 2)
+            payload["responses"][0]["revised_confidence"] = 0.5
+            payload["responses"][0]["critique_key"] = "deliberation.multi_round.bull_trend.further_softened"
+            payload["summary"]["confidence_adjustment"] = -0.09
+            payload["summary"]["confidence_adjustment_reason_key"] = (
+                "deliberation.confidence.multi_round_more_conservative"
+            )
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(
+            MultiRoundDeliberationMediator(fake_round, max_rounds=2),
+        )
+
+        self.assertEqual(synthesis["final_signal"], "hold")
+        self.assertAlmostEqual(synthesis["confidence"], 0.59)
+        deliberation = synthesis["deliberation"]
+        self.assertEqual(deliberation["mode"], "multi_round_v4")
+        self.assertEqual(deliberation["rounds"], 2)
+        self.assertEqual(deliberation["round_history"][0]["status"], "baseline")
+        self.assertEqual(deliberation["round_history"][1]["status"], "accepted")
+        self.assertEqual(
+            deliberation["summary"]["confidence_adjustment_reason_key"],
+            "deliberation.confidence.multi_round_more_conservative",
+        )
+        self.assertEqual(synthesis["revision_projection"]["source_mode"], "multi_round_v4")
+        self.assertFalse(synthesis["revision_projection"]["final_signal_overridden"])
+
+    def test_multi_round_mediator_rejects_confidence_increase_and_keeps_baseline(self):
+        def fake_round(round_index, messages):
+            request = _request_payload(messages)
+            payload = request["current_deliberation"]
+            payload["responses"][0]["revised_confidence"] = 0.99
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(
+            MultiRoundDeliberationMediator(fake_round, max_rounds=2),
+        )
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertEqual(synthesis["deliberation"]["rounds"], 1)
+        self.assertNotIn("round_history", synthesis["deliberation"])
+        self.assertAlmostEqual(synthesis["confidence"], 0.62)
+
+    def test_multi_round_mediator_cannot_undo_baseline_softening(self):
+        def fake_round(round_index, messages):
+            request = _request_payload(messages)
+            payload = request["current_deliberation"]
+            for response in payload["responses"]:
+                response["revision"] = "unchanged"
+                response["revised_signal"] = response["original_signal"]
+                response["revised_confidence"] = response["original_confidence"]
+            return json.dumps(payload)
+
+        synthesis = _synthesize_high_conflict(
+            MultiRoundDeliberationMediator(fake_round, max_rounds=2),
+        )
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertEqual(synthesis["revision_projection"]["changed_skill_count"], 2)
+
+    def test_multi_round_mediator_respects_max_rounds_one(self):
+        def fail_if_called(round_index, messages):
+            raise AssertionError("round_completion should not be called")
+
+        opinions = [
+            StrategyOpinion(skill_id="bull_trend", signal="strong_buy", confidence=0.82),
+            StrategyOpinion(skill_id="hot_theme", signal="strong_sell", confidence=0.78),
+        ]
+        conflicts = ConflictDetector().detect(opinions, final_signal="hold")
+        mediator = MultiRoundDeliberationMediator(fail_if_called, max_rounds=1)
+
+        synthesis = StrategySynthesizer(deliberation_mediator=mediator).synthesize(
+            opinions,
+            weighted_score=3.0,
+            final_signal="hold",
+            weighted_confidence=0.8,
+            conflicts=conflicts,
+        )
+
+        self.assertEqual(synthesis["deliberation"]["mode"], "mediator_v0")
+        self.assertEqual(synthesis["deliberation"]["rounds"], 1)
+
+
+def _request_payload(messages):
+    content = messages[1]["content"]
+    _, raw_json = content.split("\n\n", 1)
+    return json.loads(raw_json)
+
+
+def _synthesize_high_conflict(mediator):
+    opinions = [
+        StrategyOpinion(skill_id="bull_trend", signal="strong_buy", confidence=0.82),
+        StrategyOpinion(skill_id="hot_theme", signal="strong_sell", confidence=0.78),
+    ]
+    conflicts = ConflictDetector().detect(opinions, final_signal="hold")
+    return StrategySynthesizer(deliberation_mediator=mediator).synthesize(
+        opinions,
+        weighted_score=3.0,
+        final_signal="hold",
+        weighted_confidence=0.8,
+        conflicts=conflicts,
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,203 @@
+# -*- coding: utf-8 -*-
+"""Focused tests for concurrent strategy-skill scheduling."""
+
+from datetime import date
+import threading
+import time
+import unittest
+
+from src.agent.protocols import AgentContext, AgentOpinion, StageResult, StageStatus
+from src.agent.skills.scheduler import AgentSkillScheduler
+from src.services.history_loader import (
+    get_frozen_target_date,
+    reset_frozen_target_date,
+    set_frozen_target_date,
+)
+
+
+class _FakeSkillAgent:
+    def __init__(self, agent_name):
+        self.agent_name = agent_name
+
+
+class TestAgentSkillScheduler(unittest.TestCase):
+    def test_runs_skill_agents_concurrently_and_preserves_input_order(self):
+        active = 0
+        max_active = 0
+        lock = threading.Lock()
+        two_running = threading.Event()
+
+        def run_stage(agent, ctx, progress_callback=None, timeout_seconds=None):
+            nonlocal active, max_active
+            with lock:
+                active += 1
+                max_active = max(max_active, active)
+                if active == 2:
+                    two_running.set()
+            two_running.wait(0.2)
+            time.sleep(0.01)
+            with lock:
+                active -= 1
+            opinion = AgentOpinion(
+                agent_name=agent.agent_name,
+                signal="buy",
+                confidence=0.7,
+                reasoning=agent.agent_name,
+            )
+            ctx.add_opinion(opinion)
+            result = StageResult(stage_name=agent.agent_name, status=StageStatus.COMPLETED)
+            result.opinion = opinion
+            return result
+
+        agents = [
+            _FakeSkillAgent("skill_bull_trend"),
+            _FakeSkillAgent("skill_hot_theme"),
+            _FakeSkillAgent("skill_fund_flow"),
+        ]
+        ctx = AgentContext(query="test")
+        scheduler = AgentSkillScheduler(max_concurrency=2)
+
+        batch = scheduler.run(agents, ctx, run_stage)
+
+        self.assertEqual(max_active, 2)
+        self.assertEqual(
+            [opinion.agent_name for opinion in batch.opinions],
+            ["skill_bull_trend", "skill_hot_theme", "skill_fund_flow"],
+        )
+        self.assertEqual(ctx.opinions, [])
+
+    def test_runs_four_selected_skills_when_concurrency_is_four(self):
+        active = 0
+        max_active = 0
+        lock = threading.Lock()
+        four_running = threading.Barrier(4, timeout=2)
+
+        def run_stage(agent, ctx, progress_callback=None, timeout_seconds=None):
+            nonlocal active, max_active
+            with lock:
+                active += 1
+                max_active = max(max_active, active)
+            try:
+                four_running.wait()
+            except threading.BrokenBarrierError:
+                pass
+            with lock:
+                active -= 1
+            opinion = AgentOpinion(
+                agent_name=agent.agent_name,
+                signal="buy",
+                confidence=0.7,
+            )
+            return StageResult(
+                stage_name=agent.agent_name,
+                status=StageStatus.COMPLETED,
+                opinion=opinion,
+            )
+
+        agents = [_FakeSkillAgent(f"skill_{index}") for index in range(4)]
+        batch = AgentSkillScheduler(max_concurrency=4).run(
+            agents,
+            AgentContext(query="test"),
+            run_stage,
+        )
+
+        self.assertEqual(max_active, 4)
+        self.assertEqual(len(batch.opinions), 4)
+        self.assertEqual([item.agent_name for item in batch.opinions], [
+            "skill_0",
+            "skill_1",
+            "skill_2",
+            "skill_3",
+        ])
+
+    def test_single_worker_inherits_frozen_target_date(self):
+        self._assert_frozen_target_date_visible(agent_count=1, max_concurrency=1)
+
+    def test_concurrent_workers_inherit_frozen_target_date(self):
+        self._assert_frozen_target_date_visible(agent_count=2, max_concurrency=2)
+
+    def _assert_frozen_target_date_visible(self, *, agent_count, max_concurrency):
+        frozen_date = date(2026, 7, 22)
+        observed = []
+        all_running = threading.Barrier(agent_count, timeout=2) if agent_count > 1 else None
+
+        def run_stage(agent, ctx, progress_callback=None, timeout_seconds=None):
+            if all_running is not None:
+                try:
+                    all_running.wait()
+                except threading.BrokenBarrierError:
+                    pass
+            observed.append(get_frozen_target_date())
+            return StageResult(
+                stage_name=agent.agent_name,
+                status=StageStatus.COMPLETED,
+                opinion=AgentOpinion(
+                    agent_name=agent.agent_name,
+                    signal="buy",
+                    confidence=0.7,
+                ),
+            )
+
+        token = set_frozen_target_date(frozen_date)
+        try:
+            AgentSkillScheduler(max_concurrency=max_concurrency).run(
+                [_FakeSkillAgent(f"skill_{index}") for index in range(agent_count)],
+                AgentContext(query="test"),
+                run_stage,
+            )
+        finally:
+            reset_frozen_target_date(token)
+
+        self.assertEqual(observed, [frozen_date] * agent_count)
+
+    def test_failed_skill_becomes_diagnostic_record(self):
+        def run_stage(agent, ctx, progress_callback=None, timeout_seconds=None):
+            return StageResult(
+                stage_name=agent.agent_name,
+                status=StageStatus.FAILED,
+                error="skill timed out",
+            )
+
+        agent = _FakeSkillAgent("skill_hot_theme")
+        scheduler = AgentSkillScheduler(max_concurrency=4, timeout_per_skill=30)
+
+        batch = scheduler.run([agent], AgentContext(query="test"), run_stage)
+
+        self.assertEqual(batch.opinions, [])
+        self.assertEqual(len(batch.invalid_records), 1)
+        self.assertEqual(batch.invalid_records[0]["agent_name"], "skill_hot_theme")
+        self.assertEqual(batch.invalid_records[0]["reason"], "skill_timeout")
+        self.assertEqual(batch.timeout_per_skill, 30)
+
+    def test_success_without_opinion_becomes_skill_error(self):
+        def run_stage(agent, ctx, progress_callback=None, timeout_seconds=None):
+            return StageResult(stage_name=agent.agent_name, status=StageStatus.COMPLETED)
+
+        agent = _FakeSkillAgent("skill_hot_theme")
+        scheduler = AgentSkillScheduler(max_concurrency=1)
+
+        batch = scheduler.run([agent], AgentContext(query="test"), run_stage)
+
+        self.assertEqual(len(batch.invalid_records), 1)
+        self.assertEqual(batch.invalid_records[0]["reason"], "skill_error")
+
+    def test_legacy_unprefixed_specialist_name_is_scheduled(self):
+        def run_stage(agent, ctx, progress_callback=None, timeout_seconds=None):
+            return StageResult(
+                stage_name=agent.agent_name,
+                status=StageStatus.FAILED,
+                error="legacy specialist failed",
+            )
+
+        agent = _FakeSkillAgent("chan_theory")
+        scheduler = AgentSkillScheduler(max_concurrency=1)
+
+        batch = scheduler.run([agent], AgentContext(query="test"), run_stage)
+
+        self.assertEqual([result.stage_name for result in batch.stage_results], ["chan_theory"])
+        self.assertEqual(len(batch.invalid_records), 1)
+        self.assertEqual(batch.invalid_records[0]["agent_name"], "chan_theory")
+
+
+if __name__ == "__main__":
+    unittest.main()