Class RuntimeTaskRegistry

src/task_registry.py:72–175

Thread-safe map of ``task_id`` to ``TaskStateBase`` (or subclass). Substitute for the TS ``AppState.tasks: Record `` pattern. ``update(task_id, mutator)`` is the immutable-update analogue of the TS ``setAppState(prev => ...)`` idiom.

class RuntimeTaskRegistry:
    """Thread-safe map of ``task_id`` to ``TaskStateBase`` (or subclass).

    Substitute for the TS ``AppState.tasks: Record<string, TaskState>``
    pattern. ``update(task_id, mutator)`` is the immutable-update analogue
    of the TS ``setAppState(prev => ...)`` idiom.
    """

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._tasks: dict[str, TaskStateBase] = {}

    # -- read paths -------------------------------------------------------

    def get(self, task_id: str) -> TaskStateBase | None:
        """Return the task with the given id, or ``None`` if absent."""
        with self._lock:
            return self._tasks.get(task_id)

    def all(self) -> list[TaskStateBase]:
        """Snapshot of every registered task. Returns a list, not a view —
        callers iterating outside the lock should not see in-flight writes."""
        with self._lock:
            return list(self._tasks.values())

    def by_type(self, task_type: TaskType) -> list[TaskStateBase]:
        """Snapshot filtered to a single TaskType — used by deprecated
        compatibility views (e.g. ``ToolContext.background_bash_tasks``)."""
        with self._lock:
            return [t for t in self._tasks.values() if t.type == task_type]

    def __contains__(self, task_id: str) -> bool:
        with self._lock:
            return task_id in self._tasks

    def __len__(self) -> int:
        with self._lock:
            return len(self._tasks)

    def __iter__(self) -> Iterator[TaskStateBase]:
        # Iterate over the snapshot, not the live dict, to avoid
        # RuntimeError if writers add entries during iteration.
        return iter(self.all())

    # -- write paths ------------------------------------------------------

    def upsert(self, task: TaskStateBase) -> None:
        """Insert or replace the entry for ``task.id``."""
        with self._lock:
            self._tasks[task.id] = task

    def remove(self, task_id: str) -> bool:
        """Drop the entry; returns True iff something was removed."""
        with self._lock:
            return self._tasks.pop(task_id, None) is not None

    def update(
        self,