Anubis.Server.Session.Tasks (anubis_mcp v1.14.0)

Tasks runtime for a session (MCP spec 2025-11-25).

Owns the lifecycle of task-augmented tools/call executions: capability checks, task creation and worker spawn, status transitions, cancellation, TTL expiry, result storage through the configured Anubis.Server.TaskStore adapter, and replies to tasks/result waiters.

State fields (tasks, task_refs, task_store) live in the Session's state map; this module transforms them and performs the side effects (worker spawn, monitors, timers, store calls, waiter replies) at the edges. Anubis.Server.Session delegates to this module and supplies a frame_fn to prepare frames lazily.

Summary

Types

frame_fn()

state()

store()

task_runtime()

task_waiter()

Functions

augmented_tools_call?(arg1)

Returns true if the request is a tools/call carrying task augmentation params.

build_store(opts)

Builds the task store configuration from the :task_store session option.

create_for_tools_call(request, ctx, from, state, frame_fn)

Handles a task-augmented tools/call: validates support, tool existence, and tool task policy, then spawns the worker task and replies with the CreateTaskResult.

dispatch_request(request, ctx, from, state, frame_fn)

Dispatches a tasks/* request (tasks/get, tasks/result, tasks/cancel, tasks/list), replying inline or registering a result waiter for non-terminal tasks.

emit_status_notification(state, task_id)

Sends a notifications/tasks/status notification for the given task to the client transport.

handle_expired(task_id, state)

Handles TTL expiry of a live task: terminates the worker, releases waiters with an expiry error, and deletes the task from the store.

handle_worker_completion(task_id, callback_result, state)

Finalizes the task runtime for a completed worker, deriving the terminal status from the handler result.

handle_worker_down(task_id, reason, state)

Finalizes the task runtime as failed when its worker crashes.

task_id_for_ref(state, ref)

Returns the task ID owning the given worker monitor ref, if any.

terminate_all(map, error)

Terminates all live task workers, demonitors them, and releases their waiters with the given error. Used during session termination.

Types

frame_fn()

@type frame_fn() :: (state() -> Anubis.Server.Frame.t())

state()

@type state() :: map()

store()

@type store() :: %{adapter: module(), name: term()}

task_runtime()

@type task_runtime() :: %{
  worker_ref: reference() | nil,
  worker_pid: pid() | nil,
  ttl_timer: reference() | nil,
  waiters: [task_waiter()],
  request_id: String.t() | integer()
}

task_waiter()

@type task_waiter() ::
  {from :: GenServer.from(), request_id :: String.t() | integer()}

Functions

augmented_tools_call?(arg1)

@spec augmented_tools_call?(map()) :: boolean()

Returns true if the request is a tools/call carrying task augmentation params.

build_store(opts)

@spec build_store(keyword() | nil) :: store() | nil

Builds the task store configuration from the :task_store session option.

create_for_tools_call(request, ctx, from, state, frame_fn)

@spec create_for_tools_call(map(), map(), GenServer.from(), state(), frame_fn()) ::
  {:reply, {:ok, binary()}, state()}

Handles a task-augmented tools/call: validates support, tool existence, and tool task policy, then spawns the worker task and replies with the CreateTaskResult.

dispatch_request(request, ctx, from, state, frame_fn)

@spec dispatch_request(map(), map(), GenServer.from(), state(), frame_fn()) ::
  {:reply, {:ok, binary()}, state()} | {:noreply, state()}

Dispatches a tasks/* request (tasks/get, tasks/result, tasks/cancel, tasks/list), replying inline or registering a result waiter for non-terminal tasks.

emit_status_notification(state, task_id)

@spec emit_status_notification(state(), String.t()) :: :ok | {:error, term()}

Sends a notifications/tasks/status notification for the given task to the client transport.

handle_expired(task_id, state)

@spec handle_expired(String.t(), state()) :: {:noreply, state()}

Handles TTL expiry of a live task: terminates the worker, releases waiters with an expiry error, and deletes the task from the store.

handle_worker_completion(task_id, callback_result, state)

@spec handle_worker_completion(String.t(), term(), state()) :: {:noreply, state()}

Finalizes the task runtime for a completed worker, deriving the terminal status from the handler result.

handle_worker_down(task_id, reason, state)

@spec handle_worker_down(String.t(), term(), state()) :: {:noreply, state()}

Finalizes the task runtime as failed when its worker crashes.

task_id_for_ref(state, ref)

@spec task_id_for_ref(state(), reference()) :: String.t() | nil

Returns the task ID owning the given worker monitor ref, if any.

terminate_all(map, error)

@spec terminate_all(state(), Anubis.MCP.Error.t()) :: :ok

Terminates all live task workers, demonitors them, and releases their waiters with the given error. Used during session termination.