From c768c56fd737663cb6111388d2d276f3eb6ce9f4 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 11:51:29 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=9A=E8=BF=9B=E7=A8=8B=E5=8D=8F=E8=B0=83?= =?UTF-8?q?=E8=A1=A5=E9=BD=90=E5=87=A0=E5=A4=84=EF=BC=9A=E6=89=A7=E8=A1=8C?= =?UTF-8?q?=E6=97=B6=E9=95=BF=E4=BB=8E=E6=8B=BF=E5=88=B0=E5=90=8D=E9=A2=9D?= =?UTF-8?q?=E7=AE=97=E8=B5=B7=EF=BC=88=E6=8E=92=E9=98=9F=E7=AD=89=E5=90=8D?= =?UTF-8?q?=E9=A2=9D=E4=B8=8D=E7=AE=97=E6=89=A7=E8=A1=8C=EF=BC=8C=E5=A2=99?= =?UTF-8?q?=E9=92=9F=E6=9C=AC=E6=9D=A5=E5=B0=B1=E4=BB=8E=E5=BC=80=E5=A7=8B?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E7=AE=97=EF=BC=89=EF=BC=9B=E8=93=9D=E5=9B=BE?= =?UTF-8?q?=E7=95=99=E7=97=95=E3=80=81=E8=93=9D=E5=9B=BE=E5=8A=A9=E6=89=8B?= =?UTF-8?q?=E5=BC=80=E8=BD=AE=E5=85=88=E5=8D=A0=E4=BD=8F=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E8=A1=8C=E5=86=8D=E5=8F=96=E6=9C=80=E5=A4=A7=E5=80=BC=E5=BE=80?= =?UTF-8?q?=E4=B8=8B=E6=8E=A5=EF=BC=9B=E9=97=AE=E6=95=B0=E6=8D=AE=E7=AD=89?= =?UTF-8?q?=E5=88=AB=E7=9A=84=E8=BF=9B=E7=A8=8B=E6=8E=A5=E7=9D=80=E8=B7=91?= =?UTF-8?q?=E7=9A=84=E8=BF=90=E8=A1=8C=E6=97=B6=E5=9B=9E=E5=BA=93=E9=87=8C?= =?UTF-8?q?=E7=AD=89=E5=B0=81=E5=AD=98=EF=BC=9B=E6=8C=89=E9=85=8D=E6=96=B9?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=E7=9A=84=20AI=20=E8=B5=B7=E8=8D=89=E3=80=81?= =?UTF-8?q?=E7=9F=A5=E8=AF=86=E5=BA=93=E9=87=8D=E5=BB=BA=E7=B4=A2=E5=BC=95?= =?UTF-8?q?=E5=8F=A6=E5=8F=96=E8=B7=A8=E8=BF=9B=E7=A8=8B=E7=9A=84=E9=94=81?= =?UTF-8?q?=EF=BC=88=E5=A4=9A=E8=BF=9B=E7=A8=8B=E6=97=B6=E9=87=8D=E5=BB=BA?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E8=AE=B0=E5=9C=A8=E8=AE=BE=E7=BD=AE=E8=A1=A8?= =?UTF-8?q?=E9=87=8C=EF=BC=8C=E5=88=AB=E7=9A=84=E8=BF=9B=E7=A8=8B=E8=AF=BB?= =?UTF-8?q?=E5=BE=97=E5=88=B0=E3=80=81=E5=B1=9E=E4=B8=BB=E4=B8=8D=E5=9C=A8?= =?UTF-8?q?=E4=BA=86=E7=85=A7=E5=AE=9E=E8=AF=B4=E6=B2=A1=E5=81=9A=E5=AE=8C?= =?UTF-8?q?=EF=BC=89=EF=BC=9B=E8=B0=83=E5=BA=A6=E5=BE=AA=E7=8E=AF=E4=B8=80?= =?UTF-8?q?=E6=8B=8D=E7=9A=84=E9=97=B4=E9=9A=94=E5=8F=AF=E9=85=8D=EF=BC=88?= =?UTF-8?q?AGENTLAB=5FSCHEDULER=5FTICK=5FS=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/copilot_blueprint.py | 3 ++ backend/app/api/knowledge.py | 81 ++++++++++++++++++++++++++-- backend/app/api/settings.py | 3 +- backend/app/blueprints/store.py | 8 ++- backend/app/chat/drive.py | 17 ++++++ backend/app/cluster/mutex.py | 34 ++++++++++++ backend/app/core/config.py | 2 + backend/app/data/recipe_imports.py | 16 ++++-- backend/app/engine/runner.py | 2 + backend/app/scheduler/__init__.py | 4 +- 10 files changed, 158 insertions(+), 12 deletions(-) diff --git a/backend/app/api/copilot_blueprint.py b/backend/app/api/copilot_blueprint.py index 32cc2064..515f0ba2 100644 --- a/backend/app/api/copilot_blueprint.py +++ b/backend/app/api/copilot_blueprint.py @@ -207,9 +207,12 @@ async def _conversation(session: AsyncSession, blueprint: Any, actor: str | None async def _open_turn(session: AsyncSession, conversation: Any, question: str) -> str: from sqlalchemy import func, select + from app.chat.job import hold_conversation from app.db.base import utcnow from app.db.models import ConversationTurn + # 先占住这段对话再取最大值往下接（同 api/conversations 开一轮）：多进程时同时两次开轮不撞号 + await hold_conversation(session, conversation.id) seq = int((await session.execute(select(func.coalesce(func.max(ConversationTurn.seq), -1)) .where(ConversationTurn.conversation_id == conversation.id))).scalar_one()) + 1 turn = ConversationTurn(conversation_id=conversation.id, seq=seq, question=question, status="running") diff --git a/backend/app/api/knowledge.py b/backend/app/api/knowledge.py index 2288f1e6..5e858428 100644 --- a/backend/app/api/knowledge.py +++ b/backend/app/api/knowledge.py @@ -477,8 +477,10 @@ class _ReindexJob: } -#: 最近一次重建。只在进程里记：重建是幂等的，服务重启丢了进度，再点一次就是 +#: 最近一次重建（本进程里的）。重建是幂等的，服务重启丢了进度，再点一次就是。多进程时进度另记在设置表里（REINDEX_KEY， +#: 带上做它的进程），别的进程上的「有没有在跑」「到哪了」读那一份；同一时间只跑一个靠跨进程的锁（kb-reindex） _REINDEX: _ReindexJob | None = None +REINDEX_KEY = "cluster:kb_reindex" #: 后台任务要有人拿着引用，否则跑到一半可能被回收 _REINDEX_TASK: Any = None @@ -487,9 +489,59 @@ def reindex_job() -> _ReindexJob | None: return _REINDEX -async def _run_reindex(job: _ReindexJob) -> None: +async def _save_state(job: _ReindexJob) -> None: + """把进度记进设置表（多进程时别的进程读它；SQLite 只有一个进程，不记）。写不进去只记日志。""" + from app.cluster import counters, identity + + if not counters.shared(): + return + from app.db.base import SessionLocal + from app.db.models import Setting + + try: + async with SessionLocal() as session: + row = await session.get(Setting, REINDEX_KEY) + value = {**job.view(), "owner": identity.owner()} + if row is None: + session.add(Setting(key=REINDEX_KEY, value=value)) + else: + row.value = value + await session.commit() + except Exception: # noqa: BLE001 + logger.exception("重建索引的进度没记进库") + + +async def _stored_state() -> dict[str, Any] | None: + """别的进程记的那一份：还写着在跑、做它的进程却不在了的，照实说没做完。SQLite 只有一个进程，没有这一份。""" + from app.cluster import counters, state + + if not counters.shared(): + return None + from app.db.base import SessionLocal + from app.db.models import Setting + + async with SessionLocal() as session: + row = await session.get(Setting, REINDEX_KEY) + value = dict(row.value) if row is not None and isinstance(row.value, dict) else None + if value is None: + return None + owner = value.pop("owner", None) + if value.get("state") == "running" and not (await state.live_for([owner])).alive(owner, local=False): + value.update(state="failed", error="重建索引未完成：做这件事的服务进程停止了", hint="再点一次重建就是") + return value + + +async def _run_reindex(job: _ReindexJob, lock: Any = None) -> None: + import asyncio + from app.db.base import SessionLocal, utcnow + async def report() -> None: + while True: + await _save_state(job) + await asyncio.sleep(1.0) + + reporter = asyncio.ensure_future(report()) try: async with SessionLocal() as session: out = await kb.reindex(session, job.collection, on_progress=job.chunk_step) @@ -503,6 +555,10 @@ async def _run_reindex(job: _ReindexJob) -> None: job.state, job.error, job.hint = "failed", f"重建索引未完成：{reason}", hint or None finally: job.finished_at = utcnow() + reporter.cancel() + await _save_state(job) + if lock is not None: + await lock.give_back() @kb_router.post("/reindex") @@ -542,12 +598,22 @@ async def reindex( 409, f"正在重建索引（{_REINDEX.collection or '全部集合'}，" f"已完成 {running['done']}/{running['total']}），请等待完成后再操作", ) + # 多进程：别的进程上正在重建的，跨进程的锁占不到（app/cluster/mutex），同样 409，说法照它记在库里的进度 + from app.cluster import mutex + + lock = mutex.lock("kb-reindex") + if not await lock.try_take(): + other = await _stored_state() or {} + raise HTTPException( + 409, f"正在重建索引（{other.get('collection') or '全部集合'}，" + f"已完成 {other.get('done', 0)}/{other.get('total', 0)}），请等待完成后再操作", + ) job = _REINDEX = _ReindexJob(collection, chunks, memories) if background: - _REINDEX_TASK = asyncio.create_task(_run_reindex(job)) + _REINDEX_TASK = asyncio.create_task(_run_reindex(job, lock)) return JSONResponse(job.view(), status_code=202) - await _run_reindex(job) + await _run_reindex(job, lock) if job.state == "failed": raise HTTPException(502, job.error + (f"。{job.hint}" if job.hint else "")) return job.result @@ -557,8 +623,13 @@ async def reindex( async def reindex_status() -> dict[str, Any]: """最近一次重建的进度：{state, phase, total, done, chunks, memories, result, error…}。 - 还没重建过返回 {state: "idle"}。 + 还没重建过返回 {state: "idle"}。多进程时本进程没在做的，读做它的进程记在库里的那一份。 """ + if _REINDEX is not None and _REINDEX.state == "running": + return _REINDEX.view() + stored = await _stored_state() + if stored is not None and (_REINDEX is None or str(stored.get("started_at") or "") > _REINDEX.view()["started_at"]): + return stored return _REINDEX.view() if _REINDEX is not None else {"state": "idle"} diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 46c7ee2d..e3217874 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -914,7 +914,8 @@ def _internal(key: str) -> bool: """通知自己记的账（每天汇总的认领行、自动建过邮件渠道的人，notify/digest.STATE_PREFIX）：不是用户设置，不列、不收。""" from app.notify.digest import STATE_PREFIX - return key.startswith(STATE_PREFIX) + # cluster: 开头的是多进程协调自己记的（重建索引的进度之类，app/cluster），同样不是用户设置 + return key.startswith(STATE_PREFIX) or key.startswith("cluster:") DEFAULT_SETTINGS: dict[str, Any] = { diff --git a/backend/app/blueprints/store.py b/backend/app/blueprints/store.py index 377a0b3f..cdfba70f 100644 --- a/backend/app/blueprints/store.py +++ b/backend/app/blueprints/store.py @@ -152,8 +152,14 @@ async def _verified_reuse(session: AsyncSession, op: dict[str, Any]) -> dict[str async def _next_seq(session: AsyncSession, blueprint_id: str) -> int: - from app.db.models_blueprints import BlueprintEvent + """这个蓝图下一条留痕的序号。先占住蓝图那一行（PostgreSQL 上是行锁，SQLite 上拿到写锁）再取最大值往下接：多进程时同一个 + 蓝图同时两处记留痕（卡片上改、助手提的），一个接一个，不撞 (blueprint_id, seq) 的唯一约束。updated_at 不动。""" + from sqlalchemy import update + from app.db.models_blueprints import Blueprint, BlueprintEvent + + await session.execute(update(Blueprint).where(Blueprint.id == blueprint_id) + .values(updated_at=Blueprint.updated_at).execution_options(synchronize_session=False)) return int((await session.execute(select(func.coalesce(func.max(BlueprintEvent.seq), 0)) .where(BlueprintEvent.blueprint_id == blueprint_id))).scalar_one()) + 1 diff --git a/backend/app/chat/drive.py b/backend/app/chat/drive.py index f9f1f899..95b4fe82 100644 --- a/backend/app/chat/drive.py +++ b/backend/app/chat/drive.py @@ -100,6 +100,20 @@ def add_usage(into: dict[str, Any] | None, more: Mapping[str, Any] | None) -> di return out +async def _sealed(run_id: str, timeout: float) -> None: + """等终态的运行封存好（manifest_hash 写上），最多 timeout 秒。""" + from app.db.base import SessionLocal + from app.db.models import Run + + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + async with SessionLocal() as session: + if await session.scalar(select(Run.manifest_hash).where(Run.id == run_id)): + return + await asyncio.sleep(0.1) + + class TurnDriver: """一轮问数据。drive() 跑完一轮，结果在 turn 上；on_phase(阶段) 在每一步开始时调。子类给差异（见模块说明）。 @@ -244,6 +258,9 @@ class TurnDriver: if row is None or row.status in ("succeeded", "failed", "cancelled", "interrupted"): # 封存在终态之后：等执行任务收完尾（含封存），评分只读封存范围 await run_manager.wait_idle(run_id, timeout=CANCEL_GRACE_S) + if row is not None and row.status != "interrupted" and not run_manager.is_active(run_id): + # 多进程：运行在别的进程里执行（审批之后被别的进程接着跑了），本进程等不到它的执行任务，回库里等封存 + await _sealed(run_id, CANCEL_GRACE_S) async with SessionLocal() as session: return await session.get(Run, run_id) await asyncio.sleep(0.2) diff --git a/backend/app/cluster/mutex.py b/backend/app/cluster/mutex.py index bf71a30a..2de45376 100644 --- a/backend/app/cluster/mutex.py +++ b/backend/app/cluster/mutex.py @@ -45,6 +45,40 @@ class ClusterLock: def _waiters(self) -> Any: return self._local._waiters + async def try_take(self) -> bool: + """不等：本进程、别的进程都没人占着就占上，交回占没占到。占到了用 give_back 放（可以在另一个任务里放：后台任务做完才放）。""" + if self._local.locked(): + return False + await self._local.acquire() + if settings.db_dialect != "postgresql": + return True + try: + from sqlalchemy import text + + from app.db.base import engine + + conn = await engine.connect() + try: + conn = await conn.execution_options(isolation_level="AUTOCOMMIT") + got = bool(await conn.scalar(text("SELECT pg_try_advisory_lock(:k)"), {"k": self.key})) + except BaseException: + await conn.close() + raise + if not got: + await conn.close() + self._local.release() + return False + self._conn = conn + except BaseException: + if self._local.locked() and self._conn is None: + self._local.release() + raise + return True + + async def give_back(self) -> None: + """放掉 try_take 占到的。""" + await self.__aexit__(None, None, None) + async def __aenter__(self) -> ClusterLock: await self._local.acquire() if settings.db_dialect != "postgresql": diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 96fd430f..8d5a37fc 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -135,6 +135,8 @@ class Settings(BaseSettings): # （检查脚本的 --stacks、沙箱）必须关：库里的定时任务会照常到点发起运行、发出 webhook， # 拷贝里的任务和原库的是同一批，等于同一件事做两遍 scheduler: bool = True + # 调度循环一拍的间隔（秒）：定时认领、事件派发、流程的时限和核实都按它查一次。缺省 15；验收脚本（多进程端到端）调小 + scheduler_tick_s: float = 15.0 # --- 工具 --- http_tool_timeout: int = 20 diff --git a/backend/app/data/recipe_imports.py b/backend/app/data/recipe_imports.py index e3d4598e..59d29f1c 100644 --- a/backend/app/data/recipe_imports.py +++ b/backend/app/data/recipe_imports.py @@ -1703,8 +1703,9 @@ def _adopt_ai_work(p: Pipeline, book: _Book, snap: _Snap, work: Any) -> _Eval: return _evaluate(p, book, snap, work) -#: 正在 AI 起草的暂存区（进程内）：同一暂存区不并发起草，否则两次发送各花一份钱，同意记录、用量还会互相覆盖 -#: （SE-11）。事件循环单线程，查和加之间没有 await，不会有两个请求同时通过；进程重启后自然清空 +#: 正在 AI 起草的暂存区（本进程里）：同一暂存区不并发起草，否则两次发送各花一份钱，同意记录、用量还会互相覆盖 +#: （SE-11）。事件循环单线程，查和加之间没有 await，不会有两个请求同时通过；进程重启后自然清空。多进程时另取跨进程的锁 +#: （run_ai_draft） _AI_RUNNING: set[str] = set() @@ -1716,9 +1717,18 @@ async def run_ai_draft(session: AsyncSession, staging: ImportStaging, *, consent raise ImportRefused(422, "ai_consent_required", "需要先查看并同意发送的内容，才能让 AI 起草") if staging.id in _AI_RUNNING: raise ImportRefused(409, "ai_in_progress", "这次导入正在由 AI 起草，请等它完成后再试") + # 多进程：另一个进程上正在起草同一个暂存区的，跨进程的锁占不到，同样回 409（app/cluster/mutex） + from app.cluster import mutex + + lock = mutex.lock(f"recipe-ai:{staging.id}") _AI_RUNNING.add(staging.id) try: - return await _run_ai_draft(session, staging, preview_sha256=preview_sha256, signed_by=signed_by) + if not await lock.try_take(): + raise ImportRefused(409, "ai_in_progress", "这次导入正在由 AI 起草，请等它完成后再试") + try: + return await _run_ai_draft(session, staging, preview_sha256=preview_sha256, signed_by=signed_by) + finally: + await lock.give_back() finally: _AI_RUNNING.discard(staging.id) diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 9e2dc47a..73280840 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -792,6 +792,8 @@ class RunManager: # active + wait 比墙钟还长（机器一忙就差出几十毫秒） paused: float | None = None async with slots.hold(run_id, origin_kind): + # 执行时长从拿到名额算起：排队等名额的那一段不算执行（墙钟从 started_at 算，started_at 也是拿到名额之后才写） + started = time.perf_counter() try: async with SessionLocal() as session: row = await session.get(Run, run_id) diff --git a/backend/app/scheduler/__init__.py b/backend/app/scheduler/__init__.py index d6a48183..3b4250e8 100644 --- a/backend/app/scheduler/__init__.py +++ b/backend/app/scheduler/__init__.py @@ -34,9 +34,9 @@ async def start(*, interval: float | None = None, config: Settings | None = None if not scheduler_enabled(config): logger.info("定时任务调度已关闭（AGENTLAB_SCHEDULER），不会自动发起运行") return False - from app.scheduler.loop import TICK_SECONDS, runner + from app.scheduler.loop import runner - await runner.start(interval if interval is not None else TICK_SECONDS) + await runner.start(interval if interval is not None else float((config or settings).scheduler_tick_s)) return True