diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index d993beab..b0c24e06 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -100,6 +100,34 @@ def principal_of(snapshot: Mapping[str, Any] | None) -> Any: mode=str(s.get("mode") or "off"), via=str(s.get("via") or "session")) +async def current_principal(snapshot: Mapping[str, Any] | None) -> tuple[Any, str | None]: + """发起时的身份快照 → 接手这一刻的身份：(身份, 不能接着做的原因)。 + + 关着登录时发起的（快照里 mode 是 off、没有用户名）照快照：没有账号可核对。开着登录的按用户名读现在的账号：不在了、 + 停用了交回原因；角色、组、显示名换成现在的（降了角色就按降了之后的做）；按现在的身份已经用不了问数据（使用者、管理员 + 没开问数据，同 api/deps._chat_scope）也交回原因。认证方式、是否核实过照快照。""" + from app.api.deps import Principal + from app.auth import user_can_chat + from app.auth.sessions import groups_of + from app.db.models_identity import User + + s = dict(snapshot or {}) + name = s.get("username") + if not s or str(s.get("mode") or "off") == "off" or not name: + return principal_of(snapshot), None + async with SessionLocal() as session: + user = (await session.execute(select(User).where(User.username == name))).scalar_one_or_none() + if user is None or user.disabled: + return None, f"发起这一轮的账号「{name}」已{'停用' if user is not None else '不存在'}，本轮没有接着收尾" + groups = await groups_of(session, user.id) + principal = Principal(username=user.username, role=user.role, display_name=user.display_name or user.username, + user_id=user.id, groups=groups, verified=bool(s.get("verified")), + mode=str(s.get("mode") or "off"), via=str(s.get("via") or "session")) + if not principal.at_least("builder") and not (principal.at_least("user") and user_can_chat()): + return None, f"发起这一轮的账号「{name}」已经没有问数据的权限，本轮没有接着收尾" + return principal, None + + # -------------------------------------------------------------------------- # 驱动：问数据和评测不一样的那几项 # -------------------------------------------------------------------------- @@ -362,8 +390,13 @@ class ChatTurnExecution(runtime.JobRun): self.request = dict(job.request or {}) self.turn_seq = row.seq self.run_id = job.run_id - # 这一轮按发起的人的身份跑：answer_ops 里的 restricted()、对话归属，发起运行时的遮罩、只读都看它 - set_principal(principal_of(job.principal)) + # 这一轮按发起的人的身份跑：answer_ops 里的 restricted()、对话归属，发起运行时的遮罩、只读都看它。按这一刻的账号重新核对 + # （current_principal）：服务重启后接着收尾时，快照是发起那一刻的，之后停用了、降了角色的不能照快照接着做 + principal, refused = await current_principal(job.principal) + if refused is not None: + await self._identity_refused(refused) + return + set_principal(principal) scope = [s.get("id") for s in self.request.get("scope") or [] if isinstance(s, Mapping) and s.get("id")] driver = self.driver = ChatDriver(self, question=row.question, datasource_ids=scope) driver.turn.attempts = int(job.attempt or 0) @@ -427,6 +460,16 @@ class ChatTurnExecution(runtime.JobRun): await self._end("cancelled") await self._job(status="cancelled", stage=None, finished_at=utcnow()) + async def _identity_refused(self, reason: str) -> None: + """发起这一轮的账号接手时已经停用、或者已经没有问数据的权限：照实标中断、写清原因，不再往下做（不建图、不等运行、 + 不复核）。已经发起的那次运行不动——停在审批上的照旧等人，在记录页照常看得到；这一轮不替一个已经无权的人收它的尾。""" + await self._settle() + if not self.ended: + await self.save(status="error", error=reason, + meta={"cause": {"kind": "identity", "message": reason}, **_timing(self.request)}) + await self._end("error") + await self._job(status="interrupted", stage=None, error=reason, finished_at=utcnow()) + async def _failed(self, reason: str) -> None: await self._crash_out(reason) @@ -1166,5 +1209,5 @@ async def shutdown() -> None: __all__ = ["APPROVAL_POLL_S", "AUTO_RETRY_MODES", "BUSY", "CONCURRENCY_CLASS", "ChatDriver", "ChatTurnExecution", "MODES", "REVIEW_WAIT_S", "TURN_BUSY", "TURN_FINISHED", "TURN_INVALID", "TurnRefused", "cancel", "forget", - "job_view", "late_pending", "manager", "principal_of", "principal_snapshot", "shutdown", "start", "startup", - "state", "stop_conversation", "stream_status"] + "current_principal", "job_view", "late_pending", "manager", "principal_of", "principal_snapshot", "shutdown", + "start", "startup", "state", "stop_conversation", "stream_status"] diff --git a/backend/tests/test_chat_turns.py b/backend/tests/test_chat_turns.py index b0d2c12d..46d7066e 100644 --- a/backend/tests/test_chat_turns.py +++ b/backend/tests/test_chat_turns.py @@ -333,8 +333,9 @@ async def test_stop_while_running_cancels_the_run(client, monkeypatch, source): # -------------------------------------------------------------------------- -async def _stranded(conversation_id: str, *, run: Run | None, stage: str, mode: str = "ask") -> str: - """造一轮「上次进程没了时还在跑」的：那一行 running、任务 running（不在这个进程里）。""" +async def _stranded(conversation_id: str, *, run: Run | None, stage: str, mode: str = "ask", + principal: dict[str, Any] | None = None) -> str: + """造一轮「上次进程没了时还在跑」的：那一行 running、任务 running（不在这个进程里）。principal 是发起时的身份快照。""" async with SessionLocal() as session: if run is not None: session.add(run) @@ -344,7 +345,7 @@ async def _stranded(conversation_id: str, *, run: Run | None, stage: str, mode: session.add(turn) await session.flush() session.add(ChatTurnJob(id=turn.id, conversation_id=conversation_id, status="running", stage=stage, mode=mode, - run_id=run.id if run else None, request={"started_ms": 0}, principal={})) + run_id=run.id if run else None, request={"started_ms": 0}, principal=principal or {})) await session.commit() return turn.id @@ -1064,3 +1065,89 @@ async def test_the_cleanup_runs_in_the_background_tick(monkeypatch): again = await _finished_turn(days_ago=turn_log.RETAIN_DAYS + 2) await loop.process_due(utcnow()) assert len(await _events(again)) == 3, "一小时内不再清第二次" + + +# -------------------------------------------------------------------------- +# 服务重启后接着收尾：按接手时的账号重新核对 +# -------------------------------------------------------------------------- + + +async def _snapshot_of(name: str) -> dict[str, Any]: + from app.db.models_identity import User + + async with SessionLocal() as session: + user = (await session.execute(select(User).where(User.username == name))).scalar_one() + return {"username": user.username, "role": user.role, "display_name": user.display_name, "user_id": user.id, + "groups": [], "verified": True, "mode": "local", "via": "session"} + + +async def _set_user(name: str, **values: Any) -> None: + from sqlalchemy import update as sa_update + + from app.db.models_identity import User + + async with SessionLocal() as session: + await session.execute(sa_update(User).where(User.username == name).values(**values)) + await session.commit() + + +def _finished_run(by: str) -> Run: + return Run(status="succeeded", output={"答案": "华东 300。"}, graph={"nodes": [], "edges": []}, run_class="exploratory", + origin_kind="chat", started_by=by) + + +async def test_resuming_after_a_restart_checks_the_account_again(people, monkeypatch, source): + """服务重启后接着收尾一轮（已发起运行的接着等、复核），按接手这一刻的账号重新核对，不再照发起时的身份快照接着做： + 发起之后账号被停用的，照实标中断、写清原因；那一行不再往下改答案。""" + es.use_copilot(monkeypatch, es.Copilot(source["name"])) + es.use_report(monkeypatch) + mine = await _conversation(people["uma"], "uma 的对话") + turn_id = await _stranded(mine, run=_finished_run("uma"), stage="review", principal=await _snapshot_of("uma")) + await _set_user("uma", disabled=True) + await chat_job.startup(resume=True) + await chat_job.manager.wait(turn_id, timeout=30) + row = await _row(turn_id) + assert row.status == "error" and row.meta["cause"]["kind"] == "identity", row.meta + assert row.error == "发起这一轮的账号「uma」已停用，本轮没有接着收尾" and not row.answer + async with SessionLocal() as session: + assert (await session.get(ChatTurnJob, turn_id)).status == "interrupted" + assert (await _events(turn_id))[-1]["data"]["outcome"] == "error" + + +async def test_a_downgraded_account_without_chat_is_interrupted(people, monkeypatch, source): + """发起时是搭建者，之后被降成使用者、管理员又没给使用者开问数据：已经没有权限了，照实标中断。""" + es.use_copilot(monkeypatch, es.Copilot(source["name"])) + es.use_report(monkeypatch) + theirs = await _conversation(people["bob"], "bob 的对话") + turn_id = await _stranded(theirs, run=_finished_run("bob"), stage="review", principal=await _snapshot_of("bob")) + await _set_user("bob", role="user") + assert (await people["admin"].put("/api/auth/settings", json={"user_can_chat": False})).status_code == 200 + await chat_job.startup(resume=True) + await chat_job.manager.wait(turn_id, timeout=30) + row = await _row(turn_id) + assert row.status == "error" and row.meta["cause"]["kind"] == "identity" + assert row.error == "发起这一轮的账号「bob」已经没有问数据的权限，本轮没有接着收尾" + + +async def test_a_downgraded_account_that_can_still_chat_finishes_as_who_they_are_now(people, monkeypatch, source): + """降了角色、但仍然用得了问数据（使用者开着问数据）：照常收尾，按现在的身份（角色、组）接着做，不按快照里的搭建者。""" + es.use_copilot(monkeypatch, es.Copilot(source["name"])) + es.use_report(monkeypatch) + seen: list[Any] = [] + real = chat_job.ChatDriver.drive + + async def spy(self, *a, **k): + from app.api.deps import current_principal + + seen.append(current_principal()) + return await real(self, *a, **k) + + monkeypatch.setattr(chat_job.ChatDriver, "drive", spy) + theirs = await _conversation(people["bob"], "bob 的对话") + turn_id = await _stranded(theirs, run=_finished_run("bob"), stage="review", principal=await _snapshot_of("bob")) + await _set_user("bob", role="user") + await chat_job.startup(resume=True) + await chat_job.manager.wait(turn_id, timeout=30) + row = await _row(turn_id) + assert row.status == "done" and row.answer == "华东 300。" + assert seen and seen[-1].username == "bob" and seen[-1].role == "user" diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index 340eb304..3f6e4a90 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -136,6 +136,8 @@ export type TurnCause = | { kind: 'run'; error?: string; detail?: string | null } | { kind: 'gone' } | { kind: 'build_interrupted' } + /** 服务重启后接着收尾时，发起这一轮的账号已停用、或者已经没有问数据的权限（服务端 chat/job.current_principal） */ + | { kind: 'identity'; message?: string } | { kind: 'crash'; message?: string } /** @@ -649,6 +651,12 @@ function causeFailure(cause: TurnCause, scope?: ScopeSource[]): Failure { return GONE case 'build_interrupted': return BUILD_INTERRUPTED + case 'identity': + return { + title: '上次未完成', + reason: cause.message || '发起这一轮的账号已无法接着收尾，本轮没有接着做。', + hint: '如需继续，请由现在有问数据权限的人重新提问。', + } case 'crash': return failureOf(cause.message || '发生错误') }