feat(position-roundtable): add standalone position collaboration

This commit is contained in:
fengxiangdebaozi committed 2026-07-21 17:50:24 +08:00
1 parent b93b82fcaf
commit 85e119db08
22 files changed
+3494 -201

No files matched your search

@@ -0,0 +1,508 @@
# 岗位智能体协同页面 · 分步实现计划
> 页面:`/page/strategy/qa/position-roundtable`
>
> 核心模式:**不使用总控智能体**。情报分析岗先完成意图识别;其他岗位直接与业务链中分配给本岗位的智能体问答。每个节点独立保存对话和产物,并在问答时自动获得上游节点的有效结果。
---
## 0. 已确认的产品决策
- 新功能使用独立路由、独立业务状态和独立后端存储,不修改原多智能体会商页面的运行逻辑。
- 岗位先使用前端固定枚举;下拉框、选择器等表单控件使用 `tdesign-react`。
- **情报分析岗**负责任务意图识别,复用现有 `useStep1Intent`、`Step1Panel` 和意图接口。
- 其他岗位不经过总控派活,直接使用节点配置的 `agent_id` 调用对应智能体。
- 不调用原会商的 `multi-agent/init`、总控循环、后台圆桌作业和自动收口流程。
- 每个业务链节点对应一条独立智能体对话线程;同一节点可以在该线程内多轮追问。
- 当前岗位只能向分配给本岗位的节点提问;其他节点允许查看,但输入框只读。
- 节点问答时自动获得任务、意图和已完成上游节点的问答结果与产物。
- 左侧流程图从上到下排列,节点可点击,画布支持鼠标拖动;连接线保留但不显示依赖文字。
- 右侧保留“智能体配置 / 分析结果”两个标签页。
### 0.1 默认实现口径
以下口径在产品未另行调整前作为实现默认值:
- 业务链上一阶段的所有节点至少产生一次有效回答后,下一阶段节点进入 `ready` 状态。
- 同一阶段的节点互不依赖,可以分别问答。
- 上游节点产生新回答或新产物后,已经执行过的下游节点标记为 `stale`,但不删除历史。
- “分析结果”同时收录智能体最终回答和文件产物;文件产物优先展示。
- 第一版岗位下拉只是页面工作视角,不代表真实用户权限;后续接入用户岗位关系后再启用服务端岗位授权。
- 一个外部任务允许创建多个岗位协同会话;页面默认打开最近更新的一条。
### 0.2 待业务最终确认
- 固定岗位的正式编码和名称清单。
- 情报分析岗完成意图识别后,是否还允许承担业务链节点问答。
- 下游解锁采用“上游首次有效回答”还是增加人工“提交完成”动作。
- 新任务接口提供的字段、外部任务 ID 和页面进入方式。
- 岗位是否最终与登录用户绑定,以及一名用户是否允许属于多个岗位。
---
## 1. 目标与非目标
### 1.1 本期目标
- 为业务链席位增加岗位分配能力,且只在从岗位协同页面进入配置时显示。
- 为岗位协同任务提供可刷新、可恢复、可跨页面重新打开的持久化会话。
- 按“任务 → 意图 → 业务链阶段 → 节点智能体”组织对话和产物。
- 当前岗位仅展示和调用本岗位被分配的智能体。
- 点击流程图节点可查看该节点完整问答和产物。
- 下游智能体自动获得上游结果,不要求用户手工复制粘贴。
- 保持现有多智能体会商、任务深链、普通智能体聊天行为不变。
### 1.2 本期非目标
- 不保留总控智能体,不做总控派活、共识判断或自动多轮研讨。
- 不把岗位协同记录写入 `roundtable_drafts` 或 `roundtable_jobs`。
- 不自动生成最终综合报告;后续可增加按需调用总结智能体的“生成汇总”功能。
- 第一版不实现复杂条件分支、人工改线或动态跳转,只遵循现有业务链阶段顺序。
- 第一版不实现完整组织权限系统;真实岗位授权作为独立阶段接入。
---
## 2. 目标业务流程
1. 页面根据外部任务参数创建或恢复一条岗位协同会话。
2. 情报分析岗查看任务信息并选择业务链。
3. 情报分析岗与意图识别智能体问答,确认结构化任务意图。
4. 系统保存任务、意图和业务链快照,初始化所有业务链节点状态。
5. 用户切换到其他岗位,只看到分配给当前岗位的可问答智能体。
6. 用户点击流程图节点:
- 本岗位节点:加载或创建该节点的智能体线程,允许问答;
- 其他岗位节点:加载已有线程和产物,只读查看;
- 尚未解锁节点:展示依赖提示,不允许发送。
7. 每次发送问题前,系统构建隐藏上下文并注入当前智能体。
8. 智能体回答完成后保存节点最新回答、产物清单和版本号,并重新计算下游状态。
9. 右侧“分析结果”按时间倒序聚合全部节点回答和产物。
---
## 3. 总体技术方案
### 3.1 复用范围
| 能力 | 方案 |
|------|------|
| 意图识别 | 复用 `useStep1Intent`、`Step1Panel`、`/api/intent/*` |
| 智能体问答 | 复用 `useThreadStream` 和现有 Agent Chat 消息/输入组件 |
| 消息存储 | 复用 LangGraph thread/checkpoint,不新增消息表 |
| 文件产物 | 复用 `/api/threads/{thread_id}/artifacts` |
| 隐藏上下文 | 复用 `task_external_context` 和 `TaskExternalContextMiddleware` |
| 智能体详情 | 复用 `AgentDetailPanel` |
| 业务链 | 扩展现有 `roundtable_chains.seats` JSON |
| 页面业务状态 | 新增岗位协同 Session/Node 存储 |
### 3.2 明确不复用的圆桌能力
- `initMultiAgent`
- `useStep2Orchestration`
- `roundtable-coordinator`
- `roundtable_jobs`
- 圆桌共识百分比与最大循环轮次
- 自动 Step 3 收口和报告绘制
### 3.3 目录建议
```text
frontend-web/src/position-roundtable/
├── api/
│ ├── sessions.ts
│ └── handoff.ts
├── components/
│ ├── PositionTaskPanel.tsx
│ ├── PositionChainFlow.tsx
│ ├── PositionAgentChat.tsx
│ ├── PositionAgentList.tsx
│ └── PositionResults.tsx
├── hooks/
│ ├── usePositionSession.ts
│ ├── usePositionNodeChat.ts
│ └── usePositionHandoff.ts
├── lib/
│ ├── positions.ts
│ ├── chain-nodes.ts
│ └── types.ts
└── pages/
└── PositionRoundtablePage.tsx
```
后端建议新增:
```text
offline-backend-20260512/backend/
├── app/gateway/routers/position_roundtable.py
└── packages/harness/deerflow/persistence/
├── position_roundtable_sessions/
└── position_roundtable_nodes/
```
---
## 4. 数据模型
### 4.1 业务链席位扩展
在现有 `ChainSeat` 中增加:
```ts
interface ChainSeat {
agent_id: string;
name: string;
description?: string;
position_id?: string | null;
}
```
后端 `ChainSeatSchema` 同步增加可选 `position_id`。席位本身存储在 JSON 中,因此无需新增数据库列,但必须确保:
- 创建、读取、更新接口都保留该字段;
- 普通编辑模式虽然不显示岗位控件,保存时也不能删除已有岗位字段;
- 老链条没有该字段时按“未配置岗位”处理;
- 原圆桌编排忽略该字段,行为保持不变。
### 4.2 岗位协同会话表
`position_roundtable_sessions` 建议字段:
| 字段 | 含义 |
|------|------|
| `id` | 岗位协同会话 ID |
| `external_task_id` | 新任务系统提供的任务 ID,可空 |
| `task_snapshot` | 任务信息 JSON 快照 |
| `chain_id` | 来源业务链 ID |
| `chain_snapshot` | 启动时业务链完整快照 |
| `intent_thread_id` | 意图识别线程 ID |
| `intent_snapshot` | `useStep1Intent` 快照及结构化意图 |
| `status` | `intent_pending / active / completed / archived` |
| `created_by` | 创建人 |
| `created_at` | 创建时间 |
| `updated_at` | 更新时间 |
### 4.3 岗位协同节点表
`position_roundtable_nodes` 建议字段:
| 字段 | 含义 |
|------|------|
| `id` | 节点记录 ID |
| `session_id` | 所属岗位协同会话 |
| `node_key` | 会话内稳定节点键 |
| `stage_index` | 所属业务链阶段 |
| `seat_index` | 阶段内顺序 |
| `agent_id` | 对应智能体 |
| `position_id` | 分配岗位 |
| `thread_id` | 对应智能体聊天线程,可空 |
| `status` | `locked / ready / running / done / stale / error` |
| `latest_answer` | 最新有效回答快照 |
| `artifact_manifest` | 当前产物清单 JSON |
| `revision` | 本节点结果版本号 |
| `upstream_revision` | 最近一次问答使用的上游版本摘要 |
| `created_at` | 创建时间 |
| `updated_at` | 更新时间 |
唯一约束:`(session_id, node_key)`。
### 4.4 Thread metadata
节点线程创建后写入:
```json
{
"thread_type": "position_roundtable",
"position_session_id": "session-id",
"position_node_id": "node-id",
"position_id": "position-code",
"chain_id": "chain-id",
"agent_id": "agent-id"
}
```
Thread 继续保存真实消息;节点表只保存业务索引、状态和便于列表展示的快照。
---
## 5. API 规划
建议前缀:`/api/position-roundtable`。
### 5.1 会话接口
| 方法 | 路径 | 用途 |
|------|------|------|
| `POST` | `/sessions` | 根据任务创建岗位协同会话 |
| `GET` | `/sessions/{id}` | 加载会话、意图和全部节点状态 |
| `GET` | `/sessions?external_task_id=` | 查询任务下的会话历史 |
| `PUT` | `/sessions/{id}/task` | 更新尚未启动的任务快照 |
| `PUT` | `/sessions/{id}/intent` | 保存意图线程和意图快照 |
| `POST` | `/sessions/{id}/activate` | 冻结链条快照并初始化节点 |
| `POST` | `/sessions/{id}/archive` | 归档会话,不删除线程和产物 |
### 5.2 节点接口
| 方法 | 路径 | 用途 |
|------|------|------|
| `GET` | `/sessions/{id}/nodes/{node_key}` | 获取节点、线程、状态和产物 |
| `POST` | `/sessions/{id}/nodes/{node_key}/thread` | 幂等创建节点智能体线程 |
| `POST` | `/sessions/{id}/nodes/{node_key}/context` | 构建本轮隐藏上游上下文 |
| `POST` | `/sessions/{id}/nodes/{node_key}/complete-turn` | 回答结束后同步回答、产物和版本 |
| `GET` | `/sessions/{id}/results` | 聚合全部岗位分析结果 |
### 5.3 幂等与并发要求
- 节点线程接口必须幂等,同一节点并发请求只能创建一个有效 thread。
- `complete-turn` 使用节点 `revision` 做乐观锁,避免旧请求覆盖新结果。
- 会话激活后不允许直接替换链条;需要重新建会话或显式“基于新链条重建”。
- 删除业务链不能删除已有会话中的链条快照。
---
## 6. 上游结果交接规则
### 6.1 上游节点计算
- 线性链:当前节点之前的所有已完成节点均为上游。
- 分层链:只读取当前阶段之前的已完成阶段;同阶段节点不互相引用。
- 注入顺序按 `stage_index`、`seat_index` 排列,确保每次构建稳定。
### 6.2 隐藏上下文结构
每次发送问题前,通过 `task_external_context` 注入:
```json
{
"kind": "position_roundtable_handoff",
"task": {},
"intent": {},
"current_node": {},
"upstream_nodes": [
{
"agent_id": "...",
"position_id": "...",
"latest_answer": "...",
"artifacts": []
}
],
"upstream_revision": "..."
}
```
### 6.3 产物传递
- Markdown、TXT、JSON、CSV、YAML 等文本产物:读取正文并按总字符上限注入。
- PDF、Word、Excel、图片等非纯文本产物:复制到当前节点线程的只读上游资料目录,并在上下文中提供可读虚拟路径。
- 单文件和总上下文均设置大小限制;发生截断时在上下文中标记 `truncated=true`。
- 当前节点生成的文件仍写入自己的 `/mnt/user-data/outputs/`,不写回上游线程。
### 6.4 上游更新传播
1. 节点完成一次回答后 `revision + 1`。
2. 系统重新计算所有下游节点的上游版本摘要。
3. 已问答下游节点的摘要不一致时,将状态改为 `stale`。
4. 用户再次向该节点提问后,使用最新上下文并把状态恢复为 `done`。
---
## 7. 分阶段实施计划
### 阶段 0:边界固定与基线测试
目标:明确新旧功能边界,给后续改造建立回归保护。
- [x] 固定岗位枚举及意图岗位编码(当前为 `intelligence` / 情报分析岗)。
- [x] 固定新任务数据适配接口:当前使用可替换的模拟任务源,后续只需替换 `mock-task-source.ts`。
- [ ] 为原圆桌路由、岗位协同路由分别增加基本渲染测试。
- [ ] 记录原圆桌业务链创建、编辑、运行的基线行为。
- [ ] 确认岗位协同模块不导入 `useStep2Orchestration` 或 `initMultiAgent`。
完成标准:原会商页面无行为变化;岗位协同页面仍可独立打开。
### 阶段 1:业务链岗位配置
目标:业务链节点能够持久化岗位,但普通业务链编辑体验不受影响。
- [x] `ChainSeat`、前端 API 映射增加 `position_id`。
- [x] 后端 `ChainSeatSchema` 增加 `position_id` 并补充长度校验。
- [x] 业务链编辑器识别 `mode=position`。
- [x] 岗位模式下为每个节点增加 TDesign 岗位选择器。
- [ ] 增加“未配置岗位”提示、岗位图例和保存校验。
- [x] 普通模式隐藏岗位控件,但完整保留原岗位字段。
- [x] 岗位协同页面增加“配置岗位”入口和返回参数。
- [ ] 增加老链条、普通保存、岗位保存的前后端测试。
完成标准:从岗位页面进入时可以配置岗位;从原入口进入时看不到岗位控件;两种入口交替保存不丢字段。
### 阶段 2:岗位协同持久化底座
目标:任务、意图、链条快照和节点状态可以跨刷新恢复。
- [ ] 新增 Session/Node ORM、Repository 和数据库迁移。
- [ ] 新增会话创建、查询、更新意图、激活和归档接口。
- [ ] 激活时校验链条、岗位配置和意图结果。
- [ ] 激活时根据链条快照一次性生成全部 Node 记录。
- [ ] 新增前端 `sessions.ts` API 和 `usePositionSession`。
- [ ] 用服务端 Session 替换 `POSITION_TASK_DRAFT_KEY` 主路径。
- [ ] 页面刷新后恢复任务、意图、选中链条、节点状态和当前会话。
完成标准:刷新浏览器后不丢任务和意图;业务链后续编辑不改变已启动会话。
### 阶段 3:情报分析岗意图流程
目标:把情报分析岗固定为岗位协同的入口阶段。
- [ ] 将当前“组织筹划岗负责意图”调整为“情报分析岗负责意图”。
- [ ] 仅意图岗位渲染 `Step1Panel` 和意图输入能力。
- [ ] 将任务信息自动作为首轮隐藏背景或开场消息提供给意图智能体。
- [ ] 意图变化时自动保存 `intent_snapshot`。
- [ ] 增加“确认意图并启动业务链”动作。
- [ ] 启动前校验业务链已选、节点岗位已配置、意图已完成。
- [ ] 其他岗位在未启动时显示只读等待状态。
完成标准:情报分析岗确认意图后成功生成节点;其他岗位不能修改意图。
### 阶段 4:岗位直连智能体问答
目标:去除总控,当前岗位直接与自己节点的智能体多轮问答。
- [ ] 抽取或封装可嵌入的 Agent Chat 主区,复用现有消息列表和输入框。
- [ ] 右侧智能体列表改为当前链条、当前岗位的交集。
- [ ] 点击智能体或流程节点时切换 `activeNodeKey`。
- [ ] 节点首次发送前幂等创建 thread,`assistantId = node.agent_id`。
- [ ] 写入岗位协同 thread metadata。
- [ ] 本岗位且状态为 `ready/running/done/stale` 时允许输入。
- [ ] 其他岗位节点和 `locked` 节点使用同一消息列表只读展示。
- [ ] 移除流程图“总控协调”节点,首部展示“情报分析岗 · 意图识别”。
完成标准:页面不调用任何总控接口;不同节点对话互不串线;岗位不能向未分配节点发送问题。
### 阶段 5:节点查看与流程状态
目标:流程图成为对话、产物和状态的统一导航。
- [ ] 节点增加 `locked/ready/running/done/stale/error` 可视状态。
- [ ] 当前节点增加选中态,并与中间聊天和右侧智能体详情同步。
- [ ] 点击任意已有节点加载完整消息历史。
- [ ] 节点运行时防止重复发送并支持停止当前生成。
- [ ] 计算上一阶段完成情况并自动解锁下一阶段。
- [ ] 画布拖动时阻止节点误点击,双击空白区域复位。
- [ ] 保持左侧滚动条、固定画布高度和父容器裁剪。
完成标准:用户能从流程图定位任意节点,并准确区分可问答、只读、待上游和结果过期状态。
### 阶段 6:上游问答与产物自动交接
目标:下游智能体真正获得上游信息,而不只是页面能够查看。
- [ ] 新增节点上下文构建接口。
- [ ] 读取上游节点最新有效回答和结构化意图。
- [ ] 读取上游文本产物并执行字符数限制与截断标记。
- [ ] 为非文本产物实现安全复制或只读挂载。
- [ ] 发送消息时通过 `task_external_context` 注入上下文。
- [ ] 回答结束后同步 `latest_answer`、产物清单和 `revision`。
- [ ] 上游更新时传播 `stale` 状态。
- [ ] 增加上下文顺序、截断、文件隔离和版本传播测试。
完成标准:下游回答能够引用上游真实结论;不需要用户复制粘贴;不同 thread 的文件不串写。
### 阶段 7:分析结果聚合
目标:右侧结果页展示全部岗位、节点的真实结果。
- [ ] 新增 Session 结果聚合接口。
- [ ] 回答结论生成“问答结果”记录。
- [ ] 每个实际文件生成独立产物记录。
- [ ] 卡片显示智能体名称、产物名称、岗位、生成时间和状态。
- [ ] 支持按岗位、节点、产物类型筛选。
- [ ] 点击结果跳转并高亮对应流程节点。
- [ ] 文本产物复用现有预览组件,其他文件复用下载地址。
- [ ] `stale` 结果显示“上游已更新”提示,不自动隐藏旧结果。
完成标准:结果数量、节点文件清单和实际线程产物一致;点击结果能回到对应对话。
### 阶段 8:嵌入、权限与恢复
目标:满足外部系统嵌入及多人岗位使用场景。
- [ ] 支持从 URL/外部任务接口恢复 `external_task_id` 和 Session。
- [ ] 复用现有 `embed=1` 模式隐藏全局侧边栏和导航外壳。
- [ ] 深浅主题下核对左、中、右三栏背景和 TDesign 控件。
- [ ] 明确岗位下拉在嵌入系统中的来源和默认值。
- [ ] 若接入真实用户岗位,新增 Session 成员/岗位授权模型。
- [ ] 服务端校验当前用户是否能向目标节点发送问题。
- [ ] 定义会话归档、任务关闭和只读历史规则。
完成标准:嵌入刷新不丢会话;不同岗位用户只能写自己的节点,同时可以按规则查看上游节点。
### 阶段 9:测试、回归与发布
目标:新功能可上线,原功能零回归。
- [ ] 前端类型检查、Chrome 109 CSS 兼容检查和生产构建通过。
- [ ] 后端 Repository、Router、上下文构建和并发创建测试通过。
- [ ] 覆盖线性链、分层链、同阶段并行节点、无岗位配置和智能体被删除场景。
- [ ] 覆盖刷新恢复、重复点击、新旧上游版本、文件过大和产物为空场景。
- [ ] 回归原业务链编辑、原多智能体会商和普通 Agent Chat。
- [ ] 使用真实数据完成浅色、深色、嵌入三种视图的截图对比。
- [ ] 编写接口说明、部署迁移说明和回滚说明。
完成标准:自动化测试通过;真实链条端到端走通;原页面和原接口行为不变。
---
## 8. 建议的提交拆分
每个提交保持可独立审查和回滚:
1. `feat(chain): add optional position assignment to seats`
2. `feat(position-roundtable): add session persistence`
3. `feat(position-roundtable): persist intent workflow`
4. `feat(position-roundtable): add direct agent node chats`
5. `feat(position-roundtable): add interactive chain navigation`
6. `feat(position-roundtable): inject upstream handoff context`
7. `feat(position-roundtable): aggregate node artifacts`
8. `test(position-roundtable): cover workflow and regressions`
---
## 9. 验收清单
- [ ] 岗位协同页面与原会商页面使用不同路由和不同业务存储。
- [ ] 页面运行过程中没有创建或调用总控智能体。
- [ ] 情报分析岗能够完成并持久化意图识别。
- [ ] 业务链岗位配置只在岗位模式入口显示。
- [ ] 当前岗位的智能体列表只包含被分配的链条节点。
- [ ] 非本岗位节点可查看但不能发送消息。
- [ ] 点击节点可以恢复对应的完整聊天记录和文件产物。
- [ ] 下游智能体能够获得上游问答结果和产物内容。
- [ ] 上游更新后下游结果能够提示过期。
- [ ] 刷新、重新进入和嵌入模式下数据不丢失。
- [ ] 原业务链配置、原多智能体会商和普通智能体问答均无回归。
---
## 10. 第一轮推荐实施范围
第一轮建议完成阶段 0~4,形成可实际操作的最小闭环:
```text
岗位配置 → 创建会话 → 情报分析岗确认意图 → 生成业务链节点
→ 切换岗位 → 点击本岗位节点 → 直接调用对应智能体问答 → 刷新恢复
```
第二轮完成阶段 5~7,把流程图查看、上游自动交接和结果聚合补齐;第三轮处理真实岗位权限、外部任务接口和上线回归。
## 11. 当前实现状态(2026-07-21)
已完成的首个可运行闭环:
- 业务链席位已支持可选 `position_id`;只有从岗位协同页面进入业务链编辑器时才显示 TDesign 岗位选择器,普通会商配置会完整保留该字段且不显示该控件。未配置岗位的席位会在业务链选择区标明,并在启动前阻止进入节点执行。
- 新任务入口已先以 `mock-task-source.ts` 提供 3 条模拟任务;页面通过 `taskId` 识别任务、自动恢复该任务最近一条未归档会话,支持“新建会话”和会话归档。正式任务接口确认后只替换这一数据源即可。
- 新增独立的 `position_roundtable_sessions` / `position_roundtable_nodes` 持久化、迁移和 `/api/position-roundtable` 接口;它不使用原会商的草稿、总控、作业或自动收口流程。
- 情报分析岗复用既有意图识别对话。确认意图后会冻结业务链、创建节点、将会话 ID 写入新页面 URL,刷新可恢复节点与其已绑定的标准聊天线程。
- 业务链节点直接调用其配置的 `agent_id`,复用原消息列表和输入框。节点问答会注入任务、意图、已完成上游的回答和产物引用;同阶段全部完成后才解锁下一阶段,上游重答会把下游标记为 `stale`。
- 节点交接接口会额外读取上游节点自身线程里的文本产物正文(Markdown、文本、JSON、CSV 等),按总量和单文件长度截断后注入下游智能体;二进制产物仍仅传递安全引用,且不会跨线程写入或共享工作目录。
- 左侧流程图、右侧岗位智能体列表和结果卡片均以当前会话节点为准;跨岗位和未解锁节点保持只读。分析结果会把每条问答、每个实际文件分别列卡,支持按岗位、节点、产物类型筛选,并对 `stale` 结果标明“上游已更新”。
- 选中流程节点或分析结果卡片后,中间对话区会同时展示该节点的产物清单;点击文件复用既有预览组件,读取的始终是该节点绑定线程下的产物。归档会话保持完整可读,但前后端均拒绝继续写入、绑定节点或生成上下文。
仍需在产品/组织接口确定后替换:新任务系统的正式字段与进入方式、真实用户岗位授权、上游非文本产物的安全挂载或复制,以及使用真实登录数据的完整端到端回归。
@@ -0,0 +1,264 @@
/** Client for the independent position-collaboration Session / Node API.
*
* Messages themselves stay in LangGraph threads. These endpoints only persist
* task/intent snapshots, frozen chain topology, node states and indexes that
* let the UI recover the correct thread after a refresh.
*/
import { apiFetch } from "@/core/api/fetch-client";
import { getBackendBaseURL } from "@/core/config";
export type PositionNodeStatus =
| "locked"
| "ready"
| "running"
| "done"
| "stale"
| "error";
export interface PositionRoundtableNode {
id: string;
sessionId: string;
nodeKey: string;
stageIndex: number;
seatIndex: number;
agentId: string;
positionId: string | null;
threadId: string | null;
status: PositionNodeStatus;
latestAnswer: string | null;
artifactManifest: Array<Record<string, unknown>>;
revision: number;
upstreamRevision: Record<string, unknown> | null;
createdAt?: string;
updatedAt?: string;
}
export interface PositionRoundtableSession {
id: string;
externalTaskId: string | null;
taskSnapshot: Record<string, unknown>;
chainId: string | null;
chainSnapshot: Record<string, unknown> | null;
intentThreadId: string | null;
intentSnapshot: Record<string, unknown> | null;
status: "intent_pending" | "active" | "completed" | "archived";
createdAt?: string;
updatedAt?: string;
nodes?: PositionRoundtableNode[];
}
interface ServerNode {
id: string;
session_id: string;
node_key: string;
stage_index: number;
seat_index: number;
agent_id: string;
position_id: string | null;
thread_id: string | null;
status: PositionNodeStatus;
latest_answer: string | null;
artifact_manifest?: Array<Record<string, unknown>>;
revision?: number;
upstream_revision?: Record<string, unknown> | null;
created_at?: string;
updated_at?: string;
}
interface ServerSession {
id: string;
external_task_id: string | null;
task_snapshot?: Record<string, unknown>;
chain_id: string | null;
chain_snapshot?: Record<string, unknown> | null;
intent_thread_id: string | null;
intent_snapshot?: Record<string, unknown> | null;
status: PositionRoundtableSession["status"];
created_at?: string;
updated_at?: string;
nodes?: ServerNode[];
}
const BASE = () => `${getBackendBaseURL()}/api/position-roundtable`;
async function readJson<T>(response: Response, label: string): Promise<T> {
if (!response.ok) {
const detail = await response.text().catch(() => "");
throw new Error(`${label} failed (${response.status}): ${detail}`);
}
return response.json() as Promise<T>;
}
function toNode(node: ServerNode): PositionRoundtableNode {
return {
id: node.id,
sessionId: node.session_id,
nodeKey: node.node_key,
stageIndex: node.stage_index,
seatIndex: node.seat_index,
agentId: node.agent_id,
positionId: node.position_id,
threadId: node.thread_id,
status: node.status,
latestAnswer: node.latest_answer,
artifactManifest: node.artifact_manifest ?? [],
revision: node.revision ?? 0,
upstreamRevision: node.upstream_revision ?? null,
createdAt: node.created_at,
updatedAt: node.updated_at,
};
}
function toSession(session: ServerSession): PositionRoundtableSession {
return {
id: session.id,
externalTaskId: session.external_task_id,
taskSnapshot: session.task_snapshot ?? {},
chainId: session.chain_id,
chainSnapshot: session.chain_snapshot ?? null,
intentThreadId: session.intent_thread_id,
intentSnapshot: session.intent_snapshot ?? null,
status: session.status,
createdAt: session.created_at,
updatedAt: session.updated_at,
nodes: session.nodes?.map(toNode),
};
}
export async function listPositionSessions(externalTaskId?: string): Promise<PositionRoundtableSession[]> {
const params = externalTaskId ? `?external_task_id=${encodeURIComponent(externalTaskId)}` : "";
const result = await readJson<ServerSession[]>(
await apiFetch(`${BASE()}/sessions${params}`),
"list position sessions",
);
return result.map(toSession);
}
export async function createPositionSession(input: {
externalTaskId?: string;
taskSnapshot: Record<string, unknown>;
}): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await apiFetch(`${BASE()}/sessions`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
external_task_id: input.externalTaskId || null,
task_snapshot: input.taskSnapshot,
}),
}),
"create position session",
);
return toSession(result);
}
export async function getPositionSession(sessionId: string): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await apiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}`),
"get position session",
);
return toSession(result);
}
export async function savePositionIntent(
sessionId: string,
input: { intentThreadId?: string | null; intentSnapshot: object | null },
): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await apiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/intent`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
intent_thread_id: input.intentThreadId || null,
intent_snapshot: input.intentSnapshot,
}),
}),
"save position intent",
);
return toSession(result);
}
export async function activatePositionSession(
sessionId: string,
chainId: string,
): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await apiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/activate`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ chain_id: chainId }),
}),
"activate position session",
);
return toSession(result);
}
export async function archivePositionSession(sessionId: string): Promise<PositionRoundtableSession> {
const result = await readJson<ServerSession>(
await apiFetch(`${BASE()}/sessions/${encodeURIComponent(sessionId)}/archive`, {
method: "POST",
}),
"archive position session",
);
return toSession(result);
}
export async function bindPositionNodeThread(
sessionId: string,
nodeKey: string,
threadId: string,
): Promise<PositionRoundtableNode> {
const result = await readJson<ServerNode>(
await apiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/thread`,
{
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ thread_id: threadId }),
},
),
"bind position node thread",
);
return toNode(result);
}
/** Fetch the bounded hidden task/intent/upstream handoff before one node turn. */
export async function getPositionNodeContext(
sessionId: string,
nodeKey: string,
): Promise<Record<string, unknown>> {
return readJson<Record<string, unknown>>(
await apiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/context`,
{ method: "POST" },
),
"build position node context",
);
}
export async function completePositionNodeTurn(
sessionId: string,
nodeKey: string,
input: {
latestAnswer: string;
artifactManifest?: Array<Record<string, unknown>>;
upstreamRevision?: Record<string, unknown> | null;
},
): Promise<PositionRoundtableNode> {
const result = await readJson<ServerNode>(
await apiFetch(
`${BASE()}/sessions/${encodeURIComponent(sessionId)}/nodes/${encodeURIComponent(nodeKey)}/complete-turn`,
{
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
latest_answer: input.latestAnswer,
artifact_manifest: input.artifactManifest ?? [],
upstream_revision: input.upstreamRevision ?? null,
}),
},
),
"complete position node turn",
);
return toNode(result);
}
@@ -0,0 +1,285 @@
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { Bot, FileText, LockKeyhole, Loader2 } from "lucide-react";
import { toast } from "sonner";
import type { PromptInputMessage } from "@/components/ai-elements/prompt-input";
import { InputBox } from "@/components/workspace/input-box";
import { ThreadContext } from "@/components/workspace/messages/context";
import { MessageList } from "@/components/workspace/messages";
import { useLocalSettings } from "@/core/settings";
import { useThreadStream } from "@/core/threads/hooks";
import { textOfMessage } from "@/core/threads/utils";
import { uuid } from "@/core/utils/uuid";
import {
bindPositionNodeThread,
completePositionNodeTurn,
getPositionNodeContext,
type PositionRoundtableNode,
} from "@/position-roundtable/api/sessions";
import { listSessionArtifacts } from "@/roundtable-planning/api/multi-agent";
import { ArtifactPreviewModal } from "@/roundtable-planning/components/ArtifactPreviewModal";
interface PositionNodeChatProps {
sessionId: string;
chainId: string | null;
node: PositionRoundtableNode;
currentPositionId: string;
sessionStatus: "intent_pending" | "active" | "completed" | "archived";
onNodeUpdate: (node: PositionRoundtableNode) => void;
}
interface NodeArtifact {
name: string;
path: string;
mimeType: string | null;
}
function readNodeArtifacts(manifest: Array<Record<string, unknown>>): NodeArtifact[] {
return manifest.flatMap((artifact) => {
const path = typeof artifact.path === "string" ? artifact.path : "";
if (!path) return [];
const name =
typeof artifact.name === "string" && artifact.name.trim()
? artifact.name
: path.split("/").filter(Boolean).at(-1) || "未命名产物";
const mimeType =
typeof artifact.mime_type === "string"
? artifact.mime_type
: typeof artifact.mimeType === "string"
? artifact.mimeType
: null;
return [{ name, path, mimeType }];
});
}
/**
* Direct single-agent chat for one frozen business-chain node.
*
* It deliberately uses the regular `useThreadStream` + `MessageList` +
* `InputBox` stack. The position Session merely records which existing thread
* belongs to the node; it is not a second message persistence mechanism.
*/
export function PositionNodeChat({
sessionId,
chainId,
node,
currentPositionId,
sessionStatus,
onNodeUpdate,
}: PositionNodeChatProps) {
const [settings, setSettings] = useLocalSettings();
const [boundThreadId, setBoundThreadId] = useState<string | null>(node.threadId);
const boundThreadIdRef = useRef<string | null>(node.threadId);
const draftThreadIdRef = useRef(node.threadId ?? uuid());
const upstreamRevisionRef = useRef<Record<string, unknown> | null>(null);
const [previewArtifact, setPreviewArtifact] = useState<NodeArtifact | null>(null);
const nodeArtifacts = useMemo(() => readNodeArtifacts(node.artifactManifest), [node.artifactManifest]);
const canWrite =
sessionStatus !== "archived" &&
node.positionId === currentPositionId &&
node.status !== "locked" &&
node.status !== "error";
useEffect(() => {
setBoundThreadId(node.threadId);
boundThreadIdRef.current = node.threadId;
}, [node.threadId]);
const bindCreatedThread = useCallback(
(threadId: string) => {
boundThreadIdRef.current = threadId;
setBoundThreadId(threadId);
void bindPositionNodeThread(sessionId, node.nodeKey, threadId)
.then(onNodeUpdate)
.catch((error) => {
toast.error(`节点对话绑定失败:${error instanceof Error ? error.message : String(error)}`);
});
},
[node.nodeKey, onNodeUpdate, sessionId],
);
const [thread, sendMessage] = useThreadStream({
threadId: boundThreadId,
assistantId: node.agentId,
context: { ...settings.context, agent_id: node.agentId },
threadMetadata: {
thread_type: "position_roundtable",
position_session_id: sessionId,
position_node_id: node.id,
position_id: node.positionId,
chain_id: chainId,
agent_id: node.agentId,
},
onStart: bindCreatedThread,
onFinish: (state) => {
const threadId = boundThreadIdRef.current;
const latestAi = [...(state.messages ?? [])]
.reverse()
.find((message) => message.type === "ai");
const latestAnswer = latestAi ? textOfMessage(latestAi) ?? "" : "";
if (!threadId || !latestAnswer.trim()) return;
void (async () => {
let artifactManifest: Array<Record<string, unknown>> = [];
try {
const artifacts = await listSessionArtifacts(threadId);
artifactManifest = artifacts.files.map((file) => ({
name: file.name,
path: file.path,
mime_type: file.mimeType,
modified_at: file.modifiedAt,
}));
} catch {
// Artifact listing is best-effort: a valid text answer must still be
// persisted and unlock downstream work even if files cannot be read.
}
try {
const updated = await completePositionNodeTurn(sessionId, node.nodeKey, {
latestAnswer,
artifactManifest,
upstreamRevision: upstreamRevisionRef.current,
});
onNodeUpdate(updated);
} catch (error) {
toast.error(`节点结果保存失败:${error instanceof Error ? error.message : String(error)}`);
}
})();
},
});
const handleSubmit = useCallback(
async (message: PromptInputMessage) => {
if (!canWrite) {
toast.info(
sessionStatus === "archived"
? "该会话已归档,仅可查看历史记录"
: node.status === "locked"
? "该节点尚未解锁"
: "当前岗位只能查看该节点对话",
);
return;
}
try {
const taskExternalContext = await getPositionNodeContext(sessionId, node.nodeKey);
const upstreamRevision = taskExternalContext.upstream_revision;
upstreamRevisionRef.current =
upstreamRevision && typeof upstreamRevision === "object" && !Array.isArray(upstreamRevision)
? (upstreamRevision as Record<string, unknown>)
: null;
await sendMessage(boundThreadIdRef.current ?? draftThreadIdRef.current, message, {
agent_id: node.agentId,
task_external_context: taskExternalContext,
});
} catch (error) {
toast.error(`发送失败:${error instanceof Error ? error.message : String(error)}`);
}
},
[canWrite, node.agentId, node.nodeKey, node.status, sendMessage, sessionId, sessionStatus],
);
const status = thread.error ? "error" : thread.isLoading ? "streaming" : "ready";
const displayThreadId = boundThreadId ?? `position-node-${node.id}`;
const readOnlyHint =
sessionStatus === "archived"
? "该会话已归档,仅可查看历史对话和产物"
: node.status === "locked"
? "等待上游岗位完成后解锁"
: node.positionId !== currentPositionId
? "当前岗位仅可查看该节点的对话和产物"
: "";
return (
<ThreadContext.Provider value={{ thread }}>
<div className="flex min-h-0 flex-1 flex-col">
<header className="flex h-12 shrink-0 items-center gap-2 border-b border-slate-100 px-4 dark:border-[#2b8ffe]/20">
<span className="flex size-7 items-center justify-center rounded-lg bg-primary/10 text-primary">
{node.status === "locked" ? <LockKeyhole className="size-3.5" /> : <Bot className="size-4" />}
</span>
<div className="min-w-0 flex-1">
<h2 className="truncate text-sm font-extrabold text-slate-900 dark:text-[#e8edf5]">
{node.agentId}
</h2>
<p className="truncate text-[10px] text-muted-foreground dark:text-[#8aa0c4]">
{readOnlyHint || `节点状态:${node.status}`}
</p>
</div>
{thread.isLoading && <Loader2 className="size-4 animate-spin text-primary" />}
</header>
{nodeArtifacts.length > 0 && (
<section className="shrink-0 border-b border-slate-100 bg-slate-50/70 px-4 py-2.5 dark:border-[#2b8ffe]/20 dark:bg-[#032650]/50">
<div className="mb-1.5 flex items-center gap-1.5 text-[11px] font-semibold text-slate-600 dark:text-[#c5cfe0]">
<FileText className="size-3.5 text-primary" />
节点产物
<span className="rounded-full bg-primary/10 px-1.5 py-0.5 text-[10px] text-primary">
{nodeArtifacts.length}
</span>
</div>
<div className="flex max-h-20 flex-wrap gap-1.5 overflow-y-auto pr-1">
{nodeArtifacts.map((artifact) => (
<button
key={artifact.path}
type="button"
title={`预览 ${artifact.name}`}
onClick={() => setPreviewArtifact(artifact)}
className="inline-flex max-w-full items-center gap-1 rounded-md border border-slate-200 bg-white px-2 py-1 text-left text-[11px] text-slate-600 transition hover:border-primary/45 hover:text-primary dark:border-[#2b8ffe]/35 dark:bg-[#0a315d]/80 dark:text-[#c5cfe0]"
>
<FileText className="size-3 shrink-0" />
<span className="max-w-48 truncate">{artifact.name}</span>
</button>
))}
</div>
</section>
)}
{boundThreadId ? (
<MessageList
className="min-h-0 flex-1"
threadId={boundThreadId}
thread={thread}
paddingBottom={24}
onEditResubmit={(text) => void handleSubmit({ text, files: [] })}
/>
) : (
<div className="flex min-h-0 flex-1 flex-col items-center justify-center px-8 text-center">
<span className="mb-3 flex size-11 items-center justify-center rounded-2xl bg-primary/10 text-primary">
<Bot className="size-5" />
</span>
<p className="text-sm font-semibold text-slate-800 dark:text-[#e8edf5]">
{canWrite ? "向该岗位智能体发起第一次问答" : "该节点尚无可查看的对话"}
</p>
<p className="mt-1 max-w-sm text-xs leading-5 text-muted-foreground dark:text-[#8aa0c4]">
{canWrite
? "系统会自动传入任务意图和已完成上游节点的问答结果。"
: readOnlyHint || "请选择已启动的岗位节点。"}
</p>
</div>
)}
<div className="shrink-0 border-t border-slate-100 px-4 py-3 dark:border-[#2b8ffe]/20">
<InputBox
className="w-full"
isNewThread={!boundThreadId}
threadId={displayThreadId}
assistantId={node.agentId}
status={status}
context={settings.context}
disabled={!canWrite}
hideAdminRecommendedQuestions
onContextChange={(context) => setSettings("context", context)}
onSubmit={(message) => void handleSubmit(message)}
onStop={() => void thread.stop()}
/>
</div>
{previewArtifact && boundThreadId && (
<ArtifactPreviewModal
threadId={boundThreadId}
path={previewArtifact.path}
name={previewArtifact.name}
mimeType={previewArtifact.mimeType}
onClose={() => setPreviewArtifact(null)}
/>
)}
</div>
</ThreadContext.Provider>
);
}
@@ -0,0 +1,72 @@
/**
* Temporary task source for the standalone position-collaboration workspace.
*
* The real task-system adapter should replace only this module. The page and
* Session API consume the normalized `PositionTask` shape, so they do not need
* to know whether data came from an external task platform or this local demo.
*/
export interface PositionTask {
id: string;
title: string;
description: string;
direction: string;
source: "mock";
}
export const MOCK_POSITION_TASKS: readonly PositionTask[] = [
{
id: "mock-position-task-001",
title: "重点区域态势研判与协同处置方案",
description:
"围绕重点区域近期态势,梳理风险信号、影响范围与处置约束,形成可执行的协同方案及评估建议。",
direction: "态势研判",
source: "mock",
},
{
id: "mock-position-task-002",
title: "重大活动保障资源统筹任务",
description:
"针对重大活动保障场景,识别关键保障目标、现有资源和跨部门协同难点,输出分阶段资源统筹方案。",
direction: "组织筹划",
source: "mock",
},
{
id: "mock-position-task-003",
title: "专项行动效果复盘与优化建议",
description:
"汇总专项行动过程信息与阶段产物,研判实际效果、剩余风险和优化路径,形成后续行动建议。",
direction: "综合评估",
source: "mock",
},
] as const;
export const DEFAULT_MOCK_POSITION_TASK_ID = MOCK_POSITION_TASKS[0].id;
export function getMockPositionTask(taskId?: string | null): PositionTask {
return (
MOCK_POSITION_TASKS.find((task) => task.id === taskId) ??
MOCK_POSITION_TASKS[0]
);
}
/** Normalizes a persisted Session task snapshot without exposing source internals. */
export function taskFromPositionSnapshot(
snapshot: Record<string, unknown> | null | undefined,
fallback: PositionTask,
): PositionTask {
if (!snapshot || Object.keys(snapshot).length === 0) return fallback;
const value = (keys: string[], defaultValue: string) => {
const found = keys
.map((key) => snapshot[key])
.find((item) => typeof item === "string" && item.trim());
return typeof found === "string" ? found : defaultValue;
};
return {
id: value(["id", "taskId", "externalTaskId"], fallback.id),
title: value(["title", "taskName", "name"], fallback.title),
description: value(["description", "taskContent", "overview", "content"], fallback.description),
direction: value(["direction", "taskDirection", "category"], fallback.direction),
source: "mock",
};
}
@@ -0,0 +1,39 @@
/**
* 岗位协同的第一版岗位目录。
*
* 当前岗位仅代表页面工作视角;后续接入组织岗位 / 权限系统时,保留这里的稳定 id,
* 只替换数据源即可。业务链条席位用 `position_id` 持久化这组 id。
*/
export const POSITION_OPTIONS = [
{
id: "intelligence",
name: "情报分析岗",
description: "识别任务意图、补齐约束并确认业务链条",
isIntentPosition: true,
},
{
id: "planning",
name: "组织筹划岗",
description: "承接上游信息并进行组织筹划",
isIntentPosition: false,
},
{
id: "solution",
name: "方案设计岗",
description: "承接上游结果并形成方案产物",
isIntentPosition: false,
},
{
id: "evaluation",
name: "综合评估岗",
description: "评估链条产物并提出综合建议",
isIntentPosition: false,
},
] as const;
export type PositionId = (typeof POSITION_OPTIONS)[number]["id"];
export const DEFAULT_POSITION_ID: PositionId = "intelligence";
export const positionNameById = (positionId: string | null | undefined): string =>
POSITION_OPTIONS.find((position) => position.id === positionId)?.name ?? "未分配岗位";
File diff suppressed because it is too large. Load diff
@@ -87,6 +87,8 @@ export interface Step1PanelProps {
onSkipQuestion: () => void;
onFollowUp: () => void;
onEnterRoundtable: () => void;
/** 保持旧文案默认不变;嵌入式新工作流可覆写确认动作名称。 */
enterButtonLabel?: string;
onManualSave: () => void;
onTriggerToast: (msg: string) => void;
submitIntentClarifications: (
@@ -111,6 +113,7 @@ export function Step1Panel(props: Step1PanelProps) {
onSkipQuestion,
onFollowUp,
onEnterRoundtable,
enterButtonLabel = "进入多智能体研讨 >",
onManualSave,
onTriggerToast,
submitIntentClarifications,
@@ -334,7 +337,7 @@ export function Step1Panel(props: Step1PanelProps) {
}`}
>
<Sparkles className="w-3.5 h-3.5 text-emerald-100" />
进入多智能体研讨 &gt;
{enterButtonLabel}
</button>
</div>
@@ -113,6 +113,8 @@ export interface UseStep1IntentReturn {
// ─── Handlers ────────────────────────────────────────────────────────
/** 清空当前 Step 1 状态并重新拉取 intent thread。 */
runStep1Init: () => void;
/** 清空当前状态但不创建线程,供宿主切换任务后重新进入 Step 1。 */
resetIntent: () => void;
/**
* 把 `userText` 作为用户气泡发给 intent agent,流式渲染回复。
* `files`:可选的本地文件,发送前先上传到 intent thread,再把引用随消息提交(后端
@@ -256,6 +258,20 @@ export function useStep1Intent(options: UseStep1IntentOptions): UseStep1IntentRe
})();
}, []);
const resetIntent = useCallback(() => {
stopIntentStream();
step1InitRef.current = false;
step1InitGenRef.current += 1;
setIntentThreadId(null);
setIntentReady(null);
setIntentInitError(null);
setIntentInitErrorDetail(null);
setPendingIntentClarificationMsgId(null);
setIntentClarificationSelections(new Map());
setInteractiveMessages([]);
setTimelineEntries([]);
}, [stopIntentStream]);
// 离开 Step 1 自动中断流;`currentStep === 1` 但还没初始化 → runStep1Init。
useEffect(() => {
if (currentStep !== 1) {
@@ -648,6 +664,7 @@ export function useStep1Intent(options: UseStep1IntentOptions): UseStep1IntentRe
timelineEntries,
setIntentClarificationSelections,
runStep1Init,
resetIntent,
sendIntentMessage,
submitIntentClarifications,
updateIntentReady,
@@ -45,6 +45,11 @@ export interface ChainSeat {
* 数组在 `api/chains.ts` 不做 camel↔snake 转换,键名须与后端 schema 逐字匹配)。
*/
seat_mode?: SeatMode | null;
/**
* 岗位协同专用的席位归属。普通圆桌会商不读取该字段;它跟随 `seats` JSON 保存,
* 使同一条业务链既能继续用于原会商,也能在岗位协同中按岗位筛选可问答的智能体。
*/
position_id?: string | null;
}
/** 一条可复用的业务链条(与后端 `/api/roundtable-chains` 记录对应)。 */
@@ -77,8 +77,13 @@ import { Switch } from "@/components/ui/switch";
import { useDesensitizeSubtree } from "@/roundtable-planning/lib/use-desensitize-subtree";
import { TagSearchBar } from "@/components/workspace/tags/tag-search-bar";
import { ModelNameSelect } from "@/components/workspace/model-name-select";
import { Select as TSelect } from "tdesign-react";
import { listAgents } from "@/core/agents";
import { getStoredAuth } from "@/core/auth";
import {
POSITION_OPTIONS,
type PositionId,
} from "@/position-roundtable/lib/positions";
import {
filterRecommendCandidates,
@@ -152,6 +157,14 @@ type ActiveDrag =
export default function BusinessChainEditorPage() {
const navigate = useNavigate();
const [searchParams] = useSearchParams();
// 岗位配置只由岗位协同页入口开启,普通圆桌编辑器完全不显示也不依赖该控件。
// `returnTo` 只接受站内绝对路径,避免把导航带往外部地址。
const positionConfigMode = searchParams.get("mode") === "position";
const requestedReturnTo = searchParams.get("returnTo");
const returnPath =
positionConfigMode && requestedReturnTo?.startsWith("/")
? requestedReturnTo
: "/page/workspace/roundtable/planning";
// 整片脱敏:链条配置页展示文本(链条名/描述/席位名/标题)。可编辑的 input/textarea 不脱敏(Hook 跳过),保证编辑用原文。
const dzRef = useDesensitizeSubtree<HTMLDivElement>();
@@ -415,6 +428,18 @@ export default function BusinessChainEditorPage() {
setStageGroups((prev) => setSeatModeOnGroups(prev, agentId, mode));
}, []);
// 岗位协同专用席位归属。保留在同一个 ChainSeat JSON 内,普通会商不会读取它。
const setSeatPositionCb = useCallback((agentId: string, positionId: PositionId | null) => {
setStageGroups((prev) =>
prev.map((group) => ({
...group,
seats: group.seats.map((seat) =>
seat.agent_id === agentId ? { ...seat, position_id: positionId } : seat,
),
})),
);
}, []);
// ── 提示词 AI 优化(总控提示词 / 各 Stage 阶段目标)────────────────────────
const [optimizingCoordinator, setOptimizingCoordinator] = useState(false);
const [goalOptimizingIdx, setGoalOptimizingIdx] = useState<number | null>(null);
@@ -617,6 +642,8 @@ export default function BusinessChainEditorPage() {
seat_skill_directive: s.seat_skill_directive ?? false,
// 席位「推理深度覆盖」:随席位一起落库(null = 跟随链条级 seatMode)。
seat_mode: s.seat_mode ?? null,
// 岗位协同专用岗位归属:普通圆桌忽略,但保存时必须原样保留。
position_id: s.position_id ?? null,
};
});
// 总控编排提示:trim 后空串落 null(清空),与后端「空=不注入」一致。
@@ -893,14 +920,14 @@ export default function BusinessChainEditorPage() {
<header className="flex shrink-0 flex-wrap items-center gap-3 border-b border-border bg-card px-6 py-3">
<button
type="button"
onClick={() => navigate("/page/workspace/roundtable/planning")}
onClick={() => navigate(returnPath)}
className={cn(
"flex items-center gap-1.5 border border-border bg-card px-3 py-1.5 text-sm font-medium text-muted-foreground transition hover:bg-muted hover:text-foreground active:scale-[0.98]",
R_CTRL,
)}
>
<ArrowLeft className="size-4" />
返回圆桌
{positionConfigMode ? "返回岗位协同" : "返回圆桌"}
</button>
{/* 品牌标识 */}
@@ -921,6 +948,17 @@ export default function BusinessChainEditorPage() {
</div>
</div>
{positionConfigMode && (
<span
className={cn(
"border border-primary/30 bg-primary/10 px-2 py-1 text-[11px] font-semibold text-primary",
R_CTRL,
)}
>
岗位协同配置
</span>
)}
{/* 链条选择下拉 */}
<div className="relative ml-2">
<button
@@ -1205,6 +1243,8 @@ export default function BusinessChainEditorPage() {
onSeatModelChange={setSeatModelCb}
onSeatSkillDirectiveChange={setSeatSkillDirectiveCb}
onSeatModeChange={setSeatModeCb}
positionConfigMode={positionConfigMode}
onSeatPositionChange={setSeatPositionCb}
onStageGoalChange={setGoal}
goalOptimizingIdx={goalOptimizingIdx}
onOptimizeGoal={handleOptimizeStageGoal}
@@ -1385,6 +1425,8 @@ function StagesColumn({
onSeatModelChange,
onSeatSkillDirectiveChange,
onSeatModeChange,
positionConfigMode,
onSeatPositionChange,
onStageGoalChange,
goalOptimizingIdx,
onOptimizeGoal,
@@ -1403,6 +1445,9 @@ function StagesColumn({
onSeatModelChange: (agentId: string, model: string | null) => void;
onSeatSkillDirectiveChange: (agentId: string, enabled: boolean) => void;
onSeatModeChange: (agentId: string, mode: SeatMode | null) => void;
/** 仅岗位协同入口启用;普通圆桌编辑器不显示岗位控件。 */
positionConfigMode: boolean;
onSeatPositionChange: (agentId: string, positionId: PositionId | null) => void;
onStageGoalChange: (stageIdx: number, goal: string) => void;
/** 正在 AI 优化阶段目标的 stage 下标(null = 没有进行中的优化)。 */
goalOptimizingIdx: number | null;
@@ -1439,6 +1484,11 @@ function StagesColumn({
<span className={cn("border border-border bg-muted px-2 py-0.5 text-[10px] font-bold text-muted-foreground", R_CTRL)}>
同层并行 · 层间串行
</span>
{positionConfigMode && (
<span className={cn("border border-primary/25 bg-primary/5 px-2 py-0.5 text-[10px] font-semibold text-primary", R_CTRL)}>
为每个席位指定岗位
</span>
)}
</div>
<div className="flex select-none items-center gap-2 text-xs">
@@ -1503,6 +1553,8 @@ function StagesColumn({
onSeatModelChange={onSeatModelChange}
onSeatSkillDirectiveChange={onSeatSkillDirectiveChange}
onSeatModeChange={onSeatModeChange}
positionConfigMode={positionConfigMode}
onSeatPositionChange={onSeatPositionChange}
onGoalChange={(goal) => onStageGoalChange(stageIdx, goal)}
goalOptimizing={goalOptimizingIdx === stageIdx}
anyGoalOptimizing={goalOptimizingIdx !== null}
@@ -1534,6 +1586,8 @@ function StageSection({
onSeatModelChange,
onSeatSkillDirectiveChange,
onSeatModeChange,
positionConfigMode,
onSeatPositionChange,
onGoalChange,
goalOptimizing,
anyGoalOptimizing,
@@ -1554,6 +1608,8 @@ function StageSection({
onSeatModelChange: (agentId: string, model: string | null) => void;
onSeatSkillDirectiveChange: (agentId: string, enabled: boolean) => void;
onSeatModeChange: (agentId: string, mode: SeatMode | null) => void;
positionConfigMode: boolean;
onSeatPositionChange: (agentId: string, positionId: PositionId | null) => void;
onGoalChange: (goal: string) => void;
/** 本 Stage 的阶段目标正在 AI 优化中。 */
goalOptimizing: boolean;
@@ -1717,6 +1773,8 @@ function StageSection({
onModelChange={(model) => onSeatModelChange(seat.agent_id, model)}
onSkillDirectiveChange={(enabled) => onSeatSkillDirectiveChange(seat.agent_id, enabled)}
onSeatModeChange={(mode) => onSeatModeChange(seat.agent_id, mode)}
showPositionConfig={positionConfigMode}
onPositionChange={(positionId) => onSeatPositionChange(seat.agent_id, positionId)}
onDetail={() => onSeatDetail(seat.agent_id, seat.name)}
/>
))}
@@ -1742,6 +1800,8 @@ function SeatStrip({
onModelChange,
onSkillDirectiveChange,
onSeatModeChange,
showPositionConfig,
onPositionChange,
onDetail,
}: {
seat: ChainSeat;
@@ -1760,6 +1820,9 @@ function SeatStrip({
onSkillDirectiveChange: (enabled: boolean) => void;
/** 席位「推理深度覆盖」变更(null = 跟随链条级 seatMode)。 */
onSeatModeChange: (mode: SeatMode | null) => void;
/** 仅岗位协同入口展示的岗位配置。 */
showPositionConfig: boolean;
onPositionChange: (positionId: PositionId | null) => void;
onDetail: () => void;
}) {
const { attributes, listeners, setNodeRef, transform, transition, isDragging } =
@@ -1810,6 +1873,29 @@ function SeatStrip({
)}
</div>
{/* 岗位协同从专用入口进来时才展示。使用 TDesign Select,未分配不会阻塞普通圆桌使用。 */}
{showPositionConfig && (
<div className="flex shrink-0 items-center gap-1.5 border-l border-border px-2.5">
<span className="shrink-0 text-[10px] font-medium text-muted-foreground">岗位</span>
<TSelect
size="small"
value={seat.position_id ?? ""}
options={[
{ label: "未分配", value: "" },
...POSITION_OPTIONS.map((position) => ({
label: position.name,
value: position.id,
})),
]}
onChange={(value) =>
onPositionChange(value ? (String(value) as PositionId) : null)
}
className="w-[132px]"
popupProps={{ overlayClassName: "z-[80]" }}
/>
</div>
)}
{/* 席位模型覆盖(右侧操作区):空 = 跟随全局所选模型;选了则该席位作为子智能体
执行时用此模型。技能重的席位配强模型、边角席位配快模型,配合技能白名单实现
「强弱搭配」。放到右侧操作区,与名称/描述解耦,卡片更整洁。 */}
@@ -375,6 +375,26 @@ 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).
### Position-collaboration foundation
The position-collaboration workspace is a separate frontend route and must not
use the original roundtable coordinator or jobs. Its shared configuration
integration point is the existing `/api/roundtable-chains` store: each `seats` JSON item
may carry an optional `position_id` (validated by `ChainSeatSchema`, no schema
migration). The original roundtable only reads its established seat fields and
therefore remains unchanged. The separate `/api/position-roundtable` router
persists task/intent/frozen-chain snapshots in `position_roundtable_sessions`
and direct-agent node indexes in `position_roundtable_nodes` (migration
`20260721_01`). Message history remains in the established LangGraph thread
store; a node completion unlocks its next chain stage and a later upstream
revision marks downstream records `stale` for refresh.
For direct-agent handoff, the router resolves artifact paths only inside the
completed upstream node's own thread, then includes bounded UTF-8 excerpts for
text outputs; binary artifacts stay as metadata-only references.
`POST /sessions/{id}/archive` turns a session into a read-only historical
record: task/intent updates, node binding, hidden-context construction, and
turn completion are rejected, while read endpoints remain available.
### 岗位 (Position) Subscription System
An admin-managed aggregation layer that decides, per user, which agents / skills
@@ -122,6 +122,8 @@ FastAPI application providing REST endpoints for frontend integration:
| `POST /api/skills/install` | Install skill from `.skill` archive |
| `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/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 |
| `GET /api/memory` | Retrieve memory data |
| `POST /api/memory/reload` | Force memory reload |
| `GET /api/memory/config` | Memory configuration |
@@ -53,9 +53,10 @@ from app.gateway.routers import (
notifications,
open_3qfx,
open_chat,
parallel_agent_tasks,
parallel_agents,
positions,
parallel_agent_tasks,
parallel_agents,
position_roundtable,
positions,
public_config,
public_embed,
public_leaderboard,
@@ -1443,8 +1444,11 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Step3 流程图结构化抽取 + 校验 (SSE) at /api/flow-extract
app.include_router(flow_extract.router)
# Roundtable-planning business chains at /api/roundtable-chains
app.include_router(roundtable_chains.router)
# Roundtable-planning business chains at /api/roundtable-chains
app.include_router(roundtable_chains.router)
# 岗位协同独立会话 / 节点状态(不使用原圆桌总控或后台作业)
app.include_router(position_roundtable.router)
# Global business ? agent / chain mapping at /api/business-mapping
app.include_router(business_mapping.router)
@@ -255,9 +255,15 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.roundtable_draft_share_store = make_roundtable_draft_share_store(sf)
from deerflow.persistence.roundtable_chains import make_roundtable_chain_store
app.state.roundtable_chain_store = make_roundtable_chain_store(sf)
from deerflow.persistence.roundtable_chains import make_roundtable_chain_store
app.state.roundtable_chain_store = make_roundtable_chain_store(sf)
# 独立的岗位协同 Session / Node 存储。它只索引业务流程状态,真实消息和
# 文件仍复用 LangGraph thread/checkpoint 与现有 artifact API。
from deerflow.persistence.position_roundtable import make_position_roundtable_store
app.state.position_roundtable_store = make_position_roundtable_store(sf)
from deerflow.persistence.business_mapping import make_business_mapping_store
@@ -0,0 +1,563 @@
"""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. Actual chat
messages and artifacts stay in the established LangGraph thread stores.
"""
from __future__ import annotations
import asyncio
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
router = APIRouter(prefix="/api/position-roundtable", tags=["position-roundtable"])
NODE_STATUSES = {"locked", "ready", "running", "done", "stale", "error"}
_HANDOFF_TEXT_BUDGET = 12_000
_MAX_INLINE_ARTIFACTS_PER_NODE = 4
_MAX_INLINE_ARTIFACT_CHARS = 1_800
_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)
revision: int = 0
upstream_revision: dict[str, Any] | 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
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
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
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]) -> 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}")
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
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 _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
refs.append(
{
"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,
}
)
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 not thread_id or 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)
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="上游节点尚未完成,当前节点未解锁")
# 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"] in {"done", "stale"}
]
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",
"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
},
}
@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,
)
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 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="业务链条不存在或无权访问")
nodes = _chain_nodes(chain)
activated = await store.activate_session(
session_id,
user_id,
chain_id=body.chain_id,
chain_snapshot=chain,
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.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("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.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, _ = 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="上游节点尚未完成,当前节点未解锁")
row = await store.update_node(
session_id,
node_key,
user_id,
status="done",
latest_answer=body.latest_answer,
artifact_manifest=body.artifact_manifest,
upstream_revision=body.upstream_revision,
increment_revision=True,
)
if row is None:
raise HTTPException(status_code=404, detail="Position roundtable node not found")
# A node has no coordinator. Its completion alone drives the frozen chain:
# all parallel seats in the current stage must finish before the next stage
# is made writable. Re-answering an already completed upstream seat keeps
# downstream records for audit but marks them stale so they must be
# refreshed against the new handoff context.
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 node.get("revision", 0) > 0:
for downstream in nodes:
if (
downstream["stage_index"] > row["stage_index"]
and downstream["status"] in {"ready", "running", "done", "stale"}
):
await store.update_node(
session_id,
downstream["node_key"],
user_id,
status="stale",
)
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 current_stage and all(item["status"] == "done" for item in current_stage) and later_stages:
next_stage = later_stages[0]
for next_node in nodes:
if next_node["stage_index"] == next_stage and next_node["status"] == "locked":
await store.update_node(
session_id,
next_node["node_key"],
user_id,
status="ready",
)
return _project_node(row)
@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]
@@ -62,6 +62,9 @@ class ChainSeatSchema(BaseModel):
# 持久化,故无需迁移;仅需在此 schema 显式声明,否则 model_dump() 会剥掉它。非法值在运行期
# 被 _SEAT_MODE_PARAMS.get() 安全忽略(回退作业级),故此处只约束长度、不强校验取值。
seat_mode: str | None = Field(default=None, max_length=16)
# 岗位协同专用的席位岗位归属。字段保存在既有 seats JSON 中,普通圆桌编排
# 忽略它;显式声明可确保 Pydantic 在 create/update 的 model_dump() 时不剥掉。
position_id: str | None = Field(default=None, max_length=64)
class ChainResponse(BaseModel):
@@ -0,0 +1,87 @@
"""Create standalone position-collaboration sessions and nodes.
The tables only index the task / intent / frozen business-chain workflow.
Conversation messages remain in LangGraph checkpoints and artifacts stay in the
existing thread artifact store, so this migration intentionally has no message
or file columns beyond lightweight snapshots and manifests.
Revision ID: 20260721_01
Revises: 20260701_01
Create Date: 2026-07-21
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
from sqlalchemy.dialects.mysql import LONGTEXT
revision = "20260721_01"
down_revision = "20260701_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
return table_name in inspect(op.get_bind()).get_table_names()
def upgrade() -> None:
long_text = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
if not _has_table("position_roundtable_sessions"):
op.create_table(
"position_roundtable_sessions",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("user_id", sa.String(64), nullable=False),
sa.Column("external_task_id", sa.String(128), nullable=True),
sa.Column("task_snapshot", long_text, nullable=True),
sa.Column("chain_id", sa.String(64), nullable=True),
sa.Column("chain_snapshot", long_text, nullable=True),
sa.Column("intent_thread_id", sa.String(128), nullable=True),
sa.Column("intent_snapshot", long_text, nullable=True),
sa.Column("status", sa.String(32), nullable=False, server_default="intent_pending"),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("updated_at", sa.DateTime(), nullable=False),
)
op.create_index("ix_position_roundtable_sessions_user_id", "position_roundtable_sessions", ["user_id"])
op.create_index(
"ix_position_roundtable_sessions_external_task_id",
"position_roundtable_sessions",
["external_task_id"],
)
op.create_index("ix_position_roundtable_sessions_chain_id", "position_roundtable_sessions", ["chain_id"])
op.create_index("ix_position_roundtable_sessions_status", "position_roundtable_sessions", ["status"])
if not _has_table("position_roundtable_nodes"):
op.create_table(
"position_roundtable_nodes",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("session_id", sa.String(64), sa.ForeignKey("position_roundtable_sessions.id", ondelete="CASCADE"), nullable=False),
sa.Column("node_key", sa.String(256), nullable=False),
sa.Column("stage_index", sa.Integer(), nullable=False),
sa.Column("seat_index", sa.Integer(), nullable=False),
sa.Column("agent_id", sa.String(128), nullable=False),
sa.Column("position_id", sa.String(64), nullable=True),
sa.Column("thread_id", sa.String(128), nullable=True, unique=True),
sa.Column("status", sa.String(32), nullable=False, server_default="locked"),
sa.Column("latest_answer", long_text, nullable=True),
sa.Column("artifact_manifest", long_text, nullable=True),
sa.Column("revision", sa.Integer(), nullable=False, server_default="0"),
sa.Column("upstream_revision", long_text, nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("updated_at", sa.DateTime(), nullable=False),
sa.UniqueConstraint("session_id", "node_key", name="uq_position_roundtable_nodes_session_key"),
)
op.create_index("ix_position_roundtable_nodes_session_id", "position_roundtable_nodes", ["session_id"])
op.create_index("ix_position_roundtable_nodes_agent_id", "position_roundtable_nodes", ["agent_id"])
op.create_index("ix_position_roundtable_nodes_position_id", "position_roundtable_nodes", ["position_id"])
op.create_index("ix_position_roundtable_nodes_status", "position_roundtable_nodes", ["status"])
def downgrade() -> None:
if _has_table("position_roundtable_nodes"):
op.drop_table("position_roundtable_nodes")
if _has_table("position_roundtable_sessions"):
op.drop_table("position_roundtable_sessions")
@@ -46,6 +46,10 @@ 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_roundtable.model import (
PositionRoundtableNodeRow,
PositionRoundtableSessionRow,
)
from deerflow.persistence.positions.model import PositionRow, PositionTagBindingRow
from deerflow.persistence.recommended_questions.model import RecommendedQuestionRow
from deerflow.persistence.research_templates.model import ResearchTemplateRow
@@ -104,6 +108,8 @@ __all__ = [
"MenuOverrideRow",
"NotificationRow",
"ParallelAgentTaskRow",
"PositionRoundtableNodeRow",
"PositionRoundtableSessionRow",
"PositionRow",
"PositionTagBindingRow",
"RecommendedQuestionRow",
@@ -0,0 +1,22 @@
"""Position-collaboration session and node persistence."""
from __future__ import annotations
from deerflow.persistence.position_roundtable.model import (
PositionRoundtableNodeRow,
PositionRoundtableSessionRow,
)
from deerflow.persistence.position_roundtable.sql import PositionRoundtableRepository
__all__ = [
"PositionRoundtableNodeRow",
"PositionRoundtableRepository",
"PositionRoundtableSessionRow",
"make_position_roundtable_store",
]
def make_position_roundtable_store(session_factory):
if session_factory is None:
return None
return PositionRoundtableRepository(session_factory)
@@ -0,0 +1,83 @@
"""Persistence models for the standalone position-collaboration workspace.
This deliberately does **not** reuse ``roundtable_drafts`` or
``roundtable_jobs``. A session captures the selected business chain and the
confirmed task intent; every chain seat then has one durable node record. The
node points at an existing LangGraph thread, which continues to own the actual
message history and generated artifacts.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableLongText
class PositionRoundtableSessionRow(Base):
__tablename__ = "position_roundtable_sessions"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
# Phase one uses normal account isolation. A future membership / role ACL
# can be added without changing the session or thread identifiers.
user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
external_task_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
task_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
chain_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
chain_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
intent_thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
intent_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
# intent_pending -> active -> completed / archived. The router owns legal
# transitions; the DB keeps it as a concise, portable string.
status: Mapped[str] = mapped_column(String(32), nullable=False, default="intent_pending", index=True)
created_at: Mapped[datetime] = mapped_column(
BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC)
)
updated_at: Mapped[datetime] = mapped_column(
BeijingDateTime(),
nullable=False,
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
class PositionRoundtableNodeRow(Base):
__tablename__ = "position_roundtable_nodes"
__table_args__ = (
UniqueConstraint("session_id", "node_key", name="uq_position_roundtable_nodes_session_key"),
)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
session_id: Mapped[str] = mapped_column(
String(64),
ForeignKey("position_roundtable_sessions.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Stable inside one frozen chain snapshot; e.g. ``stage-1-seat-2-agent``.
node_key: Mapped[str] = mapped_column(String(256), nullable=False)
stage_index: Mapped[int] = mapped_column(Integer, nullable=False)
seat_index: Mapped[int] = mapped_column(Integer, nullable=False)
agent_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
# The real messages remain in the standard LangGraph/checkpoint store.
thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True, unique=True)
# locked / ready / running / done / stale / error
status: Mapped[str] = mapped_column(String(32), nullable=False, default="locked", index=True)
latest_answer: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
artifact_manifest: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
upstream_revision: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
created_at: Mapped[datetime] = mapped_column(
BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC)
)
updated_at: Mapped[datetime] = mapped_column(
BeijingDateTime(),
nullable=False,
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
@@ -0,0 +1,279 @@
"""SQL repository for position-collaboration sessions and chain nodes."""
from __future__ import annotations
import json
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.position_roundtable.model import (
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
class PositionRoundtableRepository:
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
def _session_dict(row: PositionRoundtableSessionRow) -> dict[str, Any]:
return {
"id": row.id,
"external_task_id": row.external_task_id,
"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),
"status": row.status,
"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 [],
"revision": row.revision,
"upstream_revision": _loads(row.upstream_revision),
"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,
) -> 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_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 list_sessions(
self, user_id: str, *, external_task_id: str | None = None, limit: int = 50
) -> list[dict[str, Any]]:
stmt = select(PositionRoundtableSessionRow).where(PositionRoundtableSessionRow.user_id == user_id)
if external_task_id:
stmt = stmt.where(PositionRoundtableSessionRow.external_task_id == external_task_id)
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 row.user_id != 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,
status: str | object = _UNSET,
) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(PositionRoundtableSessionRow, session_id)
if row is None or row.user_id != user_id:
return None
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 status is not _UNSET:
row.status = str(status)
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 row.user_id != 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"),
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 parent.user_id != user_id:
return None
rows = (
await session.execute(
select(PositionRoundtableNodeRow)
.where(PositionRoundtableNodeRow.session_id == session_id)
.order_by(PositionRoundtableNodeRow.stage_index, PositionRoundtableNodeRow.seat_index)
)
).scalars().all()
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:
stmt = (
select(PositionRoundtableNodeRow)
.join(PositionRoundtableSessionRow, PositionRoundtableSessionRow.id == PositionRoundtableNodeRow.session_id)
.where(
PositionRoundtableNodeRow.session_id == session_id,
PositionRoundtableNodeRow.node_key == node_key,
PositionRoundtableSessionRow.user_id == user_id,
)
.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,
upstream_revision: dict[str, Any] | object = _UNSET,
increment_revision: bool = False,
) -> dict[str, Any] | None:
async with self._sf() as session:
stmt = (
select(PositionRoundtableNodeRow)
.join(PositionRoundtableSessionRow, PositionRoundtableSessionRow.id == PositionRoundtableNodeRow.session_id)
.where(
PositionRoundtableNodeRow.session_id == session_id,
PositionRoundtableNodeRow.node_key == node_key,
PositionRoundtableSessionRow.user_id == user_id,
)
.limit(1)
)
row = (await session.execute(stmt)).scalars().first()
if row is None:
return None
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 upstream_revision is not _UNSET:
row.upstream_revision = _dumps(upstream_revision)
if increment_revision:
row.revision += 1
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return self._node_dict(row)
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 list_results(self, session_id: str, user_id: str) -> list[dict[str, Any]] | None:
return await self.list_nodes(session_id, user_id)
@@ -0,0 +1,250 @@
"""Persistence coverage for the standalone position-collaboration workflow."""
from __future__ import annotations
from types import SimpleNamespace
import pytest
import pytest_asyncio
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.gateway.routers import position_roundtable as position_roundtable_router
from deerflow.persistence.base import Base
from deerflow.persistence.position_roundtable import PositionRoundtableRepository
@pytest_asyncio.fixture
async def store(tmp_path):
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'position-roundtable.db'}")
async with engine.begin() as conn:
# Register both tables with Base.metadata before creating the temporary DB.
import deerflow.persistence.position_roundtable.model # noqa: F401
await conn.run_sync(Base.metadata.create_all)
return PositionRoundtableRepository(async_sessionmaker(engine, expire_on_commit=False))
@pytest.mark.asyncio
async def test_session_activation_freezes_chain_and_creates_nodes_once(store):
created = await store.create_session(
"user-a",
external_task_id="task-1",
task_snapshot={"title": "测试任务"},
)
await store.update_session(
created["id"],
"user-a",
intent_thread_id="intent-thread",
intent_snapshot={"objective": "识别意图"},
)
chain = {
"id": "chain-a",
"seats": [
{"agent_id": "agent-a", "position_id": "planning"},
{"agent_id": "agent-b", "position_id": "solution"},
],
}
nodes = [
{
"node_key": "stage-1-seat-1-agent-a",
"stage_index": 0,
"seat_index": 0,
"agent_id": "agent-a",
"position_id": "planning",
"status": "ready",
},
{
"node_key": "stage-2-seat-1-agent-b",
"stage_index": 1,
"seat_index": 0,
"agent_id": "agent-b",
"position_id": "solution",
"status": "locked",
},
]
activated, created_nodes = await store.activate_session(
created["id"], "user-a", chain_id="chain-a", chain_snapshot=chain, nodes=nodes
)
assert activated["status"] == "active"
assert activated["chain_snapshot"] == chain
assert [node["status"] for node in created_nodes] == ["ready", "locked"]
# Repeated activate calls must return the first frozen snapshot/nodes unchanged.
repeated, repeated_nodes = await store.activate_session(
created["id"], "user-a", chain_id="other", chain_snapshot={"id": "other"}, nodes=[]
)
assert repeated["chain_id"] == "chain-a"
assert [node["node_key"] for node in repeated_nodes] == [node["node_key"] for node in created_nodes]
@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={})
_, nodes = await store.activate_session(
created["id"],
"user-a",
chain_id="chain-a",
chain_snapshot={"id": "chain-a"},
nodes=[
{
"node_key": "stage-1-seat-1-agent-a",
"stage_index": 0,
"seat_index": 0,
"agent_id": "agent-a",
"position_id": "planning",
"status": "ready",
}
],
)
assert len(nodes) == 1
updated = await store.update_node(
created["id"],
nodes[0]["node_key"],
"user-a",
thread_id="langgraph-thread-a",
status="done",
latest_answer="节点回答",
artifact_manifest=[{"name": "产物.md"}],
upstream_revision={"intent": 1},
increment_revision=True,
)
assert updated is not None
assert updated["thread_id"] == "langgraph-thread-a"
assert updated["latest_answer"] == "节点回答"
assert updated["artifact_manifest"] == [{"name": "产物.md"}]
assert updated["revision"] == 1
assert await store.get_node(created["id"], nodes[0]["node_key"], "user-b") is None
@pytest.mark.asyncio
async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_path, monkeypatch):
artifact = tmp_path / "handoff.md"
artifact.write_text("上游产物正文:可供下一岗位直接参考。", encoding="utf-8")
calls: list[tuple[str, str]] = []
class StubStore:
async def get_session(self, session_id, user_id):
assert (session_id, user_id) == ("session-a", "user-a")
return {"task_snapshot": {"title": "测试任务"}, "intent_snapshot": {"goal": "方案分析"}}
async def list_nodes(self, session_id, user_id):
assert (session_id, user_id) == ("session-a", "user-a")
return [
{
"node_key": "stage-1-seat-1-agent-a",
"stage_index": 0,
"seat_index": 0,
"agent_id": "agent-a",
"position_id": "planning",
"thread_id": "upstream-thread",
"status": "done",
"revision": 1,
"latest_answer": "上游问答结果",
"artifact_manifest": [
{
"name": "handoff.md",
"path": "mnt/user-data/outputs/handoff.md",
"mime_type": "text/markdown",
},
{
"name": "outside.md",
"path": "mnt/user-data/uploads/outside.md",
"mime_type": "text/markdown",
},
],
},
{
"node_key": "stage-2-seat-1-agent-b",
"stage_index": 1,
"seat_index": 0,
"agent_id": "agent-b",
"position_id": "solution",
"thread_id": "current-thread",
"status": "ready",
"revision": 0,
"latest_answer": None,
"artifact_manifest": [],
},
]
async def resolve(thread_id: str, path: str):
calls.append((thread_id, path))
assert thread_id == "upstream-thread"
assert path == "mnt/user-data/outputs/handoff.md"
return artifact
monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve)
request = SimpleNamespace(
app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=StubStore())),
state=SimpleNamespace(user=SimpleNamespace(id="user-a")),
)
context = await position_roundtable_router._handoff_context(
request, "session-a", "stage-2-seat-1-agent-b"
)
upstream = context["upstream_nodes"][0]
assert upstream["latest_answer"] == "上游问答结果"
assert upstream["artifacts"][0]["content"] == "上游产物正文:可供下一岗位直接参考。"
assert "content" not in upstream["artifacts"][1]
assert calls == [("upstream-thread", "mnt/user-data/outputs/handoff.md")]
@pytest.mark.asyncio
async def test_archived_session_rejects_node_writes_and_keeps_history_readable():
class ArchivedStore:
async def get_session(self, session_id, user_id):
assert (session_id, user_id) == ("session-a", "user-a")
return {"id": session_id, "status": "archived"}
async def get_node(self, session_id, node_key, user_id):
return {
"node_key": node_key,
"status": "ready",
"thread_id": None,
}
request = SimpleNamespace(
app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=ArchivedStore())),
state=SimpleNamespace(user=SimpleNamespace(id="user-a")),
)
with pytest.raises(HTTPException) as error:
await position_roundtable_router.bind_node_thread(
request,
"session-a",
"stage-1-seat-1-agent-a",
position_roundtable_router.NodeThreadRequest(thread_id="thread-a"),
)
assert error.value.status_code == 409
assert "已归档" in str(error.value.detail)
@pytest.mark.asyncio
async def test_locked_node_cannot_bind_a_thread_before_its_stage_is_ready():
class LockedStore:
async def get_session(self, session_id, user_id):
return {"id": session_id, "status": "active"}
async def get_node(self, session_id, node_key, user_id):
return {
"node_key": node_key,
"status": "locked",
"thread_id": None,
}
request = SimpleNamespace(
app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=LockedStore())),
state=SimpleNamespace(user=SimpleNamespace(id="user-a")),
)
with pytest.raises(HTTPException) as error:
await position_roundtable_router.bind_node_thread(
request,
"session-a",
"stage-2-seat-1-agent-b",
position_roundtable_router.NodeThreadRequest(thread_id="thread-b"),
)
assert error.value.status_code == 409
assert "未解锁" in str(error.value.detail)