diff --git a/backend/app/api/conversations.py b/backend/app/api/conversations.py
index b3d54cf0..4d969ba3 100644
--- a/backend/app/api/conversations.py
+++ b/backend/app/api/conversations.py
@@ -24,7 +24,7 @@ from typing import Any
 
 from fastapi import APIRouter, Depends, Header, HTTPException
 from pydantic import BaseModel, Field
-from sqlalchemy import and_, delete, func, select
+from sqlalchemy import and_, delete, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import actor_of, current_principal
@@ -531,7 +531,13 @@ async def create_turn(
 
 
 async def new_turn(session: AsyncSession, conversation: Conversation, question: str) -> ConversationTurn:
-    """开一轮（status=running，只有问题），不提交：开轮接口和服务端发起的一轮（app/chat/job.start）共用。"""
+    """开一轮（status=running，只有问题），不提交：开轮接口和服务端发起的一轮（app/chat/job.start）共用。
+
+    先记活跃时间、再数下一轮排第几：记活跃时间是这个事务的第一句写，占住这段对话（SQLite 上拿到写锁、PostgreSQL 上锁住
+    这一行），同时开的两轮一个接一个数，不会都拿到同一个 seq。"""
+    now = utcnow()
+    await session.execute(update(Conversation).where(Conversation.id == conversation.id).values(last_active_at=now)
+                          .execution_options(synchronize_session=False))
     next_seq = int((await session.execute(
         select(func.coalesce(func.max(ConversationTurn.seq), -1))
         .where(ConversationTurn.conversation_id == conversation.id)
@@ -543,7 +549,7 @@ async def new_turn(session: AsyncSession, conversation: Conversation, question:
     )
     session.add(turn)
 
-    conversation.last_active_at = utcnow()
+    conversation.last_active_at = now
     # 第一轮的问题就是这次对话的标题。用户改过名就不要再覆盖回去
     if next_seq == 0 and conversation.title in ("", "新对话"):
         conversation.title = title_from(question)
@@ -624,6 +630,32 @@ async def recent_turns(
     return list(reversed(rows))
 
 
+#: 运行入口的 history 里每轮的答案截到多少字（同以前问数据页 historyOf 的 slice(0, 600)）
+RUN_HISTORY_CHARS = 600
+
+
+async def run_history(session: AsyncSession, conversation_id: str | None, *, before_seq: int,
+                      limit: int = HISTORY_TURNS) -> str:
+    """问数据一轮发起运行时填进入口的 history（`{{ input.history }}`）：只取这一轮之前的（seq < before_seq），照以前问数据页
+    historyOf 的规则——答案没有就用说明，都没有的跳过，每轮截 600 字，最多 limit 轮，「问：…\n答：…」之间空一行。
+
+    以前这一份是页面按它内存里的轮次拼的；一轮搬到服务端之后在这里拼。和 history_text 不一样的地方是只看这一轮之前的：
+    重试、重新运行较早的一轮时，后面接着问过的几轮不该进它的入口——那时候它还没被问过。"""
+    if not conversation_id:
+        return ""
+    rows = list((await session.execute(
+        select(ConversationTurn.question, ConversationTurn.answer, ConversationTurn.explanation)
+        .where(ConversationTurn.conversation_id == conversation_id, ConversationTurn.seq < before_seq)
+        .order_by(ConversationTurn.seq)
+    )).all())
+    lines = []
+    for question, answer, explanation in rows:
+        text = answer or explanation or ""
+        if question and text:
+            lines.append(f"问：{question}\n答：{text[:RUN_HISTORY_CHARS]}")
+    return "\n\n".join(lines[-limit:])
+
+
 async def history_text(
     session: AsyncSession, conversation_id: str | None, limit: int = HISTORY_TURNS
 ) -> str:
diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py
index 611ff17e..2d04bf4d 100644
--- a/backend/app/chat/job.py
+++ b/backend/app/chat/job.py
@@ -8,7 +8,7 @@
 **流程和评测是同一份**（chat/drive.TurnDriver）。问数据这边的差异写在 ChatDriver：
 - 生成带着对话（历史、上一轮的图）、限定的数据源；每一条操作写进事件日志（思考按 THINKING_FLUSH_S 合并成一条，几百条增量
   不逐条落库）；
-- 发起：来源 chat（origin 记对话和这一轮），入口填问题和前几轮的问答（conversations.history_text）；开登录后 user 角色照
+- 发起：来源 chat（origin 记对话和这一轮），入口填问题和这一轮之前的几轮问答（conversations.run_history）；开登录后 user 角色照
   chat/launch.user_chat_run（白名单、自己的记忆范围、遮罩、只读）——和以前浏览器发 POST /api/runs 时一字不差；
 - 停在审批上：照旧等人（评测是取消），审批过了接着等这次运行；没有待审批的断点（服务重启挂起）这一轮收成「服务重启，本轮已
   中断」，可以「继续运行」；
@@ -32,6 +32,10 @@
 
 **并发。** 归「交互」类（总方案 §5 W2-b）：这一波还是进程内的 core/job_runtime（一个 asyncio 任务一轮，不占数据源的队），
 和以前浏览器各推各的一样不设上限；第 2 波按库里的计数认领、按类别给并发上限和优先级（问数据优先于评测），租约到期别的进程接手。
+同时来的几次发起（两个标签页、连点几下）在库里排队：发起的核对和写入在一个事务里、第一句先占住这段对话（hold_conversation），
+同一段对话只起得来一轮、同一轮只排得上一次，第一条事件也在这个事务里写；事件的 seq 由任务行上的计数器发（chat/log.add）；
+同一行的几处写入（任务、晚到的复核、下一次尝试）先占住那一行再读改写（hold_turn），用量按增量记在库里那一份上，谁也不盖掉谁；
+同一轮在这个进程里不起第二个后台任务（ChatTurnManager.start）。
 """
 from __future__ import annotations
 
@@ -43,6 +47,7 @@ from collections.abc import Mapping
 from typing import Any
 
 from sqlalchemy import delete, select, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.chat import log as turn_log
@@ -150,14 +155,19 @@ class ChatDriver(TurnDriver):
     async def on_retry(self, reason: str, *, kind: str, base_graph: dict[str, Any] | None) -> None:
         from app.chat import turn as policy
 
+        # 换下来的那次运行交出去并进 earlier；这一次从「还没有运行」起，收尾时不再把它当这一次的运行记一遍
+        run, self.last_run = (self.last_run if kind == "review" else None), None
         await self.execution.retrying(reason, kind=kind, base=policy.has_nodes(base_graph), attempt=self.turn.attempts,
-                                      run=self.last_run if kind == "review" else None)
+                                      run=run)
 
     async def run_input(self, session: AsyncSession, graph: dict[str, Any]) -> dict[str, Any]:
-        from app.api.conversations import history_text
+        """入口填问题和这一轮之前的几轮问答（conversations.run_history，同以前页面上的 historyOf）：重试、重新运行较早的
+        一轮时，后面接着问过的几轮不算进来。"""
+        from app.api.conversations import run_history
         from app.chat import turn as policy
 
-        return policy.fill_input(graph, self.question, await history_text(session, self.execution.conversation_id))
+        history = await run_history(session, self.execution.conversation_id, before_seq=self.execution.turn_seq)
+        return policy.fill_input(graph, self.question, history)
 
     async def launch_args(self, session: AsyncSession, graph: dict[str, Any]) -> dict[str, Any]:
         from app.api.deps import restricted
@@ -265,8 +275,8 @@ class ChatTurnExecution(runtime.JobRun):
         self.actor: str | None = None
         self.mode = "ask"
         self.request: dict[str, Any] = {}
-        #: 这一轮的用量 {copilot, review, earlier, run}（UsageTotal），从库里那一行接着记
-        self.usage: dict[str, Any] = {}
+        #: 这一轮在对话里排第几（运行入口的 history 只取它之前的几轮）
+        self.turn_seq = 0
         self.run_id: str | None = None
         self.run_class: str | None = None
         self.driver: ChatDriver | None = None
@@ -282,9 +292,11 @@ class ChatTurnExecution(runtime.JobRun):
         """追加一条事件（和这一轮的库写入排在同一条队里，停止时不会写到一半）。"""
         return await self._write(turn_log.append, self.turn_id, type_, data or {})
 
-    async def save(self, *, meta: Mapping[str, Any] | None = None, **fields: Any) -> None:
-        """改 conversation_turns 那一行：fields 照写，meta 并进去（值为 None 的键删掉），顺手记 meta.at。"""
-        await self._write(_save_turn, self.turn_id, fields, dict(meta or {}))
+    async def save(self, *, meta: Mapping[str, Any] | None = None, usage: list[tuple[Any, ...]] | None = None,
+                   **fields: Any) -> dict[str, Any]:
+        """改 conversation_turns 那一行：fields 照写，meta 并进去（值为 None 的键删掉），顺手记 meta.at；usage 是对 meta.usage 的
+        几步改动（见 _apply_usage），在库里现有的那一份上做。交回改完之后的 meta.usage。"""
+        return await self._write(_save_turn, self.turn_id, fields, dict(meta or {}), list(usage or []))
 
     async def _job(self, **values: Any) -> None:
         async def go(session: AsyncSession) -> None:
@@ -297,10 +309,10 @@ class ChatTurnExecution(runtime.JobRun):
         turn_log.notify(self.turn_id)
 
     async def add_usage(self, part: str, usage: Mapping[str, Any] | None) -> None:
+        """这一轮又花了一笔（助手建图、复核）：在库里那一份上加，不拿内存里的拷贝盖回去——晚到的复核也往同一份上记"""
         if not isinstance(usage, Mapping):
             return
-        self.usage[part] = record.add_total(self.usage.get(part), usage)
-        await self.save(meta={"usage": dict(self.usage)})
+        await self.save(usage=[("add", part, dict(usage))])
 
     async def on_phase(self, phase: str) -> None:
         await self._job(stage=phase)
@@ -308,22 +320,18 @@ class ChatTurnExecution(runtime.JobRun):
 
     async def launched(self, run: Any) -> None:
         self.run_id, self.run_class = run.id, run.run_class
-        self.usage.pop("run", None)
         await self._job(run_id=run.id)
-        await self.save(run_id=run.id, status="running",
+        await self.save(run_id=run.id, status="running", usage=[("drop", "run")],
                         meta={"runId": run.id, "runClass": run.run_class, "runStatus": run.status})
         await self.emit("run", {"run_id": run.id, "status": run.status, "run_class": run.run_class})
 
     async def retrying(self, reason: str, *, kind: str, base: bool, attempt: int, run: Any = None) -> None:
         """自动重来一次：换下来的那次运行的用量并进 earlier；这一行回到「还没有运行」。"""
-        if run is not None:
-            self.usage["run"] = record.usage_total(run.usage) or {}
-        self.usage = record.retire_run(self.usage)
+        ops: list[tuple[Any, ...]] = [("set", "run", record.usage_total(run.usage) or {})] if run is not None else []
         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,
-                                       "usage": dict(self.usage)})
+        usage = await self.save(run_id=None, meta={"runId": None, "runStatus": None}, usage=[*ops, ("retire",)])
+        await self.emit("turn.retry", {"kind": kind, "reason": reason, "base": base, "attempt": attempt, "usage": usage})
 
     # ---------------- 执行 ----------------
 
@@ -342,8 +350,7 @@ class ChatTurnExecution(runtime.JobRun):
             return
         self.conversation_id, self.actor, self.mode = job.conversation_id, job.actor, job.mode
         self.request = dict(job.request or {})
-        meta = dict(row.meta or {}) if isinstance(row.meta, dict) else {}
-        self.usage = dict(meta.get("usage") or {})
+        self.turn_seq = row.seq
         self.run_id = job.run_id
         # 这一轮按发起的人的身份跑：answer_ops 里的 restricted()、对话归属，发起运行时的遮罩、只读都看它
         set_principal(principal_of(job.principal))
@@ -371,8 +378,8 @@ class ChatTurnExecution(runtime.JobRun):
         """按这一轮的结局写那一行、发 end、任务记成功。"""
         driver = self.driver
         assert driver is not None
-        fields, meta, outcome, output = await _settled(driver, self)
-        await self.save(meta=meta, **fields)
+        fields, meta, outcome, output, usage = await _settled(driver, self)
+        await self.save(meta=meta, usage=usage, **fields)
         await self._end(outcome, output)
         # 晚到的复核先登记、再改任务状态：实时流看到任务结束时，要知道还得等这一次复核回来
         if driver.late_review is not None:
@@ -406,7 +413,7 @@ class ChatTurnExecution(runtime.JobRun):
         if not self.ended:
             await self.save(status="error", error=record.CANCELLED,
                             meta={"outcome": "cancelled", "runStatus": "cancelled" if run_id else None,
-                                  "usage": dict(self.usage), **_timing(self.request)})
+                                  **_timing(self.request)})
             await self._end("cancelled")
         await self._job(status="cancelled", stage=None, finished_at=utcnow())
 
@@ -423,8 +430,7 @@ class ChatTurnExecution(runtime.JobRun):
         await self._settle()
         if not self.ended:
             await self.save(status="error", error=message,
-                            meta={"cause": {"kind": "crash", "message": message}, "usage": dict(self.usage),
-                                  **_timing(self.request)})
+                            meta={"cause": {"kind": "crash", "message": message}, **_timing(self.request)})
             await self._end("error")
         await self._job(status="failed", stage=None, error=message, finished_at=utcnow())
 
@@ -433,6 +439,14 @@ class ChatTurnManager(runtime.JobManager):
     def __init__(self) -> None:
         super().__init__(ChatTurnExecution, label="问数据的一轮", name="chat-turn")
 
+    def start(self, job_id: str) -> bool:  # type: ignore[override]
+        """同一轮在这个进程里还有一个后台任务活着就不起第二个、也不把登记换成新的（换了的话，停止只停得到后起的那个，
+        先起的那个跑完把「已取消」盖成「已完成」）。起了交回 True。"""
+        if self.running(job_id):
+            return False
+        super().start(job_id)
+        return True
+
 
 manager = ChatTurnManager()
 
@@ -442,32 +456,30 @@ _late: dict[str, asyncio.Task[Any]] = {}
 
 def _track_late(turn_id: str, task: asyncio.Task[Any]) -> None:
     async def finish() -> None:
+        # 先把用量记进那一行、发 review.late，再出登记、叫醒订阅方：反过来的话，开着的实时流醒来看到「不在进行中了」，
+        # 先发了 stream.end 走掉，review.late 就到不了它
         try:
-            result = await task
-        except asyncio.CancelledError:
-            return
-        except Exception:  # noqa: BLE001
-            logger.exception("问数据：晚到的复核出错")
-            return
+            try:
+                result = await task
+            except asyncio.CancelledError:
+                return
+            except Exception:  # noqa: BLE001
+                logger.exception("问数据：晚到的复核出错")
+                return
+            usage = (result or {}).get("usage") if isinstance(result, Mapping) else None
+            if not usage:
+                return
+            async with SessionLocal() as session:
+                # 加在库里现有的那一份上（这一轮可能已经在重试，那一次的用量也在往同一份上记）
+                await _save_turn(session, turn_id, {}, {}, [("add", "review", dict(usage))], touch_meta=False)
+            async with SessionLocal() as session:
+                await turn_log.append(session, turn_id, "review.late", {"usage": usage})
+        except Exception:  # noqa: BLE001 - 记账失败只记日志
+            logger.exception("问数据：记晚到的复核用量出错")
         finally:
             if _late.get(turn_id) is watcher:
                 _late.pop(turn_id, None)
             turn_log.notify(turn_id)
-        usage = (result or {}).get("usage") if isinstance(result, Mapping) else None
-        if not usage:
-            return
-        async with SessionLocal() as session:
-            from app.db.models import ConversationTurn
-
-            row = await session.get(ConversationTurn, turn_id)
-            if row is None:
-                return
-            meta = dict(row.meta or {}) if isinstance(row.meta, dict) else {}
-            spent = dict(meta.get("usage") or {})
-            spent["review"] = record.add_total(spent.get("review"), usage)
-            row.meta = {**meta, "usage": spent}
-            await session.commit()
-            await turn_log.append(session, turn_id, "review.late", {"usage": usage})
 
     watcher = asyncio.ensure_future(finish())
     _late[turn_id] = watcher
@@ -483,12 +495,41 @@ def late_pending(turn_id: str) -> bool:
 # --------------------------------------------------------------------------
 
 
-async def _save_turn(session: AsyncSession, turn_id: str, fields: Mapping[str, Any], meta: Mapping[str, Any]) -> None:
+def _apply_usage(usage: Mapping[str, Any] | None, ops: list[tuple[Any, ...]]) -> dict[str, Any]:
+    """在一份 meta.usage 上做几步改动：("add", 哪一份, 用量) 加上；("set", 哪一份, 用量) 换成；("drop", 哪一份) 去掉；
+    ("retire",) 当前这次运行换下来了（run 并进 earlier，同 record.retire_run）。"""
+    out = dict(usage or {})
+    for op in ops:
+        kind = op[0]
+        if kind == "add":
+            out[op[1]] = record.add_total(out.get(op[1]), op[2])
+        elif kind == "set":
+            out[op[1]] = dict(op[2] or {})
+        elif kind == "drop":
+            out.pop(op[1], None)
+        elif kind == "retire":
+            out = record.retire_run(out)
+    return out
+
+
+async def hold_turn(session: AsyncSession, turn_id: str) -> None:
+    """占住这一行再读改写（同一行的几处写入——这一轮的任务、晚到的复核、发起下一次尝试——一个接一个来，谁也不盖掉谁）：
+    第一句就写它，SQLite 上拿到写锁、PostgreSQL 上锁住这一行，别的写入等它提交。"""
     from app.db.models import ConversationTurn
 
-    row = await session.get(ConversationTurn, turn_id)
+    await session.execute(update(ConversationTurn).where(ConversationTurn.id == turn_id).values(updated_at=utcnow())
+                          .execution_options(synchronize_session=False))
+
+
+async def _save_turn(session: AsyncSession, turn_id: str, fields: Mapping[str, Any], meta: Mapping[str, Any],
+                     usage: list[tuple[Any, ...]] | None = None, *, touch_meta: bool = True) -> dict[str, Any]:
+    from app.db.models import ConversationTurn
+
+    await hold_turn(session, turn_id)
+    row = await session.get(ConversationTurn, turn_id, populate_existing=True)
     if row is None:
-        return
+        await session.rollback()
+        return {}
     for key, value in fields.items():
         setattr(row, key, value)
     merged = dict(row.meta or {}) if isinstance(row.meta, dict) else {}
@@ -497,10 +538,14 @@ async def _save_turn(session: AsyncSession, turn_id: str, fields: Mapping[str, A
             merged.pop(key, None)
         else:
             merged[key] = value
+    if usage:
+        merged["usage"] = _apply_usage(merged.get("usage"), usage)
     merged["v"] = 1
-    merged["at"] = int(time.time() * 1000)
+    if touch_meta:
+        merged["at"] = int(time.time() * 1000)
     row.meta = merged
     await session.commit()
+    return dict(merged.get("usage") or {})
 
 
 async def _snapshot(session: AsyncSession, turn_id: str, run_id: str | None) -> tuple[dict[str, Any] | None,
@@ -568,8 +613,9 @@ async def _run_failed_detail(run_id: str) -> str | None:
 
 
 async def _settled(driver: ChatDriver, execution: ChatTurnExecution
-                   ) -> tuple[dict[str, Any], dict[str, Any], str, Any]:
-    """这一轮的结局 → (那一行要写的列, 并进 meta 的, outcome, 完整成果)。和以前页面在各个结局处落库写的一一对应。"""
+                   ) -> tuple[dict[str, Any], dict[str, Any], str, Any, list[tuple[Any, ...]]]:
+    """这一轮的结局 → (那一行要写的列, 并进 meta 的, outcome, 完整成果, 对 meta.usage 的改动)。和以前页面在各个结局处落库写的
+    一一对应。用量里助手、复核那几份是随做随记的（add_usage），这里只记这次运行的那一份。"""
     from app.engine.runner import END_REJECTED
 
     t = driver.turn
@@ -583,7 +629,7 @@ async def _settled(driver: ChatDriver, execution: ChatTurnExecution
         text = t.reply or ""
         fields.update(status="done", answer=text, review={})
         meta.update(noQuery=True, runId=None)
-        return fields, {**meta, "usage": dict(execution.usage)}, "reply", {"answer": text}
+        return fields, meta, "reply", {"answer": text}, []
     if path == "asked":
         asked = t.asked or {}
         question = str(asked.get("question") or "").strip()
@@ -594,11 +640,11 @@ async def _settled(driver: ChatDriver, execution: ChatTurnExecution
             # 改了图又问了用户、或者只问不改（图原样交回）：图摆着等「运行」，问题作为这一轮的内容
             fields.update(graph=driver.graph or term.get("graph"), explanation=str(term.get("explanation") or ""))
             outcome = "ready"
-        return fields, {**meta, "usage": dict(execution.usage)}, outcome, {"answer": question} if question else None
+        return fields, meta, outcome, {"answer": question} if question else None, []
     if path == "ready":
         fields.update(status="done", graph=driver.graph, explanation=str(term.get("explanation") or ""), review={})
         meta["runId"] = None
-        return fields, {**meta, "usage": dict(execution.usage)}, "ready", None
+        return fields, meta, "ready", None, []
     if path == "build_failed":
         if driver.refusal is not None:
             message = str(getattr(driver.refusal, "detail", "") or driver.refusal)
@@ -616,15 +662,13 @@ async def _settled(driver: ChatDriver, execution: ChatTurnExecution
                 fields.update(graph=driver.graph, explanation=str(term.get("explanation") or ""))
         fields.update(status="error", error=error, review={})
         meta.update(cause={k: v for k, v in cause.items() if v is not None}, runId=None)
-        return fields, {**meta, "usage": dict(execution.usage)}, "error", None
+        return fields, meta, "error", None, []
 
-    # 发起过运行的结局
-    if run is not None:
-        execution.usage["run"] = record.usage_total(run.usage) or {}
-    usage = dict(execution.usage)
+    # 发起过运行的结局：这次运行的用量记成 run（换下来的早在重来时并进了 earlier）
+    usage: list[tuple[Any, ...]] = [("set", "run", record.usage_total(run.usage) or {})] if run is not None else []
     run_id = run.id if run is not None else execution.run_id
     meta.update(runId=run_id, runClass=getattr(run, "run_class", None) or execution.run_class,
-                runStatus=getattr(run, "status", None), queries=await _queries(run_id), usage=usage)
+                runStatus=getattr(run, "status", None), queries=await _queries(run_id))
     if path == "run_failed":
         refused = driver.launch_refused
         if refused is not None:
@@ -632,27 +676,27 @@ async def _settled(driver: ChatDriver, execution: ChatTurnExecution
             fields.update(status="error", error=message, review={})
             meta.update(cause={"kind": "start", "message": message, "code": getattr(refused, "code", None),
                                "status": getattr(refused, "status_code", None)}, runId=None, runStatus=None, queries=None)
-            return fields, meta, "error", None
+            return fields, meta, "error", None, usage
         if run is None:
             fields.update(status="error", error=record.GONE, review={})
             meta.update(cause={"kind": "gone"}, runStatus="gone")
-            return fields, meta, "error", None
+            return fields, meta, "error", None, usage
         if run.status == "cancelled" and run.end_reason != END_REJECTED:
             fields.update(status="error", error=record.CANCELLED, review={})
             meta["outcome"] = "cancelled"
-            return fields, meta, "cancelled", None
+            return fields, meta, "cancelled", None, usage
         fields.update(status="error", error=run.error or "运行失败", review={})
         meta["cause"] = {k: v for k, v in {"kind": "run", "error": run.error or "",
                                            "detail": await _run_failed_detail(run.id)}.items() if v is not None}
-        return fields, meta, "error", None
+        return fields, meta, "error", None, usage
     if path == "rejected":
         fields.update(status="error", error=record.CANCELLED, review={})
         meta.update(outcome="cancelled", rejected=await _rejected_by(run.id) or {"decision": "reject"})
-        return fields, meta, "cancelled", None
+        return fields, meta, "cancelled", None, usage
     if path == "suspended":
         fields.update(status="error", error=record.SUSPENDED, review={})
         meta.update(outcome="suspended", runStatus="interrupted")
-        return fields, meta, "suspended", None
+        return fields, meta, "suspended", None, usage
 
     # 跑完了：成果以库里那份为准（事件里那份为了推送体积被切过）
     output = run.output if isinstance(run.output, dict) else {}
@@ -661,7 +705,7 @@ async def _settled(driver: ChatDriver, execution: ChatTurnExecution
     fields.update(status="done", answer=rewritten or record.answer_text(output), review=review or {})
     meta.update(issuance=record.slim_issuance(output.get("_issuance")),
                 evidence=True if record.has_evidence(output) else None)
-    return fields, meta, "done", output
+    return fields, meta, "done", output, usage
 
 
 # --------------------------------------------------------------------------
@@ -712,43 +756,132 @@ async def _scope(session: AsyncSession, ids: list[str] | None) -> list[dict[str,
     return out
 
 
+#: 只有问数据的对话能发起一轮（画布助手、应用润色、蓝图的对话各有各的助手；启动整理也只认问数据的对话）
+NOT_CHAT = "这段对话不是问数据的对话，不能在这里发起问数据"
+#: 使用者（user 角色）不能继续运行：以前 POST /api/runs/{id}/continue 对它就是 403（api/deps._run_scope 只放行停止），说法同那边
+ROLE_ONLY = "没有权限：这项操作需要搭建者及以上的角色"
+
+
+async def hold_conversation(session: AsyncSession, conversation_id: str, *, touch: bool = True) -> None:
+    """占住这段对话：发起一轮的核对和写入（这段对话忙不忙、下一轮排第几、任务、第一条事件）在同一个事务里，第一句就写它——
+    SQLite 上拿到写锁、PostgreSQL 上锁住这一行，同一段对话同时来的几次发起一个接一个，后来的看得见先来的写下的。
+
+    touch：顺手记活跃时间（人在这里开始做事）。页面打开时自动发的「接着收尾」不算，原样写回，不然旧对话一打开就顶到列表最上面。"""
+    from app.db.models import Conversation
+
+    stamp = utcnow() if touch else Conversation.last_active_at
+    await session.execute(update(Conversation).where(Conversation.id == conversation_id)
+                          .values(last_active_at=stamp).execution_options(synchronize_session=False))
+
+
 async def start(session: AsyncSession, *, conversation: Any, question: str, datasource_ids: list[str] | None,
                 actor: str | None, mode: str | None = None, retry_of: Any = None,
                 max_steps: int | None = None) -> str:
     """发起一轮（新问，或者对已有的一轮重试、重新运行、继续运行、接着收尾）。交回这一轮的 id（conversation_turns.id）。
 
     新问：建 conversation_turns 那一行（同 POST /api/conversations/{id}/turns）和任务。已有的一轮：上一次的结局归档进
-    meta.attempts（重试、重新运行、放宽步数），那一行回到 running，任务重新排队。不发起抛 TurnRefused。"""
-    from app.api.conversations import new_turn, turn_run_id
-    from app.chat import turn as policy
-    from app.db.models import Approval, Run
-    from app.db.models_chat import ChatTurnJob
+    meta.attempts（重试、重新运行、放宽步数），那一行回到 running，任务重新排队。不发起抛 TurnRefused。
+
+    **同时来的几次发起只起得来一次。** 核对和写入在同一个事务里、先占住这段对话（hold_conversation）：同一段对话同时两次新问
+    只建一轮（另一次 409），同一轮同时几次重试、两个标签页同时请服务端接着收尾只有一次排得上队（任务按状态条件更新）；
+    这一次尝试的第一条事件（turn.attempt）也在这个事务里写，写不进去就什么都不留下。继续运行要先续上那次运行（它自己按运行的
+    状态认领，同一次运行同时只续得上一次），它用自己的会话写库，不能放进占着对话的事务里；续上了、这里却没排上队的，把续上的
+    那次运行取消掉。服务重启挂起的旧运行在提交之后才取消（只有排上队的那一次去取消）。"""
+    from app.api.coded import ROLE_REQUIRED
+    from app.api.conversations import turn_run_id
+    from app.api.deps import restricted
+    from app.db.models import Run
 
     mode = mode or ("retry" if retry_of is not None else "ask")
     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 getattr(conversation, "kind", "chat") != "chat":
+        raise TurnRefused(422, NOT_CHAT, TURN_INVALID)
+    if mode == "continue" and restricted():
+        raise TurnRefused(403, ROLE_ONLY, ROLE_REQUIRED)
+    if retry_of is None and not question.strip():
+        raise TurnRefused(422, "问题不能为空", TURN_INVALID)
+    if mode == "max_steps" and (not max_steps or max_steps < 1):
+        raise TurnRefused(422, "放宽到几步要写清楚", TURN_INVALID)
+
+    continued: str | None = None
+    if mode == "continue":
+        run_id = turn_run_id(retry_of)
+        run = await session.get(Run, run_id) if run_id else None
+        if run is None:
+            raise TurnRefused(409, "这一轮没有可接着的运行，请重试本轮", TURN_INVALID)
+        await _continue_run(run.id, actor)
+        continued = run.id
+    try:
+        turn_id, stale_run = await _claim(session, conversation=conversation, question=question,
+                                          datasource_ids=datasource_ids, actor=actor, mode=mode, retry_of=retry_of,
+                                          max_steps=max_steps)
+    except BaseException:
+        await session.rollback()
+        if continued:
+            from app.engine.runner import run_manager
+
+            with contextlib.suppress(Exception):
+                await run_manager.cancel(continued, actor=actor)
+        raise
+    if stale_run:
+        # 服务重启挂起的旧运行没人会再去续它，顺手取消（停在审批上的不算：重试不会从那里发起）
+        from app.engine.runner import run_manager
+
+        try:
+            await run_manager.cancel(stale_run, actor=actor)
+        except Exception:  # noqa: BLE001
+            logger.exception("问数据：重试时取消挂起的旧运行 %s 出错", stale_run)
+    if not manager.start(turn_id):
+        # 上一次的后台任务刚记完结局、还没退出：等它退出再起
+        await manager.wait(turn_id, timeout=STOP_GRACE_S)
+        if not manager.start(turn_id):
+            async with SessionLocal() as s2:
+                from app.db.models_chat import ChatTurnJob
+
+                await s2.execute(update(ChatTurnJob).where(ChatTurnJob.id == turn_id).values(
+                    status="failed", error="上一次的后台任务还没有退出", finished_at=utcnow())
+                    .execution_options(synchronize_session=False))
+                await s2.commit()
+            raise TurnRefused(409, BUSY, TURN_BUSY)
+    return turn_id
+
+
+async def _claim(session: AsyncSession, *, conversation: Any, question: str, datasource_ids: list[str] | None,
+                 actor: str | None, mode: str, retry_of: Any, max_steps: int | None) -> tuple[str, str | None]:
+    """在一个事务里核对、写轮次、排队、写这一次尝试的第一条事件，提交。交回 (这一轮的 id, 提交之后要取消的挂起旧运行)。"""
+    from app.api.conversations import new_turn, turn_run_id
+    from app.chat import turn as policy
+    from app.db.models import Approval, ConversationTurn, Run
+    from app.db.models_chat import ChatTurnJob
+
+    await hold_conversation(session, conversation.id, touch=mode != "resume")
+    if retry_of is not None:
+        await hold_turn(session, retry_of.id)
     # 接着收尾（resume）不建图、不发起运行，不算占着这段对话：一段对话里停在半路的几轮可以一起交给服务端收尾
+    besides = retry_of.id if retry_of is not None else None
     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}
+    stale_run: str | None = None
 
     if retry_of is None:
-        if not question.strip():
-            raise TurnRefused(422, "问题不能为空", TURN_INVALID)
         scope = await _scope(session, datasource_ids)
         row = await new_turn(session, conversation, question)
         row.meta = {"v": 1, **({"scope": scope} if scope else {}), "at": now_ms}
         request["scope"] = scope
-        session.add(ChatTurnJob(id=row.id, conversation_id=conversation.id, status="queued", mode=mode, request=request,
-                                principal=principal_snapshot(), actor=actor, started_at=None))
+        job = ChatTurnJob(id=row.id, conversation_id=conversation.id, status="queued", mode=mode, request=request,
+                          principal=principal_snapshot(), actor=actor, started_at=None)
+        session.add(job)
         await session.flush()
-        turn_id = row.id
         attempts: list[Any] = []
     else:
-        row = retry_of
-        job = await session.get(ChatTurnJob, row.id)
+        # 占住之后重读：发起之前读到的那一份可能已经被同时来的另一次发起改过了
+        row = await session.get(ConversationTurn, retry_of.id, populate_existing=True)
+        if row is None:
+            raise TurnRefused(404, "该轮对话不存在，可能已被删除")
+        job = await session.get(ChatTurnJob, row.id, populate_existing=True)
         if job is not None and job.status in runtime.ACTIVE:
             raise TurnRefused(409, BUSY, TURN_BUSY)
         meta = dict(row.meta or {}) if isinstance(row.meta, dict) else {}
@@ -761,15 +894,11 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data
         if mode in ("rerun", "max_steps", "run") and not graphful:
             raise TurnRefused(409, "这一轮没有可运行的工作流，请重试本轮", TURN_INVALID)
         if mode == "max_steps":
-            if not max_steps or max_steps < 1:
-                raise TurnRefused(422, "放宽到几步要写清楚", TURN_INVALID)
-            request["max_steps"] = int(max_steps)
+            request["max_steps"] = int(max_steps or 0)
         if mode in ("continue", "resume"):
             # 接着收尾不要求运行还在：运行记录删了的，服务端照实收成「运行记录不见了」
-            if run is None and (mode == "continue" or not run_id):
+            if not run_id:
                 raise TurnRefused(409, "这一轮没有可接着的运行，请重试本轮", TURN_INVALID)
-            if mode == "continue":
-                await _continue_run(run.id, actor)
             request["follow"] = True
             # 计时：收过尾的（失败、中断之后点「继续运行」）接着上次停下的地方走——隔了一夜再点，头部不该写「已运行 14 小时」；
             # 停在半路的（页面以前没来得及收尾的）从运行开始算；运行早就结束了的（补交付）按它真正结束的时刻算，不按现在
@@ -781,18 +910,14 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data
             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)
+            row.status, row.error = "running", ""
+            row.meta = {**meta, "at": now_ms}
         else:
-            # 重试、重新运行、放宽步数、跑一下：服务重启挂起的旧运行没人会再去续它，顺手取消（停在审批上的不算：重试不会从那里发起）
             if run is not None and run.status == "interrupted" and mode != "run":
                 pending = (await session.execute(select(Approval.id).where(
                     Approval.run_id == run.id, Approval.status.in_(("pending", "answered"))).limit(1))).first()
                 if pending is None:
-                    from app.engine.runner import run_manager
-
-                    try:
-                        await run_manager.cancel(run.id, actor=actor)
-                    except Exception:  # noqa: BLE001
-                        logger.exception("问数据：重试时取消挂起的旧运行 %s 出错", run.id)
+                    stale_run = run.id
             if mode != "run":
                 archived = record.attempt_record(row, meta, n=len(attempts) + 1,
                                                  at_ms=int(row.updated_at.timestamp() * 1000) if row.updated_at else now_ms)
@@ -810,39 +935,38 @@ async def start(session: AsyncSession, *, conversation: Any, question: str, data
             row.meta = {k: v for k, v in new_meta.items() if v is not None}
             row.status, row.error, row.answer, row.review = "running", "", "", {}
             row.run_id = None
-        if mode in ("continue", "resume"):
-            row.status, row.error = "running", ""
-            row.meta = {**meta, "at": now_ms}
+        values = {"status": "queued", "stage": None, "mode": mode, "request": request,
+                  "run_id": run_id if mode in ("continue", "resume") else None, "attempt": 0,
+                  "principal": principal_snapshot(), "actor": actor, "error": None, "started_at": None,
+                  "finished_at": None}
         if job is None:
-            job = ChatTurnJob(id=row.id, conversation_id=conversation.id)
-            session.add(job)
-        job.status, job.stage, job.mode, job.request = "queued", None, mode, request
-        job.run_id = run_id if mode in ("continue", "resume") else None
-        job.attempt = 0
-        job.principal, job.actor, job.error, job.started_at, job.finished_at = (
-            principal_snapshot(), actor, None, None, None)
-        conversation.last_active_at = utcnow()
+            session.add(ChatTurnJob(id=row.id, conversation_id=conversation.id, **values))
+        else:
+            # 只在它不在排队、不在跑的时候排上：同时来的另一次已经排上了的，这一次不算
+            claimed = await session.execute(update(ChatTurnJob).where(
+                ChatTurnJob.id == row.id, ChatTurnJob.status.not_in(runtime.ACTIVE)).values(**values)
+                .execution_options(synchronize_session=False))
+            if claimed.rowcount != 1:
+                raise TurnRefused(409, BUSY, TURN_BUSY)
         await session.flush()
-        turn_id = row.id
-    await session.commit()
-
-    # 这一次尝试从这一条开始：刷新、重开页面时从这里回放
-    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 [], "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))
-        await s.commit()
-    manager.start(turn_id)
-    return turn_id
+    turn_id = row.id
+    usage_now = (row.meta or {}).get("usage") if isinstance(row.meta, dict) else None
+    # 这一次尝试从这一条开始：刷新、重开页面时从这里回放。页面据此把这一轮重置成「这一次」的样子：归档、范围、到目前为止的
+    # 用量、从什么时候算起（补交付的到运行结束为止）
+    seq = await turn_log.add(session, turn_id, "turn.attempt", {
+        "mode": mode, "attempts": attempts, "scope": request.get("scope") or [], "usage": usage_now 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 session.execute(update(ChatTurnJob).where(ChatTurnJob.id == turn_id).values(attempt_seq=seq)
+                          .execution_options(synchronize_session=False))
+    try:
+        await session.commit()
+    except IntegrityError as e:
+        # 不占锁的写入（老页面开轮）正好同时：照「这段对话正忙」答
+        raise TurnRefused(409, BUSY, TURN_BUSY) from e
+    turn_log.notify(turn_id)
+    return turn_id, stale_run
 
 
 async def _continue_run(run_id: str, actor: str | None) -> None:
diff --git a/backend/app/chat/log.py b/backend/app/chat/log.py
index 4b120a25..9e73c8b1 100644
--- a/backend/app/chat/log.py
+++ b/backend/app/chat/log.py
@@ -67,19 +67,29 @@ def notify(turn_id: str) -> None:
     waiters.notify(turn_id)
 
 
-async def append(session: AsyncSession, turn_id: str, type_: str, data: dict[str, Any] | None = None) -> int:
-    """追加一条，提交，叫醒订阅方。交回它的 seq。
+async def add(session: AsyncSession, turn_id: str, type_: str, data: dict[str, Any] | None = None) -> int:
+    """在调用方的事务里加一条（不提交），交回它的 seq。
 
-    seq 按这一轮库里已有的最大值往下接：同一轮同一时刻只有一个写的人（它的后台任务；任务不在跑时才由停止、启动整理来写），
-    唯一约束 (turn_id, seq) 兜底。任务行的 last_seq 顺手跟上（GET /api/chat/turns/{id} 照它说现在到哪了）。"""
+    seq 由任务行上的计数器发：UPDATE chat_turn_jobs SET last_seq = last_seq + 1 … RETURNING 是一句原子的写，同一轮同时追加的
+    几处（这一轮的任务、停止、晚到的复核、发起下一次尝试）各拿到不同的号，唯一约束 (turn_id, seq) 兜底。以前按库里已有的最大值
+    往下接，同时两处追加就撞号。任务行的 last_seq 也就是 GET /api/chat/turns/{id} 说的「现在到哪了」。没有任务行的（不该有）
+    退回按最大值往下接。"""
     from app.db.models_chat import ChatTurnEvent, ChatTurnJob
 
-    last = int(await session.scalar(select(func.coalesce(func.max(ChatTurnEvent.seq), 0))
-                                    .where(ChatTurnEvent.turn_id == turn_id)) or 0)
-    seq = last + 1
-    session.add(ChatTurnEvent(turn_id=turn_id, seq=seq, type=type_, ts=time.time(), data=dict(data or {})))
-    await session.execute(update(ChatTurnJob).where(ChatTurnJob.id == turn_id).values(last_seq=seq)
-                          .execution_options(synchronize_session=False))
+    seq = (await session.execute(update(ChatTurnJob).where(ChatTurnJob.id == turn_id)
+                                 .values(last_seq=ChatTurnJob.last_seq + 1).returning(ChatTurnJob.last_seq)
+                                 .execution_options(synchronize_session=False))).scalar_one_or_none()
+    if seq is None:
+        seq = int(await session.scalar(select(func.coalesce(func.max(ChatTurnEvent.seq), 0))
+                                       .where(ChatTurnEvent.turn_id == turn_id)) or 0) + 1
+    session.add(ChatTurnEvent(turn_id=turn_id, seq=int(seq), type=type_, ts=time.time(), data=dict(data or {})))
+    await session.flush()
+    return int(seq)
+
+
+async def append(session: AsyncSession, turn_id: str, type_: str, data: dict[str, Any] | None = None) -> int:
+    """追加一条，提交，叫醒订阅方。交回它的 seq（见 add）。"""
+    seq = await add(session, turn_id, type_, data)
     await session.commit()
     notify(turn_id)
     return seq
@@ -134,5 +144,5 @@ async def stream(turn_id: str, after: int, *, active: Callable[[], Awaitable[str
                 yield PING
 
 
-__all__ = ["HEARTBEAT_S", "PING", "READ_BATCH", "STREAM_END", "append", "frame", "notify", "read", "stream", "waiters",
+__all__ = ["HEARTBEAT_S", "PING", "READ_BATCH", "STREAM_END", "add", "append", "frame", "notify", "read", "stream", "waiters",
            "wire"]
diff --git a/backend/tests/test_chat_turns_races.py b/backend/tests/test_chat_turns_races.py
new file mode 100644
index 00000000..0ded95cb
--- /dev/null
+++ b/backend/tests/test_chat_turns_races.py
@@ -0,0 +1,351 @@
+"""问数据的一轮在服务端跑：并发发起、晚到的复核、使用者能做的事、入口的历史、对话的种类（复核意见 A1–A5）。
+
+守的是：
+- 同一段对话同时两次新问：只起得了一轮（另一次 409），两轮不会拿到同一个 seq、不会有两个任务同时跑；同一轮同时几次重试、
+  两个标签页同时请服务端接着收尾：只有一次起得来，别的 409，不会 500，同一轮不会建两次图；事件的 seq 并发追加也不撞；
+  发起写到一半出错时什么都不留下（这段对话不会从此一直「正忙」）；后台任务对同一轮不起第二个；
+- 使用者（user 角色）不能「继续运行」：以前 POST /api/runs/{id}/continue 对它是 403（只放行停止），问数据这里照原说法；
+- 晚到的复核：开着的实时流先收到 review.late 再收到 stream.end；它记的用量不会被紧接着的重试覆盖掉；
+- 运行入口的 history 只取这一轮之前的几轮（照以前页面 historyOf 的规则：截 600 字、跳过没答案的、最多 6 轮）；
+- 只有问数据的对话（kind=chat）能发起一轮。
+
+库、问题、名字都是虚构的；助手和报告撰写都是剧本。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import func, select
+
+from app.chat import job as chat_job
+from app.chat import log as turn_log
+from app.db.base import SessionLocal
+from app.db.models import Conversation, ConversationTurn, Run
+from app.db.models_chat import ChatTurnJob
+from app.main import app
+from tests import eval_samples as es
+from tests.test_chat_turns import (  # noqa: F401 - 夹具照用
+    _ask, _conversation, _done, _events, _row, _stage, _until, client, engine_up, people, source,
+)
+
+
+class Counting(es.Copilot):
+    """数一数生成过程起了几次（一次生成里的「接着写」「自查」不算）；gate 给了就每次起草前停在闸上。"""
+
+    def __init__(self, source: str, *, gate: asyncio.Event | None = None, **kw: Any) -> None:
+        super().__init__(source, **kw)
+        self.gate = gate
+        self.drafts = 0
+
+    async def astream(self, messages):
+        if len(messages) == 2 and "上线前自查" not in messages[1][1]:
+            self.drafts += 1
+            if self.gate is not None:
+                await self.gate.wait()
+        async for chunk in super().astream(messages):
+            yield chunk
+
+
+async def _jobs_of(conversation_id: str) -> list[ChatTurnJob]:
+    async with SessionLocal() as session:
+        return list((await session.execute(select(ChatTurnJob).where(
+            ChatTurnJob.conversation_id == conversation_id))).scalars())
+
+
+# --------------------------------------------------------------------------
+# A1 并发发起
+# --------------------------------------------------------------------------
+
+
+async def test_two_asks_at_once_start_only_one_turn(client, monkeypatch, source):
+    gate = asyncio.Event()
+    model = Counting(source["name"], gate=gate, mode="reply")
+    es.use_copilot(monkeypatch, model)
+    conversation_id = await _conversation(client)
+    body = {"conversation_id": conversation_id, "question": "上周各区域销售额"}
+    got = await asyncio.gather(*(client.post("/api/chat/turns", json=body) for _ in range(2)))
+    codes = sorted(r.status_code for r in got)
+    assert codes == [202, 409], [r.text for r in got]
+    assert [r for r in got if r.status_code == 409][0].json()["code"] == "chat_turn_busy"
+    async with SessionLocal() as session:
+        rows = list((await session.execute(select(ConversationTurn).where(
+            ConversationTurn.conversation_id == conversation_id))).scalars())
+    assert len(rows) == 1 and rows[0].seq == 0, [(r.id, r.seq) for r in rows]
+    assert len(await _jobs_of(conversation_id)) == 1
+    gate.set()
+    await _done(rows[0].id)
+    assert model.drafts == 1 and (await _row(rows[0].id)).status == "done"
+
+
+async def test_retrying_one_turn_from_several_places_at_once_starts_it_once(client, monkeypatch, source):
+    model = Counting(source["name"], mode="reply")
+    es.use_copilot(monkeypatch, model)
+    conversation_id = await _conversation(client)
+    turn_id = await _ask(client, conversation_id, "上周各区域销售额")
+    await _done(turn_id)
+    gate = asyncio.Event()
+    model.gate, model.drafts = gate, 0
+    body = {"conversation_id": conversation_id, "retry_of": turn_id, "mode": "retry"}
+    async with AsyncClient(transport=ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test") as raw:
+        got = await asyncio.gather(*(raw.post("/api/chat/turns", json=body) for _ in range(3)))
+    codes = sorted(r.status_code for r in got)
+    assert codes == [202, 409, 409], [r.text for r in got]
+    gate.set()
+    job = await _done(turn_id)
+    assert job.status == "succeeded" and model.drafts == 1, "同一轮只建了一次图"
+    events = await _events(turn_id)
+    assert [e["seq"] for e in events] == list(range(1, len(events) + 1))
+    assert [e["data"]["mode"] for e in events if e["type"] == "turn.attempt"] == ["ask", "retry"]
+    assert len((await _row(turn_id)).meta["attempts"]) == 1, "只归档了一次"
+
+
+async def test_two_tabs_resuming_the_same_turn_start_it_once(client):
+    conversation_id = await _conversation(client)
+    async with SessionLocal() as session:
+        run = Run(status="succeeded", output={"答案": "华东 300。"}, graph={"nodes": [], "edges": []}, run_class="exploratory")
+        session.add(run)
+        await session.flush()
+        turn = ConversationTurn(conversation_id=conversation_id, seq=0, question="q", status="running", run_id=run.id,
+                                meta={"v": 1, "runId": run.id})
+        session.add(turn)
+        await session.commit()
+        turn_id = turn.id
+    body = {"conversation_id": conversation_id, "retry_of": turn_id, "mode": "resume"}
+    got = await asyncio.gather(*(client.post("/api/chat/turns", json=body) for _ in range(2)))
+    assert sorted(r.status_code for r in got) == [202, 409], [r.text for r in got]
+    await _done(turn_id)
+    assert [e["type"] for e in await _events(turn_id)].count("end") == 1
+    assert (await _row(turn_id)).answer == "华东 300。"
+
+
+async def test_event_seqs_never_collide_under_concurrent_appends(client):
+    conversation_id = await _conversation(client)
+    async with SessionLocal() as session:
+        turn = ConversationTurn(conversation_id=conversation_id, seq=0, question="q", status="done")
+        session.add(turn)
+        await session.flush()
+        session.add(ChatTurnJob(id=turn.id, conversation_id=conversation_id, status="succeeded"))
+        await session.commit()
+        turn_id = turn.id
+
+    async def one(i: int) -> int:
+        async with SessionLocal() as session:
+            return await turn_log.append(session, turn_id, "op", {"op": "heartbeat", "i": i})
+
+    seqs = await asyncio.gather(*(one(i) for i in range(20)))
+    assert sorted(seqs) == list(range(1, 21)), seqs
+    async with SessionLocal() as session:
+        job = await session.get(ChatTurnJob, turn_id)
+    assert job.last_seq == 20
+
+
+async def test_a_start_that_fails_halfway_leaves_nothing_behind(client, monkeypatch, source):
+    """写 turn.attempt 时出错：轮次、任务、事件一样都不留下（不会留着一个排队中、却没有后台任务的任务，让这段对话从此一直正忙）。"""
+    es.use_copilot(monkeypatch, es.Copilot(source["name"], mode="reply"))
+    conversation_id = await _conversation(client)
+    real = turn_log.add
+    calls = {"n": 0}
+
+    async def flaky(session, turn_id, type_, data=None):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("写事件时库出错了")
+        return await real(session, turn_id, type_, data)
+
+    monkeypatch.setattr(turn_log, "add", flaky)
+    async with AsyncClient(transport=ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test") as raw:
+        r = await raw.post("/api/chat/turns", json={"conversation_id": conversation_id, "question": "第一次"})
+    assert r.status_code == 500
+    async with SessionLocal() as session:
+        assert not (await session.execute(select(ConversationTurn).where(
+            ConversationTurn.conversation_id == conversation_id))).first()
+    assert not await _jobs_of(conversation_id)
+    turn_id = await _ask(client, conversation_id, "再问一次")
+    await _done(turn_id)
+    assert (await _row(turn_id)).status == "done"
+
+
+async def test_the_job_manager_does_not_start_a_turn_twice(client, monkeypatch, source):
+    gate = asyncio.Event()
+    es.use_copilot(monkeypatch, Counting(source["name"], gate=gate, mode="reply"))
+    conversation_id = await _conversation(client)
+    turn_id = await _ask(client, conversation_id)
+    await _stage(turn_id, "build")
+    first = chat_job.manager._runs[turn_id]
+    assert chat_job.manager.start(turn_id) is False, "还在跑的同一轮不起第二个"
+    assert chat_job.manager._runs[turn_id] is first, "也不把登记换成新的（停止要停得到正在跑的那个）"
+    gate.set()
+    await _done(turn_id)
+
+
+# --------------------------------------------------------------------------
+# A2 使用者不能继续运行
+# --------------------------------------------------------------------------
+
+
+async def test_a_user_cannot_continue_a_run_from_chat_either(people):
+    """以前 POST /api/runs/{id}/continue 对使用者是 403（role_required，只放行停止）；问数据这里照原说法，运行不动。"""
+    uma = people["uma"]
+    conversation_id = await _conversation(uma, "uma 的对话")
+    async with SessionLocal() as session:
+        run = Run(status="failed", output={}, graph={"nodes": [], "edges": []}, run_class="exploratory",
+                  origin_kind="chat", started_by="uma", error="查询超时")
+        session.add(run)
+        await session.flush()
+        turn = ConversationTurn(conversation_id=conversation_id, seq=0, question="q", status="error", error="查询超时",
+                                run_id=run.id, meta={"v": 1, "runId": run.id, "runStatus": "failed"})
+        session.add(turn)
+        await session.commit()
+        turn_id, run_id = turn.id, run.id
+    old = await uma.post(f"/api/runs/{run_id}/continue", json={})
+    r = await uma.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id, "mode": "continue"})
+    assert r.status_code == 403 and r.json()["code"] == "role_required", r.text
+    assert (old.status_code, old.json()["detail"]) == (r.status_code, r.json()["detail"]), "说法同 POST /api/runs/{id}/continue"
+    async with SessionLocal() as session:
+        assert (await session.get(Run, run_id)).status == "failed"
+        assert await session.get(ChatTurnJob, turn_id) is None
+    # 重试本轮、重新运行本轮照旧能用（它们另起一次问数据运行）
+    r = await uma.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id, "mode": "rerun"})
+    assert r.status_code != 403, r.text
+
+
+# --------------------------------------------------------------------------
+# A3 晚到的复核
+# --------------------------------------------------------------------------
+
+
+async def _late_turn(client, monkeypatch, source) -> tuple[str, str, asyncio.Event]:
+    import app.api.copilot as copilot
+
+    es.use_copilot(monkeypatch, es.Copilot(source["name"]))
+    es.use_report(monkeypatch)
+    monkeypatch.setattr(chat_job, "REVIEW_WAIT_S", 0.05)
+    release = asyncio.Event()
+
+    async def slow(session, run, **kw):
+        await release.wait()
+        return {"verdict": "ok", "signals": [], "retry_instruction": None, "usage": {"calls": 1, "cost_usd": 0.004}}
+
+    monkeypatch.setattr(copilot, "review_of_run", slow)
+    conversation_id = await _conversation(client)
+    turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
+    await _done(turn_id)
+    assert chat_job.late_pending(turn_id)
+    return conversation_id, turn_id, release
+
+
+async def test_an_open_stream_gets_the_late_review_before_it_ends(client, monkeypatch, source):
+    _, turn_id, release = await _late_turn(client, monkeypatch, source)
+    last = (await _events(turn_id))[-1]["seq"]
+
+    async def active() -> str | None:
+        return await chat_job.stream_status(turn_id)
+
+    async def follow() -> list[dict[str, Any]]:
+        got = []
+        async for frame in turn_log.stream(turn_id, last, active=active, heartbeat=0.2):
+            if frame.startswith("data: "):
+                got.append(json.loads(frame[6:]))
+        return got
+
+    listening = asyncio.ensure_future(follow())
+    await asyncio.sleep(0.3)
+    release.set()
+    frames = await asyncio.wait_for(listening, timeout=10)
+    assert [f["type"] for f in frames] == ["review.late", "stream.end"], frames
+    assert frames[0]["data"]["usage"] == {"calls": 1, "cost_usd": 0.004}
+
+
+async def test_a_retry_right_after_does_not_lose_the_late_review_usage(client, monkeypatch, source):
+    """晚到的复核还没回来就点了重试：它回来时记的用量不会被新的一次尝试覆盖（这一轮的花费照实是两次加起来）。"""
+    conversation_id, turn_id, release = await _late_turn(client, monkeypatch, source)
+    gate = asyncio.Event()
+    es.use_copilot(monkeypatch, Counting(source["name"], gate=gate, mode="reply"))
+    r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id, "mode": "retry"})
+    assert r.status_code == 202, r.text
+    await _stage(turn_id, "build")
+    release.set()
+    await _until(lambda: _not_late(turn_id))
+    gate.set()
+    await _done(turn_id)
+    row = await _row(turn_id)
+    assert row.status == "done" and row.meta["usage"].get("review", {}).get("cost_usd") == 0.004, row.meta["usage"]
+
+
+async def _not_late(turn_id: str) -> bool:
+    return not chat_job.late_pending(turn_id)
+
+
+# --------------------------------------------------------------------------
+# A4 运行入口的 history 只取这一轮之前的
+# --------------------------------------------------------------------------
+
+
+async def test_history_only_takes_the_turns_before_this_one():
+    """照以前页面 historyOf 的规则：这一轮之前的、有答案（答案或说明）的，每轮截 600 字，最多 6 轮。"""
+    from app.api.conversations import run_history
+
+    async with SessionLocal() as session:
+        conversation = Conversation(kind="chat", title="历史")
+        session.add(conversation)
+        await session.flush()
+        rows = [("第一问", "答一", "", "done"), ("没答上的一问", "", "", "error"), ("第三问", "", "只有说明", "error"),
+                ("长的一问", "长" * 700, "", "done"), ("这一轮", "", "", "running"), ("后面的一问", "后面的答案", "", "done")]
+        for seq, (q, a, e, st) in enumerate(rows):
+            session.add(ConversationTurn(conversation_id=conversation.id, seq=seq, question=q, answer=a, explanation=e,
+                                         status=st))
+        await session.commit()
+        text = await run_history(session, conversation.id, before_seq=4)
+        assert text == "\n\n".join(["问：第一问\n答：答一", "问：第三问\n答：只有说明", "问：长的一问\n答：" + "长" * 600])
+        assert await run_history(session, conversation.id, before_seq=0) == ""
+        for seq in range(6, 14):
+            session.add(ConversationTurn(conversation_id=conversation.id, seq=seq, question=f"问{seq}", answer=f"答{seq}",
+                                         status="done"))
+        await session.commit()
+        assert (await run_history(session, conversation.id, before_seq=14)).count("问：") == 6
+
+
+async def test_rerunning_an_earlier_turn_does_not_feed_it_later_turns(client, monkeypatch, source):
+    es.use_copilot(monkeypatch, es.Copilot(source["name"], mode="reply"))
+    conversation_id = await _conversation(client)
+    first = await _ask(client, conversation_id, "第一问")
+    await _done(first)
+    es.use_copilot(monkeypatch, es.Copilot(source["name"]))
+    es.use_report(monkeypatch)
+    second = await _ask(client, conversation_id, "第二问", datasource_ids=[source["id"]])
+    await _done(second)
+    es.use_copilot(monkeypatch, es.Copilot(source["name"], mode="reply"))
+    third = await _ask(client, conversation_id, "第三问")
+    await _done(third)
+    r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": second, "mode": "rerun"})
+    assert r.status_code == 202, r.text
+    job = await _done(second)
+    async with SessionLocal() as session:
+        run = await session.get(Run, job.run_id)
+    history = run.input.get("history", "")
+    assert "第一问" in history and "第三问" not in history, history
+
+
+# --------------------------------------------------------------------------
+# A5 只有问数据的对话能发起一轮
+# --------------------------------------------------------------------------
+
+
+async def test_only_chat_conversations_can_start_a_turn(client):
+    async with SessionLocal() as session:
+        canvas = Conversation(kind="canvas", title="画布助手的对话")
+        app_conv = Conversation(kind="app", title="应用润色")
+        session.add_all([canvas, app_conv])
+        await session.commit()
+        ids = [canvas.id, app_conv.id]
+    for conversation_id in ids:
+        r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "question": "q"})
+        assert r.status_code == 422 and r.json()["code"] == "chat_turn_invalid", r.text
+        async with SessionLocal() as session:
+            n = await session.scalar(select(func.count()).select_from(ConversationTurn).where(
+                ConversationTurn.conversation_id == conversation_id))
+        assert n == 0
+