页面嵌套修改

This commit is contained in:
董宏凯 committed 2026-07-29 21:53:32 +08:00
1 parent 213dd841b5
commit 4450f342d8
38 files changed
+2020 -309

No files matched your search

@@ -123,6 +123,11 @@ cascade pragmas. The router is mounted canonically below
schema-hidden compatibility prefix `/api/multi-agent/position-roundtable`.
Clients may retry the compatibility route only for a framework/proxy-level
`404 Not Found`; domain-level missing-session responses must remain real 404s.
`/api/intent/init` must create the empty Step-1 thread through
`multi_agent._create_thread_direct`, exactly like multi-agent initialization;
do not restore the old HTTP loopback to `/api/threads`, because deployed
Gateway ports/proxies may make the automatic page initialization fail while
later manual turns still work. Streaming stays on the existing loopback path.
### Harness / App Split
@@ -407,6 +412,10 @@ Each agent/skill carries a single `square_id` column (`AgentRow` / `SkillRow`, `
Backs the zzJC / zzXD agent-chat left sidebar (「按类型聚合智能体 + 聊天记录」). An admin picks which agent **tags** belong to a module; the module's left sidebar then shows only agents under those tags (empty/unconfigured = show all visible agents). Stored in `system_settings.json` → `agent_modules` (`AgentModuleSettings`: `modules: {module_key → [tag_id, ...]}`); pure helpers `get_module_tag_ids` / `set_module_tag_ids` (de-dupe + strip). Endpoints (in `app/gateway/routers/system_settings.py`): `GET /api/system-settings/agent-modules/{key}` (readable by any user — the sidebar needs it to filter), `PUT /api/system-settings/agent-modules/{key}` (admin only, full replacement of the tag-id list). Module keys are frontend-defined (`AGENT_MODULES` in `components/workspace/agents/agent-module-sidebar.tsx`, keyed by the module's landing agent id: `6bf`→`zzjc`, `10d69…`→`zzxd`). Tests: `tests/test_agent_modules.py`. Frontend: `core/system-settings` (`useAgentModuleConfig`/`useUpdateAgentModuleConfig`), admin `AgentModuleSettingsDialog` (gear button in the agent-chat header).
### Business-mapping authorization
`/api/business-mapping` remains globally readable. Updates to 3Q, 6BF, and 7BF require `system_role == "admin"`; the 8BF public business-chain mapping is intentionally different and can be updated only by the trusted authenticated account whose email prefix is `lqq`. Do not rely on a username supplied in an external login URL. The frontend follows the same email-prefix rule when deciding whether to render the 8BF configuration button, while the router (`app/gateway/routers/business_mapping.py`) remains the enforcement point. Tests: `tests/test_business_mapping.py`.
### Position-collaboration foundation
The position-collaboration workspace is a separate frontend route and must not
@@ -1093,6 +1102,18 @@ Both the `view_image` tool and `ViewImageMiddleware` are registered whenever `vi
## Documentation
### Position-roundtable role directory
`position_roles` is the global shared directory for position-roundtable work
views. It is deliberately separate from organisation/RBAC `positions`.
`app.gateway.deps.langgraph_runtime()` seeds the four stable built-in ids from
`app/gateway/position_role_defaults.py` via `ensure_seed()` (missing rows only;
never overwrite operator edits). The API is `GET/PUT /api/position-roles`;
writes are admin-only. Exactly one enabled `intent` role is required and owns
the initial intent workspace plus the summary/action-plan closing agents. Do
not delete a built-in id because historical chain nodes persist those ids;
custom role ids may be added and removed.
See `docs/` directory for detailed documentation:
- [../README.md](../README.md) - Offline Docker wheelhouse preparation and startup workflow. The pinned versions in `../offline-runtime-requirements.txt` and `../start_backend.sh` must stay synchronized. `start_backend.sh` stays aligned with the deployment script and adds only the Redis wheel fallback line. The download target defaults to CPython 3.12 on Linux x86_64 and verifies dependency closure using wheelhouse only.
- [CONFIGURATION.md](docs/CONFIGURATION.md) - Configuration options
+20 -1
View File
@@ -124,6 +124,7 @@ FastAPI application providing REST endpoints for frontend integration:
| `GET/POST /api/agents` | List visible agents or create a user-owned agent |
| `GET/PUT/DELETE /api/agents/{id}` | Read, update, or delete an id-addressed agent |
| `GET/POST/PUT/DELETE /api/roundtable-chains` | Manage user-owned roundtable business chains; each seat may optionally carry `position_id` for the separate position-collaboration workspace (the original roundtable ignores it) |
| `GET/PUT /api/business-mapping` | Read global business mappings. Updates to 3Q/6BF/7BF require an administrator; the 8BF public business-chain mapping is maintained exclusively by the trusted `lqq` account (derived from the authenticated email prefix). |
| `GET/POST/PUT /api/position-roundtable/sessions` | Standalone position-collaboration sessions: freeze task/intent/chain snapshots, index one direct-agent node per seat, build upstream handoff context (including bounded UTF-8 excerpts of text artifacts), and make archived sessions read-only without invoking the original roundtable coordinator or jobs. Built-in summary/action-plan completion mirrors visible report text into a Markdown artifact when a model omits its final `write_file` call. `POST /sessions/{id}/nodes/{nodeKey}/reject` accepts an optional reason, marks the rejected delivery for rework, and invalidates affected downstream deliveries so they are regenerated in order. The same router is also mounted at `/api/multi-agent/position-roundtable/*` for intranet gateways that only forward the established multi-agent namespace; the compatibility mount is intentionally omitted from OpenAPI. |
| `GET /api/memory` | Retrieve memory data |
| `POST /api/memory/reload` | Force memory reload |
@@ -478,7 +479,25 @@ checkpoint or sandbox-output volume is replaced. Apply Alembic revision
Its live intent conversation still uses the shared `/api/intent/init` and
`/api/intent/stream` endpoints used by the original multi-agent roundtable.
Only the recovery snapshot layer is position-specific. Deployments must expose
`/api/intent/init` creates its empty thread directly through the in-process
thread store/checkpointer, matching multi-agent initialization and avoiding a
deployment-sensitive `127.0.0.1:<gateway-port>` loopback. Only the recovery
snapshot layer is position-specific. Deployments must expose
either `/api/position-roundtable/*` or the compatibility mirror
`/api/multi-agent/position-roundtable/*`; updated clients probe both and only
retry when the first response is the framework/proxy-level `404 Not Found`.
## Position-roundtable role directory
The position-roundtable work views are persisted separately from organisation
positions/RBAC in `position_roles`. Gateway startup idempotently inserts the
four built-in ids (testing analysis, organisation test, solution design and
comprehensive test), but only when an id is missing: administrator changes to
the name, order, enabled state or role type are never overwritten.
`GET /api/position-roles` is consumed by the workbench and `PUT
/api/position-roles` is administrator-only. Exactly one enabled `intent` role
is required. It is the initial intent-identification workspace and owns the
summary/action-plan closing agents. Built-in rows retain stable ids so
historical chain nodes remain readable; they can be edited or disabled but not
deleted. Custom roles may be freely added or removed.
@@ -57,6 +57,7 @@ from app.gateway.routers import (
parallel_agents,
position_roundtable,
positions,
position_roles,
public_config,
public_embed,
public_leaderboard,
@@ -1485,7 +1486,10 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
)
# Global business ? agent / chain mapping at /api/business-mapping
app.include_router(business_mapping.router)
app.include_router(business_mapping.router)
# Position-roundtable work-view role configuration at /api/position-roles.
app.include_router(position_roles.router)
# Light-app registry (?????) at /api/light-apps
app.include_router(light_apps.router)
@@ -264,6 +264,14 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
from deerflow.persistence.position_roundtable import make_position_roundtable_store
app.state.position_roundtable_store = make_position_roundtable_store(sf)
# 岗位会商工作视角(名称 / 顺序 / 特殊意图岗)是全局配置,和组织/RBAC
# 岗位分开保存。启动只补齐缺失内置项,不覆盖管理员的既有修改。
from app.gateway.position_role_defaults import POSITION_ROLE_SEED
from deerflow.persistence.position_roles import make_position_role_store
app.state.position_role_store = make_position_role_store(sf)
await app.state.position_role_store.ensure_seed(POSITION_ROLE_SEED)
from deerflow.persistence.business_mapping import make_business_mapping_store
@@ -0,0 +1,47 @@
"""Built-in roles for the position-roundtable workbench.
These ids are part of the persisted business-chain contract. Administrators
may rename, reorder or disable the roles, but a startup seed only inserts
missing entries and therefore never overwrites production configuration.
"""
from __future__ import annotations
POSITION_ROLE_SEED: list[dict[str, object]] = [
{
"id": "intelligence",
"name": "测试分析岗",
"description": "识别任务意图、补齐约束并确认业务链条",
"role_type": "intent",
"enabled": True,
"is_builtin": True,
"sort_order": 0,
},
{
"id": "planning",
"name": "组织测试岗",
"description": "承接上游信息并进行组织测试",
"role_type": "standard",
"enabled": True,
"is_builtin": True,
"sort_order": 1,
},
{
"id": "solution",
"name": "方案设计岗",
"description": "承接上游结果并形成方案产物",
"role_type": "standard",
"enabled": True,
"is_builtin": True,
"sort_order": 2,
},
{
"id": "evaluation",
"name": "综合测试岗",
"description": "测试链条产物并提出综合建议",
"role_type": "standard",
"enabled": True,
"is_builtin": True,
"sort_order": 3,
},
]
@@ -26,6 +26,9 @@ from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/business-mapping", tags=["business-mapping"])
_EIGHTBF_BUSINESS_CODE = "8BF"
_EIGHTBF_MANAGER_USERNAME = "lqq"
# ── schemas ────────────────────────────────────────────────────────────────
@@ -70,6 +73,20 @@ def _is_admin(request: Request) -> bool:
return user is None or getattr(user, "system_role", None) == "admin"
def _username_of(request: Request) -> str:
"""Read the trusted login account name, not an externally supplied username."""
user = getattr(request.state, "user", None)
email = str(getattr(user, "email", "") or "")
return email.split("@", 1)[0].strip().lower()
def _can_update_mapping(request: Request, business_code: str) -> bool:
"""8BF's public chain is maintained exclusively by the lqq account."""
if business_code.strip().upper() == _EIGHTBF_BUSINESS_CODE:
return _username_of(request) == _EIGHTBF_MANAGER_USERNAME
return _is_admin(request)
def _current_user_id(request: Request) -> str:
user = getattr(request.state, "user", None)
if user is not None:
@@ -119,17 +136,14 @@ async def _validate_public_chain_mapping(
async def list_mappings(request: Request) -> BusinessMappingListResponse:
store = _get_store(request)
rows = await store.list_mappings()
return BusinessMappingListResponse(
mappings=[BusinessMappingResponse(**r) for r in rows]
)
return BusinessMappingListResponse(mappings=[BusinessMappingResponse(**r) for r in rows])
@router.put("/{business_code}", response_model=BusinessMappingResponse)
async def update_mapping(
request: Request, business_code: str, body: BusinessMappingUpdateRequest
) -> BusinessMappingResponse:
if not _is_admin(request):
raise HTTPException(status_code=403, detail="Only administrators can update business mappings")
async def update_mapping(request: Request, business_code: str, body: BusinessMappingUpdateRequest) -> BusinessMappingResponse:
if not _can_update_mapping(request, business_code):
detail = "Only the lqq account can update the 8BF business mapping" if business_code.strip().upper() == _EIGHTBF_BUSINESS_CODE else "Only administrators can update business mappings"
raise HTTPException(status_code=403, detail=detail)
store = _get_store(request)
# Only forward fields the caller explicitly sent (distinguish omitted vs null).
fields = body.model_dump(exclude_unset=True)
@@ -51,11 +51,11 @@ SSE 协议(与前端 ``streamIntent`` 对齐)
from __future__ import annotations
import asyncio
import json
import logging
import re
from typing import Any, AsyncIterator
from collections.abc import AsyncIterator
from typing import Any
import httpx
from fastapi import APIRouter, Body, HTTPException, Request
@@ -69,7 +69,7 @@ from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_ag
from app.gateway.routers.multi_agent import (
_ROUNDTABLE_THREAD_METADATA,
_auth_headers,
_create_thread,
_create_thread_direct,
_diagnose_exception,
_find_dispatch_message,
_find_trailing_ai_text,
@@ -311,21 +311,18 @@ async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str,
# 缺失则就地用嵌入模板补建,避免下一步 /stream 报 500。
ensure_roundtable_functional_agents()
base = _loopback_base(request)
headers = _auth_headers(request)
# TCP loopback 到 127.0.0.1:<gateway port>(详见 multi_agent._make_loopback_client)。
# 不要在这里替换成 httpx.ASGITransport——曾试过,会整段 buffer SSE 流导致前端
# 看不到逐 token 增量。
# Keep Step-1 initialization aligned with multi-agent initialization:
# create the empty thread in-process rather than looping back through
# 127.0.0.1:<gateway-port>. Intranet proxies and port mappings can make
# that loopback fail only for automatic page initialization. Streaming
# below still uses the TCP loopback so SSE remains incrementally flushed.
raw_task_context = payload.get("task_context")
task_context = raw_task_context if isinstance(raw_task_context, dict) else None
try:
async with _make_loopback_client(request) as client:
thread_id = await _create_thread(
client,
base,
headers,
metadata=_intent_thread_metadata(task_context),
)
thread_id = await _create_thread_direct(
request,
metadata=_intent_thread_metadata(task_context),
)
except HTTPException:
# 上游已用结构化方式失败,原样传给前端不需要二次包装。
raise
@@ -0,0 +1,148 @@
"""Global position-roundtable role configuration API.
This directory is intentionally different from ``/api/positions``: the
latter is the organisation/RBAC position model, whereas this router controls
the work views that appear in the 岗位多智能体会商 page.
"""
from __future__ import annotations
import re
from datetime import datetime
from typing import Literal
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field, field_validator
from app.gateway.deps import get_current_user, get_optional_user_from_request
from app.gateway.position_role_defaults import POSITION_ROLE_SEED
from deerflow.persistence.position_roles import PositionRoleStore
router = APIRouter(prefix="/api/position-roles", tags=["position-roles"])
ROLE_ID_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_-]{0,127}$")
class PositionRole(BaseModel):
id: str = Field(..., min_length=1, max_length=128)
name: str = Field(..., min_length=1, max_length=128)
description: str = Field(default="", max_length=512)
role_type: Literal["standard", "intent"] = "standard"
enabled: bool = True
# Response values are preserved by the repository; clients must not use
# this flag to promote a custom role into a protected built-in role.
is_builtin: bool = False
sort_order: int = Field(default=0, ge=0)
@field_validator("id")
@classmethod
def validate_id(cls, value: str) -> str:
value = value.strip()
if not ROLE_ID_PATTERN.fullmatch(value):
raise ValueError("岗位标识须以字母开头,只能包含字母、数字、- 或 _")
return value
@field_validator("name")
@classmethod
def normalize_name(cls, value: str) -> str:
value = value.strip()
if not value:
raise ValueError("岗位名称不能为空")
return value
class PositionRoleResponse(PositionRole):
updated_by: str | None = None
updated_at: datetime | str | None = None
class PositionRoleListResponse(BaseModel):
roles: list[PositionRoleResponse]
class PositionRoleReplaceRequest(BaseModel):
roles: list[PositionRole] = Field(default_factory=list)
def _get_store(request: Request) -> PositionRoleStore:
store = getattr(request.app.state, "position_role_store", None)
if store is None:
raise HTTPException(status_code=503, detail="Position role store not available")
return store
async def _require_admin(request: Request) -> None:
"""Reject non-admins but retain the local auth-disabled developer mode."""
user = await get_optional_user_from_request(request)
if user is None:
return
if getattr(user, "system_role", None) != "admin":
raise HTTPException(status_code=403, detail="岗位角色配置仅限管理员")
def _validate_payload(
rows: list[PositionRole],
existing: list[dict[str, object]],
) -> list[dict[str, object]]:
"""Apply directory invariants before the atomic replacement.
Keeping built-ins prevents old business-chain sessions from becoming
unreadable. They remain fully editable (including disable/retype); custom
roles can be removed freely.
"""
if not rows:
raise HTTPException(status_code=422, detail="至少保留一个启用的岗位")
ids = [row.id for row in rows]
if len(ids) != len(set(ids)):
raise HTTPException(status_code=422, detail="岗位标识不能重复")
names = [row.name.casefold() for row in rows]
if len(names) != len(set(names)):
raise HTTPException(status_code=422, detail="岗位名称不能重复")
enabled = [row for row in rows if row.enabled]
intent_rows = [row for row in enabled if row.role_type == "intent"]
if len(intent_rows) != 1:
raise HTTPException(
status_code=422,
detail="请保留且仅保留一个启用的意图识别岗(它承载会商入口和收口智能体)",
)
builtin_ids = {str(item["id"]) for item in existing if item.get("is_builtin")}
missing_builtin = builtin_ids.difference(ids)
if missing_builtin:
raise HTTPException(
status_code=422,
detail="内置岗位不能删除;如暂不使用,请将它停用",
)
existing_builtin = {str(item["id"]): bool(item.get("is_builtin")) for item in existing}
return [
{
**row.model_dump(),
"is_builtin": existing_builtin.get(row.id, False),
"sort_order": index,
}
for index, row in enumerate(rows)
]
@router.get("", response_model=PositionRoleListResponse)
async def list_position_roles(request: Request) -> PositionRoleListResponse:
# Startup performs the same idempotent seed. Repeating it here makes an
# upgraded running instance recover gracefully when the table was created
# after startup or a deployment starts with an empty database.
rows = await _get_store(request).ensure_seed(POSITION_ROLE_SEED)
return PositionRoleListResponse(roles=[PositionRoleResponse(**row) for row in rows])
@router.put("", response_model=PositionRoleListResponse)
async def replace_position_roles(
request: Request,
body: PositionRoleReplaceRequest,
) -> PositionRoleListResponse:
await _require_admin(request)
store = _get_store(request)
normalized = _validate_payload(body.roles, await store.list_roles())
user_id = await get_current_user(request)
rows = await store.replace_all(normalized, updated_by=user_id)
return PositionRoleListResponse(roles=[PositionRoleResponse(**row) for row in rows])
@@ -331,15 +331,57 @@ def _delivered_nodes(nodes: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [node for node in nodes if node.get("status") == "done"]
def _all_nodes_delivered(nodes: list[dict[str, Any]]) -> bool:
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) or _node_has_markdown_delivery(node))
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]) -> bool:
return str(node.get("position_id") or "").strip().lower() in {"intelligence", "intel", "intent"}
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:
@@ -419,12 +461,17 @@ def _has_confirmed_intent(intent: Any) -> bool:
return bool(objective and isinstance(constraints, list))
def _prime_intent_nodes(nodes: list[dict[str, Any]], intent: Any) -> list[dict[str, Any]]:
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):
if _is_intent_position_node(next_node, intent_position_id):
next_node.update(
{
"status": "done",
@@ -461,7 +508,7 @@ def _build_node_context_text(
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)
md_required = not _is_intent_position_node(current, _session_intent_position_id(session))
lines = [
"岗位多智能体会商上下文(隐藏注入,不要向用户复述本标签):",
"",
@@ -716,7 +763,7 @@ async def _handoff_context(request: Request, session_id: str, node_key: str) ->
),
"delivery_requirement": {
"type": "markdown_report",
"required": not _is_intent_position_node(current),
"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 {},
@@ -1137,16 +1184,26 @@ async def activate_session(
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,
chain_snapshot,
available_agent_ids=await _available_agent_ids(request, seat_ids),
)
nodes = _prime_intent_nodes(nodes, current.get("intent_snapshot"))
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,
chain_snapshot=chain_snapshot,
nodes=nodes,
)
if activated is None:
@@ -1215,7 +1272,7 @@ async def update_node_conversation(
node_key: str,
body: ConversationStateUpdateRequest,
) -> NodeResponse:
store, user_id, _ = await _require_writable_session(request, session_id)
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")
@@ -1247,7 +1304,7 @@ async def build_node_context(request: Request, session_id: str, node_key: str) -
async def complete_node_turn(
request: Request, session_id: str, node_key: str, body: CompleteTurnRequest
) -> NodeResponse:
store, user_id, _ = await _require_writable_session(request, session_id)
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")
@@ -1268,7 +1325,7 @@ async def complete_node_turn(
requires_fresh_delivery = node.get("status") in {"rejected", "stale"}
next_status = (
"done"
if _is_intent_position_node(node)
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"
@@ -1280,7 +1337,7 @@ async def complete_node_turn(
new_effective_delivery = (
next_status == "done"
and (
_is_intent_position_node(node)
_is_intent_position_node(node, _session_intent_position_id(session))
or node.get("status") != "done"
or markdown_changed
)
@@ -1384,7 +1441,7 @@ async def _prepare_builtin_run(
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):
if not _all_nodes_delivered(nodes, intent_position_id=_session_intent_position_id(session)):
raise HTTPException(status_code=409, detail="所有岗位产物完成后才能生成或更新内置交付")
if kind == "action_plan":
@@ -1428,7 +1485,7 @@ async def _complete_builtin_run(
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):
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:
@@ -35,9 +35,10 @@ from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/roundtable-chains", tags=["roundtable-chains"])
# Seat-count bounds mirror the roundtable selection bounds (2 ≤ N ≤ 8).
# Stored business chains support up to 20 seats. The frontend may choose a
# lower deployment-specific cap, but must never exceed this server contract.
MIN_SEATS = 2
MAX_SEATS = 8
MAX_SEATS = 20
# ── schemas ────────────────────────────────────────────────────────────────
@@ -183,7 +184,7 @@ def _get_store(request: Request):
def _validate_seats(seats: list[ChainSeatSchema]) -> None:
"""Enforce 2 ≤ N ≤ 8 and reject duplicate agent ids within a chain."""
"""Enforce 2 ≤ N ≤ 20 and reject duplicate agent ids within a chain."""
if len(seats) < MIN_SEATS or len(seats) > MAX_SEATS:
raise HTTPException(
status_code=400,
@@ -0,0 +1,45 @@
"""Add configurable position-roundtable roles.
Revision ID: 20260729_01
Revises: 20260723_05
Create Date: 2026-07-29
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260729_01"
down_revision = "20260723_05"
branch_labels = None
depends_on = None
def _has_table(name: str) -> bool:
return name in inspect(op.get_bind()).get_table_names()
def upgrade() -> None:
if _has_table("position_roles"):
return
op.create_table(
"position_roles",
sa.Column("id", sa.String(length=128), primary_key=True),
sa.Column("name", sa.String(length=128), nullable=False, unique=True),
sa.Column("description", sa.String(length=512), nullable=False, server_default=""),
sa.Column("role_type", sa.String(length=32), nullable=False, server_default="standard"),
sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.true()),
sa.Column("is_builtin", sa.Boolean(), nullable=False, server_default=sa.false()),
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
sa.Column("updated_by", sa.String(length=64), nullable=True),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_position_roles_name", "position_roles", ["name"])
op.create_index("ix_position_roles_role_type", "position_roles", ["role_type"])
def downgrade() -> None:
if _has_table("position_roles"):
op.drop_table("position_roles")
@@ -15,18 +15,6 @@ its storage implementation lives in ``deerflow.runtime.events.store.db`` and
there is no matching entity directory.
"""
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
from deerflow.persistence.business_mapping.model import BusinessMappingRow
from deerflow.persistence.concurrency.model import ConcurrencySampleRow
from deerflow.persistence.dashboard_sessions.model import DashboardSessionRow
from deerflow.persistence.memory_config.model import UserMemoryConfigRow
from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.knowledge.models import (
KnowledgeEmbeddingRow,
KnowledgeEntityRow,
@@ -40,12 +28,26 @@ from deerflow.knowledge.models import (
KnowledgeSourceRow,
KnowledgeTagRow,
)
from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.business_mapping.model import BusinessMappingRow
from deerflow.persistence.concurrency.model import ConcurrencySampleRow
from deerflow.persistence.dashboard_sessions.model import DashboardSessionRow
from deerflow.persistence.embed_sessions.model import EmbedSessionRow
from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.persistence.light_apps.model import LightAppRow
from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
from deerflow.persistence.memory_config.model import UserMemoryConfigRow
from deerflow.persistence.menu_overrides.model import MenuOverrideRow
from deerflow.persistence.models.run_event import RunEventRow
from deerflow.persistence.notifications.model import NotificationRow
from deerflow.persistence.parallel_agent_tasks.model import ParallelAgentTaskRow
from deerflow.persistence.position_roles.model import PositionRoleRow
from deerflow.persistence.position_roundtable.model import (
PositionRoundtableNodeRow,
PositionRoundtableSessionRow,
@@ -54,16 +56,15 @@ from deerflow.persistence.positions.model import PositionRow, PositionTagBinding
from deerflow.persistence.recommended_questions.model import RecommendedQuestionRow
from deerflow.persistence.research_templates.model import ResearchTemplateRow
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.roundtable_diagnostics.model import RoundtableDiagnosticRow
from deerflow.persistence.roundtable_draft_shares.model import RoundtableDraftShareRow
from deerflow.persistence.roundtable_drafts.model import (
RoundtableDraftRow,
RoundtableRecommendHistoryRow,
)
from deerflow.persistence.roundtable_diagnostics.model import RoundtableDiagnosticRow
from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow
from deerflow.persistence.roundtable_task_drafts.model import RoundtableTaskDraftRow
from deerflow.persistence.run.model import RunRow
from deerflow.persistence.sensitive_words.model import SensitiveWordRow
from deerflow.persistence.scheduled_tasks.model import (
ScheduledTaskDeliveryProfileRow,
ScheduledTaskRow,
@@ -71,13 +72,13 @@ from deerflow.persistence.scheduled_tasks.model import (
ScheduledTaskSubscriptionRow,
SchedulerThreadRow,
)
from deerflow.persistence.sensitive_words.model import SensitiveWordRow
from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillFavoriteRow, SkillRow
from deerflow.persistence.task_buttons.model import TaskButtonRow
from deerflow.persistence.taskcop_tasks.model import TaskCopTaskRow
from deerflow.persistence.task_reports.model import TaskReportRow
from deerflow.persistence.tags.model import TagAssignmentRow, TagRow
from deerflow.persistence.task_buttons.model import TaskButtonRow
from deerflow.persistence.task_reports.model import TaskReportRow
from deerflow.persistence.taskcop_tasks.model import TaskCopTaskRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.embed_sessions.model import EmbedSessionRow
from deerflow.persistence.thread_shares.model import ThreadShareRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
from deerflow.persistence.user.model import UserRow
@@ -123,6 +124,7 @@ __all__ = [
"RoundtableJobRow",
"RoundtableRecommendHistoryRow",
"RoundtableTaskDraftRow",
"PositionRoleRow",
"RunEventRow",
"RunRow",
"ScheduledTaskDeliveryProfileRow",
@@ -0,0 +1,27 @@
"""Position-roundtable role configuration persistence.
This is deliberately separate from :mod:`deerflow.persistence.positions`.
The latter represents organisational positions and resource grants, while this
package describes the work views used inside the position-roundtable product.
"""
from __future__ import annotations
from deerflow.persistence.position_roles.base import PositionRoleStore
from deerflow.persistence.position_roles.memory import MemoryPositionRoleStore
from deerflow.persistence.position_roles.model import PositionRoleRow
from deerflow.persistence.position_roles.sql import PositionRoleRepository
__all__ = [
"MemoryPositionRoleStore",
"PositionRoleRepository",
"PositionRoleRow",
"PositionRoleStore",
"make_position_role_store",
]
def make_position_role_store(session_factory):
if session_factory is None:
return MemoryPositionRoleStore()
return PositionRoleRepository(session_factory)
@@ -0,0 +1,30 @@
"""Abstract storage for position-roundtable role configuration."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class PositionRoleStore(ABC):
"""Global configuration shared by every position-roundtable session."""
@abstractmethod
async def list_roles(self) -> list[dict[str, Any]]:
"""Return roles in presentation order."""
raise NotImplementedError
@abstractmethod
async def replace_all(
self,
rows: list[dict[str, Any]],
*,
updated_by: str | None = None,
) -> list[dict[str, Any]]:
"""Atomically replace the editable role directory."""
raise NotImplementedError
@abstractmethod
async def ensure_seed(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Insert only missing built-in roles, without overwriting edits."""
raise NotImplementedError
@@ -0,0 +1,45 @@
"""In-memory fallback for position-roundtable role configuration."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from deerflow.persistence.position_roles.base import PositionRoleStore
def _normalize(row: dict[str, Any], updated_by: str | None = None) -> dict[str, Any]:
return {
"id": str(row["id"]),
"name": str(row.get("name") or ""),
"description": str(row.get("description") or ""),
"role_type": str(row.get("role_type") or "standard"),
"enabled": bool(row.get("enabled", True)),
"is_builtin": bool(row.get("is_builtin", False)),
"sort_order": int(row.get("sort_order", 0)),
"updated_by": updated_by,
"updated_at": datetime.now(UTC).isoformat(),
}
class MemoryPositionRoleStore(PositionRoleStore):
def __init__(self) -> None:
self._rows: dict[str, dict[str, Any]] = {}
async def list_roles(self) -> list[dict[str, Any]]:
return [dict(row) for row in sorted(self._rows.values(), key=lambda item: (item["sort_order"], item["id"]))]
async def replace_all(self, rows: list[dict[str, Any]], *, updated_by: str | None = None) -> list[dict[str, Any]]:
self._rows = {
str(row["id"]): _normalize(row, updated_by)
for row in rows
if row.get("id")
}
return await self.list_roles()
async def ensure_seed(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
for row in rows:
row_id = str(row.get("id") or "")
if row_id and row_id not in self._rows:
self._rows[row_id] = _normalize(row)
return await self.list_roles()
@@ -0,0 +1,37 @@
"""ORM model for configurable position-roundtable roles."""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class PositionRoleRow(Base):
"""A work-view role in the position-roundtable product.
``role_type`` is intentionally small and stable: ``standard`` denotes a
normal business-chain position; ``intent`` denotes the single special
position responsible for initial intent identification and closing-agent
entry. It is a configuration concern, not an organisation/RBAC position.
"""
__tablename__ = "position_roles"
id: Mapped[str] = mapped_column(String(128), primary_key=True)
name: Mapped[str] = mapped_column(String(128), unique=True, index=True)
description: Mapped[str] = mapped_column(String(512), default="")
role_type: Mapped[str] = mapped_column(String(32), default="standard", index=True)
enabled: Mapped[bool] = mapped_column(Boolean, default=True)
is_builtin: Mapped[bool] = mapped_column(Boolean, default=False, server_default="0")
sort_order: Mapped[int] = mapped_column(Integer, default=0)
updated_by: Mapped[str | None] = mapped_column(String(64))
updated_at: Mapped[datetime] = mapped_column(
BeijingDateTime(),
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
@@ -0,0 +1,66 @@
"""SQLAlchemy storage for position-roundtable role configuration."""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.position_roles.base import PositionRoleStore
from deerflow.persistence.position_roles.model import PositionRoleRow
def _to_dict(row: PositionRoleRow) -> dict[str, Any]:
data = row.to_dict()
if isinstance(data.get("updated_at"), datetime):
data["updated_at"] = data["updated_at"].isoformat()
return data
def _row_from(data: dict[str, Any], updated_by: str | None = None) -> PositionRoleRow:
return PositionRoleRow(
id=str(data["id"]),
name=str(data.get("name") or ""),
description=str(data.get("description") or ""),
role_type=str(data.get("role_type") or "standard"),
enabled=bool(data.get("enabled", True)),
is_builtin=bool(data.get("is_builtin", False)),
sort_order=int(data.get("sort_order", 0)),
updated_by=updated_by,
)
class PositionRoleRepository(PositionRoleStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
async def list_roles(self) -> list[dict[str, Any]]:
statement = select(PositionRoleRow).order_by(PositionRoleRow.sort_order.asc(), PositionRoleRow.id.asc())
async with self._sf() as session:
rows = (await session.execute(statement)).scalars()
return [_to_dict(row) for row in rows]
async def replace_all(self, rows: list[dict[str, Any]], *, updated_by: str | None = None) -> list[dict[str, Any]]:
# Keep the final occurrence of duplicate ids so malformed UI payloads
# cannot violate the primary-key constraint.
by_id = {str(row["id"]): row for row in rows if row.get("id")}
async with self._sf() as session:
await session.execute(delete(PositionRoleRow))
created = [_row_from(row, updated_by) for row in by_id.values()]
session.add_all(created)
await session.commit()
for row in created:
await session.refresh(row)
return [_to_dict(row) for row in sorted(created, key=lambda item: (item.sort_order, item.id))]
async def ensure_seed(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Idempotently insert new built-ins without changing operator edits."""
async with self._sf() as session:
existing_ids = set((await session.execute(select(PositionRoleRow.id))).scalars())
inserts = [_row_from(row) for row in rows if row.get("id") and str(row["id"]) not in existing_ids]
if inserts:
session.add_all(inserts)
await session.commit()
return await self.list_roles()
@@ -38,7 +38,12 @@ class _ChainStore:
return self.chain
def _request(role: str, chain: dict[str, object] | None = None):
def _request(
role: str,
chain: dict[str, object] | None = None,
*,
email: str = "lqq@example.com",
):
return SimpleNamespace(
app=SimpleNamespace(
state=SimpleNamespace(
@@ -46,14 +51,29 @@ def _request(role: str, chain: dict[str, object] | None = None):
roundtable_chain_store=_ChainStore(chain),
)
),
state=SimpleNamespace(user=SimpleNamespace(id="admin-id", system_role=role)),
state=SimpleNamespace(user=SimpleNamespace(id="admin-id", email=email, system_role=role)),
)
@pytest.mark.asyncio
async def test_only_administrators_can_update_global_business_mappings() -> None:
async def test_only_administrators_can_update_non_8bf_business_mappings() -> None:
request = _request("user")
with pytest.raises(HTTPException) as error:
await business_mapping_router.update_mapping(
request,
"6BF",
business_mapping_router.BusinessMappingUpdateRequest(chain_id="chain-8bf"),
)
assert error.value.status_code == 403
assert request.app.state.business_mapping_store.calls == []
@pytest.mark.asyncio
async def test_only_lqq_can_update_the_8bf_business_mapping() -> None:
request = _request("admin", email="admin@example.com")
with pytest.raises(HTTPException) as error:
await business_mapping_router.update_mapping(
request,
@@ -65,6 +85,20 @@ async def test_only_administrators_can_update_global_business_mappings() -> None
assert request.app.state.business_mapping_store.calls == []
@pytest.mark.asyncio
async def test_lqq_can_update_the_8bf_business_mapping_without_admin_role() -> None:
request = _request("user", email="lqq@example.com")
response = await business_mapping_router.update_mapping(
request,
"8BF",
business_mapping_router.BusinessMappingUpdateRequest(),
)
assert response.business_code == "8BF"
assert request.app.state.business_mapping_store.calls == [("8BF", {})]
@pytest.mark.asyncio
async def test_global_mapping_requires_a_public_chain_and_canonicalizes_its_title() -> None:
request = _request("admin", {"id": "chain-8bf", "title": "岗位协同链", "is_public": True})
@@ -80,9 +114,7 @@ async def test_global_mapping_requires_a_public_chain_and_canonicalizes_its_titl
assert response.chain_id == "chain-8bf"
assert response.chain_title == "岗位协同链"
assert request.app.state.business_mapping_store.calls == [
("8BF", {"chain_id": "chain-8bf", "chain_title": "岗位协同链"})
]
assert request.app.state.business_mapping_store.calls == [("8BF", {"chain_id": "chain-8bf", "chain_title": "岗位协同链"})]
@pytest.mark.asyncio
@@ -9,8 +9,8 @@ import pytest_asyncio
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.gateway.app import create_app
from app.gateway import services as gateway_services
from app.gateway.app import create_app
from app.gateway.routers import intent as intent_router
from app.gateway.routers import position_roundtable as position_roundtable_router
from deerflow.persistence.base import Base
@@ -50,6 +50,42 @@ def test_position_roundtable_router_has_intranet_compatibility_prefix():
assert position_roundtable_router.router.prefix == "/position-roundtable"
@pytest.mark.asyncio
async def test_intent_init_creates_thread_directly_without_loopback(monkeypatch):
"""Step 1 initialization must not depend on the deployed Gateway port."""
captured: dict[str, object] = {}
monkeypatch.setattr(intent_router, "ensure_roundtable_functional_agents", lambda: None)
async def fake_create_thread_direct(request, *, metadata):
captured["request"] = request
captured["metadata"] = metadata
return "intent-thread-direct"
monkeypatch.setattr(intent_router, "_create_thread_direct", fake_create_thread_direct)
request = SimpleNamespace()
result = await intent_router.init_intent(
request,
{"task_context": {"task": {"id": "task-001"}}},
)
assert result == {
"status": "success",
"agent_id": intent_router.INTENT_AGENT_ID,
"thread_id": "intent-thread-direct",
}
assert captured["request"] is request
assert captured["metadata"] == {
**intent_router._ROUNDTABLE_THREAD_METADATA,
"taskId": "task-001",
"position_roundtable_task_id": "task-001",
"position_task_id": "task-001:intelligence",
"position_id": "intelligence",
"position_roundtable_phase": "intent",
}
@pytest.mark.asyncio
async def test_session_activation_freezes_chain_and_creates_nodes_once(store):
created = await store.create_session(