diff --git a/README.md b/README.md index 2c2f2f2c..92895acc 100644 --- a/README.md +++ b/README.md @@ -1022,6 +1022,10 @@ pip install -e 'backend[microvm,dev,db,docs]' -c backend/requirements.lock # `.secret_key` 加密密钥。 删除 `data/` 即恢复出厂设置。 +问数据每一轮的事件日志（`agentlab.db` 的 `chat_turn_events`：助手的操作、阶段、运行、复核、终局，供页面先回放再接实时）只为 +进行中的轮次服务：已结束（成功、失败、取消、中断）超过 14 天的轮次由调度循环每小时清一次、整轮删除；重开已结束的一轮读的是 +对话里那一行（答案、图、复核、用量、几次尝试），不受影响。助手等模型时的心跳只在实时流上发、不落库。 + --- ## 常见问题 diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index 2d04bf4d..d993beab 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -16,7 +16,7 @@ **事件日志**（chat/log.py；每条 {seq, type, data, ts}）： turn.attempt {mode, attempts, scope, max_steps?} 这一次尝试开始了（发起、重试本轮、重新运行、跑一下、继续运行、接着收尾） - op {...} 助手的一条操作原样（含心跳、查看、final、usage） + op {...} 助手的一条操作原样（含查看、final、usage；心跳不落库，只在实时流上发，见 chat/log.beat） stage {stage} 到了哪一步：build 建图、run 运行、waiting 停在审批上等人、review 复核 run {run_id, status, run_class} 发起了一次运行（页面据此接上运行自己的实时流看过程） review {result} 复核结论原样（含 usage、retry_instruction） @@ -132,6 +132,10 @@ class ChatDriver(TurnDriver): await self._flush_thinking() return await self._flush_thinking() + if kind == "heartbeat": + # 纯保活：只在实时流上发、不落库（chat/log.beat；以前每 3 秒落一条） + turn_log.beat(self.execution.turn_id, op) + return await self.execution.emit("op", op) if kind == "look" and (look := record.look_event(op)) is not None: self.looks.append(look) @@ -336,6 +340,12 @@ class ChatTurnExecution(runtime.JobRun): # ---------------- 执行 ---------------- async def _run(self) -> None: + try: + await self._execute() + finally: + turn_log.forget(self.turn_id) # 进程里记的最近一拍心跳（chat/log.beat）跟着这一次执行收掉 + + async def _execute(self) -> None: from app.api.deps import set_principal from app.chat import turn as policy from app.db.models import ConversationTurn diff --git a/backend/app/chat/log.py b/backend/app/chat/log.py index 9e73c8b1..c41fdd59 100644 --- a/backend/app/chat/log.py +++ b/backend/app/chat/log.py @@ -11,28 +11,49 @@ 线上格式（SSE，每条一行 data: ）：{seq, type, data, ts}；回放完、这一轮也不在进行中了，补一条不带 seq 的 {type: stream.end, status}（任务的状态）再关；隔一拍没有新事件发一行注释心跳「: ping」（页面据此知道连接还活着）。 事件种类见 app/chat/job.py 的模块说明。 + +**助手的心跳不落库。** 模型几秒没出声时助手发一拍 {op: heartbeat, phase, elapsed_ms}（页面据此写阶段、已等多久）：它是纯保活， +以前每 3 秒落一条，一轮几百条。现在只记在本进程里（beat：每一轮最近的一拍），流上发一条不带 seq 的 {type: op, data}；重连时流 +先补最近的那一拍。之后落了任何一条事件，那一拍就作废（心跳只说「上一条事件之后还在等」，重连时补旧的会让页面的进度退回去）。 +其余事件照落：第 2 波跨进程的实时流从库里按 seq 追，别的进程的心跳那时由流按任务状态自己合成。 + +**清理。** 已结束（成功、失败、取消、中断）超过 RETAIN_DAYS 天的轮次整轮删事件日志（purge，挂在调度循环的那一拍里，一小时最多一次）： +页面重开已结束的一轮读的是 conversation_turns 那一行（答案、图、复核、用量、几次尝试都在那里），只有还在进行中的轮次才回放 +事件日志。任务行留着：last_seq 接着发号，之后在这一轮上重试、继续运行不撞号。 """ from __future__ import annotations import asyncio import contextlib +import itertools import json +import logging import time from collections import defaultdict from collections.abc import AsyncIterator, Awaitable, Callable, Iterator +from datetime import datetime, timedelta from typing import Any -from sqlalchemy import func, select, update +from sqlalchemy import delete, func, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.db.base import SessionLocal +logger = logging.getLogger(__name__) + #: 隔多久没有新事件就发一拍心跳（秒），也是订阅方回库里看一次的间隔 HEARTBEAT_S = 15.0 #: 一次回放最多读多少条 READ_BATCH = 500 #: 回放、实时推送完、这一轮不在进行中时的结束标记（不落库、不带 seq） STREAM_END = "stream.end" +#: 已结束的轮次的事件日志留几天 +RETAIN_DAYS = 14 +#: 清理多久最多跑一次（秒）、一次最多清几轮 +PURGE_EVERY_S = 3600.0 +PURGE_BATCH = 500 +#: 结束了的任务状态（core/job_runtime 的状态词里不再往下走的那几种） +FINISHED = ("succeeded", "failed", "cancelled", "interrupted") class _Waiters: @@ -67,6 +88,27 @@ def notify(turn_id: str) -> None: waiters.notify(turn_id) +#: 每一轮最近的一拍心跳：(第几拍, 操作)。只在本进程里，不落库（见模块说明） +_beats: dict[str, tuple[int, dict[str, Any]]] = {} +_beat_no = itertools.count(1) + + +def beat(turn_id: str, op: dict[str, Any]) -> None: + """助手的一拍心跳：记下最近的这一拍、叫醒订阅方，由流发出去（不落库、不占 seq）。""" + _beats[turn_id] = (next(_beat_no), dict(op)) + notify(turn_id) + + +def latest_beat(turn_id: str) -> dict[str, Any] | None: + got = _beats.get(turn_id) + return dict(got[1]) if got else None + + +def forget(turn_id: str) -> None: + """这一轮结束了：进程里记的那一拍清掉。""" + _beats.pop(turn_id, None) + + async def add(session: AsyncSession, turn_id: str, type_: str, data: dict[str, Any] | None = None) -> int: """在调用方的事务里加一条（不提交），交回它的 seq。 @@ -91,6 +133,8 @@ async def append(session: AsyncSession, turn_id: str, type_: str, data: dict[str """追加一条，提交，叫醒订阅方。交回它的 seq（见 add）。""" seq = await add(session, turn_id, type_, data) await session.commit() + # 落了新事件，上一拍心跳作废：重连时不再补它（心跳只说「上一条事件之后还在等」） + _beats.pop(turn_id, None) notify(turn_id) return seq @@ -124,8 +168,10 @@ async def stream(turn_id: str, after: int, *, active: Callable[[], Awaitable[str """SSE 的每一帧：先回放 seq > after 的，再推实时的；这一轮不在进行中、也没有新的了，补一条 stream.end 再结束。 active() 交回这一轮还在进行中时为 None，不在进行中时交回任务的状态（stream.end 里带的就是它）。""" - beat = HEARTBEAT_S if heartbeat is None else heartbeat + interval = HEARTBEAT_S if heartbeat is None else heartbeat last = after + #: 这条连接发过的最近一拍助手心跳（_beats 的第几拍）：新连上的从 0 起，先补最近的那一拍 + sent = 0 while True: with waiters.watch(turn_id) as woke: rows = await read(turn_id, last) @@ -138,11 +184,47 @@ async def stream(turn_id: str, after: int, *, active: Callable[[], Awaitable[str if status is not None: yield frame({"type": STREAM_END, "status": status, "data": {"status": status}}) return + got = _beats.get(turn_id) + if got is not None and got[0] > sent: + sent = got[0] + # 不带 seq：页面按 seq 补齐时不算它（frontend/src/api/chatTurns.ts） + yield frame({"type": "op", "data": got[1], "ts": time.time()}) + continue try: - await asyncio.wait_for(woke.wait(), timeout=beat) + await asyncio.wait_for(woke.wait(), timeout=interval) except asyncio.TimeoutError: yield PING -__all__ = ["HEARTBEAT_S", "PING", "READ_BATCH", "STREAM_END", "add", "append", "frame", "notify", "read", "stream", "waiters", - "wire"] +async def purge(now: datetime, *, retain_days: float = RETAIN_DAYS, limit: int = PURGE_BATCH) -> list[str]: + """结束超过 retain_days 天的轮次整轮删事件日志（见模块说明），交回清了哪几轮。任务行、conversation_turns 那一行不动。 + 结束的时刻按任务的 finished_at，没记的按 updated_at。""" + from app.db.models_chat import ChatTurnEvent, ChatTurnJob + + cutoff = now - timedelta(days=retain_days) + async with SessionLocal() as session: + ended = select(ChatTurnJob.id).where(ChatTurnJob.status.in_(FINISHED), + func.coalesce(ChatTurnJob.finished_at, ChatTurnJob.updated_at) < cutoff) + turns = list((await session.execute(select(ChatTurnEvent.turn_id).where(ChatTurnEvent.turn_id.in_(ended)) + .distinct().limit(limit))).scalars()) + if turns: + await session.execute(delete(ChatTurnEvent).where(ChatTurnEvent.turn_id.in_(turns)) + .execution_options(synchronize_session=False)) + await session.commit() + return turns + + +_last_purge = 0.0 + + +async def purge_tick(now: datetime) -> list[str]: + """调度循环每一拍调（scheduler/loop.process_due）：一小时最多清一次。""" + global _last_purge + if time.monotonic() - _last_purge < PURGE_EVERY_S and _last_purge: + return [] + _last_purge = time.monotonic() + return await purge(now) + + +__all__ = ["FINISHED", "HEARTBEAT_S", "PING", "READ_BATCH", "RETAIN_DAYS", "STREAM_END", "add", "append", "beat", "forget", + "frame", "latest_beat", "notify", "purge", "purge_tick", "read", "stream", "waiters", "wire"] diff --git a/backend/app/scheduler/loop.py b/backend/app/scheduler/loop.py index 59948cb6..6d231434 100644 --- a/backend/app/scheduler/loop.py +++ b/backend/app/scheduler/loop.py @@ -141,16 +141,19 @@ async def process_due(now: datetime | None = None) -> None: """流程的一拍（app/process）：待办的提醒、升级、超时驳回（永不自动批准）；结果未知的写入按退避去查； 上传了却一直没提交的文件过了时限清掉（app/files，每小时一次）；访问密钥的请求记录批量写库、过了保留期的清掉 （auth/keys.py，每小时清一次：每个请求同步写一行会和运行抢 SQLite 的写锁，所以攒到这一拍写）；每天到点发待办汇总、 - 访问密钥到期前 7 天和 1 天各提醒一次（notify/digest.py，04 细案 B3：按日期认领、按审计去重，推送放在后台，不拖这一拍）。 + 访问密钥到期前 7 天和 1 天各提醒一次（notify/digest.py，04 细案 B3：按日期认领、按审计去重，推送放在后台，不拖这一拍）； + 已结束超过保留期的问数据轮次整轮删事件日志（chat/log.purge_tick，每小时一次）。 各自出错只记日志：这一拍出错不该让调度循环停下，下一拍照常再查。""" from app.auth import keys + from app.chat import log as chat_log from app.files import store as files_store from app.notify import digest from app.process import actions, tasks for name, step in (("待办的时限处理", tasks.tick), ("写入结果的查询", actions.tick), ("没有提交的上传文件清理", files_store.cleanup_tick), ("访问密钥的请求记录", keys.tick), - ("每天的待办汇总", digest.digest_due), ("访问密钥快到期的提醒", digest.keys_due)): + ("每天的待办汇总", digest.digest_due), ("访问密钥快到期的提醒", digest.keys_due), + ("已结束的问数据轮次的事件日志清理", chat_log.purge_tick)): try: done = await step(now) if isinstance(done, list) and done: diff --git a/backend/tests/test_chat_turns.py b/backend/tests/test_chat_turns.py index fff8502e..b0d2c12d 100644 --- a/backend/tests/test_chat_turns.py +++ b/backend/tests/test_chat_turns.py @@ -239,7 +239,8 @@ async def test_the_stream_replays_then_follows_and_reconnects_by_seq(client, mon first: list[dict[str, Any]] = [] stream = turn_log.stream(turn_id, 0, active=active, heartbeat=0.2) async for frame in stream: - if frame.startswith("data: "): + # 助手的心跳不带 seq、不落库（chat/log.beat）：按 seq 补齐时不算它，同页面 + if frame.startswith("data: ") and "seq" in json.loads(frame[6:]): first.append(json.loads(frame[6:])) if any(e.get("type") == "stage" for e in first): break @@ -253,7 +254,9 @@ async def test_the_stream_replays_then_follows_and_reconnects_by_seq(client, mon if frame == turn_log.PING: pings += 1 continue - rest.append(json.loads(frame[6:])) + event = json.loads(frame[6:]) + if "seq" in event or event.get("type") == "stream.end": + rest.append(event) assert rest[-1]["type"] == "stream.end" and rest[-1]["status"] == "succeeded" got = [e["seq"] for e in first + rest[:-1]] assert got == list(range(1, len(got) + 1)), "补齐、不重复" @@ -935,3 +938,129 @@ async def test_a_stop_that_lands_while_launching_still_cancels_the_run(client, m assert (await session.get(Run, job.run_id)).status == "cancelled", "这次运行跟着取消了，没有没人管的运行" row = await _row(turn_id) assert row.status == "error" and row.error == "已取消" and row.meta["runId"] == job.run_id + + +# -------------------------------------------------------------------------- +# 事件日志：心跳不落库、已结束的轮次按期清理 +# -------------------------------------------------------------------------- + + +async def test_heartbeats_go_out_on_the_stream_but_are_not_stored(client, monkeypatch, source): + """心跳是纯保活（模型几秒没出声时助手发的一拍 heartbeat 操作）：只在实时流上发、不落库，重连时流自己补最近的一拍； + 落库的事件（思考、操作、阶段……）照旧。以前每 3 秒落一条，一轮下来几百条心跳。""" + from app.api import copilot + + gate = asyncio.Event() + es.use_copilot(monkeypatch, Gated(source["name"], gate)) + monkeypatch.setattr(copilot, "_HEARTBEAT_SECONDS", 0.05) # use_copilot 把心跳调成了一小时一拍 + es.use_report(monkeypatch) + conversation_id = await _conversation(client) + turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]]) + await _stage(turn_id, "build") + + async def active() -> str | None: + return await chat_job.stream_status(turn_id) + + beats: list[dict[str, Any]] = [] + stream = turn_log.stream(turn_id, 0, active=active, heartbeat=0.2) + async for frame in stream: + if frame.startswith("data: "): + event = json.loads(frame[6:]) + if event.get("type") == "op" and event["data"].get("op") == "heartbeat": + beats.append(event) + if len(beats) >= 2: + break + await stream.aclose() + assert all("seq" not in b for b in beats), "心跳不占 seq：重连按 seq 补齐时不会多出来、也不会漏" + # 重连（从最后一条落库的事件接着读）：流先补一拍最近的心跳，页面接着知道还在等模型 + stored = await _events(turn_id) + again = turn_log.stream(turn_id, stored[-1]["seq"], active=active, heartbeat=0.2) + first = await again.__anext__() + await again.aclose() + assert first.startswith("data: ") and json.loads(first[6:])["data"]["op"] == "heartbeat" + + 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"], "心跳一条都不落库" + assert [e["seq"] for e in events] == list(range(1, len(events) + 1)) + assert turn_log.latest_beat(turn_id) is None, "这一轮结束了，进程里记的那一拍跟着清掉" + + +async def test_a_later_event_supersedes_the_last_heartbeat(): + """心跳只说「上一条事件之后还在等」：之后落了新事件，重连时不再补那一拍（不然页面的进度会退回去）。""" + async with SessionLocal() as session: + conversation = Conversation(kind="chat", title="虚构") + session.add(conversation) + await session.flush() + turn = ConversationTurn(conversation_id=conversation.id, question="q", status="running") + session.add(turn) + await session.flush() + session.add(ChatTurnJob(id=turn.id, conversation_id=conversation.id, status="running")) + await session.commit() + turn_id = turn.id + turn_log.beat(turn_id, {"op": "heartbeat", "phase": "planning", "elapsed_ms": 3000}) + assert turn_log.latest_beat(turn_id)["phase"] == "planning" + async with SessionLocal() as session: + await turn_log.append(session, turn_id, "stage", {"stage": "run"}) + assert turn_log.latest_beat(turn_id) is None + turn_log.forget(turn_id) + + +async def _finished_turn(*, days_ago: float, status: str = "succeeded", events: int = 3) -> str: + from datetime import timedelta + + from app.db.base import utcnow + + async with SessionLocal() as session: + conversation = Conversation(kind="chat", title="虚构的旧会话") + session.add(conversation) + await session.flush() + turn = ConversationTurn(conversation_id=conversation.id, question="虚构的问题", answer="虚构的回答", + status="done" if status == "succeeded" else "error", meta={"outcome": "done"}) + session.add(turn) + await session.flush() + when = utcnow() - timedelta(days=days_ago) + session.add(ChatTurnJob(id=turn.id, conversation_id=conversation.id, status=status, last_seq=events, + finished_at=None if status in ("queued", "running") else when)) + for seq in range(1, events + 1): + session.add(ChatTurnEvent(turn_id=turn.id, seq=seq, type="op", ts=0.0, data={"op": "thinking"})) + await session.commit() + return turn.id + + +async def test_finished_turns_lose_their_event_log_after_the_retention_period(): + """已结束超过保留期的轮次整轮删事件日志：页面重开已结束的一轮读的是 conversation_turns 那一行（答案、图、复核、用量都在 + 那里），不读事件日志；任务行留着，last_seq 接着发号，之后再在这一轮上重试、继续运行不撞号。还在跑的、刚结束的不动。""" + from app.db.base import utcnow + + old = await _finished_turn(days_ago=turn_log.RETAIN_DAYS + 1) + stuck = await _finished_turn(days_ago=turn_log.RETAIN_DAYS + 1, status="interrupted") + recent = await _finished_turn(days_ago=1) + live = await _finished_turn(days_ago=turn_log.RETAIN_DAYS + 1, status="running") + purged = await turn_log.purge(utcnow()) + assert set(purged) >= {old, stuck} and not {recent, live} & set(purged) + assert await _events(old) == [] and await _events(stuck) == [] + assert len(await _events(recent)) == 3 and len(await _events(live)) == 3 + async with SessionLocal() as session: + job = await session.get(ChatTurnJob, old) + row = await session.get(ConversationTurn, old) + assert job is not None and job.last_seq == 3, "任务行留着：seq 接着往下发" + assert row.answer == "虚构的回答", "那一行不动" + async with SessionLocal() as session: + assert await turn_log.add(session, old, "turn.attempt", {"mode": "retry"}) == 4 + await session.rollback() + + +async def test_the_cleanup_runs_in_the_background_tick(monkeypatch): + """清理挂在调度循环的那一拍里（scheduler/loop.process_due），一小时最多清一次。""" + from app.db.base import utcnow + from app.scheduler import loop + + old = await _finished_turn(days_ago=turn_log.RETAIN_DAYS + 2) + monkeypatch.setattr(turn_log, "_last_purge", 0.0) + await loop.process_due(utcnow()) + assert await _events(old) == [] + again = await _finished_turn(days_ago=turn_log.RETAIN_DAYS + 2) + await loop.process_due(utcnow()) + assert len(await _events(again)) == 3, "一小时内不再清第二次" diff --git a/frontend/src/api/chatTurns.ts b/frontend/src/api/chatTurns.ts index e3990c9b..34bf8c3d 100644 --- a/frontend/src/api/chatTurns.ts +++ b/frontend/src/api/chatTurns.ts @@ -125,7 +125,12 @@ export function streamChatTurn( controller.abort() return } - if (typeof event.seq !== 'number' || event.seq <= lastSeq) continue // 重连时重叠的那段只要一份 + if (typeof event.seq !== 'number') { + // 助手等模型时的心跳不落库、不带 seq（服务端 chat/log.beat）：照收、不动 lastSeq，重连时服务端会补最近的一拍 + if (event.type === 'op' && event.data?.op === 'heartbeat' && !closed) onEvent(event as ChatTurnEvent) + continue + } + if (event.seq <= lastSeq) continue // 重连时重叠的那段只要一份 lastSeq = event.seq got = true if (!closed) onEvent(event as ChatTurnEvent) diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index cabac70e..340eb304 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -1438,7 +1438,8 @@ function play(flow: Flow, serverId: string, after: number) { } return case 'op': - patch((t) => ({ ...opPatch(t, d as CopilotOp), turnSeq: e.seq })) + // 心跳不带 seq（不落库，chat/log.beat）：不动收到的最后一个 seq + patch((t) => ({ ...opPatch(t, d as CopilotOp), ...(typeof e.seq === 'number' ? { turnSeq: e.seq } : {}) })) return case 'stage': { const stage = String(d.stage ?? '') diff --git a/scripts/check-chat.mjs b/scripts/check-chat.mjs index ddd6238d..f1d5ab23 100644 --- a/scripts/check-chat.mjs +++ b/scripts/check-chat.mjs @@ -1274,7 +1274,9 @@ async function driveTurn(T, epoch) { for (const op of got.ops) { if (op.gate) { await until(() => ctl[op.gate] || gone(T, epoch), 15000); continue } if (gone(T, epoch)) return - emit(T, 'op', op) + // 心跳不落库、不带 seq（服务端 chat/log.beat）：只记最近的一拍，流上发一条不带 seq 的；之后落了新事件就作废 + if (op.op === 'heartbeat') T.beat = { op, at: T.events.length } + else 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) @@ -1500,6 +1502,9 @@ async function streamTurn(route, T, after) { await until(() => T.events.length > after || !turnActive(T) || T.gen !== srv.gen, 6000) const fresh = T.events.filter((e) => e.seq > after) let body = fresh.map((e) => `data: ${JSON.stringify(e)}\n\n`).join('') + if (turnActive(T) && T.beat && T.beat.at === T.events.length) { + body += `data: ${JSON.stringify({ type: 'op', data: T.beat.op, ts: Date.now() / 1000 })}\n\n` + } if (!turnActive(T) && after + fresh.length >= T.events.length) { body += `data: ${JSON.stringify({ type: 'stream.end', status: T.job.status, data: { status: T.job.status } })}\n\n` }