"""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