"""SQL repository for position-collaboration sessions and chain nodes.""" from __future__ import annotations import hashlib import json from datetime import UTC, datetime from typing import Any from uuid import NAMESPACE_URL, uuid4, uuid5 from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.position_roundtable.delivery import ( is_intent_position_node, is_markdown_artifact, markdown_delivery_changed, session_intent_position_id, ) from deerflow.persistence.position_roundtable.model import ( PositionRoundtableCommandRow, PositionRoundtableNodeRow, PositionRoundtableSessionRow, ) _UNSET = object() def _dumps(value: Any) -> str | None: if value is None: return None return json.dumps(value, ensure_ascii=False) def _loads(value: Any) -> Any: if not isinstance(value, str) or not value: return None try: return json.loads(value) except Exception: return None 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)) def _has_confirmed_intent_snapshot(intent: Any) -> bool: """Return whether a persisted Step-1 snapshot may activate a chain. This duplicates only the storage-level shape check used by the router so activation can re-check the prerequisite *after* locking the session row. A router-only pre-check leaves a TOCTOU window where another user can overwrite the intent before the chain topology is frozen. """ if not isinstance(intent, dict): return False required_sections = ("coreGoals", "riskWarnings", "keyPoints", "strategicSignificance") if any(key in intent for key in required_sections): if not all(isinstance(intent.get(key), list) for key in required_sections): return False return bool(intent["coreGoals"] or intent["strategicSignificance"]) objective = intent.get("objective") or intent.get("goal") return bool(isinstance(objective, str) and objective.strip() and isinstance(intent.get("constraints"), list)) class NodeConcurrentWriteError(Exception): """节点被并发改写，调用方持有的 version 已过期（乐观锁冲突）。 由 ``update_node(expected_version=...)`` / ``recompute_chain_state`` 在检测到 version 不匹配时抛出。调用方可捕获后重新读取最新状态重算（幂等重入）。 """ def __init__( self, node_key: str, *, expected: int | None = None, actual: int | None = None, ) -> None: self.node_key = node_key self.expected = expected self.actual = actual super().__init__( f"concurrent write detected on position node {node_key}" + (f" (expected version {expected}, got {actual})" if expected is not None else "") ) class SessionWriteConflictError(Exception): """Session-level state changed after a route's optimistic pre-check. The public routes translate this into HTTP 409. Keeping the exception at the persistence boundary prevents a stale intent/task save from reviving an archived session or changing the frozen input of an active chain. """ def __init__(self, detail: str) -> None: self.detail = detail super().__init__(detail) class CommandConflictError(Exception): """命令在锁内判定为业务冲突（→ HTTP 409），事务已回滚、未改任何状态。 与「幂等重放」严格区分：重放返回上次成功结果；冲突说明状态机拒绝该命令 （节点未解锁 / 已绑定其它线程 / 同 command_id 携带了不同参数等）。 ``detail`` 是面向用户的提示，router 直接转成 409。 """ def __init__(self, detail: str) -> None: self.detail = detail super().__init__(detail) def _payload_hash(command_type: str, payload: dict[str, Any]) -> str: """规范化命令参数的 sha256，用于「同 command_id 是否同一参数」判定。""" normalized = json.dumps( {"type": command_type, "payload": payload}, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(normalized.encode("utf-8")).hexdigest() class PositionRoundtableRepository: def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: self._sf = session_factory @staticmethod def _can_access(row: PositionRoundtableSessionRow, user_id: str) -> bool: """History is partitioned by creator; even task-bound rows stay private.""" return row.user_id == user_id @staticmethod def _session_dict(row: PositionRoundtableSessionRow) -> dict[str, Any]: return { "id": row.id, "external_task_id": row.external_task_id, "task_scoped": bool(row.task_scoped), "task_snapshot": _loads(row.task_snapshot) or {}, "chain_id": row.chain_id, "chain_snapshot": _loads(row.chain_snapshot), "intent_thread_id": row.intent_thread_id, "intent_snapshot": _loads(row.intent_snapshot), "intent_conversation": _loads(row.intent_conversation), "summary_thread_id": row.summary_thread_id, "summary_snapshot": _loads(row.summary_snapshot), "summary_conversation": _loads(row.summary_conversation), "action_plan_thread_id": row.action_plan_thread_id, "action_plan_snapshot": _loads(row.action_plan_snapshot), "action_plan_conversation": _loads(row.action_plan_conversation), "status": row.status, "version": row.version, "created_at": _time(row.created_at), "updated_at": _time(row.updated_at), } @staticmethod def _node_dict(row: PositionRoundtableNodeRow) -> dict[str, Any]: return { "id": row.id, "session_id": row.session_id, "node_key": row.node_key, "stage_index": row.stage_index, "seat_index": row.seat_index, "agent_id": row.agent_id, "position_id": row.position_id, "thread_id": row.thread_id, "status": row.status, "latest_answer": row.latest_answer, "artifact_manifest": _loads(row.artifact_manifest) or [], "conversation_snapshot": _loads(row.conversation_snapshot), "revision": row.revision, "version": row.version, "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), } async def create_session( self, user_id: str, *, external_task_id: str | None, task_snapshot: dict[str, Any] | None, task_scoped: bool | None = None, ) -> dict[str, Any]: now = datetime.now(UTC) row = PositionRoundtableSessionRow( id=uuid4().hex, user_id=user_id, external_task_id=external_task_id or None, task_scoped=bool(external_task_id) if task_scoped is None else task_scoped, task_snapshot=_dumps(task_snapshot or {}), status="intent_pending", created_at=now, updated_at=now, ) async with self._sf() as session: session.add(row) await session.commit() await session.refresh(row) return self._session_dict(row) async def ensure_demo_session( self, user_id: str, *, demo_key: str, session_payload: dict[str, Any], nodes: list[dict[str, Any]], ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """Create one deterministic, user-scoped demonstration session. The browser owns the canonical JSON fixture so the same payload can be rendered locally when the gateway is unavailable. When the gateway is reachable this method imports that payload exactly once. Stable ids make repeated initialisation and React StrictMode double effects safe, while the user id in every UUID seed avoids cross-account collisions on the globally unique LangGraph thread columns. """ def stable_id(label: str) -> str: return uuid5( NAMESPACE_URL, f"deerflow:position-roundtable-demo:{user_id}:{demo_key}:{label}", ).hex session_id = stable_id("session") existing = await self.get_session(session_id, user_id) if existing is not None: return existing, (await self.list_nodes(session_id, user_id)) or [] now = datetime.now(UTC) intent_thread_id = ( stable_id("intent-thread") if session_payload.get("intent_thread_id") else None ) summary_thread_id = ( stable_id("summary-thread") if session_payload.get("summary_thread_id") else None ) action_plan_thread_id = ( stable_id("action-plan-thread") if session_payload.get("action_plan_thread_id") else None ) def with_thread_id(value: Any, thread_id: str | None) -> Any: if not isinstance(value, dict) or not thread_id: return value return {**value, "thread_id": thread_id} row = PositionRoundtableSessionRow( id=session_id, user_id=user_id, external_task_id=str(session_payload.get("external_task_id") or "").strip() or None, task_scoped=False, task_snapshot=_dumps(session_payload.get("task_snapshot") or {}), chain_id=session_payload.get("chain_id") or None, chain_snapshot=_dumps(session_payload.get("chain_snapshot")), intent_thread_id=intent_thread_id, intent_snapshot=_dumps(session_payload.get("intent_snapshot")), intent_conversation=_dumps(session_payload.get("intent_conversation")), summary_thread_id=summary_thread_id, summary_snapshot=_dumps( with_thread_id(session_payload.get("summary_snapshot"), summary_thread_id) ), summary_conversation=_dumps(session_payload.get("summary_conversation")), action_plan_thread_id=action_plan_thread_id, action_plan_snapshot=_dumps( with_thread_id( session_payload.get("action_plan_snapshot"), action_plan_thread_id ) ), action_plan_conversation=_dumps( session_payload.get("action_plan_conversation") ), status=str(session_payload.get("status") or "completed"), created_at=now, updated_at=now, ) node_rows: list[PositionRoundtableNodeRow] = [] for index, node in enumerate(nodes): node_key = str(node.get("node_key") or f"demo-node-{index + 1}") node_rows.append( PositionRoundtableNodeRow( id=stable_id(f"node:{node_key}"), session_id=session_id, node_key=node_key, stage_index=int(node.get("stage_index") or 0), seat_index=int(node.get("seat_index") or 0), agent_id=str(node.get("agent_id") or f"demo-agent-{index + 1}"), position_id=node.get("position_id") or None, thread_id=( stable_id(f"node-thread:{node_key}") if node.get("thread_id") else None ), status=str(node.get("status") or "done"), latest_answer=node.get("latest_answer") or None, artifact_manifest=_dumps(node.get("artifact_manifest") or []), conversation_snapshot=_dumps(node.get("conversation_snapshot")), revision=int(node.get("revision") or 0), upstream_revision=_dumps(node.get("upstream_revision")), rejection_count=int(node.get("rejection_count") or 0), last_rejection=_dumps(node.get("last_rejection")), invalidated_by=node.get("invalidated_by") or None, created_at=now, updated_at=now, ) ) try: async with self._sf() as session: session.add(row) session.add_all(node_rows) await session.commit() await session.refresh(row) for node_row in node_rows: await session.refresh(node_row) except IntegrityError: # Another tab may have seeded the same user/demo key between the # initial lookup and commit. The deterministic row is the desired # result, so recover it instead of returning a spurious 500. existing = await self.get_session(session_id, user_id) if existing is None: raise return existing, (await self.list_nodes(session_id, user_id)) or [] return self._session_dict(row), [ self._node_dict(node_row) for node_row in node_rows ] async def list_sessions( self, user_id: str, *, external_task_id: str | None = None, limit: int = 50 ) -> list[dict[str, Any]]: """List the caller's own task history or legacy personal history.""" if external_task_id: stmt = select(PositionRoundtableSessionRow).where( PositionRoundtableSessionRow.external_task_id == external_task_id, PositionRoundtableSessionRow.task_scoped.is_(True), PositionRoundtableSessionRow.user_id == user_id, ) else: stmt = select(PositionRoundtableSessionRow).where( PositionRoundtableSessionRow.user_id == user_id, PositionRoundtableSessionRow.task_scoped.is_(False), ) stmt = stmt.order_by(PositionRoundtableSessionRow.updated_at.desc()).limit(limit) async with self._sf() as session: rows = (await session.execute(stmt)).scalars().all() return [self._session_dict(row) for row in rows] async def get_session(self, session_id: str, user_id: str) -> dict[str, Any] | None: async with self._sf() as session: row = await session.get(PositionRoundtableSessionRow, session_id) if row is None or not self._can_access(row, user_id): return None return self._session_dict(row) async def update_session( self, session_id: str, user_id: str, *, task_snapshot: dict[str, Any] | object = _UNSET, intent_thread_id: str | None | object = _UNSET, intent_snapshot: dict[str, Any] | None | object = _UNSET, intent_conversation: dict[str, Any] | None | object = _UNSET, summary_thread_id: str | None | object = _UNSET, summary_snapshot: dict[str, Any] | None | object = _UNSET, summary_conversation: dict[str, Any] | None | object = _UNSET, action_plan_thread_id: str | None | object = _UNSET, action_plan_snapshot: dict[str, Any] | None | object = _UNSET, action_plan_conversation: dict[str, Any] | None | object = _UNSET, status: str | object = _UNSET, allowed_statuses: set[str] | None = None, ) -> dict[str, Any] | None: """Update one session snapshot under the same row lock as commands. Snapshot persistence used to do an unlocked ORM ``get`` followed by a write. That permitted a stale "confirm intent" save to land after a concurrent activation/archival, or allowed two different snapshot surfaces to overwrite each other's unseen values. Acquiring the parent row lock serializes all session writes with node/session commands. ``allowed_statuses`` additionally freezes task/intent inputs once the business chain has become active. """ async with self._sf() as session: row = ( await session.execute( select(PositionRoundtableSessionRow) .where(PositionRoundtableSessionRow.id == session_id) .with_for_update() ) ).scalar_one_or_none() if row is None or not self._can_access(row, user_id): return None if row.status == "archived": raise SessionWriteConflictError("该岗位协同会话已归档，仅可查看历史记录") if allowed_statuses is not None and row.status not in allowed_statuses: raise SessionWriteConflictError("会话已启动，不能再修改已冻结的任务或意图") if task_snapshot is not _UNSET: row.task_snapshot = _dumps(task_snapshot) if intent_thread_id is not _UNSET: row.intent_thread_id = intent_thread_id or None if intent_snapshot is not _UNSET: row.intent_snapshot = _dumps(intent_snapshot) if intent_conversation is not _UNSET: row.intent_conversation = _dumps(intent_conversation) if summary_thread_id is not _UNSET: row.summary_thread_id = summary_thread_id or None if summary_snapshot is not _UNSET: row.summary_snapshot = _dumps(summary_snapshot) if summary_conversation is not _UNSET: row.summary_conversation = _dumps(summary_conversation) if action_plan_thread_id is not _UNSET: row.action_plan_thread_id = action_plan_thread_id or None if action_plan_snapshot is not _UNSET: row.action_plan_snapshot = _dumps(action_plan_snapshot) if action_plan_conversation is not _UNSET: row.action_plan_conversation = _dumps(action_plan_conversation) if status is not _UNSET: row.status = str(status) row.version += 1 row.updated_at = datetime.now(UTC) await session.commit() await session.refresh(row) return self._session_dict(row) async def activate_session( self, session_id: str, user_id: str, *, chain_id: str, chain_snapshot: dict[str, Any], nodes: list[dict[str, Any]], ) -> tuple[dict[str, Any], list[dict[str, Any]]] | None: """Freeze a chain snapshot and create its nodes once. Repeated requests are intentionally idempotent: once nodes exist the existing snapshot and nodes are returned unchanged. """ now = datetime.now(UTC) async with self._sf() as session: row = await session.get(PositionRoundtableSessionRow, session_id) if row is None or not self._can_access(row, user_id): return None node_rows = ( await session.execute( select(PositionRoundtableNodeRow) .where(PositionRoundtableNodeRow.session_id == session_id) .order_by(PositionRoundtableNodeRow.stage_index, PositionRoundtableNodeRow.seat_index) ) ).scalars().all() if node_rows: return self._session_dict(row), [self._node_dict(node) for node in node_rows] row.chain_id = chain_id row.chain_snapshot = _dumps(chain_snapshot) row.status = "active" row.updated_at = now node_rows = [ PositionRoundtableNodeRow( id=uuid4().hex, session_id=session_id, node_key=str(node["node_key"]), stage_index=int(node["stage_index"]), seat_index=int(node["seat_index"]), agent_id=str(node["agent_id"]), position_id=node.get("position_id") or None, status=str(node.get("status") or "locked"), latest_answer=node.get("latest_answer") or None, artifact_manifest=_dumps(node.get("artifact_manifest") or []), revision=int(node.get("revision") or 0), upstream_revision=_dumps(node.get("upstream_revision")) if node.get("upstream_revision") is not None else None, created_at=now, updated_at=now, ) for node in nodes ] session.add_all(node_rows) await session.commit() await session.refresh(row) for node in node_rows: await session.refresh(node) return self._session_dict(row), [self._node_dict(node) for node in node_rows] async def list_nodes(self, session_id: str, user_id: str) -> list[dict[str, Any]] | None: async with self._sf() as session: parent = await session.get(PositionRoundtableSessionRow, session_id) if parent is None or not self._can_access(parent, 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() return [self._node_dict(row) for row in rows] async def get_node(self, session_id: str, node_key: str, user_id: str) -> dict[str, Any] | None: async with self._sf() as session: parent = await session.get(PositionRoundtableSessionRow, session_id) if parent is None or not self._can_access(parent, user_id): return None stmt = select(PositionRoundtableNodeRow).where( PositionRoundtableNodeRow.session_id == session_id, PositionRoundtableNodeRow.node_key == node_key, ).limit(1) row = (await session.execute(stmt)).scalars().first() return self._node_dict(row) if row is not None else None async def update_node( self, session_id: str, node_key: str, user_id: str, *, thread_id: str | None | object = _UNSET, status: str | object = _UNSET, latest_answer: str | None | object = _UNSET, artifact_manifest: list[dict[str, Any]] | object = _UNSET, conversation_snapshot: dict[str, Any] | None | object = _UNSET, upstream_revision: dict[str, Any] | object = _UNSET, last_rejection: dict[str, Any] | None | object = _UNSET, invalidated_by: str | None | object = _UNSET, expected_version: int | None = None, increment_revision: bool = False, increment_rejection: bool = False, ) -> dict[str, Any] | None: async with self._sf() as session: parent = await session.get(PositionRoundtableSessionRow, session_id) if parent is None or not self._can_access(parent, user_id): return None stmt = select(PositionRoundtableNodeRow).where( PositionRoundtableNodeRow.session_id == session_id, PositionRoundtableNodeRow.node_key == node_key, ).limit(1) # 乐观锁：传入 expected_version 时加行锁并校验，杜绝「检查后到提交之间 # 被并发改写」的 TOCTOU。recompute_chain_state 不走这里（它自管事务）。 if expected_version is not None: stmt = stmt.with_for_update() row = (await session.execute(stmt)).scalars().first() if row is None: return None if expected_version is not None and row.version != expected_version: raise NodeConcurrentWriteError( node_key, expected=expected_version, actual=row.version ) if thread_id is not _UNSET: row.thread_id = thread_id or None if status is not _UNSET: row.status = str(status) if latest_answer is not _UNSET: row.latest_answer = latest_answer if artifact_manifest is not _UNSET: row.artifact_manifest = _dumps(artifact_manifest) if conversation_snapshot is not _UNSET: row.conversation_snapshot = _dumps(conversation_snapshot) 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 # 任意字段变更都 bump version（乐观锁单调递增），与业务交付 revision 解耦。 row.version += 1 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 recompute_chain_state( self, session_id: str, node_key: str, user_id: str, *, next_status: str, latest_answer: str | None, artifact_manifest: list[dict[str, Any]] | None, upstream_revision: dict[str, Any] | None, new_effective_delivery: bool, ) -> dict[str, Any] | None: """在单一 ``FOR UPDATE`` 事务内原子完成「本节点交付 → 下游失效 → 解锁下一 stage」。 把原来 router 里由若干个独立提交的 ``update_node`` 拼成的状态转移收敛成一个 事务：先锁住本 session 全部节点行（MySQL ``SELECT ... FOR UPDATE``），再在锁 内重算所有节点状态，一次提交。同一 session 的并发 ``complete_node_turn`` 因此被 数据库行锁串行化，彻底消除 TOCTOU / 丢失更新导致的「下一 stage 卡 locked / invalidated_by 残留 → 下一阶段拿不到上游产物」断链。 幂等可重入：解锁不假设「最后一个 seat 负责」，而是遍历每个「全 done」stage 补 解锁其紧邻下一 stage 的 locked 节点；多次调用结果一致。``next_status`` / ``new_effective_delivery`` 等业务判定由 router 基于调用前快照算好传入，本方法 只负责把它们原子地落到节点状态机上。 """ async with self._sf() as session: parent = await session.get(PositionRoundtableSessionRow, session_id) if parent is None or not self._can_access(parent, user_id): return None # 锁住本 session 全部节点行，串行化并发状态转移（行锁持有至 commit）。 rows = ( await session.execute( select(PositionRoundtableNodeRow) .where(PositionRoundtableNodeRow.session_id == session_id) .order_by( PositionRoundtableNodeRow.stage_index, PositionRoundtableNodeRow.seat_index, ) .with_for_update() ) ).scalars().all() if not rows: return None by_key = {r.node_key: r for r in rows} current = by_key.get(node_key) if current is None: return None now = datetime.now(UTC) # 本次交付前的 revision（用于判断「此前已交付过 → 新交付需失效下游」）。 old_revision = current.revision # ① 落本节点交付状态。 current.status = next_status current.latest_answer = latest_answer if artifact_manifest is not None: current.artifact_manifest = _dumps(artifact_manifest) if upstream_revision is not None: current.upstream_revision = _dumps(upstream_revision) current.invalidated_by = None if new_effective_delivery: current.revision += 1 current.version += 1 current.updated_at = now # ② + ③ 与命令式状态机共用同一套下游失效 / 幂等解锁实现，避免逻辑漂移。 self._invalidate_downstream(rows, current, new_effective_delivery, old_revision, now) self._recompute_unlocks(rows, now) await session.commit() await session.refresh(current) return self._node_dict(current) # ── command-style state machine ────────────────────────────────────────── async def apply_node_command( self, session_id: str, node_key: str, user_id: str, *, command_id: str, command_type: str, payload: dict[str, Any] | None = None, expected_session_version: int | None = None, ) -> dict[str, Any]: """命令式状态机：所有节点级写命令（complete/bind/conversation/reject）的唯一入口。 单事务内依次完成： ① ``SELECT ... FOR UPDATE`` 锁住 session 行（可选 ``expected_session_version`` CAS，命中失败即 409）； ② 命令幂等判定——``UNIQUE(session_id, node_key, command_id)`` 命中已有记录： 参数一致（payload_hash 相同）→ 原样重放上次结果；参数不同 → 409； ③ 锁住本 session 全部节点行，**基于锁内状态**做所有业务判定 （能否完成 / 是否新交付 / 能否驳回），绝不信任调用方传入的过期快照； ④ 应用状态转移（本节点更新 → 下游失效 → 幂等重算解锁）； ⑤ 写入命令幂等记录（含结果 JSON），一次提交。 返回 dict： - ``ok=True`` — ``replayed`` 标记是否幂等重放；``node`` 为目标节点新状态， ``reject`` 命令额外返回 ``nodes``（全部节点）与 ``session``； - ``ok=False`` — ``http_status``（404/409）+ ``detail``；事务已回滚，状态未变。 一致性只来自数据库：并发命令被 session/节点行锁串行化，重试/双击/beacon 由 命令唯一键去重。进程内任何缓存都不参与正确性。 """ payload = payload or {} payload_hash = _payload_hash(command_type, payload) now = datetime.now(UTC) async def _run(session: AsyncSession) -> dict[str, Any]: parent = ( await session.execute( select(PositionRoundtableSessionRow) .where(PositionRoundtableSessionRow.id == session_id) .with_for_update() ) ).scalar_one_or_none() if parent is None or not self._can_access(parent, user_id): return {"ok": False, "http_status": 404, "detail": "Position roundtable session not found"} if expected_session_version is not None and parent.version != expected_session_version: return { "ok": False, "http_status": 409, "detail": "会话状态已更新，请刷新后重试", "session": self._session_dict(parent), } # 归档后的会话拒绝一切写命令（锁内复检，杜绝路由预检与提交之间的窗口）。 if parent.status == "archived": return { "ok": False, "http_status": 409, "detail": "该岗位协同会话已归档，仅可查看历史记录", } # ② 命令幂等：同 command_id 已执行过 → 重放上次结果（或 409）。 existing = ( await session.execute( select(PositionRoundtableCommandRow).where( PositionRoundtableCommandRow.session_id == session_id, PositionRoundtableCommandRow.node_key == (node_key or ""), PositionRoundtableCommandRow.command_id == command_id, ) ) ).scalars().first() if existing is not None: if existing.payload_hash != payload_hash: return { "ok": False, "http_status": 409, "detail": "相同操作标识携带了不同参数，已拒绝执行", } return { "ok": True, "replayed": True, "command_type": command_type, **(_loads(existing.result_json) or {}), } # ③ 锁住全部节点行（行锁持有至 commit），串行化同 session 并发命令。 rows = ( await session.execute( select(PositionRoundtableNodeRow) .where(PositionRoundtableNodeRow.session_id == session_id) .order_by( PositionRoundtableNodeRow.stage_index, PositionRoundtableNodeRow.seat_index, ) .with_for_update() ) ).scalars().all() by_key = {r.node_key: r for r in rows} if command_type == "complete": result = self._apply_complete(by_key, node_key, parent, payload, now) elif command_type == "bind": result = self._apply_bind(by_key, node_key, payload, now) elif command_type == "conversation": result = self._apply_conversation(by_key, node_key, payload, now) elif command_type == "reject": result = self._apply_reject(by_key, node_key, parent, payload, user_id, now) else: return {"ok": False, "http_status": 400, "detail": f"unknown command type: {command_type}"} if not result.get("ok", True): return result # ⑤ 写命令幂等记录（与状态转移同事务提交）。 stored = {k: v for k, v in result.items() if k != "ok"} session.add( PositionRoundtableCommandRow( id=uuid4().hex, session_id=session_id, node_key=node_key or "", command_id=command_id, command_type=command_type, payload_hash=payload_hash, result_json=_dumps(stored), created_at=now, ) ) parent.version += 1 parent.updated_at = now result["replayed"] = False result["command_type"] = command_type return result try: async with self._sf() as session: outcome = await _run(session) if outcome.get("ok", False): await session.commit() else: await session.rollback() return outcome except IntegrityError: # 命令唯一键并发竞态：另一连接已用同一 command_id 提交。读回其结果重放。 async with self._sf() as session: existing = ( await session.execute( select(PositionRoundtableCommandRow).where( PositionRoundtableCommandRow.session_id == session_id, PositionRoundtableCommandRow.node_key == (node_key or ""), PositionRoundtableCommandRow.command_id == command_id, ) ) ).scalars().first() if existing is not None: if existing.payload_hash != payload_hash: return { "ok": False, "http_status": 409, "detail": "相同操作标识携带了不同参数，已拒绝执行", } return { "ok": True, "replayed": True, "command_type": command_type, **(_loads(existing.result_json) or {}), } raise # ── command handlers（全部在 FOR UPDATE 锁内调用，入参为锁定的行） ───────── def _apply_complete( self, by_key: dict[str, PositionRoundtableNodeRow], node_key: str, parent: PositionRoundtableSessionRow, payload: dict[str, Any], now: datetime, ) -> dict[str, Any]: """complete-turn：锁内判定能否完成 / 是否新交付，原子落状态机。""" current = by_key.get(node_key) if current is None: return {"ok": False, "http_status": 404, "detail": "Position roundtable node not found"} if current.status == "locked": return {"ok": False, "http_status": 409, "detail": "上游节点尚未完成，当前节点未解锁"} if current.invalidated_by: return { "ok": False, "http_status": 409, "detail": "上游产物已被驳回或重做，当前节点需等待前序节点重新完成", } # 请求未带产物清单 → 沿用锁内的当前清单（绝不用调用方的过期快照兜底）。 incoming = payload.get("artifact_manifest") current_manifest = _loads(current.artifact_manifest) or [] effective_artifacts = incoming if isinstance(incoming, list) and incoming else current_manifest latest_answer = payload.get("latest_answer") upstream_revision = payload.get("upstream_revision") has_markdown = any(is_markdown_artifact(item) for item in effective_artifacts) markdown_changed = markdown_delivery_changed(current_manifest, effective_artifacts) requires_fresh_delivery = current.status in {"rejected", "stale"} intent_position_id = session_intent_position_id(_loads(parent.chain_snapshot)) is_intent_node = is_intent_position_node(current.position_id, intent_position_id) next_status = ( "done" if is_intent_node else "done" if has_markdown and (not requires_fresh_delivery or markdown_changed) else "running" ) # 有效 revision 是「交付 revision」，不是每一次对话应答。已完成节点上的普通 # 追问，不能仅因其旧 Markdown 仍出现在清单里就失效所有下游 stage。 new_effective_delivery = ( next_status == "done" and (is_intent_node or current.status != "done" or markdown_changed) ) old_revision = current.revision # ④ 落本节点交付状态。 current.status = next_status current.latest_answer = latest_answer current.artifact_manifest = _dumps(effective_artifacts) if upstream_revision is not None: current.upstream_revision = _dumps(upstream_revision) current.invalidated_by = None if new_effective_delivery: current.revision += 1 current.version += 1 current.updated_at = now self._invalidate_downstream(by_key.values(), current, new_effective_delivery, old_revision, now) self._recompute_unlocks(by_key.values(), now) return {"ok": True, "node": self._node_dict(current)} def _apply_bind( self, by_key: dict[str, PositionRoundtableNodeRow], node_key: str, payload: dict[str, Any], now: datetime, ) -> dict[str, Any]: """bind-thread：锁内校验解锁/失效/线程占用后绑定。""" current = by_key.get(node_key) if current is None: return {"ok": False, "http_status": 404, "detail": "Position roundtable node not found"} thread_id = str(payload.get("thread_id") or "").strip() if not thread_id: return {"ok": False, "http_status": 400, "detail": "thread_id is required"} if current.status == "locked": return {"ok": False, "http_status": 409, "detail": "上游节点尚未完成，当前节点未解锁"} if current.invalidated_by: return { "ok": False, "http_status": 409, "detail": "上游产物已被驳回或重做，当前节点需等待前序节点重新完成", } if current.thread_id and current.thread_id != thread_id: return {"ok": False, "http_status": 409, "detail": "该节点已绑定另一条对话线程"} current.thread_id = thread_id if current.status == "ready": current.status = "running" current.version += 1 current.updated_at = now return {"ok": True, "node": self._node_dict(current)} def _apply_conversation( self, by_key: dict[str, PositionRoundtableNodeRow], node_key: str, payload: dict[str, Any], now: datetime, ) -> dict[str, Any]: """conversation：对话快照持久化（UI 恢复用），锁内校验线程归属。""" current = by_key.get(node_key) if current is None: return {"ok": False, "http_status": 404, "detail": "Position roundtable node not found"} thread_id = payload.get("thread_id") if thread_id and current.thread_id and current.thread_id != thread_id: return {"ok": False, "http_status": 409, "detail": "Position node is bound to another thread"} if thread_id and not current.thread_id: current.thread_id = thread_id current.conversation_snapshot = _dumps(payload.get("conversation_snapshot")) current.version += 1 current.updated_at = now return {"ok": True, "node": self._node_dict(current)} def _apply_reject( self, by_key: dict[str, PositionRoundtableNodeRow], node_key: str, parent: PositionRoundtableSessionRow, payload: dict[str, Any], user_id: str, now: datetime, ) -> dict[str, Any]: """reject：驳回一个已交付节点并失效全部下游 stage（锁内判定可驳回性）。""" rows = sorted( by_key.values(), key=lambda r: (r.stage_index, r.seat_index) ) target = by_key.get(node_key) if target is None: return {"ok": False, "http_status": 404, "detail": "Position roundtable node not found"} if target.status == "locked": return {"ok": False, "http_status": 409, "detail": "当前节点尚未解锁，暂无可驳回的交付产物"} if ( target.revision <= 0 and not target.latest_answer and not _loads(target.artifact_manifest) ): return {"ok": False, "http_status": 409, "detail": "当前节点暂无可驳回的交付产物"} rejection = { "title": str(payload.get("title") or "").strip() or "产物驳回", "reason": str(payload.get("reason") or "").strip(), "rejected_by": user_id, "rejected_at": now.isoformat(), } summary_snapshot = _loads(parent.summary_snapshot) if isinstance(summary_snapshot, dict) and summary_snapshot.get("status") == "done": parent.summary_snapshot = _dumps( { **summary_snapshot, "status": "stale", "invalidated_by": target.node_key, "invalidated_at": now.isoformat(), } ) action_plan_snapshot = _loads(parent.action_plan_snapshot) if isinstance(action_plan_snapshot, dict) and action_plan_snapshot.get("status") == "done": parent.action_plan_snapshot = _dumps( { **action_plan_snapshot, "status": "stale", "invalidated_by": target.node_key, "invalidated_at": now.isoformat(), } ) if parent.status == "completed": parent.status = "active" target.status = "rejected" target.last_rejection = _dumps(rejection) target.rejection_count += 1 target.invalidated_by = None target.version += 1 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.version += 1 row.updated_at = now return { "ok": True, "session": self._session_dict(parent), "nodes": [self._node_dict(row) for row in rows], } @staticmethod def _invalidate_downstream( rows: Any, current: PositionRoundtableNodeRow, new_effective_delivery: bool, old_revision: int, now: datetime, ) -> None: """新交付且本节点此前已交付过 → 后续 stage 全部失效（需重做）。""" if not (new_effective_delivery and old_revision > 0): return for downstream in rows: if ( downstream.stage_index > current.stage_index and downstream.status in {"ready", "running", "done", "stale", "rejected", "error"} ): downstream.status = "stale" if _has_material(downstream) else "locked" downstream.invalidated_by = current.node_key downstream.version += 1 downstream.updated_at = now @staticmethod def _recompute_unlocks(rows: Any, now: datetime) -> None: """幂等重算 stage 解锁：每个「全部 done」stage 解锁其紧邻下一 stage。 locked → ready；stale 保持 stale（产物过期需重做）但清掉 invalidated_by （上游已就绪）。多次调用结果一致，不依赖调用时序。 """ row_list = list(rows) stage_indices = sorted({r.stage_index for r in row_list}) for stage in stage_indices: stage_rows = [r for r in row_list if r.stage_index == stage] if not stage_rows or not all(r.status == "done" for r in stage_rows): continue later = [t for t in stage_indices if t > stage] if not later: continue next_stage = min(later) for node in row_list: if node.stage_index != next_stage: continue if node.status == "locked": node.status = "ready" node.invalidated_by = None node.version += 1 node.updated_at = now elif node.status == "stale": node.invalidated_by = None node.version += 1 node.updated_at = now async def apply_session_command( self, session_id: str, user_id: str, *, command_id: str, command_type: str, payload: dict[str, Any] | None = None, expected_session_version: int | None = None, ) -> dict[str, Any]: """session 级命令（activate / archive / delete）的命令式入口。 与 ``apply_node_command`` 同一套原则：单事务锁 session 行（``FOR UPDATE``） + 命令唯一键幂等 + 锁内判定。返回结构同 ``apply_node_command`` （``ok`` / ``replayed`` / ``http_status`` / ``detail`` / ``session`` / ``nodes``）。 - ``activate`` — payload 需带 ``chain_id`` / ``chain_snapshot`` / ``nodes``。 锁内判定：已激活且链相同 → 幂等返回现有状态；链不同 → 409。 - ``archive`` — 已归档 → 幂等返回；否则置 archived。 - ``delete`` — 删除节点与会话（命令记录无外键，保留供重放）。 """ payload = payload or {} payload_hash = _payload_hash(command_type, payload) now = datetime.now(UTC) async def _replay_check(session: AsyncSession) -> dict[str, Any] | None: existing = ( await session.execute( select(PositionRoundtableCommandRow).where( PositionRoundtableCommandRow.session_id == session_id, PositionRoundtableCommandRow.node_key == "", PositionRoundtableCommandRow.command_id == command_id, ) ) ).scalars().first() if existing is None: return None if existing.payload_hash != payload_hash: return { "ok": False, "http_status": 409, "detail": "相同操作标识携带了不同参数，已拒绝执行", } return { "ok": True, "replayed": True, "command_type": command_type, **(_loads(existing.result_json) or {}), } async def _run(session: AsyncSession) -> dict[str, Any]: # delete 之后 session 行已不存在，重试必须在锁 session 之前查命令记录。 replayed = await _replay_check(session) if replayed is not None: return replayed parent = ( await session.execute( select(PositionRoundtableSessionRow) .where(PositionRoundtableSessionRow.id == session_id) .with_for_update() ) ).scalar_one_or_none() if parent is None or not self._can_access(parent, user_id): return {"ok": False, "http_status": 404, "detail": "Position roundtable session not found"} if expected_session_version is not None and parent.version != expected_session_version: return { "ok": False, "http_status": 409, "detail": "会话状态已更新，请刷新后重试", "session": self._session_dict(parent), } if command_type == "activate": result = await self._apply_activate(session, parent, payload, now) elif command_type == "archive": if parent.status == "archived": result = { "ok": True, "session": self._session_dict(parent), "nodes": [], "already": True, } else: parent.status = "archived" parent.version += 1 parent.updated_at = now result = { "ok": True, "session": self._session_dict(parent), "nodes": [], } elif command_type == "delete": await session.execute( delete(PositionRoundtableNodeRow).where( PositionRoundtableNodeRow.session_id == session_id ) ) await session.execute( delete(PositionRoundtableSessionRow).where( PositionRoundtableSessionRow.id == session_id ) ) result = {"ok": True, "deleted": True} else: return {"ok": False, "http_status": 400, "detail": f"unknown command type: {command_type}"} if not result.get("ok", True): return result stored = {k: v for k, v in result.items() if k != "ok"} session.add( PositionRoundtableCommandRow( id=uuid4().hex, session_id=session_id, node_key="", command_id=command_id, command_type=command_type, payload_hash=payload_hash, result_json=_dumps(stored), created_at=now, ) ) result["replayed"] = False result["command_type"] = command_type return result try: async with self._sf() as session: outcome = await _run(session) if outcome.get("ok", False): await session.commit() else: await session.rollback() return outcome except IntegrityError: # 命令唯一键并发竞态：读回已提交的那条命令结果重放。 async with self._sf() as session: replayed = await _replay_check(session) if replayed is not None: return replayed raise async def _apply_activate( self, session: AsyncSession, parent: PositionRoundtableSessionRow, payload: dict[str, Any], now: datetime, ) -> dict[str, Any]: """激活业务链：锁内判定「已激活 → 链相同幂等 / 链不同 409」，否则建节点。""" existing_rows = ( await session.execute( select(PositionRoundtableNodeRow) .where(PositionRoundtableNodeRow.session_id == parent.id) .order_by( PositionRoundtableNodeRow.stage_index, PositionRoundtableNodeRow.seat_index, ) .with_for_update() ) ).scalars().all() if parent.status == "archived": return { "ok": False, "http_status": 409, "detail": "该岗位协同会话已归档，仅可查看历史记录", } chain_id = str(payload.get("chain_id") or "") if existing_rows: # 已激活：同一业务链 → 幂等返回现有状态；不同链 → 冲突。 if parent.chain_id == chain_id: return { "ok": True, "session": self._session_dict(parent), "nodes": [self._node_dict(row) for row in existing_rows], "already": True, } return { "ok": False, "http_status": 409, "detail": "该会话已激活其他业务链条，不能重复激活", } if not _has_confirmed_intent_snapshot(_loads(parent.intent_snapshot)): return { "ok": False, "http_status": 400, "detail": "请先完成并保存任务意图", } chain_snapshot = payload.get("chain_snapshot") or {} nodes = payload.get("nodes") or [] if not nodes: return {"ok": False, "http_status": 400, "detail": "activate requires nodes"} parent.chain_id = chain_id parent.chain_snapshot = _dumps(chain_snapshot) parent.status = "active" parent.version += 1 parent.updated_at = now node_rows = [ PositionRoundtableNodeRow( id=uuid4().hex, session_id=parent.id, node_key=str(node["node_key"]), stage_index=int(node["stage_index"]), seat_index=int(node["seat_index"]), agent_id=str(node["agent_id"]), position_id=node.get("position_id") or None, status=str(node.get("status") or "locked"), latest_answer=node.get("latest_answer") or None, artifact_manifest=_dumps(node.get("artifact_manifest") or []), revision=int(node.get("revision") or 0), upstream_revision=( _dumps(node.get("upstream_revision")) if node.get("upstream_revision") is not None else None ), created_at=now, updated_at=now, ) for node in nodes ] session.add_all(node_rows) # 同事务内 flush 拿到完整行状态，供命令记录与返回值使用。 await session.flush() return { "ok": True, "session": self._session_dict(parent), "nodes": [self._node_dict(row) for row in node_rows], } 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 not self._can_access(parent, 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") async def delete_session(self, session_id: str, user_id: str) -> bool: """Delete a task-shared row or the caller's personal legacy row.""" async with self._sf() as session: parent = ( await session.execute( select(PositionRoundtableSessionRow) .where( PositionRoundtableSessionRow.id == session_id, ) .limit(1) ) ).scalar_one_or_none() if parent is None or not self._can_access(parent, user_id): return False # Do not rely on database-specific FK cascade settings. SQLite # deployments may not have PRAGMA foreign_keys enabled, which # would otherwise leave orphaned conversation/artifact snapshots. await session.execute( delete(PositionRoundtableNodeRow).where( PositionRoundtableNodeRow.session_id == session_id ) ) await session.execute( delete(PositionRoundtableSessionRow).where( PositionRoundtableSessionRow.id == session_id, ) ) await session.commit() return True async def list_results(self, session_id: str, user_id: str) -> list[dict[str, Any]] | None: return await self.list_nodes(session_id, user_id)