feat(position-roundtable): add delivery rejection workflow

frontend-web/src/position-roundtable/api/sessions.ts

@@ -30,6 +30,9 @@ export interface PositionRoundtableNode {
   artifactManifest: Array<Record<string, unknown>>;
   revision: number;
   upstreamRevision: Record<string, unknown> | null;
+  rejectionCount: number;
+  lastRejection: Record<string, unknown> | null;
+  invalidatedBy: string | null;
   createdAt?: string;
   updatedAt?: string;
 }
@@ -42,6 +45,8 @@ export interface PositionRoundtableSession {
   chainSnapshot: Record<string, unknown> | null;
   intentThreadId: string | null;
   intentSnapshot: Record<string, unknown> | null;
+  summarySnapshot: Record<string, unknown> | null;
+  actionPlanSnapshot: Record<string, unknown> | null;
   status: "intent_pending" | "active" | "completed" | "archived";
   createdAt?: string;
   updatedAt?: string;
@@ -62,6 +67,9 @@ interface ServerNode {
   artifact_manifest?: Array<Record<string, unknown>>;
   revision?: number;
   upstream_revision?: Record<string, unknown> | null;
+  rejection_count?: number;
+  last_rejection?: Record<string, unknown> | null;
+  invalidated_by?: string | null;
   created_at?: string;
   updated_at?: string;
 }
@@ -74,6 +82,8 @@ interface ServerSession {
   chain_snapshot?: Record<string, unknown> | null;
   intent_thread_id: string | null;
   intent_snapshot?: Record<string, unknown> | null;
+  summary_snapshot?: Record<string, unknown> | null;
+  action_plan_snapshot?: Record<string, unknown> | null;
   status: PositionRoundtableSession["status"];
   created_at?: string;
   updated_at?: string;
@@ -105,6 +115,9 @@ function toNode(node: ServerNode): PositionRoundtableNode {
     artifactManifest: node.artifact_manifest ?? [],
     revision: node.revision ?? 0,
     upstreamRevision: node.upstream_revision ?? null,
+    rejectionCount: node.rejection_count ?? 0,
+    lastRejection: node.last_rejection ?? null,
+    invalidatedBy: node.invalidated_by ?? null,
     createdAt: node.created_at,
     updatedAt: node.updated_at,
   };
@@ -119,6 +132,8 @@ function toSession(session: ServerSession): PositionRoundtableSession {
     chainSnapshot: session.chain_snapshot ?? null,
     intentThreadId: session.intent_thread_id,
     intentSnapshot: session.intent_snapshot ?? null,
+    summarySnapshot: session.summary_snapshot ?? null,
+    actionPlanSnapshot: session.action_plan_snapshot ?? null,
     status: session.status,
     createdAt: session.created_at,
     updatedAt: session.updated_at,
@@ -263,3 +278,45 @@ export async function completePositionNodeTurn(
   );
   return toNode(result);
 }
+
+export async function rejectPositionNode(
+  sessionId: string,
+  nodeKey: string,
+  input: { title: string; reason: string },
+): Promise<PositionRoundtableSession> {
+  const result = await readJson<ServerSession>(
+    await apiFetch(
+      `${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/reject`,
+      {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          title: input.title,
+          reason: input.reason,
+        }),
+      },
+    ),
+    "reject position node",
+  );
+  return toSession(result);
+}
+
+export async function generatePositionSummary(sessionId: string): Promise<PositionRoundtableSession> {
+  const result = await readJson<ServerSession>(
+    await apiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/summary/start`, {
+      method: "POST",
+    }),
+    "generate position summary",
+  );
+  return toSession(result);
+}
+
+export async function generatePositionActionPlan(sessionId: string): Promise<PositionRoundtableSession> {
+  const result = await readJson<ServerSession>(
+    await apiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/action-plan/start`, {
+      method: "POST",
+    }),
+    "generate position action plan",
+  );
+  return toSession(result);
+}

frontend-web/src/position-roundtable/components/PositionNodeChat.tsx

@@ -78,7 +78,8 @@ export function PositionNodeChat({
     sessionStatus !== "archived" &&
     node.positionId === currentPositionId &&
     node.status !== "locked" &&
-    node.status !== "error";
+    node.status !== "error" &&
+    !node.invalidatedBy;
 
   useEffect(() => {
     setBoundThreadId(node.threadId);
@@ -152,6 +153,8 @@ export function PositionNodeChat({
         toast.info(
           sessionStatus === "archived"
             ? "该会话已归档，仅可查看历史记录"
+            : node.invalidatedBy
+              ? "上游产物已被驳回或重做，当前节点需等待前序节点重新完成"
             : node.status === "locked"
               ? "该节点尚未解锁"
               : "当前岗位只能查看该节点对话",
@@ -181,6 +184,8 @@ export function PositionNodeChat({
   const readOnlyHint =
     sessionStatus === "archived"
       ? "该会话已归档，仅可查看历史对话和产物"
+      : node.invalidatedBy
+      ? "上游产物已变更，需等待前序节点重新完成后再重做"
       : node.status === "locked"
       ? "等待上游岗位完成后解锁"
       : node.positionId !== currentPositionId

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

@@ -9,6 +9,7 @@ messages and artifacts stay in the established LangGraph thread stores.
 from __future__ import annotations
 
 import asyncio
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -52,6 +53,9 @@ class NodeResponse(BaseModel):
     artifact_manifest: list[dict[str, Any]] = Field(default_factory=list)
     revision: int = 0
     upstream_revision: dict[str, Any] | None = None
+    rejection_count: int = 0
+    last_rejection: dict[str, Any] | None = None
+    invalidated_by: str | None = None
     created_at: str | None = None
     updated_at: str | None = None
 
@@ -64,6 +68,8 @@ class SessionResponse(BaseModel):
     chain_snapshot: dict[str, Any] | None = None
     intent_thread_id: str | None = None
     intent_snapshot: dict[str, Any] | None = None
+    summary_snapshot: dict[str, Any] | None = None
+    action_plan_snapshot: dict[str, Any] | None = None
     status: str
     created_at: str | None = None
     updated_at: str | None = None
@@ -101,6 +107,11 @@ class CompleteTurnRequest(BaseModel):
     upstream_revision: dict[str, Any] | None = None
 
 
+class RejectNodeRequest(BaseModel):
+    title: str = Field(min_length=1, max_length=120)
+    reason: str = Field(min_length=1, max_length=5_000)
+
+
 def _current_user_id(request: Request) -> str:
     user = getattr(request.state, "user", None)
     return str(user.id) if user is not None else get_effective_user_id()
@@ -229,6 +240,175 @@ def _truncate_text(value: Any, limit: int) -> str:
     return f"{text[:limit]}\n…（已按交接上下文长度截断）"
 
 
+def _text_from_keys(source: dict[str, Any], keys: list[str], fallback: str = "") -> str:
+    for key in keys:
+        value = source.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return fallback
+
+
+def _seat_meta(session: dict[str, Any], node: dict[str, Any]) -> tuple[str, str]:
+    chain = session.get("chain_snapshot") if isinstance(session.get("chain_snapshot"), dict) else {}
+    seats = chain.get("seats") if isinstance(chain.get("seats"), list) else []
+    seat = next(
+        (
+            item
+            for item in seats
+            if isinstance(item, dict) and str(item.get("agent_id") or "") == str(node.get("agent_id") or "")
+        ),
+        {},
+    )
+    agent_name = str(seat.get("name") or node.get("agent_id") or "未命名智能体")
+    position_id = str(node.get("position_id") or seat.get("position_id") or "未配置岗位")
+    return agent_name, position_id
+
+
+def _delivered_nodes(nodes: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [node for node in nodes if node.get("status") == "done"]
+
+
+def _all_nodes_delivered(nodes: list[dict[str, Any]]) -> bool:
+    return bool(nodes) and all(node.get("status") == "done" for node in nodes)
+
+
+def _artifact_lines(node: dict[str, Any]) -> list[str]:
+    artifacts = node.get("artifact_manifest")
+    if not isinstance(artifacts, list) or not artifacts:
+        return []
+    lines: list[str] = []
+    for artifact in artifacts[:8]:
+        if not isinstance(artifact, dict):
+            continue
+        name = artifact.get("name") or artifact.get("filename") or artifact.get("path") or "未命名产物"
+        path = artifact.get("path") or artifact.get("artifact_url") or ""
+        lines.append(f"  - {name}{f'（{path}）' if path else ''}")
+    return lines
+
+
+def _build_summary_snapshot(session: dict[str, Any], nodes: list[dict[str, Any]]) -> dict[str, Any]:
+    now = datetime.now(UTC).isoformat()
+    task = session.get("task_snapshot") if isinstance(session.get("task_snapshot"), dict) else {}
+    intent = session.get("intent_snapshot") if isinstance(session.get("intent_snapshot"), dict) else {}
+    title = _text_from_keys(task, ["title", "taskName", "name"], "未命名任务")
+    direction = _text_from_keys(task, ["direction", "taskDirection", "category"], "未填写")
+    description = _text_from_keys(task, ["description", "taskContent", "overview", "content"], "暂无任务描述")
+    objective = _text_from_keys(intent, ["objective", "goal"], "暂无意图结论")
+    constraints = intent.get("constraints") if isinstance(intent.get("constraints"), list) else []
+
+    lines = [
+        "# 岗位会商方案总结报告",
+        "",
+        f"生成时间：{now}",
+        "",
+        "## 一、任务概况",
+        "",
+        f"- 任务名称：{title}",
+        f"- 任务方向：{direction}",
+        f"- 任务描述：{description}",
+        "",
+        "## 二、意图理解",
+        "",
+        f"- 核心目标：{objective}",
+    ]
+    if constraints:
+        lines.append("- 关键约束：")
+        lines.extend(f"  - {item}" for item in constraints if isinstance(item, str) and item.strip())
+    lines.extend(["", "## 三、岗位交付汇总", ""])
+    for index, node in enumerate(_delivered_nodes(nodes), start=1):
+        agent_name, position_id = _seat_meta(session, node)
+        answer = _truncate_text(node.get("latest_answer"), 1_200) or "暂无文本回答"
+        lines.extend(
+            [
+                f"### {index}. {agent_name}",
+                "",
+                f"- 岗位：{position_id}",
+                f"- 节点版本：第 {node.get('revision', 0)} 版",
+                "",
+                answer,
+                "",
+            ]
+        )
+        artifacts = _artifact_lines(node)
+        if artifacts:
+            lines.extend(["产物文件：", *artifacts, ""])
+    lines.extend(
+        [
+            "## 四、综合判断",
+            "",
+            "本报告基于当前所有有效岗位交付产物自动汇总生成，可作为后续行动规划的输入材料。若任一前序岗位产物被驳回，本报告将自动标记为失效并需要重新生成。",
+        ]
+    )
+    return {
+        "kind": "summary",
+        "status": "done",
+        "title": "岗位会商方案总结报告",
+        "content": "\n".join(lines),
+        "generated_at": now,
+        "source_revisions": {node["node_key"]: node.get("revision", 0) for node in _delivered_nodes(nodes)},
+    }
+
+
+def _build_action_plan_snapshot(session: dict[str, Any], nodes: list[dict[str, Any]]) -> dict[str, Any]:
+    now = datetime.now(UTC).isoformat()
+    summary = session.get("summary_snapshot") if isinstance(session.get("summary_snapshot"), dict) else {}
+    delivered = _delivered_nodes(nodes)
+    tasks: list[dict[str, Any]] = []
+    for index, node in enumerate(delivered, start=1):
+        agent_name, position_id = _seat_meta(session, node)
+        tasks.append(
+            {
+                "id": f"subtask-{index:02d}",
+                "title": f"落实{agent_name}交付建议",
+                "owner_position": position_id,
+                "basis_node": node.get("node_key"),
+                "deliverable": f"{agent_name}整改/执行清单",
+                "acceptance": "补齐输入依据，形成可验证的阶段成果并回填岗位会商产物。",
+            }
+        )
+    lines = [
+        "# 岗位会商行动规划",
+        "",
+        f"生成时间：{now}",
+        "",
+        "## 一、规划依据",
+        "",
+        f"- 方案总结：{summary.get('title') or '岗位会商方案总结报告'}",
+        f"- 有效岗位产物数量：{len(delivered)}",
+        "",
+        "## 二、子任务拆解",
+        "",
+    ]
+    for task in tasks:
+        lines.extend(
+            [
+                f"### {task['id']} {task['title']}",
+                "",
+                f"- 建议责任岗位：{task['owner_position']}",
+                f"- 输入依据节点：{task['basis_node']}",
+                f"- 交付物：{task['deliverable']}",
+                f"- 验收标准：{task['acceptance']}",
+                "",
+            ]
+        )
+    lines.extend(
+        [
+            "## 三、执行提示",
+            "",
+            "若前序岗位产物被驳回，本行动规划会自动失效；请先完成被驳回节点及后续节点的重新交付，再重新生成行动规划。",
+        ]
+    )
+    return {
+        "kind": "action_plan",
+        "status": "done",
+        "title": "岗位会商行动规划",
+        "content": "\n".join(lines),
+        "generated_at": now,
+        "source_revisions": {node["node_key"]: node.get("revision", 0) for node in delivered},
+        "subtasks": tasks,
+    }
+
+
 def _artifact_refs(artifacts: Any) -> list[dict[str, Any]]:
     if not isinstance(artifacts, list):
         return []
@@ -342,6 +522,8 @@ async def _handoff_context(request: Request, session_id: str, node_key: str) ->
         raise HTTPException(status_code=404, detail="Position roundtable node not found")
     if current["status"] == "locked":
         raise HTTPException(status_code=409, detail="上游节点尚未完成，当前节点未解锁")
+    if current.get("invalidated_by"):
+        raise HTTPException(status_code=409, detail="上游产物已被驳回或重做，当前节点需等待前序节点重新完成")
 
     # Same-stage nodes are intentionally excluded. The business chain only
     # hands off completed previous stages, in stage/seat order.
@@ -350,7 +532,7 @@ async def _handoff_context(request: Request, session_id: str, node_key: str) ->
         node
         for node in nodes
         if node["stage_index"] < current["stage_index"]
-        and node["status"] in {"done", "stale"}
+        and node["status"] == "done"
     ]
     upstream_nodes: list[dict[str, Any]] = []
     for index, node in enumerate(upstream_candidates):
@@ -509,6 +691,8 @@ async def bind_node_thread(
         raise HTTPException(status_code=404, detail="Position roundtable node not found")
     if current["status"] == "locked":
         raise HTTPException(status_code=409, detail="上游节点尚未完成，当前节点未解锁")
+    if current.get("invalidated_by"):
+        raise HTTPException(status_code=409, detail="上游产物已被驳回或重做，当前节点需等待前序节点重新完成")
     if current.get("thread_id") and current["thread_id"] != body.thread_id:
         raise HTTPException(status_code=409, detail="该节点已绑定另一条对话线程")
     row = await store.update_node(
@@ -537,6 +721,8 @@ async def complete_node_turn(
         raise HTTPException(status_code=404, detail="Position roundtable node not found")
     if node["status"] == "locked":
         raise HTTPException(status_code=409, detail="上游节点尚未完成，当前节点未解锁")
+    if node.get("invalidated_by"):
+        raise HTTPException(status_code=409, detail="上游产物已被驳回或重做，当前节点需等待前序节点重新完成")
     row = await store.update_node(
         session_id,
         node_key,
@@ -545,6 +731,7 @@ async def complete_node_turn(
         latest_answer=body.latest_answer,
         artifact_manifest=body.artifact_manifest,
         upstream_revision=body.upstream_revision,
+        invalidated_by=None,
         increment_revision=True,
     )
     if row is None:
@@ -562,13 +749,19 @@ async def complete_node_turn(
         for downstream in nodes:
             if (
                 downstream["stage_index"] > row["stage_index"]
-                and downstream["status"] in {"ready", "running", "done", "stale"}
+                and downstream["status"] in {"ready", "running", "done", "stale", "rejected", "error"}
             ):
+                has_material = (
+                    downstream.get("revision", 0) > 0
+                    or bool(downstream.get("latest_answer"))
+                    or bool(downstream.get("artifact_manifest"))
+                )
                 await store.update_node(
                     session_id,
                     downstream["node_key"],
                     user_id,
-                    status="stale",
+                    status="stale" if has_material else "locked",
+                    invalidated_by=row["node_key"],
                 )
         nodes = await store.list_nodes(session_id, user_id) or []
 
@@ -579,16 +772,93 @@ async def complete_node_turn(
     if current_stage and all(item["status"] == "done" for item in current_stage) and later_stages:
         next_stage = later_stages[0]
         for next_node in nodes:
-            if next_node["stage_index"] == next_stage and next_node["status"] == "locked":
+            if next_node["stage_index"] == next_stage and next_node["status"] in {"locked", "stale"}:
                 await store.update_node(
                     session_id,
                     next_node["node_key"],
                     user_id,
-                    status="ready",
+                    status="ready" if next_node["status"] == "locked" else "stale",
+                    invalidated_by=None,
                 )
     return _project_node(row)
 
 
+@router.post("/sessions/{session_id}/nodes/{node_key}/reject", response_model=SessionDetailResponse)
+async def reject_node(
+    request: Request, session_id: str, node_key: str, body: RejectNodeRequest
+) -> SessionDetailResponse:
+    title = body.title.strip()
+    reason = body.reason.strip()
+    if not title or not reason:
+        raise HTTPException(status_code=400, detail="驳回标题和理由不能为空")
+    store, user_id, _ = await _require_writable_session(request, session_id)
+    current = await store.get_node(session_id, node_key, user_id)
+    if current is None:
+        raise HTTPException(status_code=404, detail="Position roundtable node not found")
+    if current["status"] == "locked":
+        raise HTTPException(status_code=409, detail="当前节点尚未解锁，暂无可驳回的交付产物")
+    if current.get("revision", 0) <= 0 and not current.get("latest_answer") and not current.get("artifact_manifest"):
+        raise HTTPException(status_code=409, detail="当前节点暂无可驳回的交付产物")
+    rejected = await store.reject_node(
+        session_id,
+        node_key,
+        user_id,
+        title=title,
+        reason=reason,
+        rejected_by=user_id,
+    )
+    if rejected is None:
+        raise HTTPException(status_code=404, detail="Position roundtable node not found")
+    row, node_rows = rejected
+    return SessionDetailResponse(**row, nodes=[_project_node(node) for node in node_rows])
+
+
+@router.post("/sessions/{session_id}/summary/start", response_model=SessionDetailResponse)
+async def start_summary(request: Request, session_id: str) -> SessionDetailResponse:
+    store, user_id, session = await _require_writable_session(request, session_id)
+    nodes = await store.list_nodes(session_id, user_id)
+    if nodes is None:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    if not _all_nodes_delivered(nodes):
+        raise HTTPException(status_code=409, detail="所有岗位产物完成后才能生成方案总结")
+    snapshot = _build_summary_snapshot(session, nodes)
+    row = await store.update_session(
+        session_id,
+        user_id,
+        summary_snapshot=snapshot,
+        action_plan_snapshot=None,
+        status="active",
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    node_rows = await store.list_nodes(session_id, user_id) or []
+    return SessionDetailResponse(**row, nodes=[_project_node(node) for node in node_rows])
+
+
+@router.post("/sessions/{session_id}/action-plan/start", response_model=SessionDetailResponse)
+async def start_action_plan(request: Request, session_id: str) -> SessionDetailResponse:
+    store, user_id, session = await _require_writable_session(request, session_id)
+    nodes = await store.list_nodes(session_id, user_id)
+    if nodes is None:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    if not _all_nodes_delivered(nodes):
+        raise HTTPException(status_code=409, detail="所有岗位产物完成后才能生成行动规划")
+    summary = session.get("summary_snapshot")
+    if not isinstance(summary, dict) or summary.get("status") != "done":
+        raise HTTPException(status_code=409, detail="请先生成有效的方案总结")
+    snapshot = _build_action_plan_snapshot(session, nodes)
+    row = await store.update_session(
+        session_id,
+        user_id,
+        action_plan_snapshot=snapshot,
+        status="completed",
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    node_rows = await store.list_nodes(session_id, user_id) or []
+    return SessionDetailResponse(**row, nodes=[_project_node(node) for node in node_rows])
+
+
 @router.get("/sessions/{session_id}/results", response_model=list[NodeResponse])
 async def list_results(request: Request, session_id: str) -> list[NodeResponse]:
     rows = await _get_store(request).list_results(session_id, _current_user_id(request))

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260721_02_position_roundtable_rejections.py

@@ -0,0 +1,54 @@
+"""Add rejection metadata to position roundtable nodes.
+
+Revision ID: 20260721_02
+Revises: 20260721_01
+Create Date: 2026-07-21
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+
+revision = "20260721_02"
+down_revision = "20260721_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    return table_name in inspect(op.get_bind()).get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    return column_name in {column["name"] for column in inspect(op.get_bind()).get_columns(table_name)}
+
+
+def upgrade() -> None:
+    table = "position_roundtable_nodes"
+    if not _has_table(table):
+        return
+    long_text = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
+    if not _has_column(table, "rejection_count"):
+        op.add_column(
+            table,
+            sa.Column("rejection_count", sa.Integer(), nullable=False, server_default="0"),
+        )
+    if not _has_column(table, "last_rejection"):
+        op.add_column(table, sa.Column("last_rejection", long_text, nullable=True))
+    if not _has_column(table, "invalidated_by"):
+        op.add_column(table, sa.Column("invalidated_by", sa.String(256), nullable=True))
+
+
+def downgrade() -> None:
+    table = "position_roundtable_nodes"
+    if not _has_table(table):
+        return
+    for column in ("invalidated_by", "last_rejection", "rejection_count"):
+        if _has_column(table, column):
+            op.drop_column(table, column)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260721_03_position_roundtable_builtin_snapshots.py

@@ -0,0 +1,49 @@
+"""Add builtin summary/action-plan snapshots to position roundtable sessions.
+
+Revision ID: 20260721_03
+Revises: 20260721_02
+Create Date: 2026-07-21
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+
+revision = "20260721_03"
+down_revision = "20260721_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    return table_name in inspect(op.get_bind()).get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    return column_name in {column["name"] for column in inspect(op.get_bind()).get_columns(table_name)}
+
+
+def upgrade() -> None:
+    table = "position_roundtable_sessions"
+    if not _has_table(table):
+        return
+    long_text = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
+    if not _has_column(table, "summary_snapshot"):
+        op.add_column(table, sa.Column("summary_snapshot", long_text, nullable=True))
+    if not _has_column(table, "action_plan_snapshot"):
+        op.add_column(table, sa.Column("action_plan_snapshot", long_text, nullable=True))
+
+
+def downgrade() -> None:
+    table = "position_roundtable_sessions"
+    if not _has_table(table):
+        return
+    for column in ("action_plan_snapshot", "summary_snapshot"):
+        if _has_column(table, column):
+            op.drop_column(table, column)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/position_roundtable/model.py

@@ -31,6 +31,8 @@ class PositionRoundtableSessionRow(Base):
     chain_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     intent_thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
     intent_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    summary_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    action_plan_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     # intent_pending -> active -> completed / archived. The router owns legal
     # transitions; the DB keeps it as a concise, portable string.
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="intent_pending", index=True)
@@ -66,12 +68,15 @@ class PositionRoundtableNodeRow(Base):
     position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     # The real messages remain in the standard LangGraph/checkpoint store.
     thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True, unique=True)
-    # locked / ready / running / done / stale / error
+    # locked / ready / running / done / stale / rejected / error
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="locked", index=True)
     latest_answer: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     artifact_manifest: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     upstream_revision: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    rejection_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
+    last_rejection: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    invalidated_by: Mapped[str | None] = mapped_column(String(256), nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC)
     )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/position_roundtable/sql.py

@@ -37,6 +37,10 @@ def _time(value: Any) -> str | Any:
     return value.isoformat() if isinstance(value, datetime) else value
 
 
+def _has_material(row: PositionRoundtableNodeRow) -> bool:
+    return row.revision > 0 or bool(row.latest_answer) or bool(_loads(row.artifact_manifest))
+
+
 class PositionRoundtableRepository:
     def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
         self._sf = session_factory
@@ -51,6 +55,8 @@ class PositionRoundtableRepository:
             "chain_snapshot": _loads(row.chain_snapshot),
             "intent_thread_id": row.intent_thread_id,
             "intent_snapshot": _loads(row.intent_snapshot),
+            "summary_snapshot": _loads(row.summary_snapshot),
+            "action_plan_snapshot": _loads(row.action_plan_snapshot),
             "status": row.status,
             "created_at": _time(row.created_at),
             "updated_at": _time(row.updated_at),
@@ -72,6 +78,9 @@ class PositionRoundtableRepository:
             "artifact_manifest": _loads(row.artifact_manifest) or [],
             "revision": row.revision,
             "upstream_revision": _loads(row.upstream_revision),
+            "rejection_count": row.rejection_count or 0,
+            "last_rejection": _loads(row.last_rejection),
+            "invalidated_by": row.invalidated_by,
             "created_at": _time(row.created_at),
             "updated_at": _time(row.updated_at),
         }
@@ -125,6 +134,8 @@ class PositionRoundtableRepository:
         task_snapshot: dict[str, Any] | object = _UNSET,
         intent_thread_id: str | None | object = _UNSET,
         intent_snapshot: dict[str, Any] | None | object = _UNSET,
+        summary_snapshot: dict[str, Any] | None | object = _UNSET,
+        action_plan_snapshot: dict[str, Any] | None | object = _UNSET,
         status: str | object = _UNSET,
     ) -> dict[str, Any] | None:
         async with self._sf() as session:
@@ -137,6 +148,10 @@ class PositionRoundtableRepository:
                 row.intent_thread_id = intent_thread_id or None
             if intent_snapshot is not _UNSET:
                 row.intent_snapshot = _dumps(intent_snapshot)
+            if summary_snapshot is not _UNSET:
+                row.summary_snapshot = _dumps(summary_snapshot)
+            if action_plan_snapshot is not _UNSET:
+                row.action_plan_snapshot = _dumps(action_plan_snapshot)
             if status is not _UNSET:
                 row.status = str(status)
             row.updated_at = datetime.now(UTC)
@@ -239,7 +254,10 @@ class PositionRoundtableRepository:
         latest_answer: str | None | object = _UNSET,
         artifact_manifest: list[dict[str, Any]] | object = _UNSET,
         upstream_revision: dict[str, Any] | object = _UNSET,
+        last_rejection: dict[str, Any] | None | object = _UNSET,
+        invalidated_by: str | None | object = _UNSET,
         increment_revision: bool = False,
+        increment_rejection: bool = False,
     ) -> dict[str, Any] | None:
         async with self._sf() as session:
             stmt = (
@@ -265,13 +283,100 @@ class PositionRoundtableRepository:
                 row.artifact_manifest = _dumps(artifact_manifest)
             if upstream_revision is not _UNSET:
                 row.upstream_revision = _dumps(upstream_revision)
+            if last_rejection is not _UNSET:
+                row.last_rejection = _dumps(last_rejection)
+            if invalidated_by is not _UNSET:
+                row.invalidated_by = invalidated_by or None
             if increment_revision:
                 row.revision += 1
+            if increment_rejection:
+                row.rejection_count += 1
             row.updated_at = datetime.now(UTC)
             await session.commit()
             await session.refresh(row)
             return self._node_dict(row)
 
+    async def reject_node(
+        self,
+        session_id: str,
+        node_key: str,
+        user_id: str,
+        *,
+        title: str,
+        reason: str,
+        rejected_by: str,
+    ) -> tuple[dict[str, Any], list[dict[str, Any]]] | None:
+        """Reject one delivered node and invalidate every later stage.
+
+        Historical answers and artifact manifests are deliberately preserved.
+        ``invalidated_by`` gates downstream rework until the rejected upstream
+        stage is completed again.
+        """
+        now = datetime.now(UTC)
+        async with self._sf() as session:
+            parent = await session.get(PositionRoundtableSessionRow, session_id)
+            if parent is None or parent.user_id != user_id:
+                return None
+            rows = (
+                await session.execute(
+                    select(PositionRoundtableNodeRow)
+                    .where(PositionRoundtableNodeRow.session_id == session_id)
+                    .order_by(PositionRoundtableNodeRow.stage_index, PositionRoundtableNodeRow.seat_index)
+                )
+            ).scalars().all()
+            target = next((row for row in rows if row.node_key == node_key), None)
+            if target is None:
+                return None
+
+            rejection = {
+                "title": title,
+                "reason": reason,
+                "rejected_by": rejected_by,
+                "rejected_at": now.isoformat(),
+            }
+            summary_snapshot = _loads(parent.summary_snapshot)
+            if isinstance(summary_snapshot, dict) and summary_snapshot.get("status") == "done":
+                summary_snapshot = {
+                    **summary_snapshot,
+                    "status": "stale",
+                    "invalidated_by": target.node_key,
+                    "invalidated_at": now.isoformat(),
+                }
+                parent.summary_snapshot = _dumps(summary_snapshot)
+            action_plan_snapshot = _loads(parent.action_plan_snapshot)
+            if isinstance(action_plan_snapshot, dict) and action_plan_snapshot.get("status") == "done":
+                action_plan_snapshot = {
+                    **action_plan_snapshot,
+                    "status": "stale",
+                    "invalidated_by": target.node_key,
+                    "invalidated_at": now.isoformat(),
+                }
+                parent.action_plan_snapshot = _dumps(action_plan_snapshot)
+            if parent.status == "completed":
+                parent.status = "active"
+            target.status = "rejected"
+            target.last_rejection = _dumps(rejection)
+            target.rejection_count += 1
+            target.invalidated_by = None
+            target.updated_at = now
+
+            for row in rows:
+                if row.stage_index <= target.stage_index:
+                    continue
+                if _has_material(row) or row.status in {"running", "done", "stale", "rejected", "error"}:
+                    row.status = "stale"
+                else:
+                    row.status = "locked"
+                row.invalidated_by = target.node_key
+                row.updated_at = now
+
+            parent.updated_at = now
+            await session.commit()
+            await session.refresh(parent)
+            for row in rows:
+                await session.refresh(row)
+            return self._session_dict(parent), [self._node_dict(row) for row in rows]
+
     async def archive_session(self, session_id: str, user_id: str) -> dict[str, Any] | None:
         return await self.update_session(session_id, user_id, status="archived")
 

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

@@ -139,6 +139,176 @@ async def test_node_updates_are_user_scoped_and_keep_thread_out_of_messages(stor
     assert await store.get_node(created["id"], nodes[0]["node_key"], "user-b") is None
 
 
+@pytest.mark.asyncio
+async def test_reject_node_marks_current_and_invalidates_later_stages(store):
+    created = await store.create_session("user-a", external_task_id=None, task_snapshot={})
+    _, nodes = await store.activate_session(
+        created["id"],
+        "user-a",
+        chain_id="chain-a",
+        chain_snapshot={"id": "chain-a"},
+        nodes=[
+            {
+                "node_key": "stage-1-seat-1-agent-a",
+                "stage_index": 0,
+                "seat_index": 0,
+                "agent_id": "agent-a",
+                "position_id": "planning",
+                "status": "ready",
+            },
+            {
+                "node_key": "stage-2-seat-1-agent-b",
+                "stage_index": 1,
+                "seat_index": 0,
+                "agent_id": "agent-b",
+                "position_id": "solution",
+                "status": "ready",
+            },
+            {
+                "node_key": "stage-3-seat-1-agent-c",
+                "stage_index": 2,
+                "seat_index": 0,
+                "agent_id": "agent-c",
+                "position_id": "command",
+                "status": "ready",
+            },
+        ],
+    )
+    await store.update_node(
+        created["id"],
+        nodes[0]["node_key"],
+        "user-a",
+        status="done",
+        latest_answer="第一阶段产物",
+        increment_revision=True,
+    )
+    await store.update_node(
+        created["id"],
+        nodes[1]["node_key"],
+        "user-a",
+        status="done",
+        latest_answer="第二阶段产物",
+        increment_revision=True,
+    )
+
+    rejected = await store.reject_node(
+        created["id"],
+        nodes[0]["node_key"],
+        "user-a",
+        title="依据不足",
+        reason="第一阶段判断依据需要补充。",
+        rejected_by="user-a",
+    )
+
+    assert rejected is not None
+    _, rejected_nodes = rejected
+    by_key = {node["node_key"]: node for node in rejected_nodes}
+    first = by_key[nodes[0]["node_key"]]
+    second = by_key[nodes[1]["node_key"]]
+    third = by_key[nodes[2]["node_key"]]
+    assert first["status"] == "rejected"
+    assert first["rejection_count"] == 1
+    assert first["last_rejection"]["title"] == "依据不足"
+    assert first["invalidated_by"] is None
+    assert second["status"] == "stale"
+    assert second["latest_answer"] == "第二阶段产物"
+    assert second["invalidated_by"] == nodes[0]["node_key"]
+    assert third["status"] == "locked"
+    assert third["invalidated_by"] == nodes[0]["node_key"]
+
+
+@pytest.mark.asyncio
+async def test_builtin_summary_and_action_plan_snapshots_are_invalidated_by_reject(store):
+    created = await store.create_session(
+        "user-a",
+        external_task_id=None,
+        task_snapshot={
+            "title": "测试任务",
+            "description": "形成协同方案",
+            "direction": "态势研判",
+        },
+    )
+    await store.update_session(
+        created["id"],
+        "user-a",
+        intent_snapshot={"objective": "明确处置意图", "constraints": ["时间紧"]},
+    )
+    chain = {
+        "id": "chain-a",
+        "seats": [
+            {"agent_id": "agent-a", "name": "研判智能体", "position_id": "planning"},
+            {"agent_id": "agent-b", "name": "处置智能体", "position_id": "solution"},
+        ],
+    }
+    _, nodes = await store.activate_session(
+        created["id"],
+        "user-a",
+        chain_id="chain-a",
+        chain_snapshot=chain,
+        nodes=[
+            {
+                "node_key": "stage-1-seat-1-agent-a",
+                "stage_index": 0,
+                "seat_index": 0,
+                "agent_id": "agent-a",
+                "position_id": "planning",
+                "status": "ready",
+            },
+            {
+                "node_key": "stage-2-seat-1-agent-b",
+                "stage_index": 1,
+                "seat_index": 0,
+                "agent_id": "agent-b",
+                "position_id": "solution",
+                "status": "ready",
+            },
+        ],
+    )
+    for node in nodes:
+        await store.update_node(
+            created["id"],
+            node["node_key"],
+            "user-a",
+            status="done",
+            latest_answer=f"{node['agent_id']} 产物",
+            increment_revision=True,
+        )
+    session = await store.get_session(created["id"], "user-a")
+    assert session is not None
+    nodes = await store.list_nodes(created["id"], "user-a") or []
+    summary = position_roundtable_router._build_summary_snapshot(session, nodes)
+    assert "岗位会商方案总结报告" in summary["content"]
+    assert "研判智能体" in summary["content"]
+    session = await store.update_session(created["id"], "user-a", summary_snapshot=summary)
+    assert session is not None
+    action_plan = position_roundtable_router._build_action_plan_snapshot(session, nodes)
+    session = await store.update_session(
+        created["id"],
+        "user-a",
+        action_plan_snapshot=action_plan,
+        status="completed",
+    )
+    assert session is not None
+    assert session["summary_snapshot"]["status"] == "done"
+    assert session["action_plan_snapshot"]["status"] == "done"
+
+    rejected = await store.reject_node(
+        created["id"],
+        nodes[0]["node_key"],
+        "user-a",
+        title="重做第一阶段",
+        reason="前序依据变化",
+        rejected_by="user-a",
+    )
+    assert rejected is not None
+    rejected_session, _ = rejected
+    assert rejected_session["status"] == "active"
+    assert rejected_session["summary_snapshot"]["status"] == "stale"
+    assert rejected_session["summary_snapshot"]["invalidated_by"] == nodes[0]["node_key"]
+    assert rejected_session["action_plan_snapshot"]["status"] == "stale"
+    assert rejected_session["action_plan_snapshot"]["invalidated_by"] == nodes[0]["node_key"]
+
+
 @pytest.mark.asyncio
 async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_path, monkeypatch):
     artifact = tmp_path / "handoff.md"
@@ -188,6 +358,18 @@ async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_p
                     "latest_answer": None,
                     "artifact_manifest": [],
                 },
+                {
+                    "node_key": "stage-1-seat-2-agent-stale",
+                    "stage_index": 0,
+                    "seat_index": 1,
+                    "agent_id": "agent-stale",
+                    "position_id": "planning",
+                    "thread_id": "stale-thread",
+                    "status": "stale",
+                    "revision": 1,
+                    "latest_answer": "失效上游产物",
+                    "artifact_manifest": [],
+                },
             ]
 
     async def resolve(thread_id: str, path: str):
@@ -207,6 +389,7 @@ async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_p
     )
 
     upstream = context["upstream_nodes"][0]
+    assert len(context["upstream_nodes"]) == 1
     assert upstream["latest_answer"] == "上游问答结果"
     assert upstream["artifacts"][0]["content"] == "上游产物正文：可供下一岗位直接参考。"
     assert "content" not in upstream["artifacts"][1]