"""API for the standalone position-collaboration workspace. Unlike the existing multi-agent roundtable, this feature has no coordinator and no background roundtable job. The API only persists the task/intent/chain snapshot and indexes one direct-agent node per business-chain seat. LangGraph remains the live execution source, while SQL also stores recovery copies of all conversations, drafts and delivered artifact bodies so checkpoint-volume replacement cannot turn an existing collaboration history into a blank page. """ from __future__ import annotations import asyncio from datetime import UTC, datetime from pathlib import Path from typing import Any from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel, Field from app.gateway.path_utils import aresolve_thread_virtual_path from deerflow.runtime.user_context import get_effective_user_id # The application mounts this router below both `/api` (canonical) and the # long-established `/api/multi-agent` namespace (intranet compatibility). router = APIRouter(prefix="/position-roundtable", tags=["position-roundtable"]) NODE_STATUSES = {"locked", "ready", "running", "done", "stale", "rejected", "error"} _HANDOFF_TEXT_BUDGET = 12_000 _MAX_INLINE_ARTIFACTS_PER_NODE = 4 _MAX_INLINE_ARTIFACT_CHARS = 1_800 # The handoff prompt remains intentionally small, but the recovery copy stored # in SQL must retain a complete normal Markdown/JSON deliverable. The ceiling # protects the database from an accidental multi-gigabyte log while covering # substantially larger reports than this workspace normally produces. _MAX_PERSISTED_ARTIFACT_CHARS = 4_000_000 _BUILTIN_MATERIAL_BUDGET = 24_000 SUMMARY_AGENT_ID = "roundtable-summary" ACTION_PLAN_AGENT_ID = "position-action-planner" _BUILTIN_MARKDOWN_OUTPUTS = { "summary": "mnt/user-data/outputs/岗位会商方案总结报告.md", "action_plan": "mnt/user-data/outputs/岗位会商行动规划.md", } _TEXT_ARTIFACT_SUFFIXES = { ".csv", ".json", ".log", ".md", ".markdown", ".rst", ".txt", ".xml", ".yaml", ".yml", } class NodeResponse(BaseModel): id: str session_id: str node_key: str stage_index: int seat_index: int agent_id: str position_id: str | None = None thread_id: str | None = None status: str latest_answer: str | None = None artifact_manifest: list[dict[str, Any]] = Field(default_factory=list) conversation_snapshot: dict[str, Any] | None = None 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 class SessionResponse(BaseModel): id: str external_task_id: str | None = None task_snapshot: dict[str, Any] = Field(default_factory=dict) chain_id: str | None = None chain_snapshot: dict[str, Any] | None = None intent_thread_id: str | None = None intent_snapshot: dict[str, Any] | None = None intent_conversation: dict[str, Any] | None = None summary_thread_id: str | None = None summary_snapshot: dict[str, Any] | None = None summary_conversation: dict[str, Any] | None = None action_plan_thread_id: str | None = None action_plan_snapshot: dict[str, Any] | None = None action_plan_conversation: dict[str, Any] | None = None status: str created_at: str | None = None updated_at: str | None = None class SessionDetailResponse(SessionResponse): nodes: list[NodeResponse] = Field(default_factory=list) class SessionCreateRequest(BaseModel): external_task_id: str | None = Field(default=None, max_length=128) task_snapshot: dict[str, Any] = Field(default_factory=dict) class TaskSnapshotUpdateRequest(BaseModel): task_snapshot: dict[str, Any] = Field(default_factory=dict) class IntentUpdateRequest(BaseModel): intent_thread_id: str | None = Field(default=None, max_length=128) intent_snapshot: dict[str, Any] | None = None conversation_snapshot: dict[str, Any] | None = None class ConversationStateUpdateRequest(BaseModel): """Durable UI recovery state written while a normal chat is in progress.""" thread_id: str | None = Field(default=None, max_length=128) conversation_snapshot: dict[str, Any] = Field(default_factory=dict) class ActivateRequest(BaseModel): chain_id: str = Field(min_length=1, max_length=64) class NodeThreadRequest(BaseModel): thread_id: str = Field(min_length=1, max_length=128) class CompleteTurnRequest(BaseModel): latest_answer: str = Field(default="", max_length=500_000) artifact_manifest: list[dict[str, Any]] = Field(default_factory=list) upstream_revision: dict[str, Any] | None = None class RejectNodeRequest(BaseModel): # The UI only asks for an optional rejection reason. Keep ``title`` for # compatible clients and for the audit record, but provide a stable # default so neither field turns a valid rejection into a 422/400. title: str = Field(default="产物驳回", max_length=120) reason: str = Field(default="", max_length=5_000) class PrepareBuiltinRunRequest(BaseModel): """Optional follow-up instruction for one persisted built-in agent.""" message: str = Field(default="", max_length=20_000) class BuiltinRunPreparationResponse(BaseModel): agent_id: str display_name: str thread_id: str | None = None prompt: str source_revisions: dict[str, int] = Field(default_factory=dict) class CompleteBuiltinRunRequest(BaseModel): thread_id: str = Field(min_length=1, max_length=128) content: str = Field(default="", max_length=500_000) artifact_manifest: list[dict[str, Any]] = Field(default_factory=list) source_revisions: dict[str, int] = Field(default_factory=dict) model: str | None = Field(default=None, max_length=256) class DemoSeedRequest(BaseModel): """Canonical frontend fixture imported into the normal SQL session model.""" demo_key: str = Field(min_length=1, max_length=96) session: dict[str, Any] = Field(default_factory=dict) nodes: list[dict[str, Any]] = Field(default_factory=list, max_length=32) 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() def _get_store(request: Request): store = getattr(request.app.state, "position_roundtable_store", None) if store is None: raise HTTPException(status_code=503, detail="Position roundtable store not available") return store def _get_chain_store(request: Request): store = getattr(request.app.state, "roundtable_chain_store", None) if store is None: raise HTTPException(status_code=503, detail="Business chain store not available") return store async def _require_writable_session( request: Request, session_id: str ) -> tuple[Any, str, dict[str, Any]]: """Load a Session that can still accept task, intent, or node updates.""" store = _get_store(request) user_id = _current_user_id(request) session = await store.get_session(session_id, user_id) if session is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") if session.get("status") == "archived": raise HTTPException(status_code=409, detail="该岗位协同会话已归档，仅可查看历史记录") return store, user_id, session def _project_session(row: dict[str, Any]) -> SessionResponse: return SessionResponse(**row) def _project_node(row: dict[str, Any]) -> NodeResponse: return NodeResponse(**row) async def _load_detail(request: Request, session_id: str) -> SessionDetailResponse: store = _get_store(request) user_id = _current_user_id(request) row = await store.get_session(session_id, user_id) if row is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") nodes = await store.list_nodes(session_id, user_id) return SessionDetailResponse(**row, nodes=[_project_node(node) for node in nodes or []]) def _chain_nodes( chain: dict[str, Any], *, available_agent_ids: set[str] | None = None, ) -> list[dict[str, Any]]: seats = chain.get("seats") if isinstance(chain.get("seats"), list) else [] if not seats: raise HTTPException(status_code=400, detail="业务链条没有可执行的智能体席位") seat_by_agent_id = {str(seat.get("agent_id")): seat for seat in seats if seat.get("agent_id")} if len(seat_by_agent_id) != len(seats): raise HTTPException(status_code=400, detail="业务链条包含无效或重复的智能体席位") raw_stages = chain.get("stages") stages: list[list[str]] if isinstance(raw_stages, list) and raw_stages: stages = [[str(agent_id) for agent_id in stage] for stage in raw_stages if isinstance(stage, list)] else: stages = [[str(seat["agent_id"])] for seat in seats] if not stages or any(not stage for stage in stages): raise HTTPException(status_code=400, detail="业务链条阶段配置无效") result: list[dict[str, Any]] = [] for stage_index, stage in enumerate(stages): for seat_index, agent_id in enumerate(stage): seat = seat_by_agent_id.get(agent_id) if seat is None: raise HTTPException(status_code=400, detail=f"业务链条阶段包含不存在的席位: {agent_id}") if available_agent_ids is not None and agent_id not in available_agent_ids: raise HTTPException( status_code=400, detail=f"业务链条席位「{seat.get('name') or agent_id}」缺少可用智能体，请先重新配置业务链条", ) position_id = seat.get("position_id") if not isinstance(position_id, str) or not position_id.strip(): raise HTTPException( status_code=400, detail=f"智能体「{seat.get('name') or agent_id}」尚未配置岗位，请先从岗位协同入口配置", ) result.append( { "node_key": f"stage-{stage_index + 1}-seat-{seat_index + 1}-{agent_id}", "stage_index": stage_index, "seat_index": seat_index, "agent_id": agent_id, "position_id": position_id.strip(), "status": "ready" if stage_index == 0 else "locked", } ) return result async def _available_agent_ids(request: Request, agent_ids: set[str]) -> set[str] | None: """Return the configured agent ids if the agent store is available. Position collaboration freezes a business-chain snapshot before execution. Validating seat agent ids here gives the user a concrete configuration error instead of failing later when a node tries to create its chat thread. """ agent_store = getattr(request.app.state, "agent_store", None) if agent_store is None: return None available: set[str] = set() for agent_id in agent_ids: if await agent_store.get_any(agent_id) is not None: available.add(agent_id) return available def _truncate_text(value: Any, limit: int) -> str: if limit <= 0: return "" text = value if isinstance(value, str) else str(value or "") if len(text) <= limit: return text 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]], *, intent_position_id: str | None = None, ) -> bool: return bool(nodes) and all( node.get("status") == "done" and (_is_intent_position_node(node, intent_position_id) or _node_has_markdown_delivery(node)) for node in nodes ) def _is_intent_position_node( node: dict[str, Any], intent_position_id: str | None = None, ) -> bool: position_id = str(node.get("position_id") or "").strip() if intent_position_id: return position_id == intent_position_id # Existing persisted sessions predate the configurable directory. Retain # their historical interpretation until they are archived/recreated. return position_id.lower() in {"intelligence", "intel", "intent"} def _session_intent_position_id(session: dict[str, Any]) -> str | None: chain = session.get("chain_snapshot") if isinstance(session.get("chain_snapshot"), dict) else {} value = chain.get("_intent_position_id") return value.strip() if isinstance(value, str) and value.strip() else None async def _configured_intent_position_id(request: Request) -> str: """Get the one enabled special role for a newly activated session. The chosen id is frozen into ``chain_snapshot`` on activation. A later configuration change therefore applies to future workspaces without reinterpreting completion rules in an in-flight historical business chain. """ store = getattr(request.app.state, "position_role_store", None) if store is not None: rows = await store.list_roles() selected = next( ( row.get("id") for row in rows if row.get("enabled") and row.get("role_type") == "intent" and isinstance(row.get("id"), str) ), None, ) if isinstance(selected, str) and selected.strip(): return selected.strip() return "intelligence" def _is_markdown_artifact(artifact: Any) -> bool: if not isinstance(artifact, dict): return False path = str(artifact.get("path") or artifact.get("artifact_url") or "").strip().lower() name = str(artifact.get("name") or artifact.get("filename") or "").strip().lower() mime = str(artifact.get("mime_type") or artifact.get("mimeType") or "").strip().lower() if path and not path.replace("\\", "/").lstrip("/").startswith("mnt/user-data/outputs/"): return False return ( "markdown" in mime or path.endswith(".md") or path.endswith(".markdown") or name.endswith(".md") or name.endswith(".markdown") ) def _node_has_markdown_delivery(node: dict[str, Any]) -> bool: artifacts = node.get("artifact_manifest") return isinstance(artifacts, list) and any(_is_markdown_artifact(item) for item in artifacts) def _markdown_delivery_changed( previous_artifacts: Any, incoming_artifacts: list[dict[str, Any]], ) -> bool: """Whether this turn rewrote a Markdown deliverable, not just chatted. A completed node can receive normal follow-up Q&A. That must not bump its effective delivery revision or invalidate every downstream stage unless a Markdown report really changed. The standard artifact listing includes `modified_at` and `size_bytes`, which together make a stable lightweight version marker without reading a potentially large file again. """ old = previous_artifacts if isinstance(previous_artifacts, list) else [] old_signatures = { ( str(item.get("path") or item.get("artifact_url") or ""), str(item.get("modified_at") or item.get("modifiedAt") or ""), str(item.get("size_bytes") or item.get("sizeBytes") or ""), ) for item in old if isinstance(item, dict) and _is_markdown_artifact(item) } new_signatures = { ( str(item.get("path") or item.get("artifact_url") or ""), str(item.get("modified_at") or item.get("modifiedAt") or ""), str(item.get("size_bytes") or item.get("sizeBytes") or ""), ) for item in incoming_artifacts if isinstance(item, dict) and _is_markdown_artifact(item) } return bool(new_signatures - old_signatures) def _intent_node_answer(intent: Any) -> str: if not isinstance(intent, dict): return "情报分析岗已完成任务意图识别。" objective = _text_from_keys(intent, ["objective", "goal"], "已完成任务意图识别") constraints = intent.get("constraints") if isinstance(intent.get("constraints"), list) else [] lines = ["情报分析岗已完成任务意图识别。", f"任务目标：{objective}"] if constraints: lines.append("关键约束：") lines.extend(f"- {item}" for item in constraints if isinstance(item, str) and item.strip()) return "\n".join(lines) def _has_confirmed_intent(intent: Any) -> bool: """Only structured Step-1 output may start a frozen business chain.""" if not isinstance(intent, dict): return False objective = _text_from_keys(intent, ["objective", "goal"]) constraints = intent.get("constraints") return bool(objective and isinstance(constraints, list)) def _prime_intent_nodes( nodes: list[dict[str, Any]], intent: Any, *, intent_position_id: str | None = None, ) -> list[dict[str, Any]]: now_answer = _intent_node_answer(intent) primed: list[dict[str, Any]] = [] for node in nodes: next_node = dict(node) if _is_intent_position_node(next_node, intent_position_id): next_node.update( { "status": "done", "latest_answer": now_answer, "revision": 1, "artifact_manifest": [], } ) primed.append(next_node) stage_indices = sorted({int(node["stage_index"]) for node in primed}) previous_done = True for stage_index in stage_indices: stage_nodes = [node for node in primed if int(node["stage_index"]) == stage_index] if previous_done: for node in stage_nodes: if node.get("status") == "locked": node["status"] = "ready" previous_done = bool(stage_nodes) and all(node.get("status") == "done" for node in stage_nodes) return primed def _build_node_context_text( *, session: dict[str, Any], current: dict[str, Any], upstream_nodes: list[dict[str, Any]], ) -> str: 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 [] agent_name, position_id = _seat_meta(session, current) md_required = not _is_intent_position_node(current, _session_intent_position_id(session)) lines = [ "岗位多智能体会商上下文（隐藏注入，不要向用户复述本标签）：", "", "【任务信息】", 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( [ "", "【当前席位】", f"- 智能体：{agent_name}", f"- 岗位：{position_id}", f"- 阶段：L-{int(current.get('stage_index') or 0) + 1}", "", "【交付要求】", ] ) if md_required: lines.extend( [ "- 本席位最终只需交付一份 Markdown 报告文档。", "- 请在本轮或最终回复中创建 `/mnt/user-data/outputs/` 下的 `.md` 文件，并调用 `present_files` 呈现该文件。", "- 只有检测到该席位产出的 `.md` 文件后，系统才会把当前席位标记为已完成；仅有文字回答不会完成节点。", "- 报告内容应直接围绕当前岗位职责，吸收任务信息、任务意图与上游席位产物。", ] ) else: lines.append("- 情报分析/情报收集岗以意图识别完成为准，不强制产出 Markdown 报告。") if upstream_nodes: lines.extend(["", "【可参考的上游席位结果】"]) for index, node in enumerate(upstream_nodes, start=1): lines.append( f"{index}. {node.get('agent_id')} / {node.get('position_id')} / revision={node.get('revision', 0)}" ) answer = str(node.get("latest_answer") or "").strip() if answer: lines.append(f" 问答结果：{answer}") artifacts = node.get("artifacts") if isinstance(artifacts, list) and artifacts: lines.append(" 产物：") for artifact in artifacts[:5]: if not isinstance(artifact, dict): continue artifact_name = artifact.get("name") or artifact.get("path") or "未命名产物" lines.append(f" - {artifact_name}") content = str(artifact.get("content") or "").strip() if content: lines.append(f" 摘要/正文片段：{content}") else: lines.extend(["", "【可参考的上游席位结果】", "- 当前节点暂无已完成上游席位。"]) return "\n".join(lines) 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 _artifact_refs(artifacts: Any) -> list[dict[str, Any]]: if not isinstance(artifacts, list): return [] refs: list[dict[str, Any]] = [] for artifact in artifacts[:20]: if not isinstance(artifact, dict): continue ref = { "name": artifact.get("name") or artifact.get("filename") or "未命名产物", "path": artifact.get("path") or artifact.get("artifact_url") or "", "mime_type": artifact.get("mimeType") or artifact.get("mime_type") or None, } # A database recovery payload is already constrained when it is first # persisted. Preserve it here so downstream agents can still consume # an upstream report after the owning sandbox/output disk is gone. content = artifact.get("content") if isinstance(content, str): ref["content"] = content if artifact.get("content_truncated"): ref["content_truncated"] = True refs.append(ref) return refs def _is_inline_text_artifact(artifact: dict[str, Any]) -> bool: """Return whether a listed output is safe and useful to pass as text.""" mime_type = artifact.get("mime_type") if isinstance(mime_type, str) and mime_type.lower().startswith("text/"): return True path = artifact.get("path") return isinstance(path, str) and Path(path).suffix.lower() in _TEXT_ARTIFACT_SUFFIXES def _read_text_excerpt(path: Path, max_chars: int) -> tuple[str, bool] | None: """Read one bounded UTF-8 text artifact without loading the whole file.""" if max_chars <= 0: return None # Four bytes per UTF-8 code point is enough to obtain the requested # character-sized excerpt without a potentially unbounded file read. with path.open("rb") as file: raw = file.read(max_chars * 4 + 1) if b"\x00" in raw: return None try: text = raw.decode("utf-8") except UnicodeDecodeError as error: if len(raw) == max_chars * 4 + 1 and error.start >= len(raw) - 4: text = raw[: error.start].decode("utf-8") else: # Generated artifacts are expected to be UTF-8. Keep a non-UTF-8 # output as a reference only instead of corrupting agent context. return None truncated = len(raw) > max_chars * 4 or len(text) > max_chars return _truncate_text(text, max_chars), truncated async def _artifact_refs_with_content( thread_id: str | None, artifacts: Any, text_budget: int ) -> tuple[list[dict[str, Any]], int]: """Attach bounded excerpts of text outputs, retaining binary files as refs. Paths are always resolved against the upstream node's own thread. This is deliberate: an artifact manifest is client-supplied metadata, so it must never be allowed to select another thread's user-data directory. """ refs = _artifact_refs(artifacts) if text_budget <= 0: return refs, 0 consumed = 0 inline_count = 0 for ref in refs: if inline_count >= _MAX_INLINE_ARTIFACTS_PER_NODE: break if not _is_inline_text_artifact(ref): continue virtual_path = ref.get("path") if not isinstance(virtual_path, str): continue normalized_path = virtual_path.replace("\\", "/").lstrip("/") if not normalized_path.startswith("mnt/user-data/outputs/"): continue available = text_budget - consumed if available <= 0: break excerpt_limit = min(_MAX_INLINE_ARTIFACT_CHARS, available) persisted_content = ref.get("content") if isinstance(persisted_content, str): content = _truncate_text(persisted_content, excerpt_limit) ref["content"] = content ref["content_truncated"] = bool( ref.get("content_truncated") or len(persisted_content) > len(content) ) consumed += len(content) inline_count += 1 continue if not thread_id: continue try: actual_path = await aresolve_thread_virtual_path(thread_id, normalized_path) if not await asyncio.to_thread(actual_path.is_file): continue excerpt = await asyncio.to_thread(_read_text_excerpt, actual_path, excerpt_limit) except (HTTPException, OSError): # The file may have been removed after the manifest was stored. # Keep its metadata so the UI still shows the original output. continue if excerpt is None: continue content, truncated = excerpt ref["content"] = content ref["content_truncated"] = truncated consumed += len(content) inline_count += 1 return refs, consumed async def _handoff_context(request: Request, session_id: str, node_key: str) -> dict[str, Any]: """Construct bounded, deterministic hidden context for one direct-agent turn.""" store = _get_store(request) user_id = _current_user_id(request) session = await store.get_session(session_id, user_id) nodes = await store.list_nodes(session_id, user_id) if session is None or nodes is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") if session.get("status") == "archived": raise HTTPException(status_code=409, detail="该岗位协同会话已归档，仅可查看历史记录") current = next((node for node in nodes if node["node_key"] == node_key), None) 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("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. remaining = _HANDOFF_TEXT_BUDGET upstream_candidates = [ node for node in nodes if node["stage_index"] < current["stage_index"] and node["status"] == "done" ] upstream_nodes: list[dict[str, Any]] = [] for index, node in enumerate(upstream_candidates): candidates_left = len(upstream_candidates) - index per_node_limit = min(2_800, max(0, (remaining // candidates_left) * 2 // 3)) answer = _truncate_text(node.get("latest_answer"), per_node_limit) remaining -= len(answer) artifact_budget = min( _MAX_INLINE_ARTIFACT_CHARS, max(0, remaining // candidates_left), ) artifacts, consumed = await _artifact_refs_with_content( node.get("thread_id"), node.get("artifact_manifest"), artifact_budget ) remaining -= consumed upstream_nodes.append( { "node_key": node["node_key"], "agent_id": node["agent_id"], "position_id": node.get("position_id"), "revision": node.get("revision", 0), "latest_answer": answer, "artifacts": artifacts, } ) if remaining <= 0: break return { "kind": "position_roundtable_handoff", "instructions": _build_node_context_text( session=session, current=current, upstream_nodes=upstream_nodes, ), "delivery_requirement": { "type": "markdown_report", "required": not _is_intent_position_node(current, _session_intent_position_id(session)), "completion_rule": "普通业务席位必须在 /mnt/user-data/outputs/ 下交付一份 .md 报告文档才会标记完成；情报分析/情报收集岗以意图识别完成为准。", }, "task": session.get("task_snapshot") or {}, "intent": session.get("intent_snapshot") or {}, "current_node": { "node_key": current["node_key"], "agent_id": current["agent_id"], "position_id": current.get("position_id"), "stage_index": current["stage_index"], "seat_index": current["seat_index"], }, "upstream_nodes": upstream_nodes, "upstream_revision": { node["node_key"]: node.get("revision", 0) for node in upstream_nodes }, } def _source_revisions(nodes: list[dict[str, Any]]) -> dict[str, int]: """Stable revision fence for a built-in report/planning run. The report agents run asynchronously in the browser. If a user rejects or re-delivers a node while a run is still streaming, this fence prevents that old run from overwriting the newer business-chain state when it completes. """ return {str(node["node_key"]): int(node.get("revision") or 0) for node in _delivered_nodes(nodes)} def _is_json_artifact(artifact: Any) -> bool: if not isinstance(artifact, dict): return False path = str(artifact.get("path") or artifact.get("artifact_url") or "").strip().lower() name = str(artifact.get("name") or artifact.get("filename") or "").strip().lower() mime = str(artifact.get("mime_type") or artifact.get("mimeType") or "").strip().lower() if path and not path.replace("\\", "/").lstrip("/").startswith("mnt/user-data/outputs/"): return False return "json" in mime or path.endswith(".json") or name.endswith(".json") def _artifact_signature(artifact: dict[str, Any]) -> tuple[str, str, str]: """Use server-returned path/mtime/size to detect an actually rewritten file.""" return ( str(artifact.get("path") or artifact.get("artifact_url") or ""), str(artifact.get("modified_at") or artifact.get("modifiedAt") or ""), str(artifact.get("size_bytes") or artifact.get("sizeBytes") or ""), ) def _artifact_manifest_refs(artifacts: list[dict[str, Any]]) -> list[dict[str, Any]]: """Persist display metadata plus any database recovery payload.""" refs: list[dict[str, Any]] = [] for artifact in artifacts: if not isinstance(artifact, dict): continue path = str(artifact.get("path") or artifact.get("artifact_url") or "").strip() if not path.replace("\\", "/").lstrip("/").startswith("mnt/user-data/outputs/"): continue ref = { "name": str(artifact.get("name") or artifact.get("filename") or path.rsplit("/", 1)[-1]), "path": path, "mime_type": artifact.get("mime_type") or artifact.get("mimeType") or None, "modified_at": artifact.get("modified_at") or artifact.get("modifiedAt") or None, "size_bytes": artifact.get("size_bytes") or artifact.get("sizeBytes") or None, } content = artifact.get("content") if isinstance(content, str): ref["content"] = content if artifact.get("content_truncated"): ref["content_truncated"] = True refs.append(ref) return refs async def _persist_artifact_contents( thread_id: str | None, artifacts: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Copy text deliverables into SQL so history survives ephemeral disks.""" refs = _artifact_manifest_refs(artifacts) if not thread_id: return refs enriched: list[dict[str, Any]] = [] for artifact in refs: path = str(artifact.get("path") or "") suffix = Path(path).suffix.lower() if suffix not in _TEXT_ARTIFACT_SUFFIXES: enriched.append(artifact) continue try: resolved = await aresolve_thread_virtual_path(thread_id, path) content = await asyncio.to_thread(resolved.read_text, encoding="utf-8", errors="replace") except Exception: # Artifact enrichment is recovery hardening, not a reason to fail # an otherwise successful delivery. Keep the manifest metadata if # the sandbox disappeared or its resolver/provider is unavailable. enriched.append(artifact) continue if len(content) > _MAX_PERSISTED_ARTIFACT_CHARS: artifact = { **artifact, "content": content[:_MAX_PERSISTED_ARTIFACT_CHARS], "content_truncated": True, } else: artifact = {**artifact, "content": content} enriched.append(artifact) return enriched async def _write_builtin_markdown_fallback( *, thread_id: str, kind: str, content: str ) -> dict[str, Any]: """Materialize a built-in agent's visible answer when it skipped ``write_file``. Position roundtable runs are direct chat streams. A model can therefore finish with a complete visible report but still omit its required artifact tool call. Keep the chat answer as the source of truth and mirror it into the owning thread's outputs directory so the report remains previewable and downloadable after the stream has ended. """ virtual_path = _BUILTIN_MARKDOWN_OUTPUTS.get(kind) report = content.strip() if virtual_path is None: raise HTTPException(status_code=400, detail=f"Unknown built-in delivery kind: {kind}") if not report: raise HTTPException(status_code=422, detail="内置智能体未返回可保存为报告的正文") try: output_path = await aresolve_thread_virtual_path(thread_id, virtual_path) await asyncio.to_thread(output_path.parent.mkdir, parents=True, exist_ok=True) await asyncio.to_thread( output_path.write_text, report if report.endswith("\n") else f"{report}\n", encoding="utf-8", ) stat = await asyncio.to_thread(output_path.stat) except OSError as error: raise HTTPException(status_code=500, detail="无法保存内置智能体 Markdown 报告") from error return { "name": output_path.name, "path": virtual_path, "mime_type": "text/markdown", "modified_at": datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(), "size_bytes": stat.st_size, } async def _effective_delivery_material( session: dict[str, Any], nodes: list[dict[str, Any]] ) -> str: """Build bounded source material from valid current deliveries only. Stale/rejected rows are deliberately excluded. The persisted node answer provides a quick synopsis; the Markdown files provide the actual deliverable body, preferably from its database recovery copy and otherwise through the owning thread's virtual-path resolver. """ remaining = _BUILTIN_MATERIAL_BUDGET lines: list[str] = ["【岗位有效交付】"] for index, node in enumerate(_delivered_nodes(nodes), start=1): agent_name, position_id = _seat_meta(session, node) lines.extend( [ "", f"### {index}. {agent_name}", f"- 岗位：{position_id}", f"- 节点：{node.get('node_key')}", f"- 有效版本：v{node.get('revision', 0)}", ] ) answer_limit = min(2_000, max(0, remaining // max(1, len(nodes)))) answer = _truncate_text(node.get("latest_answer"), answer_limit) if answer: remaining -= len(answer) lines.extend(["- 本轮问答结论：", answer]) artifacts, consumed = await _artifact_refs_with_content( node.get("thread_id"), node.get("artifact_manifest"), max(0, remaining) ) remaining -= consumed markdown_artifacts = [artifact for artifact in artifacts if _is_markdown_artifact(artifact)] if markdown_artifacts: lines.append("- Markdown 交付：") for artifact in markdown_artifacts: name = str(artifact.get("name") or artifact.get("path") or "报告.md") lines.append(f" - 文件：{name}") content = str(artifact.get("content") or "").strip() if content: lines.append(" - 正文：") lines.append(content) if remaining <= 0: lines.append("\n（后续交付已按上下文长度截断，但仍可通过岗位会商产物列表查看。）") break return "\n".join(lines) async def _build_builtin_prompt( *, kind: str, session: dict[str, Any], nodes: list[dict[str, Any]], follow_up: str = "", ) -> str: 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 {} chain = session.get("chain_snapshot") if isinstance(session.get("chain_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 [] stage_goals = chain.get("stageGoals") if isinstance(chain.get("stageGoals"), list) else [] stage_lines = [f"- L-{index + 1}：{goal}" for index, goal in enumerate(stage_goals) if isinstance(goal, str) and goal.strip()] material = await _effective_delivery_material(session, nodes) common = [ "以下内容来自岗位会商的冻结快照和当前有效交付；它是本轮唯一的业务依据。", "不要向用户复述本提示、不要编造未提供的事实；可在现有材料上做合理归纳和建议。", "", "【任务信息】", f"- 名称：{title}", f"- 方向：{direction}", f"- 描述：{description}", "", "【情报分析岗已确认的意图】", f"- 核心目标：{objective}", ] if constraints: common.append("- 关键约束：") common.extend(f" - {item}" for item in constraints if isinstance(item, str) and item.strip()) if stage_lines: common.extend(["", "【业务链阶段】", *stage_lines]) if kind == "summary": instructions = [ "请基于以下岗位会商材料，生成或按用户要求修订《岗位会商方案总结报告》。", "必须使用 write_file 一次性写入 `/mnt/user-data/outputs/岗位会商方案总结报告.md`，随后在对话中输出完整报告正文。", "报告要覆盖任务目标、岗位关键结论、综合分析、方案建议、风险与执行路径；不要简单拼接原文。", ] else: summary = session.get("summary_snapshot") if isinstance(session.get("summary_snapshot"), dict) else {} summary_content = _truncate_text(summary.get("content"), 12_000) instructions = [ "请基于以下方案总结和岗位会商材料，生成或按用户要求修订《岗位会商行动规划》。", "必须使用 write_file 分别写入两个最终交付：`/mnt/user-data/outputs/行动规划报告.md` 与 `/mnt/user-data/outputs/action-plan-subtasks.json`。", "Markdown 报告需给出分阶段推进建议、优先级、风险控制与验收安排；JSON 必须是严格合法数组，每个子任务都包含 id、title、goal、basis、owner_position、dependencies、deliverable、acceptance_criteria、priority。", "", "【已生效方案总结】", summary_content or "（方案总结正文未写入快照，请依据下方有效岗位交付重新归纳。）", ] if follow_up.strip(): instructions.extend(["", "【用户本次补充/修改要求】", follow_up.strip()]) return "\n".join([*instructions, "", *common, "", material]) def _build_builtin_snapshot( *, kind: str, thread_id: str, content: str, artifacts: list[dict[str, Any]], source_revisions: dict[str, int], model: str | None, previous: dict[str, Any] | None = None, ) -> dict[str, Any]: markdown = next((artifact for artifact in artifacts if _is_markdown_artifact(artifact)), None) json_artifact = next((artifact for artifact in artifacts if _is_json_artifact(artifact)), None) primary = markdown if kind == "summary" else markdown previous_primary = previous.get("primary_artifact") if isinstance(previous, dict) else None primary_changed = not isinstance(previous_primary, dict) or not isinstance(primary, dict) or ( _artifact_signature(previous_primary) != _artifact_signature(primary) ) previous_content = str(previous.get("content") or "") if isinstance(previous, dict) else "" return { "kind": kind, "status": "done", "title": "岗位会商方案总结报告" if kind == "summary" else "岗位会商行动规划", # A pure follow-up may only answer a question and leave the report file # unchanged. Keep the last report body in that case while retaining # the answer separately for audit/display. "content": content.strip() if primary_changed or not previous_content else previous_content, "last_response": content.strip(), "thread_id": thread_id, "artifact_manifest": _artifact_manifest_refs(artifacts), "primary_artifact": primary, "subtasks_artifact": json_artifact if kind == "action_plan" else None, "source_revisions": source_revisions, "model": model or None, "generated_at": datetime.now(UTC).isoformat(), } @router.post("/demo/seed", response_model=SessionDetailResponse) async def seed_demo_session( request: Request, body: DemoSeedRequest ) -> SessionDetailResponse: """Idempotently materialise the frontend demo JSON for the current user. The endpoint intentionally accepts a complete recovery snapshot instead of inventing a second demo-only schema. Imported rows therefore exercise the same history, position switching, artifact and conversation code paths as real collaboration sessions. """ row, nodes = await _get_store(request).ensure_demo_session( _current_user_id(request), demo_key=body.demo_key, session_payload=body.session, nodes=body.nodes, ) return SessionDetailResponse( **row, nodes=[_project_node(node) for node in nodes], ) @router.get("/sessions", response_model=list[SessionResponse]) async def list_sessions(request: Request, external_task_id: str | None = None) -> list[SessionResponse]: rows = await _get_store(request).list_sessions( _current_user_id(request), external_task_id=external_task_id ) return [_project_session(row) for row in rows] @router.post("/sessions", response_model=SessionResponse, status_code=201) async def create_session(request: Request, body: SessionCreateRequest) -> SessionResponse: row = await _get_store(request).create_session( _current_user_id(request), external_task_id=body.external_task_id, task_snapshot=body.task_snapshot, ) return _project_session(row) @router.get("/sessions/{session_id}", response_model=SessionDetailResponse) async def get_session(request: Request, session_id: str) -> SessionDetailResponse: return await _load_detail(request, session_id) @router.put("/sessions/{session_id}/task", response_model=SessionResponse) async def update_task_snapshot( request: Request, session_id: str, body: TaskSnapshotUpdateRequest ) -> SessionResponse: store, user_id, _ = await _require_writable_session(request, session_id) row = await store.update_session(session_id, user_id, task_snapshot=body.task_snapshot) if row is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") return _project_session(row) @router.put("/sessions/{session_id}/intent", response_model=SessionResponse) async def update_intent_snapshot( request: Request, session_id: str, body: IntentUpdateRequest ) -> SessionResponse: store, user_id, _ = await _require_writable_session(request, session_id) row = await store.update_session( session_id, user_id, intent_thread_id=body.intent_thread_id, intent_snapshot=body.intent_snapshot, intent_conversation=body.conversation_snapshot, ) if row is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") return _project_session(row) @router.put("/sessions/{session_id}/summary/state", response_model=SessionResponse) async def update_summary_conversation( request: Request, session_id: str, body: ConversationStateUpdateRequest ) -> SessionResponse: store, user_id, _ = await _require_writable_session(request, session_id) row = await store.update_session( session_id, user_id, summary_thread_id=body.thread_id, summary_conversation=body.conversation_snapshot, ) if row is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") return _project_session(row) @router.put("/sessions/{session_id}/action-plan/state", response_model=SessionResponse) async def update_action_plan_conversation( request: Request, session_id: str, body: ConversationStateUpdateRequest ) -> SessionResponse: store, user_id, _ = await _require_writable_session(request, session_id) row = await store.update_session( session_id, user_id, action_plan_thread_id=body.thread_id, action_plan_conversation=body.conversation_snapshot, ) if row is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") return _project_session(row) @router.post("/sessions/{session_id}/activate", response_model=SessionDetailResponse) async def activate_session( request: Request, session_id: str, body: ActivateRequest ) -> SessionDetailResponse: store, user_id, current = await _require_writable_session(request, session_id) if not _has_confirmed_intent(current.get("intent_snapshot")): raise HTTPException(status_code=400, detail="请先完成并保存任务意图") chain = await _get_chain_store(request).get_chain( body.chain_id, user_id, is_admin=getattr(getattr(request.state, "user", None), "system_role", None) == "admin", ) if chain is None: raise HTTPException(status_code=404, detail="业务链条不存在或无权访问") seat_ids = { str(seat.get("agent_id")) for seat in (chain.get("seats") if isinstance(chain.get("seats"), list) else []) if seat.get("agent_id") } # Freeze the currently configured special role with this session. This # prevents a later global config edit from changing the completion rule of # an already running historical chain. chain_snapshot = dict(chain) intent_position_id = await _configured_intent_position_id(request) chain_snapshot["_intent_position_id"] = intent_position_id nodes = _chain_nodes( chain_snapshot, available_agent_ids=await _available_agent_ids(request, seat_ids), ) nodes = _prime_intent_nodes( nodes, current.get("intent_snapshot"), intent_position_id=intent_position_id, ) activated = await store.activate_session( session_id, user_id, chain_id=body.chain_id, chain_snapshot=chain_snapshot, nodes=nodes, ) if activated is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") row, node_rows = activated return SessionDetailResponse(**row, nodes=[_project_node(node) for node in node_rows]) @router.post("/sessions/{session_id}/archive", response_model=SessionResponse) async def archive_session(request: Request, session_id: str) -> SessionResponse: row = await _get_store(request).archive_session(session_id, _current_user_id(request)) if row is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") return _project_session(row) @router.delete("/sessions/{session_id}") async def delete_session(request: Request, session_id: str) -> dict[str, bool]: deleted = await _get_store(request).delete_session( session_id, _current_user_id(request) ) if not deleted: raise HTTPException(status_code=404, detail="Position roundtable session not found") return {"deleted": True} @router.get("/sessions/{session_id}/nodes/{node_key}", response_model=NodeResponse) async def get_node(request: Request, session_id: str, node_key: str) -> NodeResponse: row = await _get_store(request).get_node(session_id, node_key, _current_user_id(request)) if row is None: raise HTTPException(status_code=404, detail="Position roundtable node not found") return _project_node(row) @router.post("/sessions/{session_id}/nodes/{node_key}/thread", response_model=NodeResponse) async def bind_node_thread( request: Request, session_id: str, node_key: str, body: NodeThreadRequest ) -> NodeResponse: 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("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( session_id, node_key, user_id, thread_id=body.thread_id, status="running" if current.get("status") == "ready" else current.get("status", "locked"), ) return _project_node(row) @router.put( "/sessions/{session_id}/nodes/{node_key}/state", response_model=NodeResponse, ) async def update_node_conversation( request: Request, session_id: str, node_key: str, body: ConversationStateUpdateRequest, ) -> NodeResponse: store, user_id, session = 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 ( body.thread_id and current.get("thread_id") and current["thread_id"] != body.thread_id ): raise HTTPException(status_code=409, detail="Position node is bound to another thread") row = await store.update_node( session_id, node_key, user_id, thread_id=body.thread_id if body.thread_id else current.get("thread_id"), conversation_snapshot=body.conversation_snapshot, ) if row is None: raise HTTPException(status_code=404, detail="Position roundtable node not found") return _project_node(row) @router.post("/sessions/{session_id}/nodes/{node_key}/context") async def build_node_context(request: Request, session_id: str, node_key: str) -> dict[str, Any]: """Return the bounded hidden task/intent/upstream context for a node turn.""" return await _handoff_context(request, session_id, node_key) @router.post("/sessions/{session_id}/nodes/{node_key}/complete-turn", response_model=NodeResponse) async def complete_node_turn( request: Request, session_id: str, node_key: str, body: CompleteTurnRequest ) -> NodeResponse: store, user_id, session = await _require_writable_session(request, session_id) node = await store.get_node(session_id, node_key, user_id) if node is None: 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="上游产物已被驳回或重做，当前节点需等待前序节点重新完成") # Artifact listing is best-effort on the client. Never erase a previously # delivered report merely because one listing request transiently failed. effective_artifacts = body.artifact_manifest or list(node.get("artifact_manifest") or []) if body.artifact_manifest: effective_artifacts = await _persist_artifact_contents( node.get("thread_id"), effective_artifacts, ) has_markdown = any(_is_markdown_artifact(item) for item in effective_artifacts) markdown_changed = _markdown_delivery_changed(node.get("artifact_manifest"), effective_artifacts) requires_fresh_delivery = node.get("status") in {"rejected", "stale"} next_status = ( "done" if _is_intent_position_node(node, _session_intent_position_id(session)) else "done" if has_markdown and (not requires_fresh_delivery or markdown_changed) else "running" ) # Effective revisions are delivery revisions, not every conversational # answer. This prevents a harmless follow-up on a completed seat from # invalidating all later stages just because its old Markdown file still # appears in the artifact list. new_effective_delivery = ( next_status == "done" and ( _is_intent_position_node(node, _session_intent_position_id(session)) or node.get("status") != "done" or markdown_changed ) ) row = await store.update_node( session_id, node_key, user_id, status=next_status, latest_answer=body.latest_answer, artifact_manifest=effective_artifacts, upstream_revision=body.upstream_revision, invalidated_by=None, increment_revision=new_effective_delivery, ) if row is None: raise HTTPException(status_code=404, detail="Position roundtable node not found") # A node has no coordinator. Its effective delivery alone drives the frozen # chain: all parallel seats in the current stage must finish before the # next stage is made writable. A plain conversational follow-up keeps the # existing downstream deliveries valid; only a newly written Markdown # report advances the revision and makes later stages stale. 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 new_effective_delivery and node.get("revision", 0) > 0: for downstream in nodes: if ( downstream["stage_index"] > row["stage_index"] 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" if has_material else "locked", invalidated_by=row["node_key"], ) nodes = await store.list_nodes(session_id, user_id) or [] current_stage = [item for item in nodes if item["stage_index"] == row["stage_index"]] later_stages = sorted( {item["stage_index"] for item in nodes if item["stage_index"] > row["stage_index"]} ) if row["status"] == "done" and 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"] in {"locked", "stale"}: await store.update_node( session_id, next_node["node_key"], user_id, 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() or "产物驳回" reason = body.reason.strip() 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]) async def _prepare_builtin_run( request: Request, session_id: str, *, kind: str, follow_up: str = "", ) -> BuiltinRunPreparationResponse: 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, intent_position_id=_session_intent_position_id(session)): raise HTTPException(status_code=409, detail="所有岗位产物完成后才能生成或更新内置交付") if kind == "action_plan": summary = session.get("summary_snapshot") if not isinstance(summary, dict) or summary.get("status") != "done": raise HTTPException(status_code=409, detail="请先生成有效的方案总结") thread_id = session.get("action_plan_thread_id") agent_id = ACTION_PLAN_AGENT_ID display_name = "行动规划智能体" else: if follow_up.strip(): summary = session.get("summary_snapshot") if not isinstance(summary, dict) or summary.get("status") != "done": raise HTTPException(status_code=409, detail="请先生成有效的方案总结后再继续问答") thread_id = session.get("summary_thread_id") agent_id = SUMMARY_AGENT_ID display_name = "方案总结智能体" return BuiltinRunPreparationResponse( agent_id=agent_id, display_name=display_name, thread_id=str(thread_id) if isinstance(thread_id, str) and thread_id else None, prompt=await _build_builtin_prompt( kind=kind, session=session, nodes=nodes, follow_up=follow_up, ), source_revisions=_source_revisions(nodes), ) async def _complete_builtin_run( request: Request, session_id: str, *, kind: str, body: CompleteBuiltinRunRequest, ) -> 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, intent_position_id=_session_intent_position_id(session)): raise HTTPException(status_code=409, detail="上游岗位产物已变更或未完成，不能写入旧的内置交付") current_revisions = _source_revisions(nodes) if body.source_revisions != current_revisions: raise HTTPException(status_code=409, detail="岗位产物版本已变化，请基于最新交付重新生成") artifacts = await _persist_artifact_contents(body.thread_id, body.artifact_manifest) markdown = [artifact for artifact in artifacts if _is_markdown_artifact(artifact)] if not markdown: # The report text is already visible in the chat stream. Do not lose # that successful result just because the model skipped its final # write_file call: mirror the exact response into the thread outputs # directory and persist it as the normal report artifact. fallback_markdown = await _write_builtin_markdown_fallback( thread_id=body.thread_id, kind=kind, content=body.content, ) fallback_markdown["content"] = body.content artifacts.append(fallback_markdown) markdown = [fallback_markdown] if kind == "action_plan" and not any(_is_json_artifact(artifact) for artifact in artifacts): raise HTTPException(status_code=422, detail="行动规划智能体还需交付 action-plan-subtasks.json 子任务清单") snapshot_field = "summary_snapshot" if kind == "summary" else "action_plan_snapshot" previous = session.get(snapshot_field) previous_snapshot = previous if isinstance(previous, dict) else None snapshot = _build_builtin_snapshot( kind=kind, thread_id=body.thread_id, content=body.content, artifacts=artifacts, source_revisions=current_revisions, model=body.model, previous=previous_snapshot, ) previous_primary = previous_snapshot.get("primary_artifact") if previous_snapshot else None new_primary = snapshot.get("primary_artifact") report_changed = not isinstance(previous_primary, dict) or not isinstance(new_primary, dict) or ( _artifact_signature(previous_primary) != _artifact_signature(new_primary) ) if kind == "summary": row = await store.update_session( session_id, user_id, summary_thread_id=body.thread_id, summary_snapshot=snapshot, # A rewritten summary invalidates the derived action plan. A # pure Q&A follow-up keeps the existing plan valid. action_plan_snapshot=None if report_changed else session.get("action_plan_snapshot"), status="active", ) else: row = await store.update_session( session_id, user_id, action_plan_thread_id=body.thread_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.post("/sessions/{session_id}/summary/prepare", response_model=BuiltinRunPreparationResponse) async def prepare_summary( request: Request, session_id: str, body: PrepareBuiltinRunRequest ) -> BuiltinRunPreparationResponse: return await _prepare_builtin_run(request, session_id, kind="summary", follow_up=body.message) @router.post("/sessions/{session_id}/summary/ask", response_model=BuiltinRunPreparationResponse) async def ask_summary( request: Request, session_id: str, body: PrepareBuiltinRunRequest ) -> BuiltinRunPreparationResponse: return await _prepare_builtin_run(request, session_id, kind="summary", follow_up=body.message) @router.post("/sessions/{session_id}/summary/complete", response_model=SessionDetailResponse) async def complete_summary( request: Request, session_id: str, body: CompleteBuiltinRunRequest ) -> SessionDetailResponse: return await _complete_builtin_run(request, session_id, kind="summary", body=body) @router.post("/sessions/{session_id}/action-plan/prepare", response_model=BuiltinRunPreparationResponse) async def prepare_action_plan( request: Request, session_id: str, body: PrepareBuiltinRunRequest ) -> BuiltinRunPreparationResponse: return await _prepare_builtin_run(request, session_id, kind="action_plan", follow_up=body.message) @router.post("/sessions/{session_id}/action-plan/ask", response_model=BuiltinRunPreparationResponse) async def ask_action_plan( request: Request, session_id: str, body: PrepareBuiltinRunRequest ) -> BuiltinRunPreparationResponse: return await _prepare_builtin_run(request, session_id, kind="action_plan", follow_up=body.message) @router.post("/sessions/{session_id}/action-plan/complete", response_model=SessionDetailResponse) async def complete_action_plan( request: Request, session_id: str, body: CompleteBuiltinRunRequest ) -> SessionDetailResponse: return await _complete_builtin_run(request, session_id, kind="action_plan", body=body) @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)) if rows is None: raise HTTPException(status_code=404, detail="Position roundtable session not found") return [_project_node(row) for row in rows]