feat(position-roundtable): add delivery workflow scaffolding

This commit is contained in:
fengxiangdebaozi committed 2026-07-21 19:48:39 +08:00
1 parent 97ce19973d
commit 2ca8a23102
4 files changed
+416 -129

No files matched your search

@@ -20,7 +20,7 @@ from deerflow.runtime.user_context import get_effective_user_id
router = APIRouter(prefix="/api/position-roundtable", tags=["position-roundtable"])
NODE_STATUSES = {"locked", "ready", "running", "done", "stale", "error"}
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
@@ -152,7 +152,11 @@ async def _load_detail(request: Request, session_id: str) -> SessionDetailRespon
return SessionDetailResponse(**row, nodes=[_project_node(node) for node in nodes or []])
def _chain_nodes(chain: dict[str, Any]) -> list[dict[str, Any]]:
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="业务链条没有可执行的智能体席位")
@@ -175,6 +179,11 @@ def _chain_nodes(chain: dict[str, Any]) -> list[dict[str, Any]]:
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(
@@ -194,6 +203,23 @@ def _chain_nodes(chain: dict[str, Any]) -> list[dict[str, Any]]:
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 ""
@@ -435,7 +461,15 @@ async def activate_session(
)
if chain is None:
raise HTTPException(status_code=404, detail="业务链条不存在或无权访问")
nodes = _chain_nodes(chain)
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")
}
nodes = _chain_nodes(
chain,
available_agent_ids=await _available_agent_ids(request, seat_ids),
)
activated = await store.activate_session(
session_id,
user_id,
@@ -79,6 +79,26 @@ async def test_session_activation_freezes_chain_and_creates_nodes_once(store):
assert [node["node_key"] for node in repeated_nodes] == [node["node_key"] for node in created_nodes]
def test_chain_node_activation_rejects_missing_agent_ids():
chain = {
"id": "chain-a",
"seats": [
{"agent_id": "agent-a", "name": "有效智能体", "position_id": "planning"},
{"agent_id": "missing-agent", "name": "缺失智能体", "position_id": "solution"},
],
}
with pytest.raises(HTTPException) as error:
position_roundtable_router._chain_nodes(
chain,
available_agent_ids={"agent-a"},
)
assert error.value.status_code == 400
assert "缺失智能体" in str(error.value.detail)
assert "缺少可用智能体" in str(error.value.detail)
@pytest.mark.asyncio
async def test_node_updates_are_user_scoped_and_keep_thread_out_of_messages(store):
created = await store.create_session("user-a", external_task_id=None, task_snapshot={})