问数据的心跳只落换了阶段的那一拍:浏览器检查发现刷新后过程里「规划」那一行没了——页面按心跳的阶段画那一行,全都不落库就回放不出来。现在换了阶段的那一拍(规划 → 查看 → 搭建……)照落,同一阶段接着的保活心跳才只在流上发、不落库;页面接着的心跳顶替上一拍、不叠成好几条。检查脚本的伪服务端照同一个规矩,「一条只收一份」比心跳之外的操作

This commit is contained in:
ThearchyHelios
2026-10-19 10:41:06 +08:00
parent 14001c0900
commit 76290baaf1
5 changed files with 44 additions and 20 deletions
+11 -4
View File
@@ -16,7 +16,8 @@
**事件日志**(chat/log.py;每条 {seq, type, data, ts}): **事件日志**(chat/log.py;每条 {seq, type, data, ts}):
turn.attempt {mode, attempts, scope, max_steps?} 这一次尝试开始了(发起、重试本轮、重新运行、跑一下、继续运行、接着收尾) turn.attempt {mode, attempts, scope, max_steps?} 这一次尝试开始了(发起、重试本轮、重新运行、跑一下、继续运行、接着收尾)
op {...} 助手的一条操作原样(含查看、final、usage;心跳不落库,只在实时流上发,见 chat/log.beat) op {...} 助手的一条操作原样(含查看、final、usage;心跳只落换了阶段的那一拍,同一阶段接着的只在实时流上发,
见 chat/log.beat)
stage {stage} 到了哪一步:build 建图、run 运行、waiting 停在审批上等人、review 复核 stage {stage} 到了哪一步:build 建图、run 运行、waiting 停在审批上等人、review 复核
run {run_id, status, run_class} 发起了一次运行(页面据此接上运行自己的实时流看过程) run {run_id, status, run_class} 发起了一次运行(页面据此接上运行自己的实时流看过程)
review {result} 复核结论原样(含 usage、retry_instruction) review {result} 复核结论原样(含 usage、retry_instruction)
@@ -141,6 +142,8 @@ class ChatDriver(TurnDriver):
self.execution = execution self.execution = execution
self._thinking = "" self._thinking = ""
self._thinking_at = time.monotonic() self._thinking_at = time.monotonic()
#: 最近一次落库的心跳的阶段:同一阶段接着的心跳不落库(on_op)
self._beat_phase: Any = None
#: 这一轮的查看记录(存进 meta.looks) #: 这一轮的查看记录(存进 meta.looks)
self.looks: list[dict[str, Any]] = [] self.looks: list[dict[str, Any]] = []
#: 等不及的那次复核(还在跑):收尾之后由它把用量记进这一轮 #: 等不及的那次复核(还在跑):收尾之后由它把用量记进这一轮
@@ -161,9 +164,12 @@ class ChatDriver(TurnDriver):
return return
await self._flush_thinking() await self._flush_thinking()
if kind == "heartbeat": if kind == "heartbeat":
# 纯保活:只在实时流上发、不落库(chat/log.beat;以前每 3 秒落一条) if op.get("phase") == self._beat_phase:
turn_log.beat(self.execution.turn_id, op) # 同一阶段接着的心跳是纯保活:只在实时流上发、不落库(chat/log.beat;以前每 3 秒落一条)
return turn_log.beat(self.execution.turn_id, op)
return
# 换了阶段的那一拍(规划 → 查看 → 搭建……)说的是到了哪一步,照落:刷新、重开时过程里那一行阶段要回放得出来
self._beat_phase = op.get("phase")
await self.execution.emit("op", op) await self.execution.emit("op", op)
if kind == "look" and (look := record.look_event(op)) is not None: if kind == "look" and (look := record.look_event(op)) is not None:
self.looks.append(look) self.looks.append(look)
@@ -179,6 +185,7 @@ class ChatDriver(TurnDriver):
self._thinking_at = time.monotonic() self._thinking_at = time.monotonic()
async def _generate(self, instruction: str, base_graph: dict[str, Any] | None) -> tuple[list[dict[str, Any]], str | None]: async def _generate(self, instruction: str, base_graph: dict[str, Any] | None) -> tuple[list[dict[str, Any]], str | None]:
self._beat_phase = None # 每一次生成从头算:第一拍心跳照落(页面按它画「规划」那一行)
try: try:
return await super()._generate(instruction, base_graph) return await super()._generate(instruction, base_graph)
finally: finally:
+5 -4
View File
@@ -12,10 +12,11 @@
{type: stream.end, status}(任务的状态)再关;隔一拍没有新事件发一行注释心跳「: ping」(页面据此知道连接还活着)。 {type: stream.end, status}(任务的状态)再关;隔一拍没有新事件发一行注释心跳「: ping」(页面据此知道连接还活着)。
事件种类见 app/chat/job.py 的模块说明。 事件种类见 app/chat/job.py 的模块说明。
**助手的心跳不落库。** 模型几秒没出声时助手发一拍 {op: heartbeat, phase, elapsed_ms}(页面据此写阶段、已等多久):它是纯保活, **助手的心跳只落换了阶段的那一拍。** 模型几秒没出声时助手发一拍 {op: heartbeat, phase, elapsed_ms}(页面据此写阶段、已等多久):
以前每 3 秒落一条,一轮几百条。现在只记在本进程里(beat:每一轮最近的一拍),流上发一条不带 seq 的 {type: op, data};重连时流 以前每 3 秒落一条,一轮几百条。现在换了阶段的那一拍(规划 → 查看 → 搭建……,说的是到了哪一步)照落,刷新、重开时过程里那一行
先补最近的那一拍。之后落了任何一条事件,那一拍就作废(心跳只说「上一条事件之后还在等」,重连时补旧的会让页面的进度退回去)。 阶段回放得出来;同一阶段接着的那些是纯保活,只记在本进程里(beat:每一轮最近的一拍),流上发一条不带 seq 的 {type: op, data};
其余事件照落:第 2 波跨进程的实时流从库里按 seq 追,别的进程的心跳那时由流按任务状态自己合成。 重连时流先补最近的那一拍。之后落了任何一条事件,那一拍就作废(心跳只说「上一条事件之后还在等」,重连时补旧的会让页面的进度
退回去)。其余事件照落:第 2 波跨进程的实时流从库里按 seq 追,别的进程的保活心跳那时由流按任务状态自己合成。
**清理。** 已结束(成功、失败、取消、中断)超过 RETAIN_DAYS 天的轮次整轮删事件日志(purge,挂在调度循环的那一拍里,一小时最多一次): **清理。** 已结束(成功、失败、取消、中断)超过 RETAIN_DAYS 天的轮次整轮删事件日志(purge,挂在调度循环的那一拍里,一小时最多一次):
页面重开已结束的一轮读的是 conversation_turns 那一行(答案、图、复核、用量、几次尝试都在那里),只有还在进行中的轮次才回放 页面重开已结束的一轮读的是 conversation_turns 那一行(答案、图、复核、用量、几次尝试都在那里),只有还在进行中的轮次才回放
+9 -5
View File
@@ -947,8 +947,9 @@ async def test_a_stop_that_lands_while_launching_still_cancels_the_run(client, m
async def test_heartbeats_go_out_on_the_stream_but_are_not_stored(client, monkeypatch, source): async def test_heartbeats_go_out_on_the_stream_but_are_not_stored(client, monkeypatch, source):
"""心跳是纯保活(模型几秒没出声时助手发的一拍 heartbeat 操作):只在实时流上发、不落库,重连时流自己补最近的一拍; """心跳(模型几秒没出声时助手发的一拍 heartbeat 操作):换了阶段的那一拍照落(刷新后过程里那一行阶段回放得出来),同一阶段
落库的事件(思考、操作、阶段……)照旧。以前每 3 秒落一条,一轮下来几百条心跳。""" 接着的是纯保活,只在实时流上发、不落库,重连时流自己补最近的一拍;别的事件(思考、操作、阶段……)照旧。以前每 3 秒落一条,
一轮下来几百条心跳。"""
from app.api import copilot from app.api import copilot
gate = asyncio.Event() gate = asyncio.Event()
@@ -969,10 +970,10 @@ async def test_heartbeats_go_out_on_the_stream_but_are_not_stored(client, monkey
event = json.loads(frame[6:]) event = json.loads(frame[6:])
if event.get("type") == "op" and event["data"].get("op") == "heartbeat": if event.get("type") == "op" and event["data"].get("op") == "heartbeat":
beats.append(event) beats.append(event)
if len(beats) >= 2: if any("seq" not in b for b in beats):
break break
await stream.aclose() await stream.aclose()
assert all("seq" not in b for b in beats), "心跳不占 seq:重连按 seq 补齐时不会多出来、也不会漏" assert [b for b in beats if "seq" not in b], "同一阶段接着的心跳不占 seq:重连按 seq 补齐时不会多出来、也不会漏"
# 重连(从最后一条落库的事件接着读):流先补一拍最近的心跳,页面接着知道还在等模型 # 重连(从最后一条落库的事件接着读):流先补一拍最近的心跳,页面接着知道还在等模型
stored = await _events(turn_id) stored = await _events(turn_id)
again = turn_log.stream(turn_id, stored[-1]["seq"], active=active, heartbeat=0.2) again = turn_log.stream(turn_id, stored[-1]["seq"], active=active, heartbeat=0.2)
@@ -983,7 +984,10 @@ async def test_heartbeats_go_out_on_the_stream_but_are_not_stored(client, monkey
gate.set() gate.set()
await _done(turn_id) await _done(turn_id)
events = await _events(turn_id) events = await _events(turn_id)
assert not [e for e in events if e["type"] == "op" and e["data"].get("op") == "heartbeat"], "心跳一条都不落库" stored = [e for e in events if e["type"] == "op" and e["data"].get("op") == "heartbeat"]
# 只落换了阶段的那一拍(规划、查看……各一拍),同一阶段接着的几十拍都不落
phases = [e["data"]["phase"] for e in stored]
assert phases and phases[0] == "planning" and all(a != b for a, b in zip(phases, phases[1:])), phases
assert [e["seq"] for e in events] == list(range(1, len(events) + 1)) assert [e["seq"] for e in events] == list(range(1, len(events) + 1))
assert turn_log.latest_beat(turn_id) is None, "这一轮结束了,进程里记的那一拍跟着清掉" assert turn_log.latest_beat(turn_id) is None, "这一轮结束了,进程里记的那一拍跟着清掉"
+3 -1
View File
@@ -1329,7 +1329,9 @@ function opPatch(t: ChatTurn, op: CopilotOp): Partial<ChatTurn> {
const last = t.ops[t.ops.length - 1] const last = t.ops[t.ops.length - 1]
const ops = op.op === 'thinking' && last?.op === 'thinking' const ops = op.op === 'thinking' && last?.op === 'thinking'
? [...t.ops.slice(0, -1), { ...last, delta: String(last.delta ?? '') + String(op.delta ?? '') }] ? [...t.ops.slice(0, -1), { ...last, delta: String(last.delta ?? '') + String(op.delta ?? '') }]
: [...t.ops, op] // 心跳只说「还在等」:接着的一拍顶替上一拍(重连时服务端会补最近的一拍,chat/log.beat),不叠成好几条
: op.op === 'heartbeat' && last?.op === 'heartbeat' ? [...t.ops.slice(0, -1), op]
: [...t.ops, op]
switch (op.op) { switch (op.op) {
case 'thinking': case 'thinking':
return { ops, phase: 'planning', thinking: (t.thinking + (op.delta ?? '')).slice(-2000) } return { ops, phase: 'planning', thinking: (t.thinking + (op.delta ?? '')).slice(-2000) }
+16 -6
View File
@@ -1271,12 +1271,17 @@ async function driveTurn(T, epoch) {
{ cause: { kind: 'refused', message, code: got.json.code ?? null }, runId: null, usage: T.usage }) { cause: { kind: 'refused', message, code: got.json.code ?? null }, runId: null, usage: T.usage })
} }
const ops = [] const ops = []
T.beatPhase = undefined // 每一次生成从头算:第一拍心跳照落
for (const op of got.ops) { for (const op of got.ops) {
if (op.gate) { await until(() => ctl[op.gate] || gone(T, epoch), 15000); continue } if (op.gate) { await until(() => ctl[op.gate] || gone(T, epoch), 15000); continue }
if (gone(T, epoch)) return if (gone(T, epoch)) return
// 心跳不落库、不带 seq(服务端 chat/log.beat):只记最近的一拍,流上发一条不带 seq 的;之后落了新事件就作废 // 心跳只落换了阶段的那一拍(服务端 chat/job.ChatDriver.on_op);同一阶段接着的不落库、不带 seq(chat/log.beat):只记
if (op.op === 'heartbeat') T.beat = { op, at: T.events.length } // 最近的一拍,流上发一条不带 seq 的;之后落了新事件就作废
else emit(T, 'op', op) if (op.op === 'heartbeat' && op.phase === T.beatPhase) T.beat = { op, at: T.events.length }
else {
if (op.op === 'heartbeat') T.beatPhase = op.phase
emit(T, 'op', op)
}
if (op.op !== 'thinking' && op.op !== 'heartbeat') ops.push(op) if (op.op !== 'thinking' && op.op !== 'heartbeat') ops.push(op)
if (op.op === 'look') T.looks.push(op) if (op.op === 'look') T.looks.push(op)
if (op.op === 'usage') T.usage.copilot = addTotal(T.usage.copilot, op) if (op.op === 'usage') T.usage.copilot = addTotal(T.usage.copilot, op)
@@ -2636,9 +2641,14 @@ for (const theme of THEMES) {
check('从不往回要:重连的 seq 只增不减', afters.every((a, i) => i === 0 || a >= afters[i - 1]), afters.join(',')) check('从不往回要:重连的 seq 只增不减', afters.every((a, i) => i === 0 || a >= afters[i - 1]), afters.join(','))
check('补齐:服务端这一轮记下的事件,页面一条不少地收到了', !!T && t.turnSeq === T.events.length, check('补齐:服务端这一轮记下的事件,页面一条不少地收到了', !!T && t.turnSeq === T.events.length,
`${t.turnSeq} / ${T?.events.length}`) `${t.turnSeq} / ${T?.events.length}`)
const ops = T?.events.filter((e) => e.type === 'op').length ?? -1 // 心跳只落换了阶段的那一拍,同一阶段接着的不带 seq(服务端 chat/log.beat):重连时服务端会补最近的一拍,页面接着的那一拍
check('不重复:助手的操作一条只收一份', t.ops.length === ops && t.ops.filter((o) => o.op === 'add_node').length === 3, // 顶替上一拍,不另起一条。按 seq 补齐的「一条只收一份」比的是心跳之外的操作
`${t.ops.length} / ${ops}`) const ops = T?.events.filter((e) => e.type === 'op' && e.data?.op !== 'heartbeat').length ?? -1
const mine = t.ops.filter((o) => o.op !== 'heartbeat').length
check('不重复:助手的操作一条只收一份', mine === ops && t.ops.filter((o) => o.op === 'add_node').length === 3,
`${mine} / ${ops}`)
check('……补过的心跳不叠成好几条', t.ops.every((o, i) => o.op !== 'heartbeat' || t.ops[i + 1]?.op !== 'heartbeat'),
t.ops.map((o) => o.op).join(','))
check('画布边收边长到收尾:图有 3 个节点,运行接上了', t.graph?.nodes?.length === 3 && !!t.run?.id) check('画布边收边长到收尾:图有 3 个节点,运行接上了', t.graph?.nodes?.length === 3 && !!t.run?.id)
check('页面不写这一轮的结局(不 PATCH)', log.patches.slice(patchesBefore).length === 0) check('页面不写这一轮的结局(不 PATCH)', log.patches.slice(patchesBefore).length === 0)
await page.getByRole('button', { name: '停止这一轮' }).click().catch(() => {}) await page.getByRole('button', { name: '停止这一轮' }).click().catch(() => {})