Function kill_workflow_task

src/tasks/local_workflow.py:131–152

Abort the whole run (cascades to every subagent) and mark it killed.

(task_id: str, registry: "RuntimeTaskRegistry")

def kill_workflow_task(task_id: str, registry: "RuntimeTaskRegistry") -> None:
    """Abort the whole run (cascades to every subagent) and mark it killed."""
    captured_run: Any = None
    fired = False

    def _kill(prev: TaskStateBase) -> TaskStateBase:
        nonlocal captured_run, fired
        if not isinstance(prev, LocalWorkflowTaskState) or is_terminal_task_status(prev.status):
            return prev
        captured_run = prev.run
        fired = True
        return _terminal_replace(prev, status="killed")

    registry.update(task_id, _kill)
    # Abort OUTSIDE the registry lock (the controller fires listeners).
    if captured_run is not None:
        try:
            captured_run.controller.abort("workflow_stopped")
        except Exception:
            logger.exception("failed to abort workflow run %s", task_id)
    if fired:
        enqueue_workflow_notification(task_id, registry, status="killed")


def skip_workflow_agent(task_id: str, agent_key: str, registry: "RuntimeTaskRegistry") -> bool: