diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index b0c24e06..f93b59ab 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -16,7 +16,8 @@ **事件日志**（chat/log.py；每条 {seq, type, data, ts}）： 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 复核 run {run_id, status, run_class} 发起了一次运行（页面据此接上运行自己的实时流看过程） review {result} 复核结论原样（含 usage、retry_instruction） @@ -141,6 +142,8 @@ class ChatDriver(TurnDriver): self.execution = execution self._thinking = "" self._thinking_at = time.monotonic() + #: 最近一次落库的心跳的阶段：同一阶段接着的心跳不落库（on_op） + self._beat_phase: Any = None #: 这一轮的查看记录（存进 meta.looks） self.looks: list[dict[str, Any]] = [] #: 等不及的那次复核（还在跑）：收尾之后由它把用量记进这一轮 @@ -161,9 +164,12 @@ class ChatDriver(TurnDriver): return await self._flush_thinking() if kind == "heartbeat": - # 纯保活：只在实时流上发、不落库（chat/log.beat；以前每 3 秒落一条） - turn_log.beat(self.execution.turn_id, op) - return + if op.get("phase") == self._beat_phase: + # 同一阶段接着的心跳是纯保活：只在实时流上发、不落库（chat/log.beat；以前每 3 秒落一条） + turn_log.beat(self.execution.turn_id, op) + return + # 换了阶段的那一拍（规划 → 查看 → 搭建……）说的是到了哪一步，照落：刷新、重开时过程里那一行阶段要回放得出来 + self._beat_phase = op.get("phase") await self.execution.emit("op", op) if kind == "look" and (look := record.look_event(op)) is not None: self.looks.append(look) @@ -179,6 +185,7 @@ class ChatDriver(TurnDriver): self._thinking_at = time.monotonic() async def _generate(self, instruction: str, base_graph: dict[str, Any] | None) -> tuple[list[dict[str, Any]], str | None]: + self._beat_phase = None # 每一次生成从头算：第一拍心跳照落（页面按它画「规划」那一行） try: return await super()._generate(instruction, base_graph) finally: diff --git a/backend/app/chat/log.py b/backend/app/chat/log.py index c41fdd59..7a104110 100644 --- a/backend/app/chat/log.py +++ b/backend/app/chat/log.py @@ -12,10 +12,11 @@ {type: stream.end, status}（任务的状态）再关；隔一拍没有新事件发一行注释心跳「: ping」（页面据此知道连接还活着）。 事件种类见 app/chat/job.py 的模块说明。 -**助手的心跳不落库。** 模型几秒没出声时助手发一拍 {op: heartbeat, phase, elapsed_ms}（页面据此写阶段、已等多久）：它是纯保活， -以前每 3 秒落一条，一轮几百条。现在只记在本进程里（beat：每一轮最近的一拍），流上发一条不带 seq 的 {type: op, data}；重连时流 -先补最近的那一拍。之后落了任何一条事件，那一拍就作废（心跳只说「上一条事件之后还在等」，重连时补旧的会让页面的进度退回去）。 -其余事件照落：第 2 波跨进程的实时流从库里按 seq 追，别的进程的心跳那时由流按任务状态自己合成。 +**助手的心跳只落换了阶段的那一拍。** 模型几秒没出声时助手发一拍 {op: heartbeat, phase, elapsed_ms}（页面据此写阶段、已等多久）： +以前每 3 秒落一条，一轮几百条。现在换了阶段的那一拍（规划 → 查看 → 搭建……，说的是到了哪一步）照落，刷新、重开时过程里那一行 +阶段回放得出来；同一阶段接着的那些是纯保活，只记在本进程里（beat：每一轮最近的一拍），流上发一条不带 seq 的 {type: op, data}； +重连时流先补最近的那一拍。之后落了任何一条事件，那一拍就作废（心跳只说「上一条事件之后还在等」，重连时补旧的会让页面的进度 +退回去）。其余事件照落：第 2 波跨进程的实时流从库里按 seq 追，别的进程的保活心跳那时由流按任务状态自己合成。 **清理。** 已结束（成功、失败、取消、中断）超过 RETAIN_DAYS 天的轮次整轮删事件日志（purge，挂在调度循环的那一拍里，一小时最多一次）： 页面重开已结束的一轮读的是 conversation_turns 那一行（答案、图、复核、用量、几次尝试都在那里），只有还在进行中的轮次才回放 diff --git a/backend/tests/test_chat_turns.py b/backend/tests/test_chat_turns.py index 46d7066e..f5f851b3 100644 --- a/backend/tests/test_chat_turns.py +++ b/backend/tests/test_chat_turns.py @@ -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): - """心跳是纯保活（模型几秒没出声时助手发的一拍 heartbeat 操作）：只在实时流上发、不落库，重连时流自己补最近的一拍； - 落库的事件（思考、操作、阶段……）照旧。以前每 3 秒落一条，一轮下来几百条心跳。""" + """心跳（模型几秒没出声时助手发的一拍 heartbeat 操作）：换了阶段的那一拍照落（刷新后过程里那一行阶段回放得出来），同一阶段 + 接着的是纯保活，只在实时流上发、不落库，重连时流自己补最近的一拍；别的事件（思考、操作、阶段……）照旧。以前每 3 秒落一条， + 一轮下来几百条心跳。""" from app.api import copilot 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:]) if event.get("type") == "op" and event["data"].get("op") == "heartbeat": beats.append(event) - if len(beats) >= 2: + if any("seq" not in b for b in beats): break 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) 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() await _done(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 turn_log.latest_beat(turn_id) is None, "这一轮结束了，进程里记的那一拍跟着清掉" diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index 3f6e4a90..5c40628f 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -1329,7 +1329,9 @@ function opPatch(t: ChatTurn, op: CopilotOp): Partial { const last = t.ops[t.ops.length - 1] const ops = op.op === 'thinking' && last?.op === 'thinking' ? [...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) { case 'thinking': return { ops, phase: 'planning', thinking: (t.thinking + (op.delta ?? '')).slice(-2000) } diff --git a/scripts/check-chat.mjs b/scripts/check-chat.mjs index f1d5ab23..ca9590ae 100644 --- a/scripts/check-chat.mjs +++ b/scripts/check-chat.mjs @@ -1271,12 +1271,17 @@ async function driveTurn(T, epoch) { { cause: { kind: 'refused', message, code: got.json.code ?? null }, runId: null, usage: T.usage }) } const ops = [] + T.beatPhase = undefined // 每一次生成从头算：第一拍心跳照落 for (const op of got.ops) { if (op.gate) { await until(() => ctl[op.gate] || gone(T, epoch), 15000); continue } if (gone(T, epoch)) return - // 心跳不落库、不带 seq（服务端 chat/log.beat）：只记最近的一拍，流上发一条不带 seq 的；之后落了新事件就作废 - if (op.op === 'heartbeat') T.beat = { op, at: T.events.length } - else emit(T, 'op', op) + // 心跳只落换了阶段的那一拍（服务端 chat/job.ChatDriver.on_op）；同一阶段接着的不落库、不带 seq（chat/log.beat）：只记 + // 最近的一拍，流上发一条不带 seq 的；之后落了新事件就作废 + 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 === 'look') T.looks.push(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('补齐：服务端这一轮记下的事件，页面一条不少地收到了', !!T && t.turnSeq === T.events.length, `${t.turnSeq} / ${T?.events.length}`) - const ops = T?.events.filter((e) => e.type === 'op').length ?? -1 - check('不重复：助手的操作一条只收一份', t.ops.length === ops && t.ops.filter((o) => o.op === 'add_node').length === 3, - `${t.ops.length} / ${ops}`) + // 心跳只落换了阶段的那一拍，同一阶段接着的不带 seq（服务端 chat/log.beat）：重连时服务端会补最近的一拍，页面接着的那一拍 + // 顶替上一拍，不另起一条。按 seq 补齐的「一条只收一份」比的是心跳之外的操作 + 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('页面不写这一轮的结局（不 PATCH）', log.patches.slice(patchesBefore).length === 0) await page.getByRole('button', { name: '停止这一轮' }).click().catch(() => {})