feat(圆桌): 查看运行中作业改为接管回前端续跑 + 后台挂起续跑方案文档
- onView 运行中/待澄清作业改走 takeOverRunningJob:cancelJob 停后端 → handleLoadDraft 恢复作业真实线程/协调名/对话 → 解除只读闸门 → resumeLiveTakeover 在作业线程上由前端继续实时流式编排 - hydrateFromDraft 同步写 threadIdsRef/coordinatorNameRef,使接管同帧可读真实线程 - 新增 docs/multi-agent-后台挂起续跑实现方案.md(仅方案,待实现) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
董宏凯andClaude Opus 4.8 committed
1 parent
25dbeb7a9a
commit
391409b6c0
3 files changed
+259
-31
No files matched your search
@@ -0,0 +1,198 @@
|
||||
# 后台挂起「在当前进度继续」实现方案
|
||||
|
||||
> 状态：**待实现**（仅方案，未改代码）
|
||||
> 关联：[[multi-agent-background-run-dev.md]]、[[multi-agent-frontend-dev.md]]、memory `project-roundtable-background`
|
||||
> 目标：点「后台挂起」后，后端作业**在前端当前研讨进度上接着跑**，而不是新建空线程从头重跑。
|
||||
|
||||
---
|
||||
|
||||
## 1. 现象与根因
|
||||
|
||||
### 现象
|
||||
在第二步研讨进行中点「后台挂起」，转入后台的作业是**从头重新开始执行**（重新初始化席位、从原始 intent 重新派活），而不是接着前端已经讨论到的进度往下走。
|
||||
|
||||
### 根因
|
||||
后端「起作业」路由 `start_job` **从不设 `is_resume`**（默认 `False`），而执行器据此**新建一套全新线程**：
|
||||
|
||||
```python
|
||||
# offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py (_run, 约 215 行)
|
||||
init = getattr(gateway, "init_threads", None)
|
||||
if init is not None and not params.is_resume:
|
||||
thread_ids, coordinator_name = await init() # ← 建全新空线程
|
||||
```
|
||||
|
||||
加上前端 `startBackgroundJob` **故意不传 `threadIds`**、用 `buildInitialIntent()` 作第一轮 leader prompt，于是后台作业 = 空线程 + 原始意图 = 从头重跑。
|
||||
|
||||
> 这是**原设计的有意取舍**。`startBackgroundJob` 注释写明：不复用前端线程，因为「那些线程可能有活跃 run，复用会撞 409」。本方案就是要在可控前提下打开「复用前端线程续跑」这条路。
|
||||
|
||||
### 后端其实已支持续跑
|
||||
澄清续跑路由 `resume_job`（`roundtable_jobs.py` 约 293 行）已经在做「在原线程上接着派活」：
|
||||
|
||||
```python
|
||||
executor.start_job(StartParams(
|
||||
...,
|
||||
seed_message=_RESUME_PREFIX + answer, # 续跑式提示，而非原始 intent
|
||||
thread_ids=row.get("thread_ids"), # 复用作业自己的线程
|
||||
coordinator_name=row.get("coordinator_name") or "roundtable-coordinator",
|
||||
is_resume=True, # ← 关键：跳过 init_threads，直接复用
|
||||
...
|
||||
))
|
||||
```
|
||||
|
||||
执行器在 `is_resume=True` 时直接用 `params.thread_ids` + `params.coordinator_name`（不调 `init_threads`），交给 `run_orchestration` 续跑。**所以能力已存在，`start_job` 只是没走这条路。**
|
||||
|
||||
---
|
||||
|
||||
## 2. 设计目标与关键事实
|
||||
|
||||
- 「同进度续跑」= 后台作业复用**前端当前会话的真实线程**（总控 + 各席位），这些线程的 checkpoint 里已存着到目前为止的完整研讨上下文，续跑 = 在原线程上 post 一条「继续」消息让总控接着派活。
|
||||
- 前端编排是**逐轮**的：每轮一个 `POST /api/multi-agent/run/stream`（leader 或某席位），**轮与轮之间线程空闲**。只有「某一轮在飞时」线程才被占用。
|
||||
- 后端复用现成线程是**已验证路径**（`resume_job` 天天在用），唯一新增的不同点是：这次复用的是「前端会话线程」而非「上一个后台作业的线程」——但线程同属一个后端线程库，复用方式完全一致。
|
||||
|
||||
---
|
||||
|
||||
## 3. 改造点（最小集）
|
||||
|
||||
### 3.1 后端：让 `start_job` 支持复用线程
|
||||
|
||||
文件：`offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py`
|
||||
|
||||
1. `StartJobRequest` 增加字段：
|
||||
```python
|
||||
resume: bool = False # true + 携带 threadIds ⇒ 复用这些线程续跑，不新建
|
||||
```
|
||||
2. `start_job` 里把它透传给 `StartParams`：
|
||||
```python
|
||||
is_resume = bool(body.resume and body.threadIds)
|
||||
executor.start_job(StartParams(
|
||||
...,
|
||||
thread_ids=body.threadIds,
|
||||
coordinator_name=body.coordinatorName,
|
||||
is_resume=is_resume, # ← 新增
|
||||
...
|
||||
))
|
||||
```
|
||||
其余执行器逻辑**不动**（它在 `is_resume=True` 时已能复用 `params.thread_ids`）。
|
||||
|
||||
> 幂等闸门（同草稿未终态作业直接返回）保持不变；续跑挂起前前端应保证该草稿没有在跑的旧作业。
|
||||
|
||||
### 3.2 前端 API：补 `resume` 入参
|
||||
|
||||
文件：`frontend-web/src/roundtable-planning/lib/job-types.ts`
|
||||
- `StartJobPayload` 增加 `resume?: boolean`。
|
||||
|
||||
文件：`frontend-web/src/roundtable-planning/api/roundtable-jobs.ts`
|
||||
- `toStartBody` 增加 `resume: p.resume ?? false`。
|
||||
|
||||
### 3.3 前端：`startBackgroundJob` 改为复用线程续跑
|
||||
|
||||
文件：`frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx`（`startBackgroundJob` 约 1119 行）
|
||||
|
||||
当前端已有现成会话（`step2.threadIds` 且 `step2.coordinatorName` 非空，说明研讨已 init）时，改传：
|
||||
|
||||
```ts
|
||||
const canContinue = !!(step2.threadIds && step2.coordinatorName);
|
||||
await startJob({
|
||||
draftId: draftId ?? undefined,
|
||||
// 续跑：用「继续」式 seed，而不是 buildInitialIntent()（后者会让总控从头理解任务）
|
||||
intent: buildInitialIntent(), // 仍传，作为无线程回退时的首轮 prompt
|
||||
seedMessage: canContinue
|
||||
? "用户已将本次研讨转入后台，请基于现有进展继续上一轮的协调与派活；若已可收口则直接综合输出结论。"
|
||||
: undefined,
|
||||
agents,
|
||||
model: selectedModel || undefined,
|
||||
orchestrationMode,
|
||||
chain: chainMeta,
|
||||
orchestrationPlan: orchestrationMode === "dag" ? orchestrationPlan : null,
|
||||
// 关键三件套：复用前端真实线程 + 协调名 + resume 标志
|
||||
threadIds: canContinue ? step2.threadIds ?? undefined : undefined,
|
||||
coordinatorName: canContinue ? step2.coordinatorName ?? undefined : undefined,
|
||||
resume: canContinue,
|
||||
});
|
||||
```
|
||||
|
||||
> 没有现成会话（极早期、线程未建）时 `canContinue=false`，自动回退到「新建线程 + 原 intent」的旧行为，安全。
|
||||
|
||||
### 3.4 前端：挂起前先停本地编排（避免交出去时线程还在飞）
|
||||
|
||||
文件：同上，`handleBackgroundSuspend`（约 1161 行）
|
||||
|
||||
当前顺序是 `startBackgroundJob()` → `resetToStep1()`（reset 里才 `stopOrchestration`）。复用线程时必须**先停再交**：
|
||||
|
||||
```ts
|
||||
const handleBackgroundSuspend = useCallback(async () => {
|
||||
if (step2.selectedAgents.length === 0) { /* toast 提示后 return */ }
|
||||
// 1. 先彻底停掉前端在飞的那一轮，尽量让线程在交给后台前变空闲
|
||||
step2.stopOrchestration();
|
||||
step2.setIsRoundTableRunning(false);
|
||||
try {
|
||||
// 2. 再起后台作业（复用线程续跑）
|
||||
await startBackgroundJob();
|
||||
} catch (err) { /* toast 失败后 return */ }
|
||||
// 3. 最后重置回第一步（后台作业继续在历史记录里跑）
|
||||
resetToStep1();
|
||||
triggerToast("🚀 已转入后台，将基于当前进度继续研讨…");
|
||||
void drafts.refresh();
|
||||
}, [...]);
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 4. 关键风险：线程占用（409）与三种处理
|
||||
|
||||
复用前端线程的唯一风险：**在某一轮在飞时挂起**，`stopOrchestration()` 只断了客户端 SSE，服务端那次 run 可能还在跑一小会儿；后台作业首轮 leader 向同一总控线程 post → 撞 409 / rollback。
|
||||
|
||||
| 方案 | 做法 | 取舍 |
|
||||
|---|---|---|
|
||||
| **A（推荐起步）** | 仅前端 stop + 后端 `is_resume` 复用，接受小概率 409 | 实现最简，真正同进度续跑；轮间挂起 100% 安全，仅「某轮在飞时挂起」有小概率 409 —— 走编排循环**现有 rollback/重试容错**兜底 |
|
||||
| **B（最稳）** | 同 A，但后端在复用线程续跑前，**先 cancel 这些线程上可能残留的 run**（参考 takeover 里 `cancelJob` 的 `task.cancel()` 思路，落到 thread-run 级别）再 post | 几乎无 409；后端要多写「按 thread_id 取消活跃 run」的逻辑 |
|
||||
| **C（不推荐）** | 不复用线程，新建空线程，但把「目前为止的研讨」塞进 seed_message 让总控续写 | 零 409，但**席位子线程上下文全丢**、不是真正同进度、leader prompt 臃肿 |
|
||||
|
||||
> 建议：先上 **A**，真机验证 409 命中率；若频繁再加 **B** 的后端线程级 cancel。
|
||||
|
||||
---
|
||||
|
||||
## 5. 适用范围（哪些挂起入口改续跑）
|
||||
|
||||
三个触发源（见 `autoSuspendRef`）：
|
||||
|
||||
1. **手动「后台挂起」按钮**（`handleBackgroundSuspend`）——**改续跑**。能在 start 前干净 stop，最适合复用线程。
|
||||
2. **路由卸载自动挂起**（切到其它页面，`fire`）——可改续跑（能走完整异步链）；卸载瞬间 stop 不一定彻底，按 A 方案的容错兜底。
|
||||
3. **刷新/关标签页 keepalive**（`beforeunload` 的 `beacon` → `startJobKeepalive`）——**建议保持新建线程**。关页时无法保证前端 run 已停，复用线程 409 风险最高；这里走「新建线程 + 原 intent」的安全回退即可（用户可在历史记录里点「查看」再用我们已实现的 takeover 接管回前端续跑）。
|
||||
|
||||
> 落地顺序建议：先只改 #1，验证稳定后再决定是否推广到 #2。
|
||||
|
||||
---
|
||||
|
||||
## 6. 与已实现「查看=接管回前端」的关系
|
||||
|
||||
二者是一对对称操作，复用同一套「线程同属一个后端库、可跨前后端续跑」的事实：
|
||||
|
||||
- **后台挂起（本方案）**：前端 → 后端。前端 stop，把前端线程交给后台 `is_resume` 续跑。
|
||||
- **查看接管（已实现，见 memory）**：后端 → 前端。`cancelJob` 停后台，`hydrateFromDraft` 恢复作业真实线程，`resumeLiveTakeover()` 由前端在那些线程上续跑。
|
||||
|
||||
实现本方案时可直接参照 takeover 的线程复用与 `is_resume` 经验。
|
||||
|
||||
---
|
||||
|
||||
## 7. 测试
|
||||
|
||||
后端（`make test`）：
|
||||
- 新增/扩展 `tests/test_roundtable_jobs.py`：`start_job` 带 `resume=true + threadIds` ⇒ `StartParams.is_resume=True` 且**不调** `init_threads`、直接用传入线程；`resume=false` 或无 `threadIds` ⇒ 走 `init_threads` 新建（回归旧行为）。
|
||||
- 回归：`tests/test_roundtable_inprocess_request.py`、`tests/test_roundtable_step3_pipeline.py`（确认续跑不破坏 Step3 双产物）。
|
||||
|
||||
前端（`pnpm typecheck`）+ 真机：
|
||||
- 研讨进行到几轮后点「后台挂起」→ 历史记录里点「查看」→ 确认后台是**接着已有对话**往下派活，而非从第 1 轮重来。
|
||||
- 轮间挂起 / 某轮在飞时挂起两种时机各验一次，观察是否出现 409（A 方案下应被 rollback 容错吸收）。
|
||||
|
||||
---
|
||||
|
||||
## 8. 落地清单（TL;DR）
|
||||
|
||||
- [ ] 后端 `StartJobRequest.resume` 字段 + `start_job` 透传 `is_resume`
|
||||
- [ ] 前端 `StartJobPayload.resume` + `toStartBody`
|
||||
- [ ] 前端 `startBackgroundJob`：有会话时传 `threadIds/coordinatorName/seedMessage/resume`
|
||||
- [ ] 前端 `handleBackgroundSuspend`：调整为 **先 stop 再 start 再 reset**
|
||||
- [ ] 后端测试（resume 复用 vs 新建分支）
|
||||
- [ ] 真机验证「接着进度跑」+ 409 命中率
|
||||
- [ ]（可选）方案 B：后端复用线程前按 thread_id 取消残留 run
|
||||
@@ -341,6 +341,13 @@ export interface UseStep2OrchestrationReturn {
|
||||
hasConsensus: boolean;
|
||||
lastLeaderContent: string;
|
||||
}) => void;
|
||||
/**
|
||||
* 「后台挂起接管」(2026-06-13)：点「查看」一个运行中的后台作业并取消它、
|
||||
* `hydrateFromDraft` 恢复作业真实线程 + 完整对话后调用 —— 在那些线程上**由前端
|
||||
* 继续实时流式编排**(leader 自由派活续轮,不从头重跑 DAG)。返回是否成功接管
|
||||
* (无线程时返回 false,调用方据此回退到只读查看)。
|
||||
*/
|
||||
resumeLiveTakeover: () => boolean;
|
||||
/** 「新建任务」清空 Step 2 数据(threadIds / dialogues / 共识等)。 */
|
||||
resetStep2State: () => void;
|
||||
/** 当前沙箱预览绑定的席位 thread id。 */
|
||||
@@ -2634,6 +2641,11 @@ export function useStep2Orchestration(
|
||||
}
|
||||
setThreadIds(snapshot.threadIds);
|
||||
setCoordinatorName(snapshot.coordinatorName);
|
||||
// 同步把线程/协调名写进 ref(平时由 [threadIds]/[coordinatorName] 的 effect 异步回写,
|
||||
// 滞后一拍)。后台挂起「接管续跑」会在 hydrate 之后**同帧**调 resumeLiveTakeover,
|
||||
// 需要 ref 立刻可读到作业的真实线程,否则读到旧值导致接管失败。
|
||||
threadIdsRef.current = snapshot.threadIds;
|
||||
coordinatorNameRef.current = snapshot.coordinatorName;
|
||||
setStep2RoundtableDialogues(snapshot.step2RoundtableDialogues);
|
||||
setLastLeaderContent(snapshot.lastLeaderContent);
|
||||
setHasConsensus(snapshot.hasConsensus);
|
||||
@@ -2671,6 +2683,32 @@ export function useStep2Orchestration(
|
||||
[],
|
||||
);
|
||||
|
||||
// 「后台挂起接管」：取消后端作业 + hydrateFromDraft(恢复作业真实线程/协调名/对话)后,
|
||||
// 由前端在这些线程上**继续实时流式编排**。续跑走 leader 自由派活循环(把 dagPlanExecuted
|
||||
// 置 true 避免从头重跑整张 DAG),并清掉残留的取消/编排锁让新一轮立刻起跑——
|
||||
// 与 handleInjectToRoundtable 末尾「直接启动新一轮编排」同构。
|
||||
const resumeLiveTakeover = useCallback((): boolean => {
|
||||
if (!threadIdsRef.current || !coordinatorNameRef.current) {
|
||||
triggerToast("⚠️ 接管失败:未找到研讨线程,已切换为只读查看");
|
||||
return false;
|
||||
}
|
||||
dagPlanExecutedRef.current = true;
|
||||
orchestrationCancelRef.current = true;
|
||||
abortAllInFlight();
|
||||
isOrchestratingRef.current = false;
|
||||
setIsOrchestrating(false);
|
||||
setPendingClarification(null);
|
||||
setClarificationDraft("");
|
||||
setIsRoundTableRunning(true);
|
||||
void runActiveOrchestration(
|
||||
"用户已从后台挂起接管本次研讨,请基于现有进展继续上一轮的协调与派活工作;若研讨已可收口则直接综合输出结论。",
|
||||
threadIdsRef.current,
|
||||
coordinatorNameRef.current,
|
||||
);
|
||||
triggerToast("🎮 已接管至前端,继续实时研讨");
|
||||
return true;
|
||||
}, [abortAllInFlight, runActiveOrchestration, triggerToast]);
|
||||
|
||||
const resetStep2State = useCallback(() => {
|
||||
orchestrationCancelRef.current = true;
|
||||
abortAllInFlight();
|
||||
@@ -2749,6 +2787,7 @@ export function useStep2Orchestration(
|
||||
handleSaveAgentAndRecalculate,
|
||||
hydrateFromDraft,
|
||||
applyLiveJobFrame,
|
||||
resumeLiveTakeover,
|
||||
resetStep2State,
|
||||
activeArtifactThreadId,
|
||||
seatStubMessages,
|
||||
|
||||
@@ -103,12 +103,11 @@ import {
|
||||
import type { OrchestrationPlan } from "@/roundtable-planning/lib/dag-types";
|
||||
import type { IntentSummary } from "@/roundtable-planning/api/intent";
|
||||
import {
|
||||
cancelJob,
|
||||
getJobByDraft,
|
||||
startJob,
|
||||
startJobKeepalive,
|
||||
streamJob,
|
||||
} from "@/roundtable-planning/api/roundtable-jobs";
|
||||
import { mapJobToLiveStep2 } from "@/roundtable-planning/lib/job-live";
|
||||
import { getChain } from "@/roundtable-planning/api/chains";
|
||||
import { filterRecommendCandidates } from "@/roundtable-planning/api/recommend";
|
||||
import { listAgents } from "@/core/agents";
|
||||
@@ -1028,40 +1027,32 @@ export default function RoundtablePlanningPage() {
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [currentStep, drafts.currentDraftId]);
|
||||
|
||||
// ── 「查看=恢复流程」（2026-06-13）：查看运行中的后台作业 → SSE 实时跟随 ──
|
||||
// 加载草稿后订阅 streamJob，把作业每帧对话/产物/共识进度灌进第二步画面（像本地
|
||||
// 编排一样实时长出气泡），终态时重载草稿衔接最终 run + step3，并停止订阅。
|
||||
// ── 「查看=接管回前端」（2026-06-13）：查看运行中的后台作业 → 接管续跑 ──
|
||||
// 取消后端作业 → 加载作业最近写回草稿的快照（含作业真实线程 + 完整对话）→ 解除
|
||||
// 「看历史」只读闸门 → 由前端在那些线程上**继续实时流式编排**（leader 自由派活续轮）。
|
||||
// 这样「后台挂起 → 查看」会把会商真正搬回前端，按钮/暂停/干预/再挂起全部回归，而不是
|
||||
// 停在只读跟随的后台态。
|
||||
const jobFollowUnsubRef = useRef<(() => void) | null>(null);
|
||||
const stopJobFollow = useCallback(() => {
|
||||
jobFollowUnsubRef.current?.();
|
||||
jobFollowUnsubRef.current = null;
|
||||
}, []);
|
||||
// 「查看运行中作业」：先 await 草稿 hydrate（建好席位/线程基础 + 切第二步），**再**
|
||||
// 订阅 SSE —— 保证实时帧总是最后权威，不被草稿快照覆盖（竞态根因）。强制停在第二步
|
||||
// 看研讨流程（作业可能已到 Step3，但用户点查看就是要看实时研讨过程）。
|
||||
const followRunningJob = useCallback(
|
||||
const takeOverRunningJob = useCallback(
|
||||
async (jobId: string, draftId: string) => {
|
||||
stopJobFollow();
|
||||
step2.setIsRoundTableRunning(false);
|
||||
step2.stopOrchestration();
|
||||
// 点查看前**立刻存一次**当前聊天（若用户在本地还跑着研讨，保住其记录再切去看作业）。
|
||||
await drafts.saveDraftNow(2).catch(() => null);
|
||||
// 1. 取消后端作业：避免前后端双跑/抢同一线程（409），线程随后由前端接管续跑。
|
||||
// executor.cancel 会 task.cancel() 中断在跑的 run，DB 同步置 cancelled。
|
||||
await cancelJob(jobId).catch(() => null);
|
||||
// 2. 加载作业最近写回草稿的快照：恢复 threadIds（作业真实线程）+ 协调名 + 完整对话。
|
||||
// hydrateFromDraft 已同步写好 threadIds/coordinatorName 的 ref，供下一步接管即时读取。
|
||||
await drafts.handleLoadDraft(draftId);
|
||||
setCurrentStep(2); // 覆盖 hydrate 可能切到的 furthestStep，停在第二步看流程
|
||||
jobFollowUnsubRef.current = streamJob(jobId, {
|
||||
onProgress: (j) => {
|
||||
step2.applyLiveJobFrame(mapJobToLiveStep2(j));
|
||||
},
|
||||
onDone: (j) => {
|
||||
stopJobFollow();
|
||||
// 终态：作业已把最终 run + step3 写回草稿 → 重载草稿无缝衔接（含报告）。
|
||||
if (j.draftId) void drafts.handleLoadDraft(j.draftId);
|
||||
void drafts.refresh();
|
||||
},
|
||||
onError: () => {
|
||||
/* 断流静默：画面保留最后一帧；用户可重新点「查看」再续 */
|
||||
},
|
||||
});
|
||||
setCurrentStep(2); // 覆盖 hydrate 可能切到的 furthestStep，停在第二步继续研讨
|
||||
// 3. 关掉「看历史」只读闸门 → 后台挂起/暂停/干预按钮回归，回到前端可交互态。
|
||||
drafts.loadedFromDraftRef.current = false;
|
||||
// 4. 前端接管：在作业线程上继续实时流式编排（无线程则回退为只读查看，不再续跑）。
|
||||
step2.resumeLiveTakeover();
|
||||
void drafts.refresh();
|
||||
},
|
||||
[stopJobFollow, step2, drafts, setCurrentStep],
|
||||
);
|
||||
@@ -1306,13 +1297,13 @@ export default function RoundtablePlanningPage() {
|
||||
if (!o) setChainModalJobId(null);
|
||||
}}
|
||||
onView={(job) => {
|
||||
// 「查看」（2026-06-13 恢复流程语义）：done → 加载草稿（含 step3）看报告；
|
||||
// **运行中/待澄清 → 草稿 hydrate 完成后 SSE 实时跟随**——第二步消息列表像
|
||||
// 本地编排一样实时长出新气泡/步骤卡，不再是静态聊天记录。
|
||||
// 「查看」（2026-06-13 接管语义）：done → 加载草稿（含 step3）看报告；
|
||||
// **运行中/待澄清 → 接管回前端**：取消后端作业、加载聊天记录、解除只读，
|
||||
// 由前端在作业线程上继续实时流式编排（不再是只读跟随后台）。
|
||||
setChainModalJobId(null);
|
||||
if (!job.draftId) return;
|
||||
if (job.status === "running" || job.status === "awaiting_input") {
|
||||
void followRunningJob(job.id, job.draftId);
|
||||
void takeOverRunningJob(job.id, job.draftId);
|
||||
return;
|
||||
}
|
||||
void drafts.handleLoadDraft(job.draftId);
|
||||
|
||||
Reference in new issue
Block a user