代码合并

offline-backend-20260512/backend/app/gateway/routers/report_collaboration.py

@@ -302,7 +302,29 @@ async def send_message(
             agent_run_id=body.agent_run_id,
             node_run_id=body.target_node_id,
         )
-        await RequirementResolver(store).handle_user_message(session_id, user_message=created, idempotency_key=key)
+        decision = await RequirementResolver(
+            store,
+            config=get_app_config().report_collaboration,
+            planner_llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
+        ).handle_user_message(session_id, user_message=created, idempotency_key=key)
+        # 确认完意图直接规划工作流（设计 v2 衔接）：ready 且尚无方案时，
+        # 同一次请求内自动发起规划，用户不用再手动点「生成候选方案」。
+        if decision is not None and decision.action == "ready":
+            record = await store.get_session_record(session_id)
+            plans = await store.list_plans(session_id)
+            if not plans and record and not record.get("selected_plan_id"):
+                command = await store.enqueue_plan_request(
+                    session_id,
+                    idempotency_key=f"{key}:autoplan",
+                    expected_revision=decision.requirement.revision,
+                )
+                if command.get("status") == "pending":
+                    await PlanProposalService(
+                        store,
+                        agent_store=getattr(request.app.state, "agent_store", None),
+                        config=get_app_config().report_collaboration,
+                        llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
+                    ).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=f"{key}:autoplan")
         return created
     except Exception as exc:
         return _map_store_error(exc)
@@ -333,6 +355,8 @@ async def request_plans(
             await PlanProposalService(
                 store,
                 agent_store=getattr(request.app.state, "agent_store", None),
+                config=get_app_config().report_collaboration,
+                llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
             ).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=key)
         return Response(status_code=204)
     except Exception as exc:

offline-backend-20260512/backend/app/report_collaboration/agentscope_runtime/model_adapter.py

@@ -91,6 +91,62 @@ class AttemptUsage:
     error_type: str | None = None
     error_message: str | None = None
 
+
+# AgentScope 的 Agent.compress_context 会读 model.context_size 并调 count_tokens。
+_DEFAULT_CONTEXT_SIZE = 128_000
+_CJK_RE = re.compile(r"[\u4e00-\u9fff\u3040-\u30ff\uac00-\ud7af]")
+
+
+def estimate_tokens_for_text(text: str) -> int:
+    """CJK-aware local token estimate: ~1 token per CJK char, ~4 bytes otherwise."""
+
+    if not text:
+        return 0
+    cjk = len(_CJK_RE.findall(text))
+    other = len(text) - cjk
+    return cjk + other // 4 + (1 if other % 4 else 0)
+
+
+def _collect_countable_text(messages: list[Any]) -> str:
+    parts: list[str] = []
+    for msg in messages or []:
+        blocks: Any = None
+        getter = getattr(msg, "get_content_blocks", None)
+        if callable(getter):
+            try:
+                blocks = getter()
+            except Exception:
+                blocks = None
+        if blocks is not None:
+            for block in blocks:
+                if isinstance(block, str):
+                    parts.append(block)
+                elif isinstance(block, dict):
+                    for key in ("text", "thinking", "hint"):
+                        value = block.get(key)
+                        if isinstance(value, str):
+                            parts.append(value)
+                        elif isinstance(value, list):
+                            parts.extend(str(item) for item in value)
+                else:
+                    text = getattr(block, "text", None) or getattr(block, "thinking", None)
+                    if text:
+                        parts.append(str(text))
+            continue
+        if isinstance(msg, dict):
+            parts.append(str(msg.get("content") or ""))
+        else:
+            raw = getattr(msg, "content", "")
+            if isinstance(raw, list):
+                for block in raw:
+                    if isinstance(block, dict):
+                        parts.append(str(block.get("text") or ""))
+                    else:
+                        parts.append(str(getattr(block, "text", block) if not isinstance(block, str) else block))
+            else:
+                parts.append(str(raw or ""))
+    return "\n".join(parts)
+
     def as_dict(self) -> dict[str, Any]:
         return redact_mapping(
             {
@@ -244,6 +300,21 @@ def to_lc_messages(messages: list[Any]) -> list[Any]:
     return converted
 
 
+def _resolve_context_size(app_config: AppConfig | None, config_name: str) -> int:
+    """Best-effort context window from model config extras; sane default otherwise."""
+
+    try:
+        cfg = (app_config or get_app_config()).get_model_config(config_name)
+        extra = cfg.model_dump() if cfg is not None else {}
+        for key in ("context_size", "context_length", "context_window", "max_context_tokens"):
+            value = extra.get(key)
+            if isinstance(value, (int, float)) and int(value) > 0:
+                return int(value)
+    except Exception:
+        pass
+    return _DEFAULT_CONTEXT_SIZE
+
+
 class DeerFlowChatModelAdapter:
     """Frozen DeerFlow-backed chat model. Construct once per node attempt from a snapshot."""
 
@@ -266,6 +337,21 @@ class DeerFlowChatModelAdapter:
         self._llm_factory = llm_factory
         self._node_run_id = node_run_id
         self.last_usage = AttemptUsage(model_name=snapshot.config_name, node_run_id=node_run_id)
+        # AgentScope compress_context 契约：读 context_size 判断是否压缩。
+        self.context_size = _resolve_context_size(app_config, snapshot.config_name)
+
+    async def count_tokens(self, messages: list[Any], tools: list[dict[str, Any]] | None = None) -> int:
+        """AgentScope 契约方法（compress_context 调用）。本地估算，不打 API。"""
+
+        import json as _json
+
+        total = estimate_tokens_for_text(_collect_countable_text(messages))
+        if tools:
+            try:
+                total += estimate_tokens_for_text(_json.dumps(tools, ensure_ascii=False, default=str))
+            except Exception:
+                total += sum(estimate_tokens_for_text(str(item)) for item in tools)
+        return total
 
     def _config(self) -> AppConfig:
         return self._app_config or get_app_config()

offline-backend-20260512/backend/app/report_collaboration/execution/executor.py

@@ -47,7 +47,18 @@ class ReportCollaborationExecutor:
         existing = self._tasks.get(params.run_id)
         if existing is not None and not existing.done():
             return False
-        self._tasks[params.run_id] = asyncio.create_task(self.execute(params), name=f"rc-run-{params.run_id}")
+        task = asyncio.create_task(self.execute(params), name=f"rc-run-{params.run_id}")
+
+        def _log_crash(finished: asyncio.Task[str]) -> None:
+            if not finished.cancelled() and finished.exception() is not None:
+                logger.error(
+                    "rc executor task crashed for run %s",
+                    params.run_id,
+                    exc_info=finished.exception(),
+                )
+
+        task.add_done_callback(_log_crash)
+        self._tasks[params.run_id] = task
         return True
 
     async def drain(self) -> None:

offline-backend-20260512/backend/app/report_collaboration/execution/ledger.py

@@ -10,7 +10,7 @@ from app.report_collaboration.execution.board import ArtifactBoard
 from app.report_collaboration.execution.completion import late_output_may_write_completed, require_can_complete_node
 from app.report_collaboration.execution.quality_gate import QualityGate, QualityGateContext, requirement_from_session_record
 from deerflow.persistence.report_collaboration.base import ReportCollaborationStore
-from deerflow.persistence.report_collaboration.codec import iso_now, make_node_run_id
+from deerflow.persistence.report_collaboration.codec import _TERMINAL_RUN_STATUSES, iso_now, make_node_run_id
 
 TaskResultKind = Literal["artifact_draft", "clarification_request", "repair_request", "failure", "advice"]
 
@@ -314,6 +314,7 @@ class TaskLedger:
         command_id: str,
         instruction: str,
         node_ids: list[str],
+        cascade_downstream: bool = False,
     ) -> list[str]:
         """Supersede affected attempts and persist an immutable command overlay.
 
@@ -321,6 +322,10 @@ class TaskLedger:
         ``superseded`` and is therefore stored only as superseded evidence.
         User-directed rework is not counted against automatic quality-repair
         limits; it creates the next monotonic attempt number.
+
+        ``cascade_downstream=True`` additionally supersedes every transitive
+        dependent of ``node_ids`` and re-plans it, so the wave loop re-runs
+        the whole affected sub-DAG in dependency order (设计 §4.2 指定重跑).
         """
 
         plan, run, latest, artifacts = await self._load(run_id)
@@ -407,8 +412,112 @@ class TaskLedger:
                 active_dedupe_key=str(run["session_id"]),
                 session_status="running",
             )
+        if cascade_downstream and applied:
+            applied.extend(
+                await self._cascade_downstream(
+                    run,
+                    plan,
+                    latest,
+                    artifacts,
+                    roots=[node_id for node_id in node_ids if node_id in applied],
+                    command_id=command_id,
+                )
+            )
         return applied
 
+    async def _cascade_downstream(
+        self,
+        run: dict[str, Any],
+        plan: dict[str, Any],
+        latest: dict[str, dict[str, Any]],
+        artifacts: list[dict[str, Any]],
+        *,
+        roots: list[str],
+        command_id: str,
+    ) -> list[str]:
+        """Supersede transitive dependents of ``roots`` and re-plan them.
+
+        Planned/reopened dependents are left alone — they have produced
+        nothing yet and pick up the fresh upstream artifacts when queued.
+        Like the rework target itself, user-directed cascade re-planning is
+        not budget-limited (monotonic attempt number).
+        """
+
+        deps = _dependency_map(plan)
+        specs = _plan_nodes(plan)
+        dependents: set[str] = set()
+        changed = True
+        while changed:
+            changed = False
+            for node_id, parents in deps.items():
+                if node_id in dependents or node_id in roots:
+                    continue
+                if any(parent in dependents or parent in roots for parent in parents):
+                    dependents.add(node_id)
+                    changed = True
+        cascaded: list[str] = []
+        for node_id in sorted(dependents):
+            node = latest.get(node_id)
+            spec = specs.get(node_id)
+            if node is None or spec is None:
+                continue
+            status = str(node.get("status") or "")
+            if status in {"planned", "reopened", "superseded", "cancelled", "cancel_requested", ""}:
+                continue
+            require_transition(status, "superseded")
+            node["status"] = "superseded"
+            node["ended_at"] = iso_now()
+            node["awaiting_input"] = None
+            await self._store.save_node_run(node)
+            for artifact in artifacts:
+                if (
+                    artifact.get("node_id") == node_id
+                    and int(artifact.get("attempt") or 0) == int(node.get("attempt") or 0)
+                    and not artifact.get("superseded_by")
+                ):
+                    await self._board.mark(artifact, validation_status="superseded", superseded_by=f"command:{command_id}")
+            await self._store.append_event(
+                session_id=str(run["session_id"]),
+                run_id=str(run["id"]),
+                event_type="node.status.changed",
+                data={"node_id": node_id, "node_run_id": node["node_run_id"], "status": "superseded", "reason": "downstream_cascade"},
+            )
+            nxt = int(node.get("attempt") or 0) + 1
+            await self._store.save_node_run(
+                {
+                    "node_run_id": make_node_run_id(str(run["id"]), node_id, nxt),
+                    "session_id": run["session_id"],
+                    "run_id": run["id"],
+                    "node_id": node_id,
+                    "attempt": nxt,
+                    "status": "planned",
+                    "label": node.get("label") or spec.get("label") or node_id,
+                    "role_display_name": node.get("role_display_name") or spec.get("role_display_name"),
+                    "angle": node.get("angle") or spec.get("angle"),
+                    "contract_json": node.get("contract_json"),
+                    "input_artifact_ids": [],
+                    "command_overlay_revision": int(node.get("command_overlay_revision") or 0),
+                    "agent_snapshot": dict(node.get("agent_snapshot") or {}),
+                    "error": None,
+                    "started_at": None,
+                    "ended_at": None,
+                }
+            )
+            await self._store.append_event(
+                session_id=str(run["session_id"]),
+                run_id=str(run["id"]),
+                event_type="node.attempt.created",
+                data={
+                    "node_id": node_id,
+                    "node_run_id": make_node_run_id(str(run["id"]), node_id, nxt),
+                    "attempt": nxt,
+                    "reason": "downstream_cascade",
+                    "command_id": command_id,
+                },
+            )
+            cascaded.append(node_id)
+        return cascaded
+
     @staticmethod
     def _add_command_overlay(node: dict[str, Any], *, command_id: str, instruction: str) -> None:
         snapshot = dict(node.get("agent_snapshot") or {})
@@ -622,8 +731,18 @@ class TaskLedger:
                 return False
         return True
 
+    async def refresh_run_terminal(self, run_id: str) -> str:
+        """Public terminal-state evaluation; safe to call repeatedly."""
+
+        return await self._refresh_run_terminal(run_id)
+
     async def _refresh_run_terminal(self, run_id: str) -> str:
-        _plan, run, latest, _artifacts = await self._load(run_id)
+        plan, run, latest, _artifacts = await self._load(run_id)
+        run_status = str(run.get("status") or "running")
+        if run_status in _TERMINAL_RUN_STATUSES:
+            # Idempotency guard: never re-finalize (the finalizer would run
+            # twice when the executor's wave loop re-checks after completion).
+            return run_status
         statuses = [item["status"] for item in latest.values()]
         if statuses and all(status == "completed" for status in statuses):
             from app.report_collaboration.reporting.finalizer import ReportFinalizationError, ReportFinalizer
@@ -636,11 +755,32 @@ class TaskLedger:
             await self._store.patch_run(run_id, status="completed", active_dedupe_key=None, session_status="completed")
             return "completed"
         if any(status in _BLOCKING or status == "queued" for status in statuses):
-            return str(run.get("status") or "running")
+            return run_status
         if any(status == "failed" for status in statuses):
-            await self._store.patch_run(run_id, status="failed", active_dedupe_key=None, session_status="failed")
+            await self._fail_run(run_id, run, reason="node_failed")
             return "failed"
-        return str(run.get("status") or "running")
+        # Zombie guard: an ``interrupted`` node whose attempt budget is
+        # exhausted can never run again — the ledger only queues
+        # planned/reopened attempts and recovery only opens a new attempt
+        # while attempt < max_attempts. Without this, the run would stay
+        # "running" forever while the dispatcher keeps re-claiming it.
+        specs = _plan_nodes(plan)
+        for node_id, item in latest.items():
+            if str(item.get("status")) != "interrupted":
+                continue
+            if int(item.get("attempt") or 1) >= _node_max_attempts(specs.get(node_id), item):
+                await self._fail_run(run_id, run, reason="node_attempts_exhausted", node_id=node_id)
+                return "failed"
+        return run_status
+
+    async def _fail_run(self, run_id: str, run: dict[str, Any], *, reason: str, node_id: str | None = None) -> None:
+        await self._store.patch_run(run_id, status="failed", active_dedupe_key=None, session_status="failed")
+        await self._store.append_event(
+            session_id=str(run.get("session_id") or ""),
+            run_id=run_id,
+            event_type="run.status.changed",
+            data={"status": "failed", "reason": reason, **({"node_id": node_id} if node_id else {})},
+        )
 
     async def _load(self, run_id: str) -> tuple[dict[str, Any], dict[str, Any], dict[str, dict[str, Any]], list[dict[str, Any]]]:
         run = await self._store.get_run(run_id)
@@ -666,6 +806,20 @@ def _plan_nodes(plan: dict[str, Any]) -> dict[str, dict[str, Any]]:
     return {str(item["id"]): item for item in (plan.get("nodes") or []) if item.get("id")}
 
 
+def _node_max_attempts(spec: dict[str, Any] | None, node_record: dict[str, Any]) -> int:
+    """Same precedence as ``_queue_node``: node contract, then plan spec, then 3."""
+
+    contract = node_record.get("contract_json") or {}
+    if isinstance(contract, str):
+        try:
+            import json as _json
+
+            contract = _json.loads(contract)
+        except Exception:
+            contract = {}
+    return int(contract.get("max_attempts") or (spec or {}).get("max_attempts") or 3)
+
+
 def _dependency_map(plan: dict[str, Any]) -> dict[str, list[str]]:
     deps: dict[str, list[str]] = {node_id: list(spec.get("depends_on") or []) for node_id, spec in _plan_nodes(plan).items()}
     for edge in plan.get("edges") or []:

offline-backend-20260512/backend/app/report_collaboration/execution/member_runner.py

@@ -195,8 +195,11 @@ class WaveExecutor:
                 await self._command_processor.process_safe_point(run_id)
             wave = await self._ledger.ready_assignments(run_id)
             if not wave:
-                run = await self._store.get_run(run_id)
-                return str((run or {}).get("status") or "queued")
+                # Nothing runnable. Evaluate terminal state: all-completed
+                # finalizes, exhausted-interrupted nodes fail the run (zombie
+                # guard). Idempotent when the run already reached a terminal
+                # status via apply_result.
+                return await self._ledger.refresh_run_terminal(run_id)
             for assignment in wave:
                 self.dispatched.append(assignment.node_id)
                 result = await self._runner.run(assignment)

offline-backend-20260512/backend/app/report_collaboration/interventions/service.py

@@ -386,6 +386,7 @@ class RuntimeCommandService:
             command_id=command["command_id"],
             instruction=instruction,
             node_ids=impact_nodes,
+            cascade_downstream=True,
         )
         if not changed:
             raise ReportCollaborationValidationError("COMMAND_NO_EFFECT", "目标节点已过期或命令已经应用")

offline-backend-20260512/backend/app/report_collaboration/planning/planner_agent.py

@@ -0,0 +1,422 @@
+"""规划师智能体（PlannerAgent）——真智能体，不是规则路由 + 模板文案。
+
+设计 v2 的需求对话与工作流生成主路径：
+- ``converse``：与用户多轮自由对话确认需求。回复文案（含开场）完全由
+  智能体生成并直接落库展示 —— 「收到您的报告需求。为了给出…」这类
+  死模板文案已从该路径移除。
+- ``design_workflow``：带着**完整对话上下文**（规划师问了什么、用户
+  答了什么）生成受词表约束的自定义工作流（``StrategyRoleSet``），
+  与确定性模板走同一条 装配 → 校验 → 绑定 链路。
+
+规则路由与模板只作为模型彻底不可用时的最后防线，不出现在用户面前。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any, Callable
+
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+from app.report_collaboration.agentscope_runtime.model_resolver import ROLE_MODEL_FIELD, RoleModelSnapshot, resolve_role_model
+from app.report_collaboration.contracts.plans import RoleDemand
+from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
+from app.report_collaboration.planning.controller import RoleSlot, StrategyRoleSet
+from app.report_collaboration.requirement.intent_router import _merge_snapshot, apply_default_assumptions
+from deerflow.config.app_config import AppConfig, get_app_config
+
+logger = logging.getLogger(__name__)
+
+# ── 词表（结构化输出的安全边界，LLM 不受约束的自由只在文案层） ─────────────
+
+MEMBER_ROLES: tuple[str, ...] = tuple(key for key in ROLE_MODEL_FIELD if key != "coordinator")
+
+ROLE_ARTIFACTS: dict[str, tuple[str, ...]] = {
+    "researcher": ("EvidenceBundle",),
+    "verifier": ("VerificationResult",),
+    "analyst": ("AngleAnalysis",),
+    "synthesizer": ("SynthesisOutline",),
+    "writer": ("ReportSectionDraft",),
+    "reviewer": ("ReviewDecision",),
+    "reviser": ("ReportSectionDraft",),
+}
+
+QUALITY_GATES: tuple[str, ...] = ("cite_required", "dual_verification", "independent_review", "coverage_matrix")
+
+# ── 智能体输出契约 ─────────────────────────────────────────────────────────
+
+
+class PlannerTurn(BaseModel):
+    """规划师一轮对话的完整产出：自由文案 + 结构化状态。"""
+
+    model_config = ConfigDict(extra="ignore")
+
+    action: str = "clarify"
+    reply: str = ""
+    questions: list[str] = Field(default_factory=list)
+    impact: list[str] = Field(default_factory=list)
+    assumptions: list[str] = Field(default_factory=list)
+    requirement: dict[str, Any] = Field(default_factory=dict)
+
+
+class StrategyNodeDraft(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    slot_id: str
+    role_key: str
+    display_name: str
+    angle: str | None = None
+    mission: str
+    depends_on: list[str] = Field(default_factory=list)
+    output_artifact_type: str
+    suggested_tools: list[str] = Field(default_factory=list)
+
+
+class StrategyDraft(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    reply: str = ""
+    strategy: str = "custom"
+    title: str
+    summary: str
+    rationale: str
+    estimated_duration_seconds: int = 900
+    estimated_cost_level: str = "medium"
+    quality_gate_ids: list[str] = Field(default_factory=lambda: ["cite_required"])
+    nodes: list[StrategyNodeDraft] = Field(default_factory=list)
+
+
+LlmFactory = Callable[[], Any]
+
+
+def planner_model_snapshot(app_config: AppConfig | None = None) -> RoleModelSnapshot:
+    """规划师身份沿用角色解析链：coordinator_model → planner_model → 默认模型。"""
+
+    return resolve_role_model("coordinator", app_config=app_config or get_app_config())
+
+
+def build_planner_llm(app_config: AppConfig | None = None) -> Any:
+    from deerflow.models.factory import create_chat_model
+
+    snapshot = planner_model_snapshot(app_config)
+    return create_chat_model(snapshot.config_name, app_config=app_config or get_app_config())
+
+
+_THINK_BLOCK_RE = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
+
+
+def _extract_json(text: str) -> dict[str, Any] | None:
+    if not text:
+        return None
+    # 思考模型（MiniMax-M2.7 等）回复可能带 <think>…</think>；JSON 在思考块之后。
+    # 先剥思考块提取；剥完没有再回退用全文（极端情况 JSON 混在思考内）。
+    for candidate in (_THINK_BLOCK_RE.sub("", text).strip(), text.strip()):
+        if not candidate:
+            continue
+        cleaned = candidate
+        if cleaned.startswith("```"):
+            cleaned = cleaned.strip("`")
+            if cleaned.lower().startswith("json"):
+                cleaned = cleaned[4:]
+            cleaned = cleaned.strip()
+        start = cleaned.find("{")
+        end = cleaned.rfind("}")
+        if start < 0 or end <= start:
+            continue
+        try:
+            parsed = json.loads(cleaned[start : end + 1])
+        except Exception:
+            continue
+        if isinstance(parsed, dict):
+            return parsed
+    return None
+
+
+async def _invoke_json(llm: Any, messages: list[Any]) -> tuple[dict[str, Any] | None, str]:
+    """调用模型并提取 JSON。返回 (解析结果, 原始文本)；解析失败结果为 None。"""
+
+    try:
+        response = await llm.ainvoke(messages)
+        raw = getattr(response, "content", response)
+        if isinstance(raw, list):
+            raw = "".join(str(getattr(block, "text", block) if not isinstance(block, str) else block) for block in raw)
+        raw = str(raw)
+        return _extract_json(raw), raw
+    except Exception as exc:  # noqa: BLE001 — 智能体故障必须显式失败
+        logger.warning("rc planner agent llm call failed: %s: %s", exc.__class__.__name__, exc)
+        return None, ""
+
+
+async def _invoke_json_with_retry(llm: Any, system: str, user_prompt: str) -> dict[str, Any] | None:
+    """带一次格式重试：模型输出不含 JSON 时，附原始输出要求只出 JSON 再问一次。"""
+
+    parsed, raw = await _invoke_json(llm, [("system", system), ("human", user_prompt)])
+    if parsed is None:
+        logger.warning("rc planner reply without JSON (raw head: %.200s)", raw[:200])
+        retry_prompt = (
+            user_prompt
+            + "\n\n【格式提醒】你上一轮的输出没有包含合法 JSON。这次必须只输出一个 JSON 对象，"
+            + f"不要在 JSON 之外输出任何文字。你上一轮的输出是：{raw[:500]}"
+        )
+        parsed, _ = await _invoke_json(llm, [("system", system), ("human", retry_prompt)])
+    return parsed
+
+
+# ── 规划师智能体 ───────────────────────────────────────────────────────────
+
+
+class PlannerAgent:
+    """需求对话与工作流设计的唯一主路径（真智能体）。"""
+
+    def __init__(self, llm: Any, *, app_config: AppConfig | None = None) -> None:
+        self._llm = llm
+        self._app_config = app_config
+
+    # ── 需求对话 ───────────────────────────────────────────────────────────
+
+    async def converse(
+        self,
+        *,
+        user_text: str,
+        history: list[dict[str, str]],
+        previous_snapshot: dict[str, Any] | None,
+        previous_revision: int,
+        clarification_round: int,
+        max_rounds: int,
+    ) -> PlannerTurn | None:
+        """一轮规划对话。返回 None 表示智能体不可用（调用方走最后防线）。"""
+
+        forced = clarification_round >= max_rounds
+        system = _CONVERSE_SYSTEM.format(
+            roles=", ".join(MEMBER_ROLES),
+            forced_note=(
+                "\n\n【重要】澄清轮次已达上限，本轮你必须判定需求就绪（action=ready）："
+                "自然地告诉用户你将按当前理解开始规划，把未能确认的点写进 assumptions。"
+                if forced
+                else ""
+            ),
+        )
+        known = {k: v for k, v in (previous_snapshot or {}).items() if v not in (None, "", [])}
+        known.pop("revision", None)
+        history_text = "\n".join(
+            f"{'用户' if item.get('role') == 'human' else '规划师'}: {str(item.get('content') or '')[:400]}"
+            for item in history[-10:]
+        )
+        user_prompt = (
+            f"当前已确认的需求快照（JSON）：{json.dumps(known, ensure_ascii=False)}\n\n"
+            f"最近的对话：\n{history_text or '（无）'}\n\n"
+            f"用户最新消息：{user_text}"
+        )
+        parsed = await _invoke_json_with_retry(self._llm, system, user_prompt)
+        if parsed is None:
+            return None
+        try:
+            turn = PlannerTurn.model_validate(parsed)
+        except ValidationError:
+            logger.warning("rc planner turn invalid: %s", str(parsed)[:200])
+            return None
+        if not (turn.reply or "").strip():
+            return None
+        if forced and turn.action != "ready":
+            turn.action = "ready"
+        if turn.action == "ready" and not turn.assumptions:
+            turn.assumptions = ["按对话中已确认的信息直接规划"]
+        return turn
+
+    def merge_requirement(
+        self,
+        turn: PlannerTurn,
+        *,
+        previous_snapshot: dict[str, Any] | None,
+        previous_revision: int,
+    ) -> ReportRequirementSnapshot:
+        """把智能体本轮带回的需求字段合并进既有快照（revision 语义同规则路由）。"""
+
+        known = {k: v for k, v in (previous_snapshot or {}).items() if k != "revision"}
+        # LLM 契约是「只填能确认的字段」；再过滤一次空值，避免空串覆盖已确认信息。
+        patch = {
+            k: v
+            for k, v in (turn.requirement or {}).items()
+            if k != "revision" and v not in (None, "", [])
+        }
+        revision = (
+            max(1, int(previous_revision) + 1)
+            if (previous_snapshot or patch)
+            else max(1, int(previous_revision) or 1)
+        )
+        if not previous_snapshot:
+            revision = 1
+        merged = _merge_snapshot(known, patch, revision=revision)
+        if turn.action == "ready":
+            merged = apply_default_assumptions(merged)[0]
+        return merged
+
+    # ── 工作流设计 ─────────────────────────────────────────────────────────
+
+    async def design_workflow(
+        self,
+        *,
+        requirement: ReportRequirementSnapshot,
+        history: list[dict[str, str]],
+        max_team_members: int = 8,
+    ) -> tuple[StrategyRoleSet, str] | None:
+        """带着完整对话上下文设计工作流；返回 (策略, 展示文案) 或 None。"""
+
+        fields = {k: v for k, v in requirement.model_dump().items() if v not in (None, "", [])}
+        fields.pop("revision", None)
+        history_text = "\n".join(
+            f"{'用户' if item.get('role') == 'human' else '规划师'}: {str(item.get('content') or '')[:300]}"
+            for item in history[-12:]
+        )
+        system = _DESIGN_SYSTEM.format(
+            roles=", ".join(MEMBER_ROLES),
+            artifacts="\n".join(f"  - {role}: {' / '.join(arts)}" for role, arts in ROLE_ARTIFACTS.items()),
+            gates=", ".join(QUALITY_GATES),
+            max_nodes=min(max_team_members, 8),
+        )
+        user_prompt = (
+            f"需求快照（JSON）：{json.dumps(fields, ensure_ascii=False)}\n\n"
+            f"需求确认阶段的对话（你与用户）：\n{history_text or '（无）'}\n\n请设计工作流。"
+        )
+        parsed = await _invoke_json_with_retry(self._llm, system, user_prompt)
+        if parsed is None:
+            return None
+        try:
+            draft = StrategyDraft.model_validate(parsed)
+        except ValidationError:
+            logger.warning("rc planner strategy draft invalid: %s", str(parsed)[:200])
+            return None
+        strategy = _validate_strategy_dag(draft)
+        if strategy is None:
+            logger.warning("rc planner strategy failed vocabulary/DAG validation: %s", str(parsed)[:200])
+            return None
+        reply = (draft.reply or "").strip() or f"已按需求设计好协作工作流「{draft.title}」。"
+        return strategy, reply
+
+    def workflow_reply_fallback(self, strategy: StrategyRoleSet) -> str:
+        return (
+            f"我按确认的需求设计了「{strategy.title}」：{strategy.summary}\n"
+            "可以直接采用开始执行；想调整就告诉我（例如「加一条竞品对比线」）。"
+        )
+
+
+# ── 提示词 ─────────────────────────────────────────────────────────────────
+
+_CONVERSE_SYSTEM = """你是「报告协作工作台」的规划师智能体。用户想让你组织一个多智能体团队协作产出一份报告，你的任务是先通过自然对话把需求弄清楚，然后（由你自己）设计协作工作流。
+
+你的对话风格：
+- 像一位资深顾问：先回应用户刚说的话，再追问。回复里**不要**出现「收到您的报告需求」这类机械开场——用你自己的话自然表达。
+- 每轮最多追问 2~3 个真正影响方案设计的问题：关注对象与范围、给谁看、必须覆盖的角度、篇幅与深度、时间范围、特殊约束。用户已说清的绝不重复问。
+- 用户回复后先简短确认你的理解（一两句），再决定是否继续问。
+- 信息足够时，自然地总结你对需求的把握并判定就绪（action=ready），告诉用户下一步你会为他们设计工作流。
+
+可用的团队角色（供你构思，不必对用户罗列）：{roles}{forced_note}
+
+每轮严格输出一个 JSON 对象（不要输出 JSON 以外的文字）：
+{{
+  "reply": "给用户看的完整回复（自然对话，直接展示）",
+  "action": "clarify" 或 "ready",
+  "questions": ["追问的问题（与 reply 一致，供系统记录）"],
+  "impact": ["为什么需要问（与 questions 对应）"],
+  "assumptions": ["ready 时：你采用的默认假设"],
+  "requirement": {{"topic": "...", "objective": "...", "audience": "...", "required_angles": ["..."], "time_range": "...", "length_target": "...", "tone": "..."}}
+}}
+
+requirement 只填你从对话中能确认的字段（含此前轮次已确认的），其余省略。reply 用中文。"""
+
+_DESIGN_SYSTEM = """你是「报告协作工作台」的规划师智能体。需求已经与你确认完毕，现在请设计多智能体协作工作流。
+
+词表约束（结构化部分必须严格遵守）：
+- role_key 只能取：{roles}
+- 各角色的 output_artifact_type 只能取：
+{artifacts}
+- quality_gate_ids 只能取：{gates}
+
+结构要求：
+- 2~{max_nodes} 个节点；至少 1 个 writer；正式报告建议加 reviewer。
+- 研究线（researcher）按角度并行；research → verify → analyze → write → review 是常见主干，可按需求裁剪。
+- depends_on 引用其他节点的 slot_id；禁止环；至少一个节点 depends_on 为空（起点）。
+- slot_id 用简短英文（如 research-market / write-main）。mission 写清楚该节点做什么、产出什么。
+
+严格输出 JSON（不要输出 JSON 以外的文字）：
+{{
+  "reply": "给用户看的介绍：说明你为什么这样设计（对应需求的哪些要求），一两句即可",
+  "strategy": "custom",
+  "title": "方案标题",
+  "summary": "一句话概述团队构成与链路",
+  "rationale": "设计理由",
+  "estimated_duration_seconds": 1500,
+  "estimated_cost_level": "low|medium|high",
+  "quality_gate_ids": ["cite_required"],
+  "nodes": [
+    {{"slot_id": "research-market", "role_key": "researcher", "display_name": "研究员", "angle": "市场份额", "mission": "...", "depends_on": [], "output_artifact_type": "EvidenceBundle", "suggested_tools": ["web_search"]}}
+  ]
+}}"""
+
+
+# ── 工作流草稿校验（词表 + DAG） ──────────────────────────────────────────
+
+
+def _validate_strategy_dag(draft: StrategyDraft) -> StrategyRoleSet | None:
+    nodes = draft.nodes
+    if not 2 <= len(nodes) <= 8:
+        return None
+    slot_ids = {node.slot_id for node in nodes}
+    if len(slot_ids) != len(nodes):
+        return None
+    if not any(node.role_key == "writer" for node in nodes):
+        return None
+    for node in nodes:
+        if node.role_key not in MEMBER_ROLES:
+            return None
+        if node.output_artifact_type not in ROLE_ARTIFACTS.get(node.role_key, ()):
+            return None
+        for dep in node.depends_on:
+            if dep not in slot_ids or dep == node.slot_id:
+                return None
+    indegree = {node.slot_id: len(node.depends_on) for node in nodes}
+    children: dict[str, list[str]] = {node.slot_id: [] for node in nodes}
+    for node in nodes:
+        for dep in node.depends_on:
+            children[dep].append(node.slot_id)
+    queue = [slot for slot, degree in indegree.items() if degree == 0]
+    seen = 0
+    while queue:
+        current = queue.pop()
+        seen += 1
+        for child in children[current]:
+            indegree[child] -= 1
+            if indegree[child] == 0:
+                queue.append(child)
+    if seen != len(nodes):
+        return None
+
+    gates = [gate for gate in draft.quality_gate_ids if gate in QUALITY_GATES] or ["cite_required"]
+    return StrategyRoleSet(
+        strategy="custom" if draft.strategy == "custom" else f"custom_{draft.strategy[:24]}",
+        title=draft.title[:80],
+        summary=draft.summary[:200],
+        rationale=draft.rationale[:300],
+        estimated_duration_seconds=max(300, min(int(draft.estimated_duration_seconds or 1500), 7200)),
+        estimated_cost_level=draft.estimated_cost_level if draft.estimated_cost_level in {"low", "medium", "high"} else "medium",
+        recommended=True,
+        slots=[
+            RoleSlot(
+                slot_id=node.slot_id,
+                demand=RoleDemand(
+                    role_key=node.role_key,
+                    display_name=node.display_name[:32] or node.role_key,
+                    angle=(node.angle or None),
+                    mission=node.mission[:400],
+                    depends_on_roles=[],
+                    output_artifact_type=node.output_artifact_type,
+                    suggested_tools=[tool for tool in node.suggested_tools if tool][:4],
+                ),
+                depends_on=list(node.depends_on),
+            )
+            for node in nodes
+        ],
+        quality_gate_ids=gates,
+    )

offline-backend-20260512/backend/app/report_collaboration/planning/service.py

@@ -1,17 +1,18 @@
 """Fulfill a durable plan_request: catalog → select → assemble → validate → persist.
 
-Does not start a run. Rule-based until the BE-7 model adapter is wired.
+Does not start a run. P1.2 起优先由规划师 LLM 生成定制策略，模板为回退。
 """
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 from uuid import uuid4
 
 from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
 from app.report_collaboration.planning.assembler import PlanAssembler
 from app.report_collaboration.planning.catalog import project_agent_catalog
-from app.report_collaboration.planning.controller import PlanController
+from app.report_collaboration.planning.planner_agent import LlmFactory
 from app.report_collaboration.planning.validator import apply_validation, validate_assembled_plan
 from app.report_collaboration.requirement.defaults import COORDINATOR_NAME, DEFAULT_TITLE
 from deerflow.config.app_config import get_app_config
@@ -22,6 +23,8 @@ from deerflow.persistence.report_collaboration import (
     ReportCollaborationValidationError,
 )
 
+logger = logging.getLogger(__name__)
+
 _BUSY = frozenset({"running", "awaiting_input", "reviewing", "completed", "failed", "cancelled"})
 
 
@@ -51,12 +54,15 @@ class PlanProposalService:
         *,
         agent_store: Any | None = None,
         config: ReportCollaborationConfig | None = None,
+        llm_factory: LlmFactory | None = None,
     ) -> None:
         self._store = store
         self._agent_store = agent_store
         self._config = config
-        self._controller = PlanController()
         self._assembler = PlanAssembler()
+        self._llm_factory = llm_factory
+        self._llm: Any | None = None
+        self._llm_tried = False
 
     def _cfg(self) -> ReportCollaborationConfig:
         return self._config or get_app_config().report_collaboration
@@ -110,12 +116,29 @@ class PlanProposalService:
         requirement = requirement_from_session(record)
         cfg = self._cfg()
         catalog = await project_agent_catalog(user_id=user_id, config=cfg, agent_store=self._agent_store)
-        strategies = self._controller.propose(
-            requirement,
-            max_parallel_tasks=cfg.max_parallel_tasks,
+        group_id = f"rpg_{uuid4().hex[:12]}"
+
+        # P1-R：规划师智能体是唯一路径 —— 带着完整对话上下文设计工作流。
+        # 模型不可用即无法继续（无模板兜底）。
+        from app.report_collaboration.planning.planner_agent import PlannerAgent
+
+        llm = self._planner_llm()
+        history = [
+            {"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")}
+            for item in await self._store.list_messages(session_id)
+        ]
+        designed = await PlannerAgent(llm).design_workflow(
+            requirement=requirement,
+            history=history,
             max_team_members=cfg.max_team_members,
         )
-        group_id = f"rpg_{uuid4().hex[:12]}"
+        if designed is None:
+            raise ReportCollaborationValidationError(
+                "PLANNER_UNAVAILABLE", "规划师智能体未能生成有效工作流，请重试或换个说法再试"
+            )
+        strategies: list[Any] = [designed[0]]
+        planner_reply: str | None = designed[1]
+
         assembled_list = [
             self._assembler.assemble(
                 strategy,
@@ -140,13 +163,44 @@ class PlanProposalService:
         for candidate in persisted:
             saved.append(await self._store.insert_plan(session_id, candidate.model_dump()))
         await self._store.update_command_status(command_id, status="completed")
+        content = planner_reply or "工作流已生成，可开始执行。"
         await self._store.create_message(
             session_id,
             role="ai",
-            content="已根据当前需求生成 3 个候选协作方案。请选择其中一个后再开始执行；选择方案不会自动开跑。",
+            content=content,
             idempotency_key=f"{idempotency_key}:plans",
             name=COORDINATOR_NAME,
-            metadata={"plan_request_command_id": command_id, "proposal_group_id": group_id},
+            metadata={"plan_request_command_id": command_id, "proposal_group_id": group_id, "plan_source": "planner_agent"},
             completed=True,
         )
         return saved
+
+    def _planner_llm(self) -> Any:
+        """规划师模型：注入优先（测试）；否则按 planner_llm_enabled 懒构建。不可用即抛错。"""
+
+        if self._llm_factory is not None:
+            llm = self._llm_factory()
+            if llm is None:
+                raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置")
+            return llm
+        try:
+            if not (self._config or get_app_config().report_collaboration).planner_llm_enabled:
+                raise ReportCollaborationValidationError(
+                    "PLANNER_UNAVAILABLE", "规划师智能体未启用（report_collaboration.planner_llm_enabled）"
+                )
+        except ReportCollaborationValidationError:
+            raise
+        except Exception as exc:
+            raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", f"规划师智能体配置不可读：{exc}") from None
+        if not self._llm_tried:
+            self._llm_tried = True
+            try:
+                from app.report_collaboration.planning.planner_agent import build_planner_llm
+
+                self._llm = build_planner_llm()
+            except Exception as exc:
+                logger.warning("rc planner agent llm build failed: %s", exc)
+                self._llm = None
+        if self._llm is None:
+            raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置")
+        return self._llm

offline-backend-20260512/backend/app/report_collaboration/requirement/__init__.py

@@ -1,6 +1,6 @@
 """Pre-plan requirement resolution package."""
 
-from app.report_collaboration.requirement.intent_router import PrePlanDecision, route_pre_plan_turn
+from app.report_collaboration.requirement.intent_router import PrePlanDecision
 from app.report_collaboration.requirement.resolver import RequirementResolver
 
-__all__ = ["PrePlanDecision", "RequirementResolver", "route_pre_plan_turn"]
+__all__ = ["PrePlanDecision", "RequirementResolver"]

offline-backend-20260512/backend/app/report_collaboration/requirement/intent_router.py

@@ -1,4 +1,9 @@
-"""Pre-plan IntentRouterAgent: understand the user turn, never dispatch work."""
+"""Pre-plan 数据结构与快照合并工具。
+
+P1-R 之后需求确认完全由规划师智能体（``planning/planner_agent.py``）驱动；
+本模块只保留被复用的数据结构与合并工具。原规则路由 ``route_pre_plan_turn``
+已按设计 v2 移除 —— 模型不可用即报错，不再有规则兜底。
+"""
 
 from __future__ import annotations
 
@@ -7,12 +12,11 @@ from typing import Any, Literal
 from pydantic import BaseModel, ConfigDict, Field
 
 from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
-from app.report_collaboration.requirement.defaults import DEFAULT_ANGLES, DEFAULT_AUDIENCE, DEFAULT_LANGUAGE, DEFAULT_SOURCE_CONSTRAINTS, DEFAULT_TONE, PLAN_AFFECTING_GAPS
-from app.report_collaboration.requirement.extractor import extract_requirement_patch, wants_direct_start
+from app.report_collaboration.requirement.defaults import DEFAULT_ANGLES, DEFAULT_AUDIENCE, DEFAULT_LANGUAGE, DEFAULT_SOURCE_CONSTRAINTS, DEFAULT_TONE
 
 
 class PrePlanDecision(BaseModel):
-    """Structured output of the pre-plan router. It never builds a graph or starts a run."""
+    """规划师智能体一轮决策。It never builds a graph or starts a run."""
 
     model_config = ConfigDict(extra="forbid")
 
@@ -24,11 +28,13 @@ class PrePlanDecision(BaseModel):
     used_defaults: bool = False
     reason: str
     confidence: float = Field(ge=0, le=1)
+    #: 规划师智能体产出的完整对话文案，直接落库展示。
+    reply: str | None = None
 
 
 def _merge_snapshot(previous: dict[str, Any] | None, patch: dict[str, Any], *, revision: int) -> ReportRequirementSnapshot:
     base = dict(previous or {})
-    base.update(patch)
+    base.update(patch or {})
     topic = str(base.get("topic") or "").strip()
     return ReportRequirementSnapshot(
         topic=topic or "未指定主题",
@@ -49,21 +55,6 @@ def _merge_snapshot(previous: dict[str, Any] | None, patch: dict[str, Any], *, r
     )
 
 
-def _gaps(requirement: ReportRequirementSnapshot) -> list[tuple[str, str, str]]:
-    missing: list[tuple[str, str, str]] = []
-    if not requirement.topic or requirement.topic == "未指定主题":
-        missing.append(("topic", "这份报告要写什么主题？请用一两句话说明对象、时间和目的。", "没有主题无法选择方案"))
-    for field, question, impact in PLAN_AFFECTING_GAPS:
-        value = getattr(requirement, field, None)
-        if field == "required_angles":
-            empty = not value
-        else:
-            empty = not (value or "").strip()
-        if empty:
-            missing.append((field, question, impact))
-    return missing
-
-
 def apply_default_assumptions(requirement: ReportRequirementSnapshot) -> tuple[ReportRequirementSnapshot, list[str]]:
     assumptions: list[str] = []
     data = requirement.model_dump()
@@ -82,62 +73,3 @@ def apply_default_assumptions(requirement: ReportRequirementSnapshot) -> tuple[R
         data["tone"] = DEFAULT_TONE
         assumptions.append(DEFAULT_TONE)
     return ReportRequirementSnapshot.model_validate(data), assumptions
-
-
-def route_pre_plan_turn(
-    text: str,
-    *,
-    previous_snapshot: dict[str, Any] | None,
-    previous_revision: int,
-    clarification_round: int,
-    max_clarification_rounds: int,
-) -> PrePlanDecision:
-    """Rule-based pre-plan routing. Does not call tools or create plans."""
-
-    patch = extract_requirement_patch(text, previous=previous_snapshot)
-    revision = max(1, int(previous_revision) + 1) if previous_snapshot or patch else max(1, int(previous_revision) or 1)
-    if not previous_snapshot:
-        revision = 1
-    requirement = _merge_snapshot(previous_snapshot, patch, revision=revision)
-    skip = wants_direct_start(text)
-    hit_cap = clarification_round >= max_clarification_rounds
-    gaps = _gaps(requirement)
-
-    if requirement.topic == "未指定主题" and skip:
-        return PrePlanDecision(
-            action="clarify",
-            requirement=requirement,
-            questions=["这份报告要写什么主题？请用一两句话说明对象、时间和目的。"],
-            impact=["没有主题无法选择方案"],
-            reason="用户要求直接开始，但还没有主题",
-            confidence=0.9,
-        )
-
-    if skip or (hit_cap and gaps):
-        filled, assumptions = apply_default_assumptions(requirement)
-        return PrePlanDecision(
-            action="ready",
-            requirement=filled,
-            assumptions=assumptions,
-            used_defaults=True,
-            reason="用户要求按默认假设继续" if skip else "已达澄清上限，使用明确假设继续",
-            confidence=0.86 if skip else 0.7,
-        )
-
-    if gaps:
-        return PrePlanDecision(
-            action="clarify",
-            requirement=requirement,
-            questions=[item[1] for item in gaps],
-            impact=[item[2] for item in gaps],
-            reason="缺少真正影响方案选择的信息",
-            confidence=0.8,
-        )
-
-    return PrePlanDecision(
-        action="ready",
-        requirement=requirement,
-        reason="必要信息已齐，可以进入规划",
-        confidence=0.84,
-        used_defaults=False,
-    )

offline-backend-20260512/backend/app/report_collaboration/requirement/resolver.py

@@ -1,18 +1,39 @@
-"""Validate and persist requirement revisions. Never starts planning or a run."""
+"""Requirement resolver: the planner agent is the only path.
+
+模型不可用即无法继续 —— 直接抛 ``PLANNER_UNAVAILABLE``，没有任何规则/
+模板兜底（设计 v2：需求确认必须由真智能体完成）。
+"""
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 from uuid import uuid4
 
 from app.report_collaboration.contracts.clarifications import ClarificationRecord
-from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
 from app.report_collaboration.requirement.defaults import COORDINATOR_NAME, DEFAULT_TITLE
-from app.report_collaboration.requirement.intent_router import PrePlanDecision, route_pre_plan_turn
+from app.report_collaboration.requirement.intent_router import PrePlanDecision
 from deerflow.config.app_config import get_app_config
-from deerflow.persistence.report_collaboration import ReportCollaborationStore
+from deerflow.persistence.report_collaboration import (
+    ReportCollaborationStore,
+    ReportCollaborationValidationError,
+)
 from deerflow.persistence.report_collaboration.codec import iso_now
 
+logger = logging.getLogger(__name__)
+
+# 仅供类型注解；运行时惰性导入避免 requirement ↔ planning 循环依赖。
+LlmFactory = Any
+
+
+def _title_from_topic(topic: str, current: str | None) -> str | None:
+    if current and current != DEFAULT_TITLE:
+        return None
+    cleaned = (topic or "").strip()
+    if not cleaned or cleaned == "未指定主题":
+        return None
+    return cleaned[:40]
+
 
 def _state_from_session(record: dict[str, Any]) -> dict[str, Any]:
     raw = record.get("requirement_json")
@@ -35,58 +56,56 @@ def _state_from_session(record: dict[str, Any]) -> dict[str, Any]:
     return {"snapshot": None, "clarifications": [], "assumptions": [], "clarification_round": 0, "ready_to_plan": False}
 
 
-def _dump_state(
-    *,
-    snapshot: ReportRequirementSnapshot,
-    clarifications: list[dict[str, Any]],
-    assumptions: list[str],
-    clarification_round: int,
-    ready_to_plan: bool,
-) -> dict[str, Any]:
-    return {
-        "snapshot": snapshot.model_dump(),
-        "clarifications": clarifications,
-        "assumptions": assumptions,
-        "clarification_round": clarification_round,
-        "ready_to_plan": ready_to_plan,
-    }
-
-
-def format_clarification_message(questions: list[str], *, allow_skip: bool = True) -> str:
-    numbered = "\n".join(f"{index}. {question}" for index, question in enumerate(questions, start=1))
-    suffix = "\n\n也可以直接回复「直接开始」，我会按默认假设记录需求。澄清完成不会自动生成方案。" if allow_skip else ""
-    return f"收到您的报告需求。为了给出最合适的协作方案，请先确认：\n\n{numbered}{suffix}"
-
-
-def format_ready_message(decision: PrePlanDecision) -> str:
-    req = decision.requirement
-    angles = "、".join(req.required_angles) if req.required_angles else "（未指定）"
-    lines = [
-        f"已写入需求版本 v{req.revision}。",
-        f"主题：{req.topic}",
-        f"读者：{req.audience or '未指定'}",
-        f"角度：{angles}",
-    ]
-    if decision.assumptions:
-        lines.append("默认假设：" + "；".join(decision.assumptions) + "。")
-    lines.append("澄清完成不等于已经生成候选方案，需要显式发起规划后才会出现方案卡。")
-    return "\n".join(lines)
-
-
-def _title_from_topic(topic: str, current: str | None) -> str | None:
-    if current and current != DEFAULT_TITLE:
-        return None
-    cleaned = (topic or "").strip()
-    if not cleaned or cleaned == "未指定主题":
-        return None
-    return cleaned[:40]
-
-
 class RequirementResolver:
     """Writes requirement revisions and clarification records. Does not enqueue plans."""
 
-    def __init__(self, store: ReportCollaborationStore) -> None:
+    def __init__(
+        self,
+        store: ReportCollaborationStore,
+        *,
+        planner_llm_factory: LlmFactory | None = None,
+        config: Any | None = None,
+    ) -> None:
         self._store = store
+        self._planner_llm_factory = planner_llm_factory
+        self._config = config
+        self._default_llm: Any = None
+        self._default_llm_tried = False
+
+    def _planner_llm(self) -> Any:
+        """规划师模型：注入优先（测试）；否则按 planner_llm_enabled 懒构建。不可用即抛错。"""
+
+        if self._planner_llm_factory is not None:
+            llm = self._planner_llm_factory()
+            if llm is None:
+                raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置")
+            return llm
+        try:
+            from deerflow.config.report_collaboration_config import ReportCollaborationConfig
+
+            cfg: Any = self._config
+            if cfg is None:
+                cfg = get_app_config().report_collaboration
+            if not ReportCollaborationConfig.model_validate(cfg).planner_llm_enabled:
+                raise ReportCollaborationValidationError(
+                    "PLANNER_UNAVAILABLE", "规划师智能体未启用（report_collaboration.planner_llm_enabled）"
+                )
+        except ReportCollaborationValidationError:
+            raise
+        except Exception as exc:
+            raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", f"规划师智能体配置不可读：{exc}") from None
+        if not self._default_llm_tried:
+            self._default_llm_tried = True
+            try:
+                from app.report_collaboration.planning.planner_agent import build_planner_llm
+
+                self._default_llm = build_planner_llm()
+            except Exception as exc:
+                logger.warning("rc planner agent llm build failed: %s", exc)
+                self._default_llm = None
+        if self._default_llm is None:
+            raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置")
+        return self._default_llm
 
     async def handle_user_message(
         self,
@@ -111,16 +130,67 @@ class RequirementResolver:
         cfg_rounds = max_clarification_rounds
         if cfg_rounds is None:
             cfg_rounds = int(get_app_config().report_collaboration.max_clarification_rounds)
-        decision = route_pre_plan_turn(
-            str(user_message.get("content") or ""),
-            previous_snapshot=state["snapshot"],
+
+        text = str(user_message.get("content") or "")
+        history = [
+            {"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")}
+            for item in messages
+            if item.get("message_id") != user_message["message_id"]
+        ]
+
+        decision = await self._agent_decision(
+            text,
+            history=history,
+            state=state,
             previous_revision=int(record.get("requirement_revision") or 0),
-            clarification_round=int(state["clarification_round"] or 0),
-            max_clarification_rounds=cfg_rounds,
+            max_rounds=cfg_rounds,
         )
+        if decision is None:
+            # 模型调用失败/输出不合法 —— 无法继续，直接报错（无兜底）。
+            raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体本轮未返回有效决策，请重试")
         await self._apply(session_id, record, state, decision, user_message=user_message, idempotency_key=idempotency_key)
         return decision
 
+    async def _agent_decision(
+        self,
+        text: str,
+        *,
+        history: list[dict[str, str]],
+        state: dict[str, Any],
+        previous_revision: int,
+        max_rounds: int,
+    ) -> PrePlanDecision | None:
+        llm = self._planner_llm()
+        from app.report_collaboration.planning.planner_agent import PlannerAgent
+
+        agent = PlannerAgent(llm)
+        turn = await agent.converse(
+            user_text=text,
+            history=history,
+            previous_snapshot=state["snapshot"],
+            previous_revision=previous_revision,
+            clarification_round=int(state["clarification_round"] or 0),
+            max_rounds=max_rounds,
+        )
+        if turn is None:
+            return None
+        requirement = agent.merge_requirement(
+            turn,
+            previous_snapshot=state["snapshot"],
+            previous_revision=previous_revision,
+        )
+        return PrePlanDecision(
+            action="ready" if turn.action == "ready" else "clarify",
+            requirement=requirement,
+            questions=[item for item in turn.questions if item and item.strip()][:3],
+            impact=[item for item in turn.impact if item][:3],
+            assumptions=[item for item in turn.assumptions if item],
+            used_defaults=bool(turn.assumptions),
+            reason="planner_agent",
+            confidence=0.9,
+            reply=turn.reply.strip(),
+        )
+
     async def _apply(
         self,
         session_id: str,
@@ -146,7 +216,7 @@ class RequirementResolver:
                 created_at=iso_now(),
             )
             clarifications.append(clarification.model_dump())
-            content = format_clarification_message(decision.questions)
+            content = decision.reply or ""
             metadata = {
                 "preplan_for": user_message["message_id"],
                 "clarification": clarification.model_dump(),
@@ -155,17 +225,22 @@ class RequirementResolver:
         else:
             for item in clarifications:
                 item["resolved"] = True
-            content = format_ready_message(decision)
-            metadata = {"preplan_for": user_message["message_id"], "ready_to_plan": True}
+            content = decision.reply or ""
+            metadata = {
+                "preplan_for": user_message["message_id"],
+                "ready_to_plan": True,
+                # 结构化需求快照：前端据此渲染「需求摘要卡」（计划模式 UI）。
+                "requirement": decision.requirement.model_dump(),
+            }
             ready = True
 
-        payload = _dump_state(
-            snapshot=decision.requirement,
-            clarifications=clarifications,
-            assumptions=list(decision.assumptions),
-            clarification_round=round_count,
-            ready_to_plan=ready,
-        )
+        payload = {
+            "snapshot": decision.requirement.model_dump(),
+            "clarifications": clarifications,
+            "assumptions": list(decision.assumptions),
+            "clarification_round": round_count,
+            "ready_to_plan": ready,
+        }
         await self._store.save_requirement_state(
             session_id,
             requirement_json=payload,

offline-backend-20260512/backend/config.yaml

@@ -748,6 +748,7 @@ workflows:
 report_collaboration:
   enabled: true
   worker_enabled: true
+  planner_llm_enabled: true
   planner_model: null
   coordinator_model: null
   research_model: null

offline-backend-20260512/backend/packages/harness/deerflow/config/report_collaboration_config.py

@@ -18,6 +18,10 @@ class ReportCollaborationConfig(BaseModel):
         default=False,
         description="When true, a dispatcher/worker may claim run leases. Gateway-only processes keep this false.",
     )
+    planner_llm_enabled: bool = Field(
+        default=False,
+        description="P1: LLM planner conversation (clarify + custom strategy). Off keeps the deterministic rule router and template plans.",
+    )
     planner_model: str | None = Field(default=None, description="Model name from AppConfig.models for planning")
     coordinator_model: str | None = Field(default=None, description="Model name from AppConfig.models for the constrained coordinator")
     research_model: str | None = Field(default=None, description="Model name from AppConfig.models for research members")

offline-backend-20260512/backend/tests/test_report_collaboration_api.py

@@ -25,6 +25,40 @@ class _DisabledCfg:
     report_collaboration = ReportCollaborationConfig(enabled=False)
 
 
+class _PlannerFakeLLM:
+    """规划师假模型：需求确认回「读者」追问；工作流设计回单份定制方案。"""
+
+    _converse = (
+        '{"reply": "好的。先确认：这份报告的主要读者是谁？", "action": "clarify",'
+        '"questions": ["这份报告的主要读者是谁？"], "impact": ["读者影响研究线划分"],'
+        '"requirement": {"topic": "新能源汽车趋势报告"}}'
+    )
+    _ready = (
+        '{"reply": "明白了，读者为企业战略决策人员。我来设计工作流。", "action": "ready",'
+        '"assumptions": [], "requirement": {"topic": "新能源汽车趋势报告", "audience": "企业战略决策人员"}}'
+    )
+    _design = (
+        '{"reply": "已按需求设计定制工作流。", "strategy": "custom", "title": "定制方案",'
+        '"summary": "研究后写作", "rationale": "贴合需求", "estimated_duration_seconds": 900,'
+        '"estimated_cost_level": "low", "quality_gate_ids": ["cite_required"],'
+        '"nodes": ['
+        '{"slot_id": "research-1", "role_key": "researcher", "display_name": "研究员", "mission": "检索证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},'
+        '{"slot_id": "write-1", "role_key": "writer", "display_name": "撰写员", "mission": "成稿", "depends_on": ["research-1"], "output_artifact_type": "ReportSectionDraft"}]}'
+    )
+
+    async def ainvoke(self, messages):
+        from types import SimpleNamespace
+
+        prompt = str(messages[-1][1]) if messages else ""
+        if "请设计工作流" in prompt:
+            payload = self._design
+        elif "企业战略决策人员" in prompt or "读者" in prompt and "用户最新消息" in prompt:
+            payload = self._ready
+        else:
+            payload = self._converse
+        return SimpleNamespace(content=payload)
+
+
 def _client(monkeypatch: pytest.MonkeyPatch, *, enabled: bool = True, user: str | None = "user_a") -> tuple[TestClient, MemoryReportCollaborationStore]:
     monkeypatch.setattr(report_collaboration, "get_app_config", lambda: _Cfg() if enabled else _DisabledCfg())
     app = FastAPI()
@@ -32,6 +66,7 @@ def _client(monkeypatch: pytest.MonkeyPatch, *, enabled: bool = True, user: str
     inner = MemoryReportCollaborationStore()
     store = PublishingReportCollaborationStore(inner, hub)
     app.state.report_collaboration_store = store
+    app.state.report_collaboration_planner_llm = _PlannerFakeLLM
     app.state.report_collaboration_live_hub = hub
     app.state.report_collaboration_sse_settings = SseStreamSettings(
         deadline_seconds=0.6,
@@ -126,14 +161,14 @@ def test_session_message_plan_run_http(monkeypatch: pytest.MonkeyPatch) -> None:
     proposed_body = proposed.json()
     assert proposed_body["session"]["status"] == "proposal_ready"
     plans = proposed_body["plans"]
-    assert {item["strategy"] for item in plans} == {"balanced_review", "parallel_depth", "focused_fast"}
+    assert {item["strategy"] for item in plans} == {"custom"}
     plan = next(item for item in plans if item.get("recommended"))
     assert plan["validation"]["ok"] is True
     assert all(node.get("output_artifact_type") and node.get("acceptance_criteria") for node in plan["nodes"])
 
     listed = client.get(f"/api/report-collaboration/sessions/{session_id}/plans")
     assert listed.status_code == 200
-    assert len(listed.json()["plans"]) == 3
+    assert len(listed.json()["plans"]) == 1
 
     selected = client.post(
         f"/api/report-collaboration/sessions/{session_id}/plans/{plan['id']}/select",

offline-backend-20260512/backend/tests/test_report_collaboration_models.py

@@ -7,7 +7,7 @@ from types import SimpleNamespace
 import pytest
 
 from app.report_collaboration.agentscope_runtime.event_adapter import ConversationReplica, MemberStreamContext, SeqAllocator, make_emitter
-from app.report_collaboration.agentscope_runtime.model_adapter import DeerFlowChatModelAdapter, ModelAdapterError, record_attempt_usage, redact_secrets
+from app.report_collaboration.agentscope_runtime.model_adapter import DeerFlowChatModelAdapter, ModelAdapterError, estimate_tokens_for_text, record_attempt_usage, redact_secrets
 from app.report_collaboration.agentscope_runtime.model_resolver import freeze_plan_models, resolve_role_model
 from deerflow.config.app_config import AppConfig
 from deerflow.config.model_config import ModelConfig
@@ -128,6 +128,33 @@ async def test_adapter_streams_text_tool_and_usage() -> None:
     assert frozen.value.code == "MODEL_FROZEN"
 
 
+@pytest.mark.asyncio
+async def test_adapter_count_tokens_and_context_size_contract() -> None:
+    """AgentScope compress_context 依赖 model.context_size 与 count_tokens(messages, tools)。"""
+
+    snapshot = resolve_role_model("researcher", app_config=_app())
+    adapter = DeerFlowChatModelAdapter(snapshot, app_config=_app())
+    assert isinstance(adapter.context_size, int) and adapter.context_size > 0
+
+    class _Msg:
+        def __init__(self, text: str) -> None:
+            self.content = text
+
+        def get_content_blocks(self):
+            return [{"type": "text", "text": self.content}]
+
+    agentscope_style = [_Msg("市场分析" * 20)]
+    tokens_msg = await adapter.count_tokens(agentscope_style, None)
+    dict_style = [{"role": "user", "content": "hello world"}]
+    tokens_dict = await adapter.count_tokens(dict_style, None)
+    tools = [{"name": "web_search", "parameters": {"type": "object"}}]
+    tokens_tools = await adapter.count_tokens(dict_style, tools)
+    assert tokens_msg > 0
+    assert tokens_dict == estimate_tokens_for_text("hello world")
+    assert tokens_tools > tokens_dict
+    assert await adapter.count_tokens([], None) == 0
+
+
 @pytest.mark.asyncio
 async def test_adapter_redacts_provider_errors_and_records_metrics() -> None:
     snapshot = resolve_role_model("writer", app_config=_app())

offline-backend-20260512/backend/tests/test_report_collaboration_planner_agent.py

@@ -0,0 +1,181 @@
+"""P1-R 规划师智能体：真对话（文案由智能体产出直接落库）+ 工作流设计 + 降级路径。"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
+from app.report_collaboration.planning.planner_agent import PlannerAgent
+from app.report_collaboration.requirement.resolver import RequirementResolver
+from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore
+
+
+class _FakeLLM:
+    def __init__(self, payload: str) -> None:
+        self._payload = payload
+
+    async def ainvoke(self, messages):
+        return SimpleNamespace(content=self._payload)
+
+
+_TURN_CLARIFY = """```json
+{"reply": "好的，智能家居是个不错的方向。为了让团队分工更准，我想先确认两点：这份报告主要给谁看？大概需要多长的篇幅？",
+ "action": "clarify",
+ "questions": ["这份报告主要给谁看？", "大概需要多长的篇幅？"],
+ "impact": ["读者决定研究线划分", "篇幅决定成员数量"],
+ "requirement": {"topic": "智能家居设备2026年市场分析", "required_angles": ["市场规模", "竞争格局", "技术趋势"]}}
+```"""
+
+_TURN_READY = """{"reply": "明白了：给公司管理层的战略规划参考，5000字左右，覆盖三个角度。我现在就可以为你们设计工作流了。",
+ "action": "ready", "assumptions": ["地域默认中国大陆"],
+ "requirement": {"topic": "智能家居设备2026年市场分析", "audience": "公司管理层", "required_angles": ["市场规模", "竞争格局", "技术趋势"], "length_target": "5000字"}}"""
+
+_DESIGN_JSON = """{"reply": "我按三个角度设计了两条并行研究线汇写再审的链路，正好对齐管理层决策的阅读习惯。",
+ "strategy": "custom", "title": "三角度并行研究工作流", "summary": "三线并行 + 汇写 + 独立审稿",
+ "rationale": "需求明确三个角度，管理层需要可信结论", "estimated_duration_seconds": 1800,
+ "estimated_cost_level": "medium", "quality_gate_ids": ["cite_required", "independent_review"],
+ "nodes": [
+  {"slot_id": "research-scale", "role_key": "researcher", "display_name": "研究员", "angle": "市场规模",
+   "mission": "检索市场规模证据", "depends_on": [], "output_artifact_type": "EvidenceBundle", "suggested_tools": ["web_search"]},
+  {"slot_id": "research-comp", "role_key": "researcher", "display_name": "研究员", "angle": "竞争格局",
+   "mission": "检索竞争格局证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},
+  {"slot_id": "research-tech", "role_key": "researcher", "display_name": "研究员", "angle": "技术趋势",
+   "mission": "检索技术趋势证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},
+  {"slot_id": "write-main", "role_key": "writer", "display_name": "撰写员",
+   "mission": "汇总三线证据成稿", "depends_on": ["research-scale", "research-comp", "research-tech"], "output_artifact_type": "ReportSectionDraft"},
+  {"slot_id": "review-1", "role_key": "reviewer", "display_name": "审稿员",
+   "mission": "独立审稿", "depends_on": ["write-main"], "output_artifact_type": "ReviewDecision"}
+ ]}"""
+
+
+# ── converse：智能体对话 ───────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_converse_returns_agent_reply_and_questions() -> None:
+    agent = PlannerAgent(_FakeLLM(_TURN_CLARIFY))
+    turn = await agent.converse(
+        user_text="帮我写一份智能家居设备2026年市场分析报告",
+        history=[],
+        previous_snapshot=None,
+        previous_revision=0,
+        clarification_round=0,
+        max_rounds=3,
+    )
+    assert turn is not None and turn.action == "clarify"
+    assert "为了让团队分工更准" in turn.reply
+    assert len(turn.questions) == 2
+    merged = agent.merge_requirement(turn, previous_snapshot=None, previous_revision=0)
+    assert merged.topic == "智能家居设备2026年市场分析"
+    assert merged.revision == 1
+
+
+@pytest.mark.asyncio
+async def test_converse_forced_ready_at_round_cap() -> None:
+    agent = PlannerAgent(_FakeLLM(_TURN_CLARIFY))
+    turn = await agent.converse(
+        user_text="再说说",
+        history=[],
+        previous_snapshot={"topic": "智能家居市场", "revision": 2},
+        previous_revision=2,
+        clarification_round=3,
+        max_rounds=3,
+    )
+    assert turn is not None and turn.action == "ready"
+    assert turn.assumptions
+
+
+@pytest.mark.asyncio
+async def test_converse_invalid_json_returns_none() -> None:
+    agent = PlannerAgent(_FakeLLM("我拒绝输出 JSON"))
+    assert (
+        await agent.converse(
+            user_text="写报告",
+            history=[],
+            previous_snapshot=None,
+            previous_revision=0,
+            clarification_round=0,
+            max_rounds=3,
+        )
+        is None
+    )
+
+
+@pytest.mark.asyncio
+async def test_resolver_persists_agent_reply_without_template_wrapper() -> None:
+    store = MemoryReportCollaborationStore()
+    session = await store.create_session(owner_id="u1", title="新会话", idempotency_key="ik1")
+    resolver = RequirementResolver(store, planner_llm_factory=lambda: _FakeLLM(_TURN_CLARIFY))
+    decision = await resolver.handle_user_message(
+        session["id"],
+        user_message={"message_id": "m1", "role": "human", "content": "帮我写一份智能家居设备2026年市场分析报告"},
+        idempotency_key="ik-m1",
+    )
+    assert decision is not None and decision.reply and "为了让团队分工更准" in decision.reply
+    messages = await store.list_messages(session["id"])
+    ai_messages = [item for item in messages if item["role"] == "ai"]
+    assert ai_messages, "agent reply message missing"
+    assert "为了让团队分工更准" in ai_messages[-1]["content"]
+    assert "收到您的报告需求" not in ai_messages[-1]["content"]  # 模板文案不得出现
+
+
+@pytest.mark.asyncio
+async def test_resolver_raises_when_agent_output_invalid() -> None:
+    """模型输出不合法 → 直接报错（无规则/模板兜底）。"""
+
+    from deerflow.persistence.report_collaboration import ReportCollaborationValidationError
+
+    store = MemoryReportCollaborationStore()
+    session = await store.create_session(owner_id="u1", title="新会话", idempotency_key="ik1")
+    resolver = RequirementResolver(store, planner_llm_factory=lambda: _FakeLLM("not json"))
+    with pytest.raises(ReportCollaborationValidationError) as exc:
+        await resolver.handle_user_message(
+            session["id"],
+            user_message={"message_id": "m1", "role": "human", "content": "帮我写一份智能家居设备2026年市场分析报告"},
+            idempotency_key="ik-m1",
+        )
+    assert exc.value.code == "PLANNER_UNAVAILABLE"
+
+
+# ── design_workflow：工作流设计 ───────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_design_workflow_returns_strategy_with_context_reply() -> None:
+    requirement = ReportRequirementSnapshot(
+        topic="智能家居设备2026年市场分析",
+        required_angles=["市场规模", "竞争格局", "技术趋势"],
+        revision=2,
+    )
+    history = [{"role": "human", "content": "给管理层看，覆盖三个角度"}]
+    designed = await PlannerAgent(_FakeLLM(_DESIGN_JSON)).design_workflow(
+        requirement=requirement, history=history
+    )
+    assert designed is not None
+    strategy, reply = designed
+    assert strategy.strategy == "custom"
+    assert {slot.demand.role_key for slot in strategy.slots} == {"researcher", "writer", "reviewer"}
+    assert "管理层" in reply or "设计" in reply
+    write = next(slot for slot in strategy.slots if slot.slot_id == "write-main")
+    assert set(write.depends_on) == {"research-scale", "research-comp", "research-tech"}
+
+
+@pytest.mark.asyncio
+async def test_design_workflow_rejects_bad_vocabulary_and_cycles() -> None:
+    requirement = ReportRequirementSnapshot(topic="测试", revision=1)
+    agent = PlannerAgent(_FakeLLM(_DESIGN_JSON))
+
+    bad_role = _DESIGN_JSON.replace('"role_key": "writer"', '"role_key": "project_manager"')
+    assert await PlannerAgent(_FakeLLM(bad_role)).design_workflow(requirement=requirement, history=[]) is None
+
+    cyclic = _DESIGN_JSON.replace(
+        '"depends_on": [], "output_artifact_type": "EvidenceBundle", "suggested_tools"',
+        '"depends_on": ["review-1"], "output_artifact_type": "EvidenceBundle", "suggested_tools"',
+    )
+    assert await PlannerAgent(_FakeLLM(cyclic)).design_workflow(requirement=requirement, history=[]) is None
+
+    bad_artifact = _DESIGN_JSON.replace('"output_artifact_type": "ReportSectionDraft"', '"output_artifact_type": "PDF"')
+    assert await PlannerAgent(_FakeLLM(bad_artifact)).design_workflow(requirement=requirement, history=[]) is None
+    _ = agent  # 保持引用清晰

offline-backend-20260512/backend/tests/test_report_collaboration_planning.py

@@ -12,7 +12,7 @@ from app.report_collaboration.planning.selector import AgentSelector
 from app.report_collaboration.planning.service import PlanProposalService
 from app.report_collaboration.planning.validator import validate_assembled_plan
 from deerflow.config.report_collaboration_config import ReportCollaborationConfig
-from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore
+from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore, ReportCollaborationValidationError
 
 
 class _AgentStore:
@@ -135,6 +135,26 @@ def test_validator_rejects_unknown_agent_id() -> None:
     asyncio.run(_run())
 
 
+class _DesignFakeLLM:
+    """规划师工作流设计假模型：返回单份合法定制方案。"""
+
+    _payload = (
+        '{"reply": "我按需求设计了两条研究线汇写再审的定制工作流。",'
+        '"strategy": "custom", "title": "定制方案", "summary": "两线并行汇写再审",'
+        '"rationale": "贴合需求", "estimated_duration_seconds": 1200,'
+        '"estimated_cost_level": "medium", "quality_gate_ids": ["cite_required"],'
+        '"nodes": ['
+        '{"slot_id": "research-a", "role_key": "researcher", "display_name": "研究员", "angle": "市场", "mission": "检索市场证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},'
+        '{"slot_id": "write-main", "role_key": "writer", "display_name": "撰写员", "mission": "成稿", "depends_on": ["research-a"], "output_artifact_type": "ReportSectionDraft"}'
+        ']}'
+    )
+
+    async def ainvoke(self, messages):
+        from types import SimpleNamespace
+
+        return SimpleNamespace(content=self._payload)
+
+
 @pytest.mark.asyncio
 async def test_service_writes_plans_without_starting_a_run() -> None:
     store = MemoryReportCollaborationStore()
@@ -146,15 +166,20 @@ async def test_service_writes_plans_without_starting_a_run() -> None:
         status="clarifying",
     )
     command = await store.enqueue_plan_request(session["id"], idempotency_key="ik-plan", expected_revision=1)
-    saved = await PlanProposalService(store).fulfill(session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan")
-    assert len(saved) == 3
+    saved = await PlanProposalService(store, llm_factory=lambda: _DesignFakeLLM()).fulfill(
+        session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan"
+    )
+    assert len(saved) == 1  # 智能体生成单份定制方案
     snapshot = await store.get_snapshot(session["id"])
     assert snapshot["session"]["status"] == "proposal_ready"
     assert snapshot["session"]["selected_plan_id"] is None
     assert snapshot["run"] is None
-    assert {item["strategy"] for item in snapshot["plans"]} == {"balanced_review", "parallel_depth", "focused_fast"}
+    assert snapshot["plans"][0]["strategy"] == "custom"
     assert all(item["requirement_revision"] == 1 for item in snapshot["plans"])
     assert snapshot["session"]["active_run_id"] is None
+    ai = [item for item in snapshot["messages"] if item["role"] == "ai"][-1]
+    assert ai["metadata"]["plan_source"] == "planner_agent"
+    assert "定制工作流" in ai["content"]
 
     selected = await store.select_plan(session["id"], saved[0]["id"], idempotency_key="ik-select", expected_revision=1)
     after_select = await store.get_snapshot(session["id"])
@@ -163,6 +188,24 @@ async def test_service_writes_plans_without_starting_a_run() -> None:
     assert after_select["session"]["active_run_id"] is None
 
 
+@pytest.mark.asyncio
+async def test_service_raises_when_planner_unavailable() -> None:
+    store = MemoryReportCollaborationStore()
+    session = await store.create_session(owner_id="u1", title="", idempotency_key="s1b")
+    await store.save_requirement_state(
+        session["id"],
+        requirement_json={"snapshot": _requirement().model_dump(), "clarifications": [], "assumptions": [], "clarification_round": 1, "ready_to_plan": True},
+        requirement_revision=1,
+        status="clarifying",
+    )
+    command = await store.enqueue_plan_request(session["id"], idempotency_key="ik-plan2", expected_revision=1)
+    with pytest.raises(ReportCollaborationValidationError) as exc:
+        await PlanProposalService(store, llm_factory=lambda: None).fulfill(
+            session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan2"
+        )
+    assert exc.value.code == "PLANNER_UNAVAILABLE"
+
+
 def test_cycle_is_rejected() -> None:
     cfg = ReportCollaborationConfig()
     candidate_nodes_cycle = AssembledPlan.model_validate(

offline-backend-20260512/backend/tests/test_report_collaboration_requirement.py

@@ -1,76 +1,35 @@
-"""RC-BE-005: pre-plan intent router and requirement resolver."""
+"""P1-R 需求确认：规划师智能体是唯一路径（无规则/模板兜底）。"""
 
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 import pytest
 
 from app.report_collaboration.requirement.extractor import extract_topic, wants_direct_start
-from app.report_collaboration.requirement.intent_router import route_pre_plan_turn
 from app.report_collaboration.requirement.resolver import RequirementResolver
-from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore
+from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore, ReportCollaborationValidationError
 
 
-def test_topic_only_asks_plan_affecting_gaps() -> None:
-    decision = route_pre_plan_turn(
-        "写一份新能源汽车主要品牌市场趋势分析报告",
-        previous_snapshot=None,
-        previous_revision=0,
-        clarification_round=0,
-        max_clarification_rounds=3,
-    )
-    assert decision.action == "clarify"
-    assert "新能源" in decision.requirement.topic
-    assert decision.requirement.revision == 1
-    assert len(decision.questions) >= 2
-    assert any("读者" in item for item in decision.questions)
-    assert any("角度" in item for item in decision.questions)
+class _ScriptedLLM:
+    """按顺序吐出预置 JSON 决策（模拟规划师智能体多轮）。"""
+
+    def __init__(self, payloads: list[str]) -> None:
+        self._payloads = list(payloads)
+
+    async def ainvoke(self, messages):
+        payload = self._payloads.pop(0) if self._payloads else self._payloads[0]
+        return SimpleNamespace(content=payload)
 
 
-def test_direct_start_records_default_assumptions() -> None:
-    first = route_pre_plan_turn(
-        "撰写新能源汽车趋势报告，直接开始",
-        previous_snapshot=None,
-        previous_revision=0,
-        clarification_round=0,
-        max_clarification_rounds=3,
-    )
-    assert first.action == "ready"
-    assert first.used_defaults is True
-    assert first.requirement.audience == "企业战略决策人员"
-    assert "市场份额" in first.requirement.required_angles
-    assert first.assumptions
+_CLARIFY = """{"reply": "新能源汽车趋势是个好题目。先确认两个关键点：这份报告主要给谁看？希望覆盖哪些角度？",
+ "action": "clarify", "questions": ["这份报告主要给谁看？", "希望覆盖哪些角度？"],
+ "impact": ["读者影响研究线", "角度影响成员分工"],
+ "requirement": {"topic": "新能源汽车趋势报告"}}"""
 
-
-def test_direct_start_without_topic_still_asks_topic() -> None:
-    decision = route_pre_plan_turn(
-        "直接开始",
-        previous_snapshot=None,
-        previous_revision=0,
-        clarification_round=0,
-        max_clarification_rounds=3,
-    )
-    assert decision.action == "clarify"
-    assert any("主题" in item for item in decision.questions)
-
-
-def test_clarification_cap_uses_explicit_assumptions() -> None:
-    first = route_pre_plan_turn(
-        "写一份行业报告",
-        previous_snapshot=None,
-        previous_revision=0,
-        clarification_round=0,
-        max_clarification_rounds=1,
-    )
-    assert first.action == "clarify"
-    capped = route_pre_plan_turn(
-        "先这样吧",
-        previous_snapshot=first.requirement.model_dump(),
-        previous_revision=first.requirement.revision,
-        clarification_round=1,
-        max_clarification_rounds=1,
-    )
-    assert capped.action == "ready"
-    assert capped.used_defaults is True
+_READY = """{"reply": "清楚了：给投资机构，覆盖市场份额、价格策略和政策环境。我来为你们设计工作流。",
+ "action": "ready", "assumptions": ["时间范围默认近两年"],
+ "requirement": {"topic": "新能源汽车趋势报告", "audience": "投资机构", "required_angles": ["市场份额", "价格策略", "政策环境"]}}"""
 
 
 def test_extract_helpers() -> None:
@@ -79,14 +38,15 @@ def test_extract_helpers() -> None:
 
 
 @pytest.mark.asyncio
-async def test_resolver_writes_clarification_without_planning() -> None:
+async def test_resolver_clarify_round_persists_agent_reply_and_record() -> None:
     store = MemoryReportCollaborationStore()
     session = await store.create_session(owner_id="u1", title="", idempotency_key="s1")
     user = await store.create_message(session["id"], role="human", content="写一份新能源汽车趋势报告", idempotency_key="m1")
-    resolver = RequirementResolver(store)
+    llm = _ScriptedLLM([_CLARIFY])
+    resolver = RequirementResolver(store, planner_llm_factory=lambda: llm)
     decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3)
-    assert decision is not None
-    assert decision.action == "clarify"
+    assert decision is not None and decision.action == "clarify"
+    assert "先确认两个关键点" in (decision.reply or "")
 
     snapshot = await store.get_snapshot(session["id"])
     assert snapshot["session"]["status"] == "clarifying"
@@ -96,10 +56,14 @@ async def test_resolver_writes_clarification_without_planning() -> None:
     assert len(ai) == 1
     assert ai[0]["completed_at"]
     assert ai[0]["name"] == "协调者"
+    # 智能体文案直接落库，模板开场不得出现
+    assert "先确认两个关键点" in ai[0]["content"]
+    assert "收到您的报告需求" not in ai[0]["content"]
     record = await store.get_session_record(session["id"])
     assert record["requirement_json"]["ready_to_plan"] is False
     assert record["requirement_json"]["clarifications"][0]["resolved"] is False
 
+    # 幂等：同一消息重放不产生第二条 AI 回复
     replay = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3)
     assert replay is None
     snapshot2 = await store.get_snapshot(session["id"])
@@ -107,52 +71,63 @@ async def test_resolver_writes_clarification_without_planning() -> None:
 
 
 @pytest.mark.asyncio
-async def test_resolver_ready_is_not_plan_ready() -> None:
+async def test_resolver_ready_writes_requirement_metadata_for_summary_card() -> None:
     store = MemoryReportCollaborationStore()
     session = await store.create_session(owner_id="u1", title="", idempotency_key="s2")
     user = await store.create_message(
-        session["id"],
-        role="human",
-        content="写一份新能源汽车趋势报告，面向战略决策层，必须覆盖市场、政策、海外和风险，直接开始",
-        idempotency_key="m2",
+        session["id"], role="human", content="写一份新能源汽车趋势报告，给投资机构看", idempotency_key="m2"
     )
-    resolver = RequirementResolver(store)
+    llm = _ScriptedLLM([_READY])
+    resolver = RequirementResolver(store, planner_llm_factory=lambda: llm)
     decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m2")
-    assert decision is not None
-    assert decision.action == "ready"
+    assert decision is not None and decision.action == "ready"
 
     snapshot = await store.get_snapshot(session["id"])
     assert snapshot["session"]["status"] == "clarifying"
-    assert snapshot["plans"] == []
+    assert snapshot["plans"] == []  # ready ≠ 已生成方案
+    ai = [item for item in snapshot["messages"] if item["role"] == "ai"][-1]
+    assert ai["metadata"]["ready_to_plan"] is True
+    assert ai["metadata"]["requirement"]["audience"] == "投资机构"
+    assert "设计工作流" in ai["content"]
     record = await store.get_session_record(session["id"])
     assert record["requirement_json"]["ready_to_plan"] is True
-    assert all(item["resolved"] for item in record["requirement_json"]["clarifications"])
-    assert "不等于已经生成" in snapshot["messages"][-1]["content"]
 
 
 @pytest.mark.asyncio
-async def test_answer_merges_and_resolves_clarification() -> None:
+async def test_resolver_answer_merges_fields_across_rounds() -> None:
     store = MemoryReportCollaborationStore()
     session = await store.create_session(owner_id="u1", title="", idempotency_key="s3")
     first = await store.create_message(session["id"], role="human", content="写一份新能源汽车趋势报告", idempotency_key="m3a")
-    resolver = RequirementResolver(store)
+    llm = _ScriptedLLM([_CLARIFY, _READY])
+    resolver = RequirementResolver(store, planner_llm_factory=lambda: llm)
     await resolver.handle_user_message(session["id"], user_message=first, idempotency_key="m3a")
 
     second = await store.create_message(
-        session["id"],
-        role="human",
-        content="读者是投资机构，必须覆盖市场份额、价格策略和政策环境",
-        idempotency_key="m3b",
+        session["id"], role="human", content="读者是投资机构，必须覆盖市场份额、价格策略和政策环境", idempotency_key="m3b"
     )
     decision = await resolver.handle_user_message(session["id"], user_message=second, idempotency_key="m3b")
-    assert decision is not None
-    assert decision.action == "ready"
-    assert decision.used_defaults is False
-    assert decision.requirement.audience == "投资机构"
+    assert decision is not None and decision.action == "ready"
     assert decision.requirement.revision == 2
+    assert decision.requirement.audience == "投资机构"
+    assert set(decision.requirement.required_angles) == {"市场份额", "价格策略", "政策环境"}
     record = await store.get_session_record(session["id"])
-    assert record["requirement_revision"] == 2
     assert all(item["resolved"] for item in record["requirement_json"]["clarifications"])
-    snapshot = await store.get_snapshot(session["id"])
-    assert snapshot["plans"] == []
-    assert snapshot["session"]["status"] == "clarifying"
+
+
+@pytest.mark.asyncio
+async def test_resolver_raises_planner_unavailable_without_fallback() -> None:
+    store = MemoryReportCollaborationStore()
+    session = await store.create_session(owner_id="u1", title="", idempotency_key="s4")
+    user = await store.create_message(session["id"], role="human", content="写报告", idempotency_key="m4")
+
+    # 工厂显式禁用（模型不可用）→ 直接报错，无规则兜底
+    disabled = RequirementResolver(store, planner_llm_factory=lambda: None)
+    with pytest.raises(ReportCollaborationValidationError) as exc:
+        await disabled.handle_user_message(session["id"], user_message=user, idempotency_key="m4")
+    assert exc.value.code == "PLANNER_UNAVAILABLE"
+
+    # 模型输出不合法 → 同样报错
+    broken = RequirementResolver(store, planner_llm_factory=lambda: _ScriptedLLM(["我说人话不行吗"]))
+    with pytest.raises(ReportCollaborationValidationError) as exc2:
+        await broken.handle_user_message(session["id"], user_message=user, idempotency_key="m4")
+    assert exc2.value.code == "PLANNER_UNAVAILABLE"

offline-backend-20260512/backend/tests/test_report_collaboration_runner.py

@@ -155,6 +155,89 @@ async def test_waves_researcher_before_writer_and_text_does_not_complete() -> No
     assert phases[f"{run_id}-research-attempt1"] != phases[f"{run_id}-write-attempt1"]
 
 
+@pytest.mark.asyncio
+async def test_command_rework_cascades_downstream_and_reruns_in_order() -> None:
+    """P0.3 指定重跑级联：重跑 research 后 write 被级联 superseded 并按依赖序重跑。"""
+
+    store = MemoryReportCollaborationStore()
+    run_id = await _boot(store)
+    kernel = ScriptedMemberKernel()
+    kernel.enqueue("research", TaskResult(kind="artifact_draft", artifact_type="EvidenceBundle", artifact_draft=_evidence()))
+    kernel.enqueue("write", TaskResult(kind="artifact_draft", artifact_type="ReportSectionDraft", artifact_draft=_draft()))
+    executor = WaveExecutor(store, kernel)
+    assert await executor.run_until_idle(run_id) == "completed"
+
+    ledger = TaskLedger(store)
+    changed = await ledger.apply_command_rework(
+        run_id,
+        command_id="cmd-cascade-1",
+        instruction="重新查，重点看价格战",
+        node_ids=["research"],
+        cascade_downstream=True,
+    )
+    assert set(changed) == {"research", "write"}
+
+    nodes = await store.list_node_runs(run_id)
+    by_node: dict[str, list[dict]] = {}
+    for item in nodes:
+        by_node.setdefault(item["node_id"], []).append(item)
+    research_attempts = {item["attempt"]: item["status"] for item in by_node["research"]}
+    write_attempts = {item["attempt"]: item["status"] for item in by_node["write"]}
+    assert research_attempts[1] == "superseded" and research_attempts[2] == "planned"
+    assert write_attempts[1] == "superseded" and write_attempts[2] == "planned"
+    events = [item for item in await store.list_events(run_id) if item["type"] == "node.status.changed"]
+    cascade_events = [item for item in events if (item.get("data") or {}).get("reason") == "downstream_cascade"]
+    assert cascade_events and cascade_events[0]["data"]["node_id"] == "write"
+
+    kernel2 = ScriptedMemberKernel()
+    kernel2.enqueue("research", TaskResult(kind="artifact_draft", artifact_type="EvidenceBundle", artifact_draft=_evidence("价格战证据")))
+    kernel2.enqueue("write", TaskResult(kind="artifact_draft", artifact_type="ReportSectionDraft", artifact_draft=_draft()))
+    executor2 = WaveExecutor(store, kernel2)
+    assert await executor2.run_until_idle(run_id) == "completed"
+    assert executor2.dispatched == ["research", "write"]
+
+
+@pytest.mark.asyncio
+async def test_exhausted_interrupted_node_fails_run_instead_of_zombie() -> None:
+    """P0.2 僵尸 run 守卫：attempt 耗尽的 interrupted 节点应让 run 进入 failed。"""
+
+    store = MemoryReportCollaborationStore()
+    run_id = await _boot(store, _plan(max_attempts=2))
+    # 模拟两次崩溃被 recovery 打断后 attempt 耗尽：research attempt2 interrupted。
+    nodes = await store.list_node_runs(run_id)
+    research = next(item for item in nodes if item["node_id"] == "research")
+    research["attempt"] = 2
+    research["status"] = "interrupted"
+    research["error"] = "lease_reclaimed"
+    await store.save_node_run(research)
+    write = next(item for item in nodes if item["node_id"] == "write")
+    write["status"] = "planned"
+
+    executor = WaveExecutor(store, ScriptedMemberKernel())
+    status = await executor.run_until_idle(run_id)
+    assert status == "failed"
+    run = await store.get_run(run_id)
+    assert run["status"] == "failed"
+    events = [item for item in await store.list_events(run_id) if item["type"] == "run.status.changed"]
+    assert events and events[-1]["data"]["reason"] == "node_attempts_exhausted"
+    assert events[-1]["data"]["node_id"] == "research"
+
+
+@pytest.mark.asyncio
+async def test_interrupted_with_budget_left_keeps_run_alive() -> None:
+    """未耗尽的 interrupted 节点仍留给 recovery 补开 attempt，不能提前 fail。"""
+
+    store = MemoryReportCollaborationStore()
+    run_id = await _boot(store, _plan(max_attempts=3))
+    nodes = await store.list_node_runs(run_id)
+    research = next(item for item in nodes if item["node_id"] == "research")
+    research["attempt"] = 1
+    research["status"] = "interrupted"
+    await store.save_node_run(research)
+    ledger = TaskLedger(store)
+    assert await ledger.refresh_run_terminal(run_id) != "failed"
+
+
 @pytest.mark.asyncio
 async def test_parallel_researchers_share_phase() -> None:
     store = MemoryReportCollaborationStore()