Class InProcessTeammateTask

src/tasks/in_process_teammate.py:323–362

``Task`` adapter for ``in_process_teammate`` entries. Polymorphic dispatch target for ``stop_task`` (Phase 5). Setting ``abort_event`` propagates the kill signal; the teammate's run loop sees ``TeammateAbortedError`` at its next yield and unwinds the lifecycle.

class InProcessTeammateTask:
    """``Task`` adapter for ``in_process_teammate`` entries.

    Polymorphic dispatch target for ``stop_task`` (Phase 5). Setting
    ``abort_event`` propagates the kill signal; the teammate's run
    loop sees ``TeammateAbortedError`` at its next yield and
    unwinds the lifecycle.
    """

    name: str = "InProcessTeammateTask"
    type: Literal["in_process_teammate"] = "in_process_teammate"

    async def kill(
        self, task_id: str, registry: "RuntimeTaskRegistry"
    ) -> None:
        aborted_event: asyncio.Event | None = None

        def _kill(prev: TaskStateBase) -> TaskStateBase:
            nonlocal aborted_event
            if not isinstance(prev, InProcessTeammateTaskState):
                return prev
            if is_terminal_task_status(prev.status):
                return prev
            aborted_event = prev.abort_event
            return replace(prev, status="killed")

        registry.update(task_id, _kill)
        # Set the event OUTSIDE the registry lock — same defense-in-depth
        # pattern as kill_async_agent. asyncio.Event is thread-safe for
        # ``set()`` (it dispatches via the loop's call_soon_threadsafe
        # internally), but we don't want a misbehaving subclass to
        # deadlock the registry.
        if aborted_event is not None:
            try:
                aborted_event.set()
            except Exception:
                import logging
                logging.getLogger(__name__).exception(
                    "failed to set abort event for killed teammate %s", task_id
                )


__all__ = [