Async Worker Pattern

The Problem

External API calls (fetching structures, running agents, querying databases) can take seconds or longer. Running them on the main thread would freeze the UI.

The Solution: QThread Workers

codemol uses Qt's QThread with signal-based callbacks:

sequenceDiagram
    participant UI as Main Thread
    participant W as Worker (QThread)
    participant API as External API

    UI->>W: worker.start()
    Note over UI: UI stays responsive
    W->>API: HTTP request
    API-->>W: response
    W-->>UI: finished.emit(result)
    UI->>UI: update console/panels

Worker Template

from PyQt5.QtCore import QThread, pyqtSignal


class MyWorker(QThread):
    finished = pyqtSignal(object)  # success result
    error = pyqtSignal(str)        # error message

    def __init__(self, param1, param2):
        super().__init__()
        self.param1 = param1
        self.param2 = param2

    def run(self):
        """Runs in background thread — no UI access here."""
        try:
            result = some_api_call(self.param1, self.param2)
            self.finished.emit(result)
        except Exception as e:
            self.error.emit(str(e))

Usage in a Manager

class MyManager:
    def __init__(self, ctx: AppContext):
        self._ctx = ctx
        self._worker = None

    def start_operation(self, param):
        self._worker = MyWorker(param)
        self._worker.finished.connect(self._on_success)
        self._worker.error.connect(self._on_error)
        self._worker.start()
        self._ctx.console.log("Operation started...")

    def _on_success(self, result):
        self._ctx.console.log(f"Done: {result}")

    def _on_error(self, msg):
        self._ctx.console.log(f"Error: {msg}")

Thread Safety

Never access Qt widgets from the worker's run() method. All UI updates must happen through signals that are received on the main thread.

Workers in codemol

Worker	File	Purpose
AgentWorker	agents/worker.py	Runs LLM agent loop
AtlasWorker	codemol/atlas_worker.py	Atlas/Nomosis API operations

Error Handling

Workers should catch all exceptions and emit them via the error signal:

def run(self):
    try:
        result = api_call()
        self.finished.emit(result)
    except ConnectionError:
        self.error.emit("Could not connect to server")
    except TimeoutError:
        self.error.emit("Request timed out")
    except Exception as e:
        self.error.emit(f"Unexpected error: {e}")

The manager then displays the error in the console — the user never sees a crash.