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

Copy Markdown

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

Functions

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

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

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

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

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

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

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

Finalizes the task runtime as failed when its worker crashes.

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

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.