问数据整轮放到服务端跑(前端):store/chat.ts 改成「发起 + 播放」,断线按 seq 重连补齐,刷新、关掉再打开接着看

为什么:一轮问数据搬到服务端之后(上一个提交),页面不该再推这一轮:建图那条流、发起运行、复核、按复核重来、落库都和页面
绑在一起,页面一断这一轮就停。现在页面只发起、只看,要不要运行、要不要重来、什么时候交付都照服务端说的。

做了什么:
- 新的 api/chatTurns.ts:发起、查询、停止(挂在 api.chatTurns),streamChatTurn 订阅这一轮的事件日志——先回放、再接实时;
  断了按收到的最后一个 seq 自动重连(收到东西马上接着连,连不上按退避),45 秒什么都没收到(服务端 15 秒一拍心跳)当连接死了。
- store/chat.ts:plan / launch / settle / retry 换成发起(launchTurn)+ 播放(play)。收到的操作照原来的渲染逻辑画(画布边收
  边长、过程步骤、思考、查看、用量);收到 run 就接上运行自己的实时流画过程(watch 只画过程,不再判结局);stage 说到了
  哪一步(停在审批上立刻去拿待审批);turn.retry 照原来自动重来那样重置;end 照服务端落库的那一行摆结局(和刷新之后从库里
  恢复的是同一份 restoreTurn)。「重试本轮」「重新运行本轮」「放宽步数」「跑一下」「继续运行」「停止」都走新接口。
- 页面不再写这一轮的结局(不再 PATCH 轮次、不再调 /copilot/review、不再发 POST /api/runs);失败的说法照服务端记的原因
  (meta.cause)按 lib/explain 讲,老轮次记的 failure 照旧认。
- 刷新、关掉再打开:服务端还在跑的一轮(ConversationTurn.job)从这一次尝试的第一条事件接着播;库里停在半路、服务端又没在跑
  的(老轮次、服务重启后停着的)请服务端接着收尾(mode=resume),不在页面上自己收尾。问数据页的显示不变。
- 页面上那份退路(MAX_ATTEMPTS、fillInput、retryInstruction、等复核的计时)去掉:决定只在服务端一处。后端对照测试改成守
  「页面上不再有这几样」。
- scripts/check-chat.mjs:伪造的服务端照 backend/app/chat/job.py 的做法用原来那几份剧本把一轮跑完(生成、发起运行、等运行的剧本
  放完、复核、拼那一行),原有各段照过;守落库的几处改成守「页面不 PATCH 这一轮」和照服务端那一行摆;新加「断线重连补齐」
  「刷新、关掉再打开接着看」「停止(建图中也停得下)」几段,并核对发起、重试、放宽步数、继续运行、停止的请求体。
- 服务端顺带:turn.attempt 带上到目前为止的用量、从什么时候算起(补交付的到运行结束为止),turn.retry 带上换下来的那次运行
  并进之后的用量(页面上的运行壳子里没有用量);接着收尾(resume)不要求运行还在,运行记录删了的照实收成「运行记录不见了」。
This commit is contained in:
ThearchyHelios
2026-10-18 16:49:14 +08:00
parent d18a496ef9
commit 5ff3e1136d
8 changed files with 1822 additions and 1278 deletions
+15 -5
View File
@@ -310,7 +310,8 @@ class ChatTurnExecution(runtime.JobRun):
self.run_id = None
await self._job(attempt=attempt, run_id=None)
await self.save(run_id=None, meta={"runId": None, "runStatus": None, "usage": dict(self.usage)})
await self.emit("turn.retry", {"kind": kind, "reason": reason, "base": base, "attempt": attempt})
await self.emit("turn.retry", {"kind": kind, "reason": reason, "base": base, "attempt": attempt,
"usage": dict(self.usage)})
# ---------------- 执行 ----------------
@@ -751,7 +752,8 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data
raise TurnRefused(422, "放宽到几步要写清楚", TURN_INVALID)
request["max_steps"] = int(max_steps)
if mode in ("continue", "resume"):
if run is None:
# 接着收尾不要求运行还在:运行记录删了的,服务端照实收成「运行记录不见了」
if run is None and (mode == "continue" or not run_id):
raise TurnRefused(409, "这一轮没有可接着的运行,请重试本轮", TURN_INVALID)
if mode == "continue":
await _continue_run(run.id, actor)
@@ -761,9 +763,10 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data
spent = meta.get("ms")
if row.status != "running" and isinstance(spent, (int, float)):
request["started_ms"] = now_ms - int(spent)
elif run.started_at is not None:
elif run is not None and run.started_at is not None:
request["started_ms"] = int(run.started_at.timestamp() * 1000)
if mode == "resume" and run.status in ("succeeded", "failed", "cancelled") and run.finished_at is not None:
if mode == "resume" and run is not None and run.status in ("succeeded", "failed", "cancelled") \
and run.finished_at is not None:
request["ended_ms"] = int(run.finished_at.timestamp() * 1000)
else:
# 重试、重新运行、放宽步数、跑一下:服务重启挂起的旧运行没人会再去续它,顺手取消(停在审批上的不算:重试不会从那里发起)
@@ -812,8 +815,15 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data
# 这一次尝试从这一条开始:刷新、重开页面时从这里回放
async with SessionLocal() as s:
from app.db.models import ConversationTurn
fresh = await s.get(ConversationTurn, turn_id)
usage = ((fresh.meta or {}) if fresh is not None and isinstance(fresh.meta, dict) else {}).get("usage")
# 页面据此把这一轮重置成「这一次」的样子:归档、范围、到目前为止的用量、从什么时候算起(补交付的到运行结束为止)
seq = await turn_log.append(s, turn_id, "turn.attempt", {
"mode": mode, "attempts": attempts, "scope": request.get("scope") or [],
"mode": mode, "attempts": attempts, "scope": request.get("scope") or [], "usage": usage or {},
"started_ms": request.get("started_ms"),
**({"ended_ms": request["ended_ms"]} if request.get("ended_ms") else {}),
**({"max_steps": request["max_steps"]} if request.get("max_steps") else {})})
await s.execute(update(ChatTurnJob).where(ChatTurnJob.id == turn_id).values(attempt_seq=seq)
.execution_options(synchronize_session=False))
+25 -32
View File
@@ -1,11 +1,10 @@
"""问数据一轮的「决定」(app/chat/turn.py;细案 01 做法 1)。
要不要运行、输入怎么填、要不要带着原图重来、重来时说什么,从问数据页下沉到后端:final 带 next,复核带 retry_instruction,
问数据页照它们走;老服务端没给这两个字段时,问数据页退回它自己那份同样的规则。这里守:
问数据的一轮后来整个搬到了服务端(chat/drive.py、chat/job.py),页面只发起、只播放。这里守:
- 各条决定的分支(next_step、fill_input、retry_after_review、retry_instruction 的两种措辞);
- 前后端对照:frontend/src/store/chat.ts 里留着的那份常量(MAX_ATTEMPTS、fillInput 认问题字段的正则、两种重来的措辞)
和这里逐字一致——写法同 test_frontend_type_unions.py,两边一改就对不上;
- 页面上不再留一份同样的决定(一轮整个搬到服务端之后,frontend/src/store/chat.ts 只发起、只播放);用户点重试本轮时的措辞;
- /copilot/review 按这一轮试过几次给 retry_instruction:可以重来时是那句话,不重来是 null。
问题、字段名都是虚构的。
@@ -112,7 +111,7 @@ def test_retry_instruction_two_wordings():
# --------------------------------------------------------------------------
# 前后端对照
# 页面不再留一份同样的决定
# --------------------------------------------------------------------------
@@ -120,30 +119,23 @@ def _ts() -> str:
return CHAT_TS.read_text(encoding="utf-8")
def test_max_attempts_matches_the_frontend():
found = re.search(r"const MAX_ATTEMPTS = (\d+)", _ts())
assert found and int(found.group(1)) == turn.MAX_ATTEMPTS
def test_the_page_no_longer_keeps_its_own_copy_of_the_decisions():
"""一轮整个搬到服务端之后(docs/design/orchestration-round §1),问数据页只发起、只播放:要不要运行、要不要重来、重来说
什么、入口怎么填都只在这里(和 chat/drive.py),页面上那份退路(MAX_ATTEMPTS、fillInput、retryInstruction)去掉了——
两份并存迟早走样,界面层也不放逻辑。以前这里逐字对照两边的常量和措辞,现在守的是页面上没有这几样。"""
ts = _ts()
for name in ("const MAX_ATTEMPTS", "function fillInput", "function retryInstruction", "streamCopilot(",
"api.copilot.review(", "api.runs.start(", "patchTurn("):
assert name not in ts, f"chat.ts 里又出现了 {name}:这一轮的决定、发起、落库应在服务端"
def test_the_question_field_pattern_matches_the_frontend():
found = re.search(r"/(question\|[^/]+)/i\.test", _ts())
assert found, "chat.ts 的 fillInput 里找不到认问题字段的正则"
assert found.group(1) == turn.QUESTION_FIELD.pattern and turn.QUESTION_FIELD.flags & re.IGNORECASE
def _template(source: str) -> str:
"""一段 TS 里由 + 连起来的模板字符串 → 一个 Python format 模板(${x} → {x},\\n → 换行)。"""
pieces = re.findall(r"`([^`]*)`", source)
text = "".join(pieces).replace("\\n", "\n")
return re.sub(r"\$\{(\w+)\}", r"{\1}", text)
def test_the_retry_wordings_match_the_frontend():
body = re.search(r"function retryInstruction\([^)]*\): string \{(.*?)\n\}", _ts(), re.S)
assert body, "chat.ts 里找不到 retryInstruction"
with_base, scratch = re.split(r"\n\s*: ", body.group(1).split("hasBase", 1)[1].split("?", 1)[1], maxsplit=1)
assert _template(with_base) == turn.RETRY_WITH_BASE
assert _template(scratch) == turn.RETRY_FROM_SCRATCH
def test_user_retry_wordings():
"""用户点「重试本轮」时交给助手的话(以前在页面 retryTurn 里拼):有图时说「没有给出可用的结果」,没图的同从零重建。"""
with_base = turn.user_retry_instruction("上周销售额", "已手动停止", True)
assert with_base == ("这个工作流上一次没有给出可用的结果:已手动停止。\n\n请针对这个原因改它(比如调大最大步数、改正工具参数、"
"换个查询方式),改完继续回答原来的问题:上周销售额")
assert turn.user_retry_instruction("上周销售额", "已手动停止", False) == turn.retry_instruction(
"上周销售额", "已手动停止", False)
# --------------------------------------------------------------------------
@@ -172,12 +164,13 @@ async def _run(output: dict, graph: dict, code: str) -> str:
return run.id
def test_the_frontend_retries_an_error_the_server_says_to_retry():
"""问数据页照服务端给的 next 走:error 操作带 next=retry(只写了思考、一个操作都没给)时带着原因重来一次,不再直接报
「助手本轮未完成」。读 chat.ts 里 error 那一支(行为另由 scripts/check-chat.mjs 的 emptyretry 一节在浏览器里守)。"""
branch = re.search(r"case 'error':(.*?)break\n", _ts(), re.S)
assert branch, "chat.ts 里找不到 error 那一支"
assert re.search(r"op\.next === 'retry'", branch.group(1)) and "retry(" in branch.group(1), branch.group(1)
def test_the_page_says_the_server_is_retrying_an_empty_generation():
"""只写了思考、一个操作都没给(error.next=retry)时,重来一次是服务端定的(chat/drive.py 照 next_step 走,问数据的一轮记一条
turn.retry,kind=empty);页面照它把这一轮重置、写「助手上一次没有给出任何操作,正在重新生成工作流…」。读 chat.ts 里
turn.retry 那一支(行为另由 tests/test_chat_turns.py 和 scripts/check-chat.mjs 的 emptyretry 一节守)。"""
branch = re.search(r"function retryPatch\((.*?)\n\}", _ts(), re.S)
assert branch, "chat.ts 里找不到 retryPatch"
assert re.search(r"d\.kind === 'empty' \? CHAT_RETRY_TEXT\.empty", branch.group(1)), branch.group(1)
async def test_answer_ops_says_to_retry_an_empty_generation(client, monkeypatch):
+23
View File
@@ -834,3 +834,26 @@ def test_attempt_records_say_what_the_page_used_to_say():
legacy = record.attempt_record(row(status="error", error="x"), {"failure": {"title": "操作超时", "reason": "查询超时。"}}, n=1)
assert legacy["summary"] == "操作超时:查询超时。"
assert record.attempt_record(row(answer="字" * 5000), {}, n=1)["answer"].endswith("…")
async def test_an_empty_generation_is_retried_once_on_the_server(client, monkeypatch):
"""助手只写了思考、一个操作都没给(error.next=retry):服务端带着原因重来一次(turn.retry kind=empty),不用页面在。"""
import app.api.copilot as copilot
from tests.test_usage_meter import Scripted
reply = json.dumps({"op": "reply", "text": "上周没有华中区的数据"}, ensure_ascii=False)
model = Scripted(model="claude-opus-5", replies=[[""], [reply]])
async def _model(*_a, **_k):
return model, "claude-opus-5"
monkeypatch.setattr(copilot, "get_chat_model", _model)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, "上周华中区销售额")
await _done(turn_id)
events = await _events(turn_id)
[retry] = [e["data"] for e in events if e["type"] == "turn.retry"]
assert retry["kind"] == "empty" and retry["reason"] == "模型没有给出任何修改,也没有说明原因"
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, "两次建图都是这一轮花的"
+153
View File
@@ -0,0 +1,153 @@
/**
* 问数据的一轮在服务端跑(docs/design/orchestration-round/00-总方案.md §1;服务端 backend/app/api/chat_turns.py),挂在
* api.chatTurns 上。页面只发起、只看:建图、运行、复核、按复核重来、落库都在服务端,页面断了这一轮也会跑完。
*
* - start:发起一轮(新问,或者对已有的一轮重试、重新运行、放宽步数、跑一下、继续运行、接着收尾),202 {turn_id, turn};
* - get:这一轮的任务现在怎样;cancel:停止(取消生成、取消运行);
* - streamChatTurn:这一轮的事件日志(SSE)。先回放 seq > after 的,再推实时;断了按收到的最后一个 seq 自动重连,补齐、
* 不重复;45 秒什么都没收到(服务端每 15 秒有一拍心跳)就当连接死了,重连。服务端说这一轮结束了(stream.end)才停。
*
* 请求函数从 ./http 导入,不从 ./client 导入:client.ts 要导入这个文件把它挂到 api 上,反过来导入就成了循环。
*/
import type { ConversationTurn } from '../types'
import { ApiError, BASE, NETWORK_MESSAGE, actorHeader, failure, get, parseBody, post, report } from './http'
/** 发起一轮的方式(服务端 app/chat/job.MODES) */
export type ChatTurnMode = 'ask' | 'retry' | 'rerun' | 'max_steps' | 'run' | 'continue' | 'resume'
export interface ChatTurnStart {
conversation_id: string
question?: string
/** 只查这几个数据源。新问不给是不限;重试不给沿用这一轮的,空数组是不限 */
datasource_ids?: string[]
retry_of?: string
mode?: ChatTurnMode
max_steps?: number
}
/** 事件日志里的一条(服务端 app/chat/job.py 的模块说明列了各种 type) */
export interface ChatTurnEvent {
seq: number
type: string
data: Record<string, any>
ts: number
}
/** 实时流结束时的情况:status 是服务端说的任务状态(stream.end);error 是这一轮读不了(不存在、没权限) */
export interface ChatTurnStreamEnd {
status?: string
error?: string
httpStatus?: number
}
const turn = (id: string) => `/chat/turns/${encodeURIComponent(id)}`
export const chatTurns = {
start: (body: ChatTurnStart) => post<{ turn_id: string; turn: ConversationTurn | null }>('/chat/turns', body),
get: (id: string) => get<{ id: string; conversation_id: string; job: ConversationTurn['job']; turn: ConversationTurn }>(turn(id)),
cancel: (id: string) => post<{ ok: boolean }>(`${turn(id)}/cancel`, {}),
}
/** 多久什么都没收到就当连接死了(服务端心跳 15 秒一拍) */
const STALL_MS = 45_000
/** 连不上时重连的退避上限 */
const RETRY_MAX_MS = 5000
/**
* 订阅这一轮的事件日志。返回取消函数(关掉连接、不再重连)。
*
* 连接正常结束却没收到 stream.end(服务端重启、代理断开):按最后一个 seq 马上重连;连不上(网络断了、服务端在重启)按
* 退避重试,直到服务端回来——这一轮在服务端照跑,回来之后补齐中间漏掉的那段
*/
export function streamChatTurn(
turnId: string,
after: number,
onEvent: (event: ChatTurnEvent) => void,
onEnd: (end: ChatTurnStreamEnd) => void,
): () => void {
let closed = false
let lastSeq = after
let failures = 0
let controller: AbortController | null = null
let stall: ReturnType<typeof setTimeout> | undefined
let again: ReturnType<typeof setTimeout> | undefined
const finish = (end: ChatTurnStreamEnd) => {
if (closed) return
closed = true
clearTimeout(stall)
onEnd(end)
}
const arm = () => {
clearTimeout(stall)
stall = setTimeout(() => controller?.abort(), STALL_MS)
}
const connect = async () => {
if (closed) return
controller = new AbortController()
/** 这一次连接收到了事件(不只是心跳) */
let got = false
let alive = false
try {
arm()
const path = `${turn(turnId)}/stream?after=${lastSeq}`
const res = await fetch(BASE + path, { headers: { Accept: 'text/event-stream', ...actorHeader() }, signal: controller.signal })
if (!res.ok || !res.body) {
const err = failure(res.status, res.statusText, await res.text().catch(() => ''), path)
// 这一轮读不了(删了、不是你的):再连也一样
if (err.kind !== 'network' && res.status < 500) {
finish({ error: err.message, httpStatus: res.status })
return
}
throw err
}
report(true)
alive = true
const reader = res.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
for (;;) {
const { done, value } = await reader.read()
if (done) break
arm()
buffer += decoder.decode(value, { stream: true })
let idx: number
while ((idx = buffer.indexOf('\n\n')) >= 0) {
const frame = buffer.slice(0, idx)
buffer = buffer.slice(idx + 2)
if (!frame.startsWith('data: ')) continue // 心跳(: ping)
let event: any
try { event = parseBody(frame.slice(6)) } catch { continue }
if (!event || typeof event !== 'object') continue
if (event.type === 'stream.end') {
finish({ status: event.data?.status ?? event.status })
controller.abort()
return
}
if (typeof event.seq !== 'number' || event.seq <= lastSeq) continue // 重连时重叠的那段只要一份
lastSeq = event.seq
got = true
if (!closed) onEvent(event as ChatTurnEvent)
}
}
} catch (e: any) {
if (closed) return
if (e instanceof TypeError) report(false, new ApiError(0, NETWORK_MESSAGE, { kind: 'network', raw: String(e) }))
}
if (closed) return
clearTimeout(stall)
// 收到了东西就马上接着连;只有心跳的稍等一下;连不上的按退避
failures = got ? 0 : alive ? 0 : failures + 1
const delay = got ? 0 : alive ? 300 : Math.min(300 * 2 ** failures, RETRY_MAX_MS)
again = setTimeout(() => void connect(), delay)
}
void connect()
return () => {
closed = true
clearTimeout(stall)
clearTimeout(again)
controller?.abort()
}
}
+5
View File
@@ -39,6 +39,8 @@ import { caliberLedger } from './caliberLedger'
import { capabilities, endpoints, notify } from './capabilities'
// 普通用户建应用(typesBlueprint.ts 是接口约定)
import { blueprints, files, personal, templates } from './blueprints'
// 问数据的一轮在服务端跑(编排完善第 1 波):发起、看、停,实时流在 chatTurns.ts 的 streamChatTurn
import { chatTurns } from './chatTurns'
// 请求层搬到了 http.ts(见那边的说明)。原先从这里导出的照旧从这里导出,调用方不用改;
// 请求函数也一并转出,给不在 api/ 下、又想直接发请求的地方用。api/ 下的命名空间文件请直接从 http.ts 导入
@@ -986,6 +988,9 @@ export const api = {
templates,
files,
personal,
// 问数据的一轮(服务端跑整轮:发起、看、停)
chatTurns,
}
/**
+596 -985
View File
File diff suppressed because it is too large Load Diff
+16
View File
@@ -1646,6 +1646,22 @@ export interface ConversationTurn {
* (conversations.BUILD_STALE),前端判断同一件事时用同一只钟。老后端没有
*/
updated_at?: string
/** 这一轮在服务端的任务(问数据整轮在服务端跑,backend/app/chat/job.job_view)。这一波之前的老轮次没有,为 null */
job?: ChatTurnJob | null
}
/** 问数据一轮在服务端的任务现在怎样。status 用后台任务的状态词;stage 是在哪一步;attempt_seq 是这一次尝试从哪一条事件开始 */
export interface ChatTurnJob {
status: 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled' | 'interrupted'
stage: 'build' | 'run' | 'waiting' | 'review' | null
mode: string
run_id: string | null
attempt: number
last_seq: number
attempt_seq: number
error: string | null
/** 等不及的那次复核还在后台跑 */
late_review?: boolean
}
/**
+989 -256
View File
File diff suppressed because it is too large Load Diff