32 lines
1.6 KiB
Python
32 lines
1.6 KiB
Python
"""A pinned online worker can identify its own orphaned in-flight records."""
|
|
|
|
import inspect
|
|
|
|
from deerflow.runtime.runs.manager import RunManager
|
|
from deerflow.runtime.runs.schemas import RunStatus
|
|
|
|
|
|
class WorkerRunManager(RunManager):
|
|
def has_active_thread(self, thread_id: str) -> bool:
|
|
return any(
|
|
record.thread_id == thread_id
|
|
and record.status in {RunStatus.pending, RunStatus.running}
|
|
and (record.task is None or not record.task.done())
|
|
for record in self._runs.values()
|
|
)
|
|
|
|
async def get_persisted(self, run_id: str, *, authorized_thread_id: str | None = None):
|
|
if "authorized_thread_id" in inspect.signature(RunManager.get_persisted).parameters:
|
|
record = await super().get_persisted(run_id, authorized_thread_id=authorized_thread_id)
|
|
else:
|
|
record = await super().get_persisted(run_id)
|
|
if record is not None and authorized_thread_id and record.thread_id != authorized_thread_id:
|
|
return None
|
|
if record is not None and self.get(run_id) is None and record.status in {RunStatus.pending, RunStatus.running}:
|
|
# Routing never migrates a live process. Missing live state means a
|
|
# previous process ended; replaying tools here would be unsafe.
|
|
record.status = RunStatus.error
|
|
record.error = "执行节点已重启，本轮已中断；已保存的聊天记录保留。请核对外部操作结果后再继续。"
|
|
await self._store.update_status(run_id, record.status.value, error=record.error)
|
|
return record
|