barrel_mcp_tasks (barrel_mcp v3.0.1)

View Source

Long-running operation registry (MCP tasks).

Tools registered with task_support => optional | required (long_running => true is the old spelling of optional) can answer with a taskId instead of a result: at once for a legacy client, after the inline window for a modern one. The worker continues in the background; clients poll via tasks/get, enumerate via tasks/list (legacy), and abort via tasks/cancel. State transitions emit notifications/tasks/status on the session's SSE channel. The mode rule lives in the transports (barrel_mcp_http_engine:handle_async_tool_call, barrel_mcp_protocol:task_plan), not here: this module is the table and the lifecycle.

Tasks live in a protected ETS table keyed by TaskId, which is crypto-random and so unique on its own. Who owns a task is a field rather than part of the key: a task id is the durable handle a client holds, and it has to be resolvable without knowing what created it.

The owner is opaque here, and differs by era. A legacy task is owned by its session id; a modern one has no session, so it is owned by the authenticated principal. Either way every lookup matches on it, so one owner cannot reach another's tasks.

A periodic sweep evicts terminal tasks (success / error / cancelled) older than ?TASK_TTL.

Sections, in file order

  • Public API: create, get, list, finish, fail, cancel, set_worker/3, await_input/5 for input_required.
  • gen_server: every transition, the sweep, generation fencing so a late worker cannot revive an expired task.
  • Internal: rendering per era (ttl versus ttlMs), the error object, expiry.

Summary

Functions

Park a task on the input its handler asked for.

Block until a task reaches a terminal state, then return what the underlying request would have returned.

Mark a task as cancelled and notify the client. Sends {cancel, RequestId} to the worker pid (if recorded) so cooperative arity-2 handlers can abort.

Create a new running task. Returns the task id.

Record failure: store the error and emit notification.

Record success: store the result and emit notifications/tasks/status.

Read a task, rendered for the handshake era.

Read a task rendered for a given era. The retention field is named ttl through 2025-11-25 and ttlMs in the extension.

Every task an owner holds, rendered for the legacy era (tasks/list exists only there).

Which of these task ids the owner actually holds.

The originating request params recorded for a task, if any.

Attach the worker to a task created before it, so cancel and expiry can reach the process; carries the request id for MRTR resumption.

Start the task table owner, registered as barrel_mcp_tasks.

Record answers a client supplied for a task through tasks/update. Merged rather than replaced, so a client answering one key at a time does not drop the others.

Functions

await_input(Owner, TaskId, Requests, HandlerState, Params)

-spec await_input(term(), binary(), map(), term(), map()) -> ok | {error, term()}.

Park a task on the input its handler asked for.

The asking worker has already returned, so everything needed to run the handler again is stored: the originating params, the handler's own state, and which keys are outstanding.

await_result(Owner, TaskId, Timeout)

-spec await_result(term(), binary(), timeout()) -> {ok, map()} | {error, not_found} | {error, timeout}.

Block until a task reaches a terminal state, then return what the underlying request would have returned.

The wait happens in the calling process, never in the tasks server: that same process has to accept the transition that ends the wait, so blocking it would deadlock. The server only records who to tell.

cancel(SessionId, TaskId)

-spec cancel(binary() | undefined, binary()) -> ok | {error, not_found}.

Mark a task as cancelled and notify the client. Sends {cancel, RequestId} to the worker pid (if recorded) so cooperative arity-2 handlers can abort.

create(SessionId, Method, Opts)

-spec create(SessionId :: binary() | undefined, Method :: binary(), Opts :: map()) ->
                {ok, binary()} | {error, too_many_tasks}.

Create a new running task. Returns the task id.

expire(Task)

fail(Owner, TaskId, Reason)

-spec fail(term(), binary(), term()) -> ok | {error, not_found}.

Record failure: store the error and emit notification.

finish(SessionId, TaskId, Result)

-spec finish(binary() | undefined, binary(), term()) -> ok | {error, not_found}.

Record success: store the result and emit notifications/tasks/status.

get(SessionId, TaskId)

-spec get(SessionId :: binary() | undefined, TaskId :: binary()) -> {ok, map()} | {error, not_found}.

Read a task, rendered for the handshake era.

get(SessionId, TaskId, Era)

-spec get(binary() | undefined, binary(), legacy | modern) -> {ok, map()} | {error, not_found}.

Read a task rendered for a given era. The retention field is named ttl through 2025-11-25 and ttlMs in the extension.

handle_call(_, From, State)

handle_cast(Msg, State)

handle_info(_, State)

init(_)

list(SessionId, Opts)

-spec list(SessionId :: binary() | undefined, map()) -> {ok, [map()]}.

Every task an owner holds, rendered for the legacy era (tasks/list exists only there).

owned(Owner, Ids)

-spec owned(term(), [binary()]) -> [binary()].

Which of these task ids the owner actually holds.

An id naming no task and an id belonging to someone else give the same answer, so a caller cannot use the difference to learn that a task exists.

params(Owner, TaskId)

-spec params(term(), binary()) -> {ok, map()} | {error, not_found}.

The originating request params recorded for a task, if any.

set_worker(SessionId, TaskId, Info)

-spec set_worker(binary() | undefined, binary(), #{worker := pid(), request_id => integer() | binary()}) ->
                    ok | {error, not_found}.

Attach the worker to a task created before it, so cancel and expiry can reach the process; carries the request id for MRTR resumption.

start_link()

Start the task table owner, registered as barrel_mcp_tasks.

terminate(Reason, State)

update(Owner, TaskId, Responses)

-spec update(term(), binary(), map()) -> ok | {resume, map()} | {error, not_found}.

Record answers a client supplied for a task through tasks/update. Merged rather than replaced, so a client answering one key at a time does not drop the others.