yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 01:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

问数据接着收尾一轮时按当前账号重新核对身份：服务重启后接手（已发起运行的接着等、复核）以前用的是发起时的身份快照，之后停用了、降了角色不会再核对。现在每次执行开始时按用户名读现在的账号（chat/job.current_principal）：不在了、停用了，或者按现在的角色已经用不了问数据的，照实标中断、写清原因（cause identity），不再往下做；还用得了的按现在的角色、组接着做。已经发起的运行不动。页面照原因说「上次未完成」

Browse Source
This commit is contained in:
ThearchyHelios
2026-10-19 09:38:07 +08:00
parent edde451d19
commit fa431da91b
3 changed files with 145 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/chat/job.py
+47 -4
View File
@@ -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"]
backend/tests/test_chat_turns.py
+90 -3
View File
@@ -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"
frontend/src/store/chat.ts
+8
View File
@@ -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 || '发生错误')
}