diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index 60cd197f..747ef734 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -716,7 +716,8 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data if mode not in MODES or (mode == "ask") != (retry_of is None): raise TurnRefused(422, "发起方式不对:新问不带 retry_of,重试、重新运行、继续运行要带上是哪一轮", TURN_INVALID) besides = retry_of.id if retry_of is not None else None - if await _busy(session, conversation.id, besides=besides): + # 接着收尾(resume)不建图、不发起运行,不算占着这段对话:一段对话里停在半路的几轮可以一起交给服务端收尾 + if mode != "resume" and await _busy(session, conversation.id, besides=besides): raise TurnRefused(409, BUSY, TURN_BUSY) now_ms = int(time.time() * 1000) request: dict[str, Any] = {"started_ms": now_ms} diff --git a/backend/tests/test_chat_turns.py b/backend/tests/test_chat_turns.py index 36f554ff..cc57e0ad 100644 --- a/backend/tests/test_chat_turns.py +++ b/backend/tests/test_chat_turns.py @@ -857,3 +857,27 @@ async def test_an_empty_generation_is_retried_once_on_the_server(client, monkeyp row = await _row(turn_id) assert row.status == "done" and row.answer == "上周没有华中区的数据" and row.meta["noQuery"] is True assert row.meta["usage"]["copilot"]["calls"] == 2, "两次建图都是这一轮花的" + + +async def test_several_stranded_turns_of_one_conversation_can_be_resumed_together(client): + """一段对话里停在半路的几轮(老轮次、服务重启后停着的)一起交给服务端收尾:接着收尾不建图、不发起运行,不算占着这段对话。""" + conversation_id = await _conversation(client) + ids = [] + for answer in ("甲", "乙"): + async with SessionLocal() as session: + run = Run(status="succeeded", output={"答案": answer}, graph={"nodes": [], "edges": []}, run_class="exploratory") + session.add(run) + await session.flush() + turn = ConversationTurn(conversation_id=conversation_id, seq=len(ids), question=f"问{answer}", status="running", + run_id=run.id, meta={"v": 1, "runId": run.id}) + session.add(turn) + await session.commit() + ids.append(turn.id) + for turn_id in ids: + r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id, + "mode": "resume"}) + assert r.status_code == 202, r.text + for turn_id, answer in zip(ids, ("甲", "乙")): + await _done(turn_id) + row = await _row(turn_id) + assert row.status == "done" and row.answer == answer diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index 88511b76..2a42744b 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -1043,6 +1043,27 @@ export const useChat = create((set, get) => ({ await resumeTurn(conversationId, turnId) return } + // 重新核对一轮还没有运行的(实时流断在建图中、读不到了):问服务端它现在怎样,在跑就接着播,收了尾就照那一行摆 + if (turn.phase === 'checking' && turn.serverId) { + patch(() => ({ hydrated: 'pending' })) + try { + const res = await api.chatTurns.get(turn.serverId) + if (res.job && LIVE_JOB.has(res.job.status)) { + patch(() => ({ hydrated: 'done', job: res.job })) + play(flowOf(conversationId, turnId), turn.serverId, Math.max(0, res.job.attempt_seq - 1)) + } else { + patch((t) => ({ ...endPatch(t, { turn: res.turn }), hydrated: 'done' })) + } + } catch (e) { + const failure: Failure = { + title: PHASE_TEXT.unknown, + reason: `查询本轮运行状态时出错(${errorMessage(e)})。运行可能仍在进行,也可能已结束。`, + hint: '请稍后点击「重新核对」;服务端重新连接后会自动再核对一次。', + } + patch(() => ({ hydrated: undefined, phase: 'unknown', status: PHASE_TEXT.unknown, failure, error: lineOf(failure) })) + } + return + } const runId = turn.run?.id if (!runId) return // 核对期间用户点了重试或停止:拿回来的是上一次的状态,作废 @@ -1262,7 +1283,7 @@ function attemptReset(t: ChatTurn, d: Record, ts: number): Partial< phase: 'running', status: PHASE_TEXT.running, attempts, scope, usage, events: [], run: null, output: null, review: null, reviewRaw: null, rawOutput: null, failure: null, error: '', lastSeq: 0, final: undefined, rejected: null, runStatus: undefined, pendingRun: false, hydrated: 'done', - startedAt, endedAt: undefined, writing: null, noQuery: false, + startedAt, endedAt: undefined, writing: null, noQuery: false, restored: false, steps: undefined, } } return { @@ -1274,6 +1295,8 @@ function attemptReset(t: ChatTurn, d: Record, ts: number): Partial< runStatus: undefined, clipped: undefined, pendingRun: false, steps: undefined, hydrated: 'done', startedAt, endedAt: undefined, attempt: 0, noQuery: false, thinking: '', writing: null, asked: undefined, + // 刷新后接着播的那一轮就是正在跑的一轮:过程照实时的样子一直摆着(不是从库里恢复、要点开才取的历史轮次) + restored: false, } } @@ -1474,8 +1497,27 @@ function play(flow: Flow, serverId: string, after: number) { const stopTurn = streamChatTurn(serverId, after, (e) => { if (!flow.alive()) return apply(e) - }, () => { - // 服务端说这一轮的事件都发完了(或者这一轮读不了了)。运行的实时流还开着的让它自己收尾 + }, (end) => { + // 服务端说这一轮的事件都发完了(或者这一轮读不了了)。运行的实时流还开着的让它自己收尾。 + // 正常情况下 end 早就到了;没到(服务端那一轮出了岔子、这一轮被删了、没权限了)就别让它一直转圈:照库里那一行摆, + // 读不到就照实说没核对上 + const cur = findTurn(flow.conversationId, flow.turnId) + if (!flow.alive() || !cur || !(BUSY_PHASES.has(cur.phase) || cur.phase === 'checking' || cur.phase === 'waiting')) return + const unknown = (said: string) => { + const failure: Failure = { + title: PHASE_TEXT.unknown, + reason: `查询本轮运行状态时出错(${said})。运行可能仍在进行,也可能已结束。`, + hint: '请稍后点击「重新核对」;服务端重新连接后会自动再核对一次。', + } + patch(() => ({ phase: 'unknown', status: PHASE_TEXT.unknown, failure, error: lineOf(failure), cancel: null, + writing: null, hydrated: undefined })) + } + if (end.error) return unknown(end.error) + void api.chatTurns.get(serverId).then((res) => { + if (!flow.alive()) return + if (res.turn && res.turn.status !== 'running') patch((t) => endPatch(t, { turn: res.turn })) + else if (!res.job || !LIVE_JOB.has(res.job.status)) unknown(PHASE_TEXT.unknown) + }, (e) => { if (flow.alive()) unknown(errorMessage(e)) }) }) const close = () => { stopTurn() diff --git a/scripts/check-chat.mjs b/scripts/check-chat.mjs index de7582c6..c0abb8a6 100644 --- a/scripts/check-chat.mjs +++ b/scripts/check-chat.mjs @@ -2666,6 +2666,8 @@ for (const theme of THEMES) { check('刷新之后:这一轮还在服务端建图,照那个样子摆(不当成断了)', live, `${t1?.phase} · ${t1?.ops?.length} 条操作`) check('……从这一次尝试的第一条回放,已经建出的两步一步不少、一步不重', t1?.ops.filter((o) => o.op === 'add_node').length === 2 && t1?.ops.filter((o) => o.op === 'plan').length === 1) + check('……建图的过程照实时的样子摆着(不是要点开才取的历史轮次)', !t1?.restored + && await page.locator('[data-turn]').last().locator('button', { hasText: '规划' }).count() > 0) check('……刷新没有再发起一次(只发起过一次)', log.turnStarts.length - startsBefore === 1, `${log.turnStarts.length - startsBefore} 次`) check('……回放从这一次尝试开始(after 是 attempt_seq 之前)', log.streams.some((x) => x.turn === serverId && x.after === 0))