| from __future__ import annotations |
|
|
| import threading |
| from concurrent.futures import ThreadPoolExecutor |
| from typing import Any, Callable, TypeVar |
|
|
|
|
| T = TypeVar("T") |
|
|
|
|
| class _NoQtInvoker: |
| def invoke(self, payload: dict[str, Any]) -> None: |
| fn = payload["fn"] |
| try: |
| payload["result"] = fn() |
| except Exception as exc: |
| payload["error"] = exc |
| finally: |
| payload["event"].set() |
|
|
|
|
| def _qt_invoker_class(): |
| try: |
| from PySide6.QtCore import QObject, Signal, Slot |
| except Exception: |
| return None |
|
|
| class _QtInvoker(QObject): |
| invoke_requested = Signal(object) |
|
|
| def __init__(self) -> None: |
| super().__init__() |
| self.invoke_requested.connect(self._invoke) |
|
|
| @Slot(object) |
| def _invoke(self, payload: dict[str, Any]) -> None: |
| fn = payload["fn"] |
| try: |
| payload["result"] = fn() |
| except Exception as exc: |
| payload["error"] = exc |
| finally: |
| payload["event"].set() |
|
|
| def invoke(self, payload: dict[str, Any]) -> None: |
| self.invoke_requested.emit(payload) |
|
|
| return _QtInvoker |
|
|
|
|
| class RemoteCommandDispatcher: |
| """Boundary between HTTP worker threads and ADAM's Qt-owned services.""" |
|
|
| def __init__(self, *, timeout_seconds: float = 30.0) -> None: |
| self.timeout_seconds = timeout_seconds |
| self.main_thread_id = threading.get_ident() |
| self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="adam-remote-plan") |
| self._invoker: Any = _NoQtInvoker() |
| invoker_class = _qt_invoker_class() |
| if invoker_class is not None: |
| try: |
| from PySide6.QtCore import QCoreApplication |
|
|
| if QCoreApplication.instance() is not None: |
| self._invoker = invoker_class() |
| except Exception: |
| self._invoker = _NoQtInvoker() |
|
|
| def shutdown(self) -> None: |
| self._executor.shutdown(wait=False, cancel_futures=True) |
|
|
| def call_background(self, fn: Callable[[], T], *, timeout: float | None = None) -> T: |
| future = self._executor.submit(fn) |
| return future.result(timeout=timeout or self.timeout_seconds) |
|
|
| def call_ui(self, fn: Callable[[], T], *, timeout: float | None = None) -> T: |
| if threading.get_ident() == self.main_thread_id: |
| return fn() |
| event = threading.Event() |
| payload: dict[str, Any] = {"fn": fn, "event": event} |
| self._invoker.invoke(payload) |
| if not event.wait(timeout or self.timeout_seconds): |
| raise TimeoutError("ADAM did not finish the remote command in time.") |
| if "error" in payload: |
| raise payload["error"] |
| return payload.get("result") |
|
|
| def submit_job(self, jobs: Any, plan: Any) -> Any: |
| return self.call_ui(lambda: jobs.submit(plan)) |
|
|
| def confirm_job(self, jobs: Any, job_id: str) -> Any: |
| return self.call_ui(lambda: jobs.confirm(job_id)) |
|
|
| def job_action(self, jobs: Any, job_id: str, action: str) -> Any: |
| def run() -> Any: |
| job = jobs.get(job_id) if hasattr(jobs, "get") else None |
| if job is None: |
| raise ValueError("ADAM could not find that job.") |
| if action == "cancel": |
| jobs.cancel(job_id) |
| return {"job": job} |
| if action == "confirm": |
| jobs.confirm(job_id) |
| return {"job": job} |
| if action == "pause": |
| jobs.pause(job_id) |
| return {"job": job} |
| if action == "resume": |
| jobs.resume(job_id) |
| return {"job": job} |
| if action == "end": |
| ended = jobs.end_task(job_id) if hasattr(jobs, "end_task") else False |
| if not ended: |
| raise ValueError("Only interrupted jobs can be ended remotely.") |
| return {"job": job} |
| if action == "retry": |
| return {"job": job, "retried": jobs.retry(job_id)} |
| raise ValueError("Unsupported remote job action.") |
|
|
| return self.call_ui(run) |
|
|
