From 1dc1966211ccebc753c0d395428288289ff860c3 Mon Sep 17 00:00:00 2001 From: David Bernheisel Date: Sun, 4 Oct 2026 19:03:59 -0400 Subject: [PATCH 1/6] Tasks: serve the MCP Tasks extension A tool answers tools/call with a task by returning {:reply, %Phantom.Tasks{}, session}. Phantom does not store tasks: the router implements get_task/2 (and optionally update_task/3, cancel_task/2, authorize_task_subscriptions/2) for tasks/get, tasks/update, and tasks/cancel under MCP 2026-07-28. Phantom.Tracker.notify_task_updated/1 sends notifications/tasks to subscriptions/listen streams, which fetch the task with get_task/2. The conformance suite's tasks scenarios now run and must pass. A guide covers storing and running tasks with Ecto, Oban, and Oban Pro. --- CHANGELOG.md | 27 ++ README.md | 10 +- guides/tasks.md | 557 ++++++++++++++++++++++++ lib/phantom/elicit.ex | 19 +- lib/phantom/plug.ex | 20 +- lib/phantom/request.ex | 47 +- lib/phantom/router.ex | 252 ++++++++++- lib/phantom/session.ex | 119 ++++- lib/phantom/tasks.ex | 214 +++++++++ lib/phantom/test.ex | 83 +++- lib/phantom/tracker.ex | 90 ++++ mix.exs | 2 +- test/conformance/conformance_test.exs | 12 +- test/conformance/router.ex | 196 +++++++++ test/phantom/distributed_test.exs | 90 ++++ test/phantom/elicit_test.exs | 26 ++ test/phantom/plug_test.exs | 49 +++ test/phantom/progress_ref_test.exs | 69 +++ test/phantom/request_test.exs | 12 + test/phantom/task_subscription_test.exs | 259 +++++++++++ test/phantom/tasks_dispatch_test.exs | 380 ++++++++++++++++ test/phantom/tasks_test.exs | 199 +++++++++ test/support/app/mcp/router.ex | 20 + 23 files changed, 2724 insertions(+), 28 deletions(-) create mode 100644 guides/tasks.md create mode 100644 lib/phantom/tasks.ex create mode 100644 test/phantom/progress_ref_test.exs create mode 100644 test/phantom/task_subscription_test.exs create mode 100644 test/phantom/tasks_dispatch_test.exs create mode 100644 test/phantom/tasks_test.exs diff --git a/CHANGELOG.md b/CHANGELOG.md index 9934f6d..49d4352 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,30 @@ +## Unreleased + +- New: the Tasks extension (`io.modelcontextprotocol/tasks`) under MCP + 2026-07-28. A tool returns `{:reply, %Phantom.Tasks{}, session}` to answer + with a task, and the router implements `c:Phantom.Router.get_task/2`, and + optionally `update_task/3`, `cancel_task/2`, and + `authorize_task_subscriptions/2`, for `tasks/get`, `tasks/update`, and + `tasks/cancel`. Phantom does not store tasks. See the Tasks guide for Ecto + and Oban examples. +- New: `Phantom.Session.tasks_supported?/1`, and + `Phantom.Tracker.notify_task_updated/1`. Each `subscriptions/listen` stream + that follows the task fetches it with `get_task/2` and sends + `notifications/tasks`. +- New: `Phantom.Test.get_task/3`, `update_task/4`, `cancel_task/3`, + `assert_task/2`, and a `:tasks` option for `call_tool/4`. +- New: `Phantom.Session.progress_ref/1` returns a JSON-safe reference to the + current request. `Phantom.Session.notify_progress/4` accepts it, so a job + on any node can report progress to a waiting `tools/call` or `prompts/get` + through PubSub. +- New: `Phantom.Elicit.to_input_request/1` renders one embedded + `elicitation/create` request to place under your own `inputRequests` key, + such as a task's input requests. +- New: `Phantom.Request.missing_task_capability/0`, the error for a tool that + only runs as a task when the client did not declare the extension. +- `Phantom.Request.missing_capability/1` accepts a `ClientCapabilities` map, + for capabilities whose names contain dots. + ## 0.6.0 (2026-10-04) - New: `Phantom.Test` for unit-testing MCP routers without an HTTP transport. diff --git a/README.md b/README.md index beba52a..cfdaf94 100644 --- a/README.md +++ b/README.md @@ -717,6 +717,10 @@ Phantom will implement these MCP requests on your behalf: `input_required` result + encrypted `requestState`; both flows go through `Phantom.Session.elicit/3` and behave identically from the tool's point of view. +- `tasks/get`, `tasks/update`, `tasks/cancel` - the Tasks extension under MCP + `2026-07-28`, when the router implements `c:Phantom.Router.get_task/2`. A tool + answers with a `Phantom.Tasks` and you store and run the task, such as with + Oban. See the [Tasks guide](guides/tasks.md). Phantom **does not yet support these methods**: @@ -872,8 +876,10 @@ the JSON-RPC body into HTTP headers so load balancers, proxies, and WAFs can route on the operation without parsing the body: - `Mcp-Method` — mirrors `method`. Required on every POST. -- `Mcp-Name` — mirrors `params.name` (for `tools/call`, `prompts/get`) - or `params.uri` (for `resources/read`). +- `Mcp-Name` — mirrors `params.name` (for `tools/call`, `prompts/get`), + `params.uri` (for `resources/read`), or `params.taskId` (for `tasks/get`, + `tasks/update`, `tasks/cancel`). Route on it when tasks live in one node's + memory. Header names are case-insensitive; values are case-sensitive (so `Mcp-Method: TOOLS/CALL` does *not* match a body method of `tools/call`). diff --git a/guides/tasks.md b/guides/tasks.md new file mode 100644 index 0000000..8e92952 --- /dev/null +++ b/guides/tasks.md @@ -0,0 +1,557 @@ +# Tasks + +The [Tasks extension](https://modelcontextprotocol.io/specification/2026-07-28/extensions/tasks) +(`io.modelcontextprotocol/tasks`) lets a tool answer `tools/call` with a task +instead of a result. The client polls the task with `tasks/get` until it +finishes, answers its input requests with `tasks/update`, and may cancel it +with `tasks/cancel`. Tasks require MCP `2026-07-28`. + +Phantom handles the protocol. You store the tasks, run the work, and move +each task through its statuses: + +| Phantom | You | +|---|---| +| Builds `CreateTaskResult` and `tasks/get` responses from `Phantom.Tasks` | Generate unguessable task IDs (`Phantom.Tasks.generate_id/0`) | +| Advertises the extension and rejects clients that did not declare it | Store a task *before* returning it, so an immediate `tasks/get` finds it | +| Validates the `Mcp-Name` routing header on `tasks/*` | Check access in `c:Phantom.Router.get_task/2` | +| Passes only outstanding `inputResponses` to `c:Phantom.Router.update_task/3` | Never reuse an input request key during a task's lifetime | +| Sends `notifications/tasks` to `subscriptions/listen` streams | Call `Phantom.Tracker.notify_task_updated/1` with the task or its ID after each change | +| | Delete tasks after their TTL | + +## Returning a task + +A tool returns a `Phantom.Tasks` struct like any other result. Only answer +with a task when the client declared the extension on this request. +Otherwise, do the work in a process, report progress, and respond when it +finishes: + +```elixir +def export_report(%{"report_id" => report_id} = params, session) do + if Phantom.Session.tasks_supported?(session) do + {:reply, start_export(params, session), session} + else + Task.async(fn -> + csv = + Reports.export!(report_id, + on_progress: &Phantom.Session.notify_progress(session, &1, &2, "Exporting") + ) + + Phantom.Session.respond(session, Tool.text(csv)) + end) + + {:noreply, session} + end +end +``` + +If a tool returns a task anyway, Phantom answers with the `-32021` missing +capability error. A tool that only works as a task can return that error itself, +before creating anything: + +```elixir +{:error, Phantom.Request.missing_task_capability(), session} +``` + +## Router callbacks + +Implementing `c:Phantom.Router.get_task/2` enables the extension: + +- `c:Phantom.Router.get_task/2` — fetch a task for `tasks/get`, `tasks/update`, + and `tasks/cancel`. Return `{:error, :not_found}` when the session may not + see it, so the client cannot tell whether it exists. +- `c:Phantom.Router.update_task/3` — receive answers to the task's + `input_requests`. Without it, Phantom acknowledges `tasks/update` and drops + the answers. +- `c:Phantom.Router.cancel_task/2` — cancel the work. Without it, Phantom + acknowledges `tasks/cancel` and the task carries on, which the spec allows. +- `c:Phantom.Router.authorize_task_subscriptions/2` — which tasks a + `subscriptions/listen` stream may follow. Defaults to the tasks + `get_task/2` returns; override it to check many with one query. Before each + notification, the stream fetches the task with `get_task/2`, which checks + access again and sends the stored task. + +## Example: Ecto and Oban + +This example stores tasks in Postgres and runs them with +[Oban](https://hexdocs.pm/oban). Inserting the task and its job in one +transaction means a task never exists without its job, or the reverse. + +### Migration + +```elixir +defmodule MyApp.Repo.Migrations.CreateMCPTasks do + use Ecto.Migration + + def change do + create table(:mcp_tasks, primary_key: false) do + add :id, :string, primary_key: true + add :user_id, references(:users, on_delete: :delete_all), null: false + add :status, :string, null: false + add :status_message, :string + add :input_requests, :map, null: false, default: %{} + add :input_responses, :map, null: false, default: %{} + add :result, :map + add :error, :map + add :job_id, :bigint + add :expires_at, :utc_datetime_usec, null: false + + timestamps(type: :utc_datetime_usec) + end + + create index(:mcp_tasks, [:user_id]) + create index(:mcp_tasks, [:expires_at]) + end +end +``` + +### Schema and context + +```elixir +defmodule MyApp.MCP.TaskRecord do + use Ecto.Schema + + @primary_key {:id, :string, autogenerate: false} + @timestamps_opts [type: :utc_datetime_usec] + + schema "mcp_tasks" do + belongs_to :user, MyApp.Accounts.User + field :status, Ecto.Enum, values: [:working, :input_required, :completed, :failed, :cancelled] + field :status_message, :string + field :input_requests, :map, default: %{} + field :input_responses, :map, default: %{} + field :result, :map + field :error, :map + field :job_id, :integer + field :expires_at, :utc_datetime_usec + + timestamps() + end +end +``` + +```elixir +defmodule MyApp.MCP.TaskRecords do + import Ecto.Query + + alias MyApp.MCP.TaskRecord + alias MyApp.Repo + + @ttl :timer.hours(6) + + def create!(user, attrs \\ %{}) do + now = DateTime.utc_now() + + %TaskRecord{ + id: Phantom.Tasks.generate_id(), + user_id: user.id, + status: :working, + expires_at: DateTime.add(now, @ttl, :millisecond) + } + |> Ecto.Changeset.change(attrs) + |> Repo.insert!() + end + + def get(id, user), do: Repo.get_by(TaskRecord, id: id, user_id: user.id) + def get!(id), do: Repo.get!(TaskRecord, id) + + def to_phantom(%TaskRecord{} = record) do + Phantom.Tasks.new( + id: record.id, + status: record.status, + status_message: record.status_message, + created_at: record.inserted_at, + last_updated_at: record.updated_at, + ttl_ms: DateTime.diff(record.expires_at, record.inserted_at, :millisecond), + poll_interval_ms: 2_000, + input_requests: record.input_requests, + result: record.result, + error: record.error + ) + end + + def working(id, message), do: transition(id, fn _ -> %{status: :working, status_message: message} end) + + def complete(id, result), + do: transition(id, fn _ -> %{status: :completed, result: result, input_requests: %{}} end) + + def fail(id, error), + do: transition(id, fn _ -> %{status: :failed, error: error, input_requests: %{}} end) + + def cancel(id), + do: transition(id, fn _ -> %{status: :cancelled, status_message: "Cancelled"} end) + + def need_input(id, requests), + do: transition(id, fn _ -> %{status: :input_required, input_requests: requests} end) + + # Answered requests are removed; the task works again once none remain. + def answer(id, responses) do + transition(id, fn record -> + remaining = Map.drop(record.input_requests, Map.keys(responses)) + + %{ + status: if(remaining == %{}, do: :working, else: :input_required), + input_requests: remaining, + input_responses: Map.merge(record.input_responses, responses) + } + end) + end + + # Only a running task changes. A finished task stays finished, so a job + # that completes after the client cancelled leaves it cancelled. + defp transition(id, fun) do + result = + Repo.transact(fn -> + query = + from t in TaskRecord, + where: t.id == ^id and t.status in [:working, :input_required], + lock: "FOR UPDATE" + + case Repo.one(query) do + nil -> {:error, :finished} + record -> record |> Ecto.Changeset.change(fun.(record)) |> Repo.update() + end + end) + + with {:ok, record} <- result do + Phantom.Tracker.notify_task_updated(to_phantom(record)) + end + + result + end + + def prune do + Repo.delete_all(from t in TaskRecord, where: t.expires_at < ^DateTime.utc_now()) + end +end +``` + +`Repo.transact/2` requires Ecto 3.13. Use `Repo.transaction/1` on older versions. + +### The tool and its job + +```elixir +defmodule MyApp.MCP.Router do + use Phantom.Router, name: "MyApp", vsn: "1.0" + + alias MyApp.MCP.TaskRecords + + @description "Export a report as CSV. Large reports run as a task." + tool :export_report do + field :report_id, :integer, required: true + end + + def export_report(%{"report_id" => report_id}, session) do + if Phantom.Session.tasks_supported?(session) do + {:ok, record} = + MyApp.Repo.transact(fn -> + record = TaskRecords.create!(session.assigns.current_user) + job = Oban.insert!(MyApp.ExportWorker.new(%{task_id: record.id, report_id: report_id})) + {:ok, MyApp.Repo.update!(Ecto.Changeset.change(record, job_id: job.id))} + end) + + {:reply, TaskRecords.to_phantom(record), session} + else + Task.async(fn -> + csv = + MyApp.Reports.export!(report_id, + on_progress: &Phantom.Session.notify_progress(session, &1, &2, "Exporting") + ) + + Phantom.Session.respond(session, Tool.text(csv)) + end) + + {:noreply, session} + end + end + + def get_task(id, session) do + case TaskRecords.get(id, session.assigns.current_user) do + nil -> {:error, :not_found} + record -> {:ok, TaskRecords.to_phantom(record)} + end + end + + # Once every request is answered, run the job that asked for input again. + def update_task(task, responses, _session) do + case TaskRecords.answer(task.id, responses) do + {:ok, %{status: :working, job_id: job_id}} -> Oban.retry_job(job_id) + _ -> :ok + end + end + + def cancel_task(task, _session) do + with {:ok, record} <- TaskRecords.cancel(task.id), do: Oban.cancel_job(record.job_id) + :ok + end +end +``` + +The worker reports through the context. A tool error is still a `completed` +task with `isError: true`; `failed` is only for JSON-RPC errors. + +```elixir +defmodule MyApp.ExportWorker do + use Oban.Worker, queue: :exports, max_attempts: 3 + + alias MyApp.MCP.TaskRecords + alias Phantom.Tool + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"task_id" => id, "report_id" => report_id}}) do + TaskRecords.working(id, "Rendering report #{report_id}") + + case MyApp.Reports.export(report_id) do + {:ok, csv} -> + TaskRecords.complete(id, Tool.response(Tool.text(csv))) + :ok + + {:error, :too_large} -> + TaskRecords.complete(id, Tool.error("The report is too large")) + :ok + + # Oban retries the job; the task stays working meanwhile. + {:error, reason} -> + {:error, reason} + end + end +end +``` + +A job that crashes on its last attempt would leave its task `working` until +its TTL. Fail the task from Oban's telemetry instead: + +```elixir +# In your application's start/2 +:telemetry.attach("mcp-task-failures", [:oban, :job, :exception], &MyApp.MCP.TaskTelemetry.handle_exception/4, nil) +``` + +```elixir +defmodule MyApp.MCP.TaskTelemetry do + alias MyApp.MCP.TaskRecords + + def handle_exception(_event, _measure, %{job: %{args: %{"task_id" => id}} = job} = meta, _config) + when meta.state == :discard or job.attempt >= job.max_attempts do + TaskRecords.fail(id, Phantom.Request.internal_error("The task stopped unexpectedly")) + end + + def handle_exception(_event, _measure, _meta, _config), do: :ok +end +``` + +Prune expired tasks with Oban's cron plugin: + +```elixir +config :my_app, Oban, + plugins: [{Oban.Plugins.Cron, crontab: [{"@hourly", MyApp.PruneMCPTasksWorker}]}] +``` + +```elixir +defmodule MyApp.PruneMCPTasksWorker do + use Oban.Worker, queue: :maintenance + + @impl Oban.Worker + def perform(_job) do + MyApp.MCP.TaskRecords.prune() + :ok + end +end +``` + +### Asking for input during a task + +A task in `input_required` lists its requests, and the client answers them +with `tasks/update`. Each key must be new for the task's lifetime. The +`update_task/3` callback above stores the answers and retries the job once +none are outstanding, so the job reads them from the record. Here, a job for +an `export_all` tool confirms first: + +```elixir +def perform(%Oban.Job{args: %{"task_id" => id}}) do + record = TaskRecords.get!(id) + + case record.input_responses do + %{"confirm" => %{"action" => "accept"}} -> + TaskRecords.complete(id, Tool.response(Tool.text(MyApp.Reports.export_all!()))) + + %{"confirm" => _declined} -> + TaskRecords.complete(id, Tool.response(Tool.text("Export skipped"))) + + _ -> + elicit = + Phantom.Elicit.form(%{ + message: "Export every report? This may take a while.", + requested_schema: [%{name: "confirm", type: :boolean, required: true}] + }) + + TaskRecords.need_input(id, %{"confirm" => Phantom.Elicit.to_input_request(elicit)}) + end + + :ok +end +``` + +## Example: Oban Pro + +With [Oban Pro](https://oban.pro), a worker hook can fail the task when the +job is discarded, in place of the telemetry handler. `after_process/3` runs +for `:complete`, `:cancel`, `:discard`, `:error`, and `:snooze`; +`c:Phantom.Router.cancel_task/2` already records cancellations: + +```elixir +defmodule MyApp.ExportWorker do + use Oban.Pro.Worker, queue: :exports, max_attempts: 3 + + alias MyApp.MCP.TaskRecords + alias Phantom.Tool + + @impl Oban.Pro.Worker + def process(%Oban.Job{args: %{"report_id" => report_id} = args}) do + mark_working(args, "Rendering report #{report_id}") + progress = &Phantom.Session.notify_progress(args["progress"], &1, &2, "Exporting") + + with {:ok, csv} <- MyApp.Reports.export(report_id, on_progress: progress) do + complete(args, csv) + {:ok, csv} + end + end + + @impl Oban.Pro.Worker + def after_process(:discard, %{args: %{"task_id" => id}}, _result), + do: TaskRecords.fail(id, Phantom.Request.internal_error("The export failed")) + + def after_process(_state, _job, _result), do: :ok + + defp mark_working(%{"task_id" => id}, message), do: TaskRecords.working(id, message) + defp mark_working(_args, _message), do: :ok + + defp complete(%{"task_id" => id}, csv), do: TaskRecords.complete(id, Tool.response(Tool.text(csv))) + defp complete(_args, _csv), do: :ok +end +``` + +The worker returns `{:ok, csv}`, so the same job also serves clients without +tasks. [`Oban.Pro.Relay`](https://oban.pro/docs/pro/Oban.Pro.Relay.html) +inserts it and awaits its result from any node, and the tool responds when it +arrives. `Phantom.Session.progress_ref/1` lets the job report progress to the +waiting request from whichever node runs it; with no ref in the arguments, +`notify_progress/4` does nothing: + +```elixir +alias Oban.Pro.Relay + +def export_report(%{"report_id" => report_id}, session) do + if Phantom.Session.tasks_supported?(session) do + # ...insert the task and the job, as above + else + Task.async(fn -> + response = + %{report_id: report_id, progress: Phantom.Session.progress_ref(session)} + |> MyApp.ExportWorker.new() + |> Relay.async() + |> Relay.await(timeout: :timer.minutes(5)) + |> case do + {:ok, csv} -> Tool.text(csv) + {:error, :timeout} -> Tool.error("The export took too long") + _failed -> Tool.error("The export failed") + end + + Phantom.Session.respond(session, response) + end) + + {:noreply, session} + end +end +``` + +Relay needs a working PubSub notifier, and the Postgres notifier limits +results to 8kb compressed; use the `PG` notifier for larger results. + +A [workflow](https://oban.pro/docs/pro/Oban.Pro.Workflow.html) splits the +task into steps. Each step reports its progress as the status message, and +the last step builds the result from the steps' recorded output: + +```elixir +alias Oban.Pro.Workflow + +def export_report(%{"report_id" => report_id}, session) do + {:ok, record} = + MyApp.Repo.transact(fn -> + record = TaskRecords.create!(session.assigns.current_user) + args = %{task_id: record.id, report_id: report_id} + + Workflow.new() + |> Workflow.add(:fetch, MyApp.FetchWorker.new(args)) + |> Workflow.add(:render, MyApp.RenderWorker.new(args), deps: :fetch) + |> Workflow.add(:finish, MyApp.FinishWorker.new(args), deps: :render) + |> Oban.insert_all() + + {:ok, record} + end) + + {:reply, TaskRecords.to_phantom(record), session} +end +``` + +Each step that the next one reads must be recorded. `MyApp.FetchWorker` +uses `recorded: true` and returns `{:ok, rows}` the same way. + +```elixir +defmodule MyApp.RenderWorker do + use Oban.Pro.Worker, queue: :exports, recorded: true + + @impl Oban.Pro.Worker + def process(%Oban.Job{args: %{"task_id" => id}} = job) do + TaskRecords.working(id, "Rendering") + rows = Workflow.get_recorded(job, :fetch) + {:ok, MyApp.Reports.to_csv(rows)} + end +end + +defmodule MyApp.FinishWorker do + use Oban.Pro.Worker, queue: :exports + + @impl Oban.Pro.Worker + def process(%Oban.Job{args: %{"task_id" => id}} = job) do + csv = Workflow.get_recorded(job, :render) + TaskRecords.complete(id, Phantom.Tool.response(Phantom.Tool.text(csv))) + :ok + end +end +``` + +To cancel a workflow, cancel its jobs in `c:Phantom.Router.cancel_task/2` +with `Workflow.cancel_jobs/3`. + +## Testing + +`Phantom.Test` sends task requests with the extension declared. Build the +session with the stateless protocol. With Oban's `testing: :inline`, jobs run +inside `Oban.insert/2`, so the task is finished by the time the tool replies: + +```elixir +setup do + Phantom.Test.start(router: MyApp.MCP.Router) + user = insert(:user) + {:ok, session: build_session(MyApp.MCP.Router, protocol_version: "2026-07-28", assigns: %{current_user: user})} +end + +test "exports a report as a task", %{session: session} do + %{taskId: id} = + session + |> call_tool(:export_report, %{report_id: 1}, tasks: true) + |> assert_task(status: :working) + + session + |> get_task(id) + |> assert_task(status: :completed) +end +``` + +Use `update_task/4` and `cancel_task/3` for the other methods. + +## Load balancers + +Clients send the task ID in the `Mcp-Name` header on `tasks/get`, +`tasks/update`, and `tasks/cancel`, and Phantom rejects a mismatch. When tasks +live in a shared database, any node can answer them. When they live in one +node's memory, route on `Mcp-Name`. diff --git a/lib/phantom/elicit.ex b/lib/phantom/elicit.ex index 5bd22fa..e943a32 100644 --- a/lib/phantom/elicit.ex +++ b/lib/phantom/elicit.ex @@ -429,13 +429,18 @@ defmodule Phantom.Elicit do Render an `Elicit` struct as the MCP `2026-07-28` `inputRequests` map, which a tool or prompt returns in an `input_required` result. """ - def to_input_requests(%__MODULE__{} = elicit) do - %{ - "elicitation" => %{ - method: "elicitation/create", - params: Map.delete(to_json(elicit), :elicitationId) - } - } + def to_input_requests(%__MODULE__{} = elicit), + do: %{"elicitation" => to_input_request(elicit)} + + @doc """ + Render an `Elicit` struct as one embedded `elicitation/create` request, to + place under a key of your choosing in `inputRequests`, such as a task's + `:input_requests` (see `Phantom.Tasks`). + + %{"confirm" => Phantom.Elicit.to_input_request(elicit)} + """ + def to_input_request(%__MODULE__{} = elicit) do + %{method: "elicitation/create", params: Map.delete(to_json(elicit), :elicitationId)} end def to_json(%__MODULE__{mode: :url} = elicit) do diff --git a/lib/phantom/plug.ex b/lib/phantom/plug.ex index 2b859c9..7ed3ca8 100644 --- a/lib/phantom/plug.ex +++ b/lib/phantom/plug.ex @@ -506,7 +506,16 @@ defmodule Phantom.Plug do header_method = mcp_header(conn, "mcp-method") body_name = name_from_params(body_method, Map.get(params, "params")) header_name = mcp_header(conn, "mcp-name") |> decode_header_value() - needs_name? = body_method in ["tools/call", "prompts/get", "resources/read"] + + needs_name? = + body_method in [ + "tools/call", + "prompts/get", + "resources/read", + "tasks/get", + "tasks/update", + "tasks/cancel" + ] cond do is_nil(header_method) -> @@ -533,6 +542,7 @@ defmodule Phantom.Plug do end defp name_from_params("resources/read", params) when is_map(params), do: params["uri"] + defp name_from_params("tasks/" <> _, params) when is_map(params), do: params["taskId"] defp name_from_params(_method, params) when is_map(params), do: params["name"] defp name_from_params(_method, _params), do: nil @@ -769,6 +779,7 @@ defmodule Phantom.Plug do {state, exceptions} :ok -> + subscribe_request(state.session, request) run_dispatch(state, request, exceptions) end end @@ -1180,6 +1191,13 @@ defmodule Phantom.Plug do defp claim_in_flight(_session_id, _request), do: :ok + # `Phantom.Session.notify_progress/4` reaches the request through this topic. + defp subscribe_request(session, %Request{id: id, method: method}) + when method in @dedupable_methods and not is_nil(id), + do: Phantom.Tracker.subscribe_request(session.pubsub, session.id, id) + + defp subscribe_request(_session, _request), do: :ok + defp release_in_flight(session_id, %Request{id: id, method: method}) when method in @dedupable_methods and not is_nil(id), do: Phantom.Tracker.untrack_in_flight(session_id, id) diff --git a/lib/phantom/request.ex b/lib/phantom/request.ex index 844d422..012e495 100644 --- a/lib/phantom/request.ex +++ b/lib/phantom/request.ex @@ -44,6 +44,9 @@ defmodule Phantom.Request do completion/complete subscriptions/listen notifications/cancelled + tasks/get + tasks/update + tasks/cancel ] @modern_cacheable_methods ~w[ @@ -101,17 +104,10 @@ defmodule Phantom.Request do A request omitted a client capability required to process it. Capabilities are named by path, such as `"sampling"` or `"elicitation.url"`, - and reported to the client as a `ClientCapabilities` object. + and reported to the client as a `ClientCapabilities` object. Pass that + object directly when a name contains dots, such as an extension. """ - def missing_capability(capabilities) do - required = - capabilities - |> List.wrap() - |> Enum.reduce(%{}, fn capability, acc -> - path = String.split(capability, ".") - update_in(acc, Enum.map(path, &Access.key(&1, %{})), & &1) - end) - + def missing_capability(required) when is_map(required) do %{ code: @missing_capability, message: "Missing required client capability", @@ -119,6 +115,32 @@ defmodule Phantom.Request do } end + def missing_capability(capabilities) do + capabilities + |> List.wrap() + |> Enum.reduce(%{}, fn capability, acc -> + path = String.split(capability, ".") + update_in(acc, Enum.map(path, &Access.key(&1, %{})), & &1) + end) + |> missing_capability() + end + + @doc false + def task_not_found, + do: %{code: @invalid_params, message: "Failed to retrieve task: Task not found"} + + @doc false + def task_expired, + do: %{code: @invalid_params, message: "Failed to retrieve task: Task has expired"} + + @doc """ + The request did not declare the Tasks extension + (`io.modelcontextprotocol/tasks`), such as when a tool only runs as a task. + See `Phantom.Tasks`. + """ + def missing_task_capability, + do: missing_capability(%{extensions: %{Phantom.Tasks.extension() => %{}}}) + @doc "The requested MCP protocol version is unsupported." def unsupported_protocol(requested) do %{ @@ -567,6 +589,11 @@ defmodule Phantom.Request do %{jsonrpc: "2.0", method: "notifications/resources/updated", params: content} end + @doc "Task status notification" + def task_updated(content) do + %{jsonrpc: "2.0", method: "notifications/tasks", params: content} + end + @doc "Tools List updated notification" def tools_updated do %{jsonrpc: "2.0", method: "notifications/tools/list_changed"} diff --git a/lib/phantom/router.ex b/lib/phantom/router.ex index b2e711b..92e419e 100644 --- a/lib/phantom/router.ex +++ b/lib/phantom/router.ex @@ -41,6 +41,7 @@ defmodule Phantom.Router do alias Phantom.Resource alias Phantom.ResourceTemplate alias Phantom.Session + alias Phantom.Tasks alias Phantom.Tool alias Phantom.Tool.JSONSchema @@ -176,6 +177,60 @@ defmodule Phantom.Router do @callback authorize_resource_subscriptions([resolved_resource()], Session.t()) :: [resolved_resource() | String.t()] | nil + @doc """ + Fetch a task the client asked about with `tasks/get`, `tasks/update`, or + `tasks/cancel`. Implementing it enables the Tasks extension + (`io.modelcontextprotocol/tasks`); see `Phantom.Tasks`. + + This is where you check that the session may access the task. Return + `{:error, :not_found}` when it may not, so the client cannot tell the task + exists. Return `{:error, :expired}` for a task past its TTL, or a JSON-RPC + error map for anything else. + """ + @callback get_task(task_id :: String.t(), Session.t()) :: + {:ok, Phantom.Tasks.t()} | {:error, :not_found | :expired | map()} + + @doc """ + Receive the client's responses to a task's `input_requests` from + `tasks/update`, such as to resume the work. + + The task is the one `c:get_task/2` returned. Only responses for keys still + outstanding on it are passed; Phantom acknowledges the request without + calling this when none are, or when this callback is not implemented. + """ + @callback update_task(Phantom.Tasks.t(), input_responses :: map(), Session.t()) :: + :ok | {:error, map()} + + @doc """ + Cancel a task, such as by cancelling its job, when the client sends + `tasks/cancel`. + + The task is the one `c:get_task/2` returned. Cancellation is cooperative: + the task may still finish with another status. Without this callback, + Phantom acknowledges the request and the task carries on. + """ + @callback cancel_task(Phantom.Tasks.t(), Session.t()) :: :ok | {:error, map()} + + @doc """ + Authorize task status notifications for task IDs a client listens to with + `subscriptions/listen`. + + Return the allowed task IDs. Returning `nil` or an empty list, raising, or + returning an invalid value rejects them all, and IDs the client did not ask + for are ignored. + + Phantom invokes this callback when a client listens. Override it to check + many tasks with one query. Before each `notifications/tasks`, Phantom fetches + the task with `c:get_task/2`, so revoked access takes effect on the next + update and the client gets the stored task. + + The default implementation allows the tasks `c:get_task/2` returns. + """ + @callback authorize_task_subscriptions([task_id :: String.t()], Session.t()) :: + [String.t()] | nil + + @optional_callbacks get_task: 2, update_task: 3, cancel_task: 2 + @dialyzer {:nowarn_function, default_vsn: 1} defp default_vsn(nil) do Mix.Project.config()[:version] @@ -240,6 +295,10 @@ defmodule Phantom.Router do def connect(session, _auth_info), do: {:ok, session} def disconnect(session), do: {:ok, session} def authorize_resource_subscriptions(resources, _session), do: resources + + def authorize_task_subscriptions(task_ids, session), + do: Phantom.Router.authorize_gettable_tasks(__MODULE__, task_ids, session) + def terminate(session), do: {:error, nil} def instructions(_session), do: {:ok, @instructions} @@ -474,6 +533,7 @@ defmodule Phantom.Router do |> Phantom.Router.completion_capability(__MODULE__, session) |> Phantom.Router.logging_capability(__MODULE__, session) |> Phantom.Router.ui_capability(__MODULE__, session) + |> Phantom.Router.tasks_capability(__MODULE__) {:reply, %{ @@ -504,8 +564,11 @@ defmodule Phantom.Router do request, session ) do - case Session.listen(session, request.id, notifications) do - {:ok, session} -> {:noreply, session} + with :ok <- Phantom.Router.validate_task_subscriptions(notifications, session), + {:ok, session} <- Session.listen(session, request.id, notifications) do + {:noreply, session} + else + {:error, error} -> {:error, error, session} :error -> {:error, Request.invalid_params(), session} end end @@ -525,6 +588,10 @@ defmodule Phantom.Router do Phantom.Router.get_tool(__MODULE__, session, params, request) end + def dispatch_method("tasks/" <> _ = method, params, _request, session) do + Phantom.Router.task_request(__MODULE__, method, params, session) + end + def dispatch_method( "completion/complete", %{ @@ -634,6 +701,7 @@ defmodule Phantom.Router do @doc false defoverridable list_resources: 2, authorize_resource_subscriptions: 2, + authorize_task_subscriptions: 2, server_info: 1, disconnect: 1, connect: 2, @@ -1289,6 +1357,76 @@ defmodule Phantom.Router do [] end + @doc false + def authorize_task_subscriptions(_router, [], _session), do: [] + + def authorize_task_subscriptions(router, task_ids, session) do + case router.authorize_task_subscriptions(task_ids, session) do + allowed when is_list(allowed) -> + allowed = MapSet.new(allowed) + Enum.filter(task_ids, &MapSet.member?(allowed, &1)) + + _invalid -> + [] + end + rescue + exception -> + Logger.error( + "Task subscription authorization failed closed in #{inspect(router)}: " <> + Exception.message(exception) + ) + + [] + catch + kind, reason -> + Logger.error( + "Task subscription authorization failed closed in #{inspect(router)}: " <> + Exception.format_banner(kind, reason) + ) + + [] + end + + # Fails closed, as a stream must not crash on the user's callback. + @doc false + def get_task_for_notification(router, task_id, session) do + case router.get_task(task_id, session) do + {:ok, %Tasks{} = task} -> {:ok, task} + _ -> :error + end + rescue + exception -> + Logger.error( + "Task notification failed closed in #{inspect(router)}: " <> Exception.message(exception) + ) + + :error + catch + kind, reason -> + Logger.error( + "Task notification failed closed in #{inspect(router)}: " <> + Exception.format_banner(kind, reason) + ) + + :error + end + + @doc false + def authorize_gettable_tasks(router, task_ids, session) do + if tasks_enabled?(router), + do: Enum.filter(task_ids, &match?({:ok, _}, router.get_task(&1, session))), + else: [] + end + + @doc false + def validate_task_subscriptions(%{"taskIds" => [_ | _]}, session) do + if Session.tasks_supported?(session), + do: :ok, + else: {:error, Request.missing_task_capability()} + end + + def validate_task_subscriptions(_notifications, _session), do: :ok + defp resolve_resource(router, session, uri) when is_binary(uri) do with {:ok, %{scheme: scheme} = uri_struct} when is_binary(scheme) <- URI.new(uri), resource_router when not is_nil(resource_router) <- @@ -1402,6 +1540,16 @@ defmodule Phantom.Router do end end + @doc false + def tasks_capability(capabilities, router) do + if tasks_enabled?(router) do + extensions = Map.get(capabilities, :extensions, %{}) + Map.put(capabilities, :extensions, Map.put(extensions, Tasks.extension(), %{})) + else + capabilities + end + end + @doc false def completion_capability(capabilities, router, session) do resource_templates = Cache.list(session, router, :resource_templates) @@ -1807,6 +1955,14 @@ defmodule Phantom.Router do end end + defp finalize_result(:tool, {:reply, %Tasks{} = task, %Session{}}, _spec, _params, session) do + if Session.tasks_supported?(session) do + task |> Tasks.to_create_result() |> respond_to_caller() + else + respond_error_to_caller(Request.missing_task_capability()) + end + end + defp finalize_result(kind, {:reply, result, %Session{}}, spec, _params, session) do formatted = format_response(kind, result, session) @@ -1837,7 +1993,7 @@ defmodule Phantom.Router do elicitations |> Enum.with_index() |> Map.new(fn {elicit, index} -> - request = elicit |> Elicit.to_input_requests() |> Map.fetch!("elicitation") + request = Elicit.to_input_request(elicit) {"elicitation-#{index}", request} end) @@ -2060,6 +2216,96 @@ defmodule Phantom.Router do defp elicit_response_args(response) when is_map(response), do: response defp elicit_response_args(_response), do: %{} + @doc false + def task_request(router, method, params, session) do + with :ok <- validate_task_method(router, method, session), + :ok <- validate_tasks_supported(session), + {:ok, task_id} <- fetch_task_id(params), + {:ok, task} <- fetch_task(router, task_id, session) do + run_task_method(method, router, task, params, session) + else + {:error, error} -> {:error, error, session} + end + end + + @doc false + def tasks_enabled?(router), do: function_exported?(router, :get_task, 2) + + @task_methods ["tasks/get", "tasks/update", "tasks/cancel"] + + # The extension exists only under MCP 2026-07-28. + defp validate_task_method(router, method, session) do + if method in @task_methods and tasks_enabled?(router) and Session.stateless?(session), + do: :ok, + else: {:error, Request.not_found()} + end + + defp validate_tasks_supported(session) do + if Session.tasks_supported?(session), + do: :ok, + else: {:error, Request.missing_task_capability()} + end + + defp fetch_task_id(%{"taskId" => task_id}) when is_binary(task_id), do: {:ok, task_id} + + defp fetch_task_id(_params), + do: {:error, Request.invalid_params(%{taskId: "must be a string"})} + + defp fetch_task(router, task_id, session) do + case router.get_task(task_id, session) do + {:ok, %Tasks{} = task} -> {:ok, task} + {:error, :not_found} -> {:error, Request.task_not_found()} + {:error, :expired} -> {:error, Request.task_expired()} + {:error, error} when is_map(error) -> {:error, error} + other -> raise_callback_result!(router, "get_task/2", other) + end + end + + defp run_task_method("tasks/get", _router, task, _params, session), + do: {:reply, Tasks.to_json(task), session} + + # The spec recommends ignoring responses to keys the task is not waiting on. + defp run_task_method("tasks/update", router, task, %{"inputResponses" => responses}, session) + when is_map(responses) do + outstanding = + if task.status == :input_required, + do: Map.take(responses, Map.keys(task.input_requests)), + else: %{} + + if map_size(outstanding) > 0 and function_exported?(router, :update_task, 3), + do: + task_callback_result( + router, + "update_task/3", + router.update_task(task, outstanding, session), + session + ), + else: {:reply, %{}, session} + end + + defp run_task_method("tasks/update", _router, _task, _params, session), + do: {:error, Request.invalid_params(%{inputResponses: "must be an object"}), session} + + defp run_task_method("tasks/cancel", router, task, _params, session) do + if function_exported?(router, :cancel_task, 2), + do: + task_callback_result(router, "cancel_task/2", router.cancel_task(task, session), session), + else: {:reply, %{}, session} + end + + defp task_callback_result(_router, _callback, :ok, session), do: {:reply, %{}, session} + + defp task_callback_result(_router, _callback, {:error, error}, session) when is_map(error), + do: {:error, error, session} + + defp task_callback_result(router, callback, other, _session), + do: raise_callback_result!(router, callback, other) + + defp raise_callback_result!(router, callback, value) do + raise ArgumentError, + "#{inspect(router)}.#{callback} returned an unexpected value: #{inspect(value)}" + end + @doc false def get_prompt(router, session, name) do Enum.find(Cache.list(session, router, :prompts), &(&1.name == name)) diff --git a/lib/phantom/session.ex b/lib/phantom/session.ex index 27627d4..6f73e2b 100644 --- a/lib/phantom/session.ex +++ b/lib/phantom/session.ex @@ -275,6 +275,21 @@ defmodule Phantom.Session do def stateless?(_), do: false + @doc """ + Whether the client declared the Tasks extension on the current request, so + a tool may answer it with a `Phantom.Tasks`. The extension requires MCP + `2026-07-28`, and only the current request counts. + """ + def tasks_supported?(%__MODULE__{request: request} = session) do + stateless?(session) and + is_map( + get_in(Request.client_capabilities(request) || %{}, [ + "extensions", + Phantom.Tasks.extension() + ]) + ) + end + defp do_elicit(session, elicitation, opts) do timeout = Keyword.get(opts, :timeout, @elicitation_timeout) @@ -388,6 +403,16 @@ defmodule Phantom.Session do |> then(&Phantom.Router.authorize_resource_subscriptions(session.router, &1, session)) |> Enum.each(&Phantom.Tracker.subscribe_resource/1) + task_ids = + Phantom.Router.authorize_task_subscriptions( + session.router, + Map.get(filter, "taskIds", []), + session + ) + + Enum.each(task_ids, &Phantom.Tracker.subscribe_task/1) + filter = filter |> Map.delete("taskIds") |> maybe_put_subscription("taskIds", task_ids) + session = %{ session | close_after_complete: false, @@ -550,17 +575,74 @@ defmodule Phantom.Session do def ping(%__MODULE__{pid: pid}), do: ping(pid) def ping(pid) when is_pid(pid), do: GenServer.cast(pid, :ping) + @doc """ + A reference to the current request that another process, on any node, can + pass to `notify_progress/4`. It is a JSON-safe map, so it can travel in job + arguments, such as an Oban job's. + + Returns `nil` when the client did not ask for progress or the session has + no PubSub, and `notify_progress/4` does nothing with `nil`. + + args = %{report_id: id, progress: Phantom.Session.progress_ref(session)} + + # In the job, on any node: + Phantom.Session.notify_progress(args["progress"], 50, 100, "Halfway") + """ + @spec progress_ref(t()) :: map() | nil + def progress_ref(%__MODULE__{pubsub: pubsub, request: %Request{id: request_id}} = session) + when is_atom(pubsub) and not is_nil(pubsub) do + case progress_token(session) do + nil -> + nil + + token -> + %{ + "pubsub" => Atom.to_string(pubsub), + "session_id" => session.id, + "request_id" => request_id, + "progress_token" => token + } + end + end + + def progress_ref(%__MODULE__{}), do: nil + @doc """ Send a progress notification to the client the `progress` and `total` can be a integer or float, but must be ever-increasing. the `total` is optional. + Pass the session, or a reference from `progress_ref/1` to notify from + another process or node. + https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress """ - @spec notify_progress(t, number(), nil | number(), String.t() | nil) :: :ok + @spec notify_progress(t | map() | nil, number(), nil | number(), String.t() | nil) :: :ok def notify_progress(session, progress, total \\ nil, message \\ nil) + def notify_progress(nil, _progress, _total, _message), do: :ok + + def notify_progress( + %{ + "pubsub" => pubsub, + "session_id" => session_id, + "request_id" => request_id, + "progress_token" => token + }, + progress, + total, + message + ) do + pubsub + |> String.to_existing_atom() + |> Phantom.Tracker.cast_request( + session_id, + request_id, + {:progress, token, progress, total, message} + ) + end + def notify_progress(%__MODULE__{} = session, progress, total, message) do {pid, _request_id, token} = route(session) @@ -869,6 +951,33 @@ defmodule Phantom.Session do end end + def handle_cast({:task_updated, task_id}, state) do + notifications = + case Phantom.Router.get_task_for_notification(state.session.router, task_id, state.session) do + {:ok, task} -> + if task.status in [:completed, :failed, :cancelled], + do: Phantom.Tracker.unsubscribe_task(task.id) + + subscription_notifications( + state.session, + "taskIds", + task.id, + Request.task_updated(Phantom.Tasks.to_json(task)) + ) + + :error -> + [] + end + + if notifications != [] do + cancel_inactivity(state) + state = stream_notifications(state, notifications) + {:noreply, state |> set_activity() |> schedule_inactivity()} + else + {:noreply, state} + end + end + def handle_cast({:resource_updated, uri}, state), do: handle_cast({:resources_updated, [uri]}, state) @@ -945,14 +1054,16 @@ defmodule Phantom.Session do defp normalize_subscription_filter(filter) do resource_subscriptions = Map.get(filter, "resourceSubscriptions", []) + task_ids = Map.get(filter, "taskIds", []) - if is_list(resource_subscriptions) and Enum.all?(resource_subscriptions, &is_binary/1) do + if string_list?(resource_subscriptions) and string_list?(task_ids) do normalized = %{} |> maybe_put_subscription("toolsListChanged", filter["toolsListChanged"] == true) |> maybe_put_subscription("promptsListChanged", filter["promptsListChanged"] == true) |> maybe_put_subscription("resourcesListChanged", filter["resourcesListChanged"] == true) |> maybe_put_subscription("resourceSubscriptions", Enum.uniq(resource_subscriptions)) + |> maybe_put_subscription("taskIds", Enum.uniq(task_ids)) {:ok, normalized} else @@ -960,6 +1071,8 @@ defmodule Phantom.Session do end end + defp string_list?(list), do: is_list(list) and Enum.all?(list, &is_binary/1) + defp maybe_put_subscription(filter, _key, false), do: filter defp maybe_put_subscription(filter, _key, []), do: filter defp maybe_put_subscription(filter, key, value), do: Map.put(filter, key, value) @@ -1203,6 +1316,8 @@ defmodule Phantom.Session do error = Request.error(request_id, Request.duplicate_request()) state.stream_fun.(state, request_id, "message", error) else + Phantom.Tracker.subscribe_request(state.session.pubsub, state.session.id, request_id) + state |> claim_in_flight(request_id) |> do_dispatch_stdio_request(request) diff --git a/lib/phantom/tasks.ex b/lib/phantom/tasks.ex new file mode 100644 index 0000000..11b4516 --- /dev/null +++ b/lib/phantom/tasks.ex @@ -0,0 +1,214 @@ +defmodule Phantom.Tasks do + @moduledoc """ + A task from the MCP Tasks extension (`io.modelcontextprotocol/tasks`), + available under MCP `2026-07-28`. + + A tool handler may answer a call with a task instead of a result, and the + client polls for the outcome. Return one from the handler and Phantom + responds with a `CreateTaskResult`: + + def export_report(params, session) do + task = MyApp.MCPTasks.create!(session, params) + {:reply, Phantom.Tasks.new(id: task.id, status: :working, created_at: task.inserted_at), session} + end + + Phantom does not store tasks. You store them, run the work, and move them + through their statuses; Phantom asks your router for them when the client + polls. Check `Phantom.Session.tasks_supported?/1` before returning a task, + since a client that did not declare the extension cannot receive one. + + Return the task from the handler. `Phantom.Session.respond/2` does not + accept one, since the task is already the asynchronous answer. Only tools + answer with tasks. + + ## Fields + + - `:id` - (required) The task ID. It acts as a bearer token, so it must be + unguessable; see `generate_id/0`. + - `:status` - (required) One of `:working`, `:input_required`, + `:completed`, `:failed`, or `:cancelled`. Strings are accepted. + - `:created_at` - (required) A `DateTime`. + - `:last_updated_at` - A `DateTime`. Defaults to `:created_at`. + - `:status_message` - An optional message describing the current status. + - `:ttl_ms` - Milliseconds from creation that the task is kept, or `nil` + for unlimited. + - `:poll_interval_ms` - Suggested milliseconds between client polls. + - `:input_requests` - Required when `:input_required`. A map of keys to + `Phantom.Elicit` structs or embedded request maps. Keys must not be + reused over the task's lifetime. + - `:result` - Required when `:completed`. The final `CallToolResult`, such + as `Phantom.Tool.response(Phantom.Tool.text("Done"))`. A tool error is + still `:completed`. + - `:error` - Required when `:failed`. A JSON-RPC error with `:code` and + `:message`, such as `Phantom.Request.internal_error("Timed out")`. + """ + + @statuses ~w[working input_required completed failed cancelled]a + @status_strings Map.new(@statuses, &{Atom.to_string(&1), &1}) + + @enforce_keys [:id, :status, :created_at, :last_updated_at] + defstruct [ + :id, + :status, + :status_message, + :created_at, + :last_updated_at, + :ttl_ms, + :poll_interval_ms, + :result, + :error, + input_requests: %{} + ] + + @type status :: :working | :input_required | :completed | :failed | :cancelled + + @type t :: %__MODULE__{ + id: String.t(), + status: status(), + status_message: String.t() | nil, + created_at: DateTime.t(), + last_updated_at: DateTime.t(), + ttl_ms: non_neg_integer() | nil, + poll_interval_ms: pos_integer() | nil, + input_requests: %{String.t() => map()}, + result: map() | nil, + error: map() | nil + } + + @doc """ + Build a task. Raises `ArgumentError` when the fields do not describe a + valid task for its status. + """ + @spec new(map() | Keyword.t()) :: t() + def new(attrs) do + attrs = Map.new(attrs) + + %__MODULE__{ + id: fetch_id!(attrs), + status: fetch_status!(attrs), + created_at: fetch_datetime!(attrs, :created_at), + last_updated_at: fetch_datetime!(attrs, :last_updated_at, attrs[:created_at]), + status_message: attrs[:status_message], + ttl_ms: fetch_duration!(attrs, :ttl_ms, 0), + poll_interval_ms: fetch_duration!(attrs, :poll_interval_ms, 1), + input_requests: Map.new(attrs[:input_requests] || %{}, &to_input_request/1), + result: attrs[:result], + error: attrs[:error] + } + |> validate_status_fields!() + end + + @doc false + def extension, do: "io.modelcontextprotocol/tasks" + + @doc "Generate an unguessable, URL-safe task ID with 256 bits of entropy." + @spec generate_id() :: String.t() + def generate_id do + 32 |> :crypto.strong_rand_bytes() |> Base.url_encode64(padding: false) + end + + # Tasks changed with struct update syntax are validated as they leave Phantom. + @doc false + def to_json(%__MODULE__{} = task) do + task = revalidate(task) + task |> base_json() |> put_status_fields(task) + end + + # The client fetches the result, error, and input requests with `tasks/get`. + @doc false + def to_create_result(%__MODULE__{} = task), + do: task |> revalidate() |> base_json() |> Map.put(:resultType, "task") + + defp revalidate(task), do: task |> Map.from_struct() |> new() + + defp base_json(task) do + %{ + taskId: task.id, + status: Atom.to_string(task.status), + createdAt: DateTime.to_iso8601(task.created_at), + lastUpdatedAt: DateTime.to_iso8601(task.last_updated_at), + ttlMs: task.ttl_ms + } + |> maybe_put(:statusMessage, task.status_message) + |> maybe_put(:pollIntervalMs, task.poll_interval_ms) + end + + defp put_status_fields(json, %{status: :input_required} = task), + do: Map.put(json, :inputRequests, task.input_requests) + + defp put_status_fields(json, %{status: :completed} = task), + do: Map.put(json, :result, task.result) + + defp put_status_fields(json, %{status: :failed} = task), + do: Map.put(json, :error, task.error) + + defp put_status_fields(json, _task), do: json + + defp fetch_id!(%{id: id}) when is_binary(id) and byte_size(id) > 0, do: id + + defp fetch_id!(attrs), + do: raise(ArgumentError, ":id must be a non-empty string, got: #{inspect(attrs[:id])}") + + defp fetch_status!(%{status: status}) when status in @statuses, do: status + + defp fetch_status!(%{status: status}) when is_map_key(@status_strings, status), + do: Map.fetch!(@status_strings, status) + + defp fetch_status!(attrs) do + raise ArgumentError, + "status must be one of #{inspect(@statuses)}, got: #{inspect(attrs[:status])}" + end + + defp fetch_datetime!(attrs, key, default \\ nil) do + case attrs[key] || default do + %DateTime{} = datetime -> datetime + other -> raise ArgumentError, "#{inspect(key)} must be a DateTime, got: #{inspect(other)}" + end + end + + defp fetch_duration!(attrs, key, min) do + case attrs[key] do + nil -> + nil + + ms when is_integer(ms) and ms >= min -> + ms + + other -> + raise ArgumentError, + "#{inspect(key)} must be nil or an integer >= #{min}, got: #{inspect(other)}" + end + end + + # Keys are strings, as the client's `inputResponses` keys are. + defp to_input_request({key, %Phantom.Elicit{} = elicit}), + do: {to_string(key), Phantom.Elicit.to_input_request(elicit)} + + defp to_input_request({key, request}), do: {to_string(key), request} + + defp validate_status_fields!(%{status: :completed, result: result}) + when not is_map(result), + do: raise(ArgumentError, "a completed task requires a :result map") + + defp validate_status_fields!(%{status: :failed, error: %{code: code, message: message}} = task) + when is_integer(code) and is_binary(message), + do: task + + defp validate_status_fields!( + %{status: :failed, error: %{"code" => code, "message" => message}} = task + ) + when is_integer(code) and is_binary(message), + do: task + + defp validate_status_fields!(%{status: :failed}), + do: raise(ArgumentError, "a failed task requires an :error with a :code and :message") + + defp validate_status_fields!(%{status: :input_required, input_requests: requests}) + when map_size(requests) == 0, + do: raise(ArgumentError, "an input_required task requires :input_requests") + + defp validate_status_fields!(task), do: task + + defp maybe_put(map, _key, nil), do: map + defp maybe_put(map, key, value), do: Map.put(map, key, value) +end diff --git a/lib/phantom/test.ex b/lib/phantom/test.ex index d62e4c3..a7640a4 100644 --- a/lib/phantom/test.ex +++ b/lib/phantom/test.ex @@ -203,6 +203,9 @@ defmodule Phantom.Test do Options: * `:timeout` - ms to wait for an async response (default: 1000) * `:progress_token` - sets `params._meta.progressToken` on the request + * `:tasks` - declares the Tasks extension on the request, so the tool may + answer with a `Phantom.Tasks`. Requires a session built with + `:protocol_version`. """ @spec call_tool(Session.t(), atom() | String.t(), map() | keyword(), keyword()) :: any() def call_tool(session, name, args \\ %{}, opts \\ []) do @@ -288,6 +291,33 @@ defmodule Phantom.Test do dispatch_blocking(session, "resources/list", %{"cursor" => cursor}, opts) end + @doc """ + Dispatch `tasks/get` for a task. Declares the Tasks extension on the + request, so the session must be built with `:protocol_version`. + + Returns the task as the client sees it, or `{:jsonrpc_error, error_map}`. + """ + @spec get_task(Session.t(), String.t(), keyword()) :: any() + def get_task(session, task_id, opts \\ []) do + dispatch_blocking(session, "tasks/get", %{"taskId" => task_id}, [tasks: true] ++ opts) + end + + @doc """ + Dispatch `tasks/update` with responses to a task's input requests. See + `get_task/3`. + """ + @spec update_task(Session.t(), String.t(), map() | keyword(), keyword()) :: any() + def update_task(session, task_id, input_responses, opts \\ []) do + params = %{"taskId" => task_id, "inputResponses" => stringify_keys(input_responses)} + dispatch_blocking(session, "tasks/update", params, [tasks: true] ++ opts) + end + + @doc "Dispatch `tasks/cancel` for a task. See `get_task/3`." + @spec cancel_task(Session.t(), String.t(), keyword()) :: any() + def cancel_task(session, task_id, opts \\ []) do + dispatch_blocking(session, "tasks/cancel", %{"taskId" => task_id}, [tasks: true] ++ opts) + end + @doc false def invoke_elicit_responder(%Phantom.Elicit{mode: mode} = elicitation) do case Process.get({__MODULE__, :elicit_responder, mode}) do @@ -347,7 +377,13 @@ defmodule Phantom.Test do defp dispatch_once(session, method, params, opts) do timeout = Keyword.get(opts, :timeout, @default_timeout) - meta = Map.merge(session_meta(session), progress_meta(opts)) + + meta = + session + |> session_meta() + |> Map.merge(progress_meta(opts)) + |> put_tasks_capability(opts[:tasks]) + request = %{build_request(method, params: Map.put(params, "_meta", meta)) | meta: meta} session = %{ @@ -371,6 +407,25 @@ defmodule Phantom.Test do defp session_meta(%Session{request: %Request{meta: meta}}) when is_map(meta), do: meta defp session_meta(_session), do: %{} + @protocol_version_key "io.modelcontextprotocol/protocolVersion" + @client_capabilities_key "io.modelcontextprotocol/clientCapabilities" + + defp put_tasks_capability(meta, true) do + if not is_map_key(meta, @protocol_version_key) do + raise ArgumentError, + "the Tasks extension requires a session built with :protocol_version, such as \"2026-07-28\"" + end + + capabilities = Map.get(meta, @client_capabilities_key, %{}) + + extensions = + capabilities |> Map.get("extensions", %{}) |> Map.put(Phantom.Tasks.extension(), %{}) + + Map.put(meta, @client_capabilities_key, Map.put(capabilities, "extensions", extensions)) + end + + defp put_tasks_capability(meta, _tasks), do: meta + defp progress_meta(opts) do case Keyword.get(opts, :progress_token) do nil -> %{} @@ -607,6 +662,32 @@ defmodule Phantom.Test do end end + @doc """ + Assert the result is a task, from a tool that answered with one or from + `get_task/3`. + + Options: `:id` matches the task ID, `:status` the status (such as + `:completed`). + """ + def assert_task(result, opts \\ []) do + case result do + %{taskId: id, status: status} -> + if expected = opts[:id] do + assert id == expected, "expected task id #{inspect(expected)}, got: #{inspect(id)}" + end + + if expected = opts[:status] do + assert status == to_string(expected), + "expected task status #{inspect(to_string(expected))}, got: #{inspect(status)}" + end + + result + + other -> + flunk("expected a task, got: #{inspect(other)}") + end + end + @doc "Assert the tool returned image content." def assert_tool_image(result, opts \\ []) do assert_content_type(result, :image, opts) diff --git a/lib/phantom/tracker.ex b/lib/phantom/tracker.ex index fff471b..81b21d4 100644 --- a/lib/phantom/tracker.ex +++ b/lib/phantom/tracker.ex @@ -27,6 +27,7 @@ defmodule Phantom.Tracker do @sessions "phantom:sessions" @requests "phantom:requests" @resources "phantom:resources" + @tasks "phantom:tasks" def available?, do: @available def resource_subscription_topic, do: @resources @@ -276,6 +277,17 @@ defmodule Phantom.Tracker do def list_resource_listeners, do: [] end + @doc "Return a list of all listening for task status notifications" + if @available do + def list_task_listeners do + Phoenix.Tracker.list(__MODULE__, @tasks) + rescue + _ -> [] + end + else + def list_task_listeners, do: [] + end + @doc "Fetch the PID of the open request by ID" def get_request(%Phantom.Request{id: request_id}), do: get_request(request_id) @@ -338,6 +350,11 @@ defmodule Phantom.Tracker do "phantom:listen:" <> Base.url_encode64(key, padding: false) end + defp request_topic(session_id, request_id) do + key = :crypto.hash(:sha256, :erlang.term_to_binary({session_id, request_id})) + "phantom:request:" <> Base.url_encode64(key, padding: false) + end + if @available do @doc false def subscribe_session(nil, _session_id), do: :ok @@ -386,6 +403,25 @@ defmodule Phantom.Tracker do ) end + @doc false + def subscribe_request(nil, _session_id, _request_id), do: :ok + + # Progress is best-effort, so a missing PubSub must not fail the request. + def subscribe_request(pubsub, session_id, request_id) do + Phoenix.PubSub.subscribe(pubsub, request_topic(session_id, request_id)) + rescue + ArgumentError -> :ok + end + + @doc false + def cast_request(pubsub, session_id, request_id, message) do + Phoenix.PubSub.broadcast( + pubsub, + request_topic(session_id, request_id), + {:"$gen_cast", message} + ) + end + @doc false def cast_session(nil, _session_id, _message), do: :ok @@ -397,6 +433,8 @@ defmodule Phantom.Tracker do def cast_session(_pubsub, _session_id, _message), do: :ok def subscribe_listen(_pubsub, _principal, _request_id), do: :ok def cast_listen(_pubsub, _principal, _request_id, _message), do: :ok + def subscribe_request(_pubsub, _session_id, _request_id), do: :ok + def cast_request(_pubsub, _session_id, _request_id, _message), do: :ok end @doc "Untrack the processe for everything" @@ -470,6 +508,28 @@ defmodule Phantom.Tracker do def subscribe_resource(_uri), do: {:error, :not_available} end + @doc false + if @available do + def subscribe_task(task_id) do + Phoenix.Tracker.track(__MODULE__, self(), @tasks, task_id, %{pid: self()}) + rescue + _ -> {:error, :tracker_not_in_supervision_tree} + end + else + def subscribe_task(_task_id), do: {:error, :not_available} + end + + @doc false + if @available do + def unsubscribe_task(task_id) do + Phoenix.Tracker.untrack(__MODULE__, self(), @tasks, task_id) + rescue + _ -> :ok + end + else + def unsubscribe_task(_task_id), do: :ok + end + @doc "Unsubscribe the process to resource notifications from the PubSub on topic #{inspect(@resources)}" if @available do def unsubscribe_resource(uri) do @@ -539,6 +599,36 @@ defmodule Phantom.Tracker do def notify_resources_updated(_uris), do: {:ok, 0} end + @doc """ + Notify clients listening to a task that it changed. Call it after each + change you store, with the task or its ID. + + Each listening stream fetches the task with `c:Phantom.Router.get_task/2` + and sends what it returns as `notifications/tasks`, so clients get the + stored task even when notifications arrive out of order. Returns the number + of listening streams. + """ + @spec notify_task_updated(Phantom.Tasks.t() | String.t()) :: + {:ok, non_neg_integer()} | {:error, :tracker_not_in_supervision_tree} + def notify_task_updated(%Phantom.Tasks{id: task_id}), do: notify_task_updated(task_id) + + if @available do + def notify_task_updated(task_id) when is_binary(task_id) do + pids = + __MODULE__ + |> Phoenix.Tracker.get_by_key(@tasks, task_id) + |> Enum.map(fn {_key, %{pid: pid}} -> pid end) + |> Enum.uniq() + + Enum.each(pids, &GenServer.cast(&1, {:task_updated, task_id})) + {:ok, length(pids)} + rescue + _ -> {:error, :tracker_not_in_supervision_tree} + end + else + def notify_task_updated(task_id) when is_binary(task_id), do: {:ok, 0} + end + @doc "Convenience wrapper for notifying sessions about one updated resource" @spec notify_resource_updated(String.t()) :: {:ok, non_neg_integer()} | {:error, :tracker_not_in_supervision_tree} diff --git a/mix.exs b/mix.exs index 5415129..ef5a0d4 100644 --- a/mix.exs +++ b/mix.exs @@ -109,7 +109,7 @@ defmodule Phantom.MixProject do defp docs do [ main: "Phantom", - extras: ["guides/mcp_apps.md", "CHANGELOG.md"], + extras: ["guides/mcp_apps.md", "guides/tasks.md", "CHANGELOG.md"], assets: %{"assets" => "assets"}, before_closing_body_tag: %{html: @mermaidjs} ] diff --git a/test/conformance/conformance_test.exs b/test/conformance/conformance_test.exs index b8d7063..562772e 100644 --- a/test/conformance/conformance_test.exs +++ b/test/conformance/conformance_test.exs @@ -11,7 +11,17 @@ defmodule Phantom.ConformanceTest do @unscored_scenarios [ {"server-session-lifecycle", "2025-11-25"}, {"json-schema-2020-12", "2025-11-25"}, - {"json-schema-2020-12", "2026-07-28"} + {"json-schema-2020-12", "2026-07-28"}, + {"tasks-lifecycle", "2026-07-28"}, + {"tasks-capability-negotiation", "2026-07-28"}, + {"tasks-wire-fields", "2026-07-28"}, + {"tasks-request-state-removal", "2026-07-28"}, + {"tasks-mrtr-input", "2026-07-28"}, + {"tasks-request-headers", "2026-07-28"}, + {"tasks-dispatch-and-envelope", "2026-07-28"}, + {"tasks-status-notifications", "2026-07-28"}, + {"tasks-required-task-error", "2026-07-28"}, + {"tasks-mrtr-composition", "2026-07-28"} ] setup_all do diff --git a/test/conformance/router.ex b/test/conformance/router.ex index 82ffe0d..068afe5 100644 --- a/test/conformance/router.ex +++ b/test/conformance/router.ex @@ -590,6 +590,202 @@ defmodule Phantom.Conformance.MCP.Router do defp roots_request, do: %{method: "roots/list", params: %{}} + ## Tasks + + # A global Agent lets every node in the cluster answer `tasks/*`. + @tasks {:global, Phantom.Conformance.Tasks} + + tool :greet, description: "Returns a greeting without a task" do + field :name, :string, required: true + end + + tool :slow_compute, description: "Sleeps for the given seconds, then returns a result" do + field :seconds, :integer, required: true + field :label, :string + end + + tool :failing_job, description: "Requires a task; ends with a tool error after about a second" + tool :protocol_error_job, description: "Runs as a task that fails with a protocol error" + + tool :confirm_delete, description: "Asks to confirm, then deletes a file" do + field :filename, :string, required: true + end + + tool :multi_input, description: "Asks for two inputs at once, then completes" + tool :test_tool_with_task, description: "Asks for a name, then greets it from a task" + + def greet(%{"name" => name}, session), do: {:reply, Tool.text("Hello, #{name}!"), session} + + def slow_compute(%{"seconds" => seconds} = params, session) do + label = params["label"] || "compute" + + if seconds > 0 and Session.tasks_supported?(session) do + task = + start_task(fn -> + Process.sleep(:timer.seconds(seconds)) + {:completed, Tool.response(Tool.text("Computed #{label}"))} + end) + + {:reply, task, session} + else + Process.sleep(:timer.seconds(seconds)) + {:reply, Tool.text("Computed #{label}"), session} + end + end + + def failing_job(_params, session) do + if Session.tasks_supported?(session) do + task = + start_task(fn -> + Process.sleep(1_000) + {:completed, Tool.error("failing_job failed")} + end) + + {:reply, task, session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def protocol_error_job(_params, session) do + if Session.tasks_supported?(session) do + task = + start_task(fn -> + {:failed, Phantom.Request.internal_error("protocol_error_job failed")} + end) + + {:reply, task, session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def confirm_delete(%{"filename" => filename}, session) do + if Session.tasks_supported?(session) do + requests = %{"confirm" => elicit_request("Delete #{filename}?", "confirm", "boolean")} + {:reply, await_input_task(requests, "Deleted #{filename}"), session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def multi_input(_params, session) do + if Session.tasks_supported?(session) do + requests = %{ + "name" => elicit_request("What is your name?", "name"), + "confirm" => elicit_request("Continue?", "confirm", "boolean") + } + + {:reply, await_input_task(requests, "Received all input"), session} + else + {:error, Phantom.Request.missing_task_capability(), session} + end + end + + def test_tool_with_task(_params, session) do + case {session.state, input_responses(session)} do + {%{step: :named}, %{"user_name" => response}} -> + name = input_text(response, "name") + task = start_task(fn -> {:completed, Tool.response(Tool.text("Hello, #{name}!"))} end) + {:reply, task, session} + + _ -> + {:reply, + Tool.input_required( + input_requests: %{"user_name" => elicit_request("What is your name?", "name")}, + state: %{step: :named} + ), session} + end + end + + def get_task(task_id, _session) do + case Agent.get(tasks(), &Map.get(&1, task_id)) do + nil -> {:error, :not_found} + %{task: task} -> {:ok, task} + end + end + + def update_task(task, responses, _session) do + update_stored_task(task.id, fn %{task: task, on_input: result} = entry -> + remaining = Map.drop(task.input_requests, Map.keys(responses)) + + task = + if map_size(remaining) == 0, + do: %{task | status: :completed, input_requests: %{}, result: Tool.response(result)}, + else: %{task | input_requests: remaining} + + %{entry | task: task} + end) + end + + def cancel_task(task, _session) do + update_stored_task(task.id, fn %{task: task} = entry -> + %{entry | task: %{task | status: :cancelled, status_message: "Cancelled by the client"}} + end) + end + + defp start_task(work) do + task = new_task(:working, %{}) + Agent.update(tasks(), &Map.put(&1, task.id, %{task: task, on_input: nil})) + + spawn(fn -> + outcome = work.() + + update_stored_task(task.id, fn + %{task: %{status: :working} = task} = entry -> %{entry | task: finish(task, outcome)} + entry -> entry + end) + end) + + task + end + + defp await_input_task(requests, result) do + task = new_task(:input_required, requests) + Agent.update(tasks(), &Map.put(&1, task.id, %{task: task, on_input: Tool.text(result)})) + task + end + + defp new_task(status, input_requests) do + Phantom.Tasks.new( + id: Phantom.Tasks.generate_id(), + status: status, + created_at: DateTime.utc_now(), + ttl_ms: :timer.minutes(5), + poll_interval_ms: 200, + input_requests: input_requests + ) + end + + defp finish(task, {:completed, result}), do: %{task | status: :completed, result: result} + defp finish(task, {:failed, error}), do: %{task | status: :failed, error: error} + + # A terminal task never changes, so a late update or cancel is a no-op. + defp update_stored_task(task_id, fun) do + updated = + Agent.get_and_update(tasks(), fn tasks -> + case Map.fetch(tasks, task_id) do + {:ok, %{task: %{status: status}} = entry} when status in [:working, :input_required] -> + entry = fun.(entry) + entry = put_in(entry.task.last_updated_at, DateTime.utc_now()) + {entry.task, Map.put(tasks, task_id, entry)} + + _ -> + {nil, tasks} + end + end) + + if updated, do: Phantom.Tracker.notify_task_updated(updated) + :ok + end + + defp tasks do + case Agent.start(fn -> %{} end, name: @tasks) do + {:ok, pid} -> pid + {:error, {:already_started, pid}} -> pid + end + end + ## Resources resource "test://static-text", :static_text, diff --git a/test/phantom/distributed_test.exs b/test/phantom/distributed_test.exs index eb95412..97d03e4 100644 --- a/test/phantom/distributed_test.exs +++ b/test/phantom/distributed_test.exs @@ -447,6 +447,96 @@ defmodule Phantom.DistributedTest do end end + describe "cross-node task notifications" do + test "a task update on node 2 reaches a subscriptions/listen stream on node 1" do + task_id = "task-cross-node" + + meta = %{ + "io.modelcontextprotocol/protocolVersion" => "2026-07-28", + "io.modelcontextprotocol/clientCapabilities" => %{ + "extensions" => %{"io.modelcontextprotocol/tasks" => %{}} + } + } + + stream_resp = + post_mcp( + @node1_port, + %{ + jsonrpc: "2.0", + id: "listen:tasks", + method: "subscriptions/listen", + params: %{"notifications" => %{"taskIds" => [task_id]}, "_meta" => meta} + }, + headers: [ + {"mcp-protocol-version", "2026-07-28"}, + {"mcp-method", "subscriptions/listen"} + ] + ) + + await_task_tracked(@node2, task_id) + + assert {:ok, 1} = :rpc.call(@node2, Phantom.Tracker, :notify_task_updated, [task_id]) + + notification = + poll_for_sse_event(stream_resp, 10_000, &(&1["method"] == "notifications/tasks")) + + assert notification, "Expected notifications/tasks on the node 1 listen stream" + assert notification["params"]["taskId"] == task_id + assert notification["params"]["status"] == "working" + end + end + + describe "cross-node progress" do + test "a progress reference reaches the request's stream from another node" do + resp = + post_mcp( + @node1_port, + %{ + jsonrpc: "2.0", + id: "progress-1", + method: "tools/call", + params: %{ + "name" => "remote_progress_tool", + "arguments" => %{}, + "_meta" => Map.put(@modern_meta, "progressToken", "remote-token") + } + }, + headers: [ + {"mcp-protocol-version", "2026-07-28"}, + {"mcp-method", "tools/call"}, + {"mcp-name", "remote_progress_tool"} + ] + ) + + notification = + poll_for_sse_event(resp, 10_000, &(&1["method"] == "notifications/progress")) + + assert notification, "Expected notifications/progress from another node" + + assert %{"progressToken" => "remote-token", "progress" => 50, "total" => 100} = + notification["params"] + + assert notification["params"]["message"] =~ "node" + refute notification["params"]["message"] =~ "#{@node1}" + end + end + + defp await_task_tracked(node, task_id, attempts \\ 100) + + defp await_task_tracked(node, task_id, attempts) when attempts > 0 do + listeners = :rpc.call(node, Phantom.Tracker, :list_task_listeners, []) + + if Enum.any?(listeners, &match?({^task_id, _}, &1)) do + :ok + else + Process.sleep(50) + await_task_tracked(node, task_id, attempts - 1) + end + end + + defp await_task_tracked(node, task_id, 0), + do: flunk("Task #{task_id} not visible on #{node} within timeout") + describe "stateless core (2026-07-28) — no Tracker, no sticky session" do # Node A returns `inputRequired` with an encrypted `requestState`. # Node B — with no prior knowledge of the original call — decodes the diff --git a/test/phantom/elicit_test.exs b/test/phantom/elicit_test.exs index 8f6b378..1fa638b 100644 --- a/test/phantom/elicit_test.exs +++ b/test/phantom/elicit_test.exs @@ -235,6 +235,32 @@ defmodule Phantom.ElicitTest do end end + describe "to_input_request/1" do + test "builds one embedded elicitation request" do + elicit = + Elicit.form(%{ + message: "Continue?", + requested_schema: [%{type: :boolean, name: "confirm", required: true}] + }) + + assert %{ + method: "elicitation/create", + params: %{mode: "form", message: "Continue?", requestedSchema: %{type: "object"}} + } = Elicit.to_input_request(elicit) + end + + test "omits the legacy elicitation id from a URL request" do + elicit = + Elicit.url(%{ + message: "Authenticate", + url: "https://example.com", + elicitation_id: "legacy-id" + }) + + refute Map.has_key?(Elicit.to_input_request(elicit).params, :elicitationId) + end + end + describe "to_input_requests/1" do test "builds the modern embedded elicitation request map" do elicit = diff --git a/test/phantom/plug_test.exs b/test/phantom/plug_test.exs index 67bc542..59f2c6c 100644 --- a/test/phantom/plug_test.exs +++ b/test/phantom/plug_test.exs @@ -1414,6 +1414,55 @@ defmodule Phantom.PlugTest do refute_receive {:response, nil, "message", %{method: "notifications/message"}} end + test "tasks methods require Mcp-Name to carry the taskId" do + for method <- ~w[tasks/get tasks/update tasks/cancel] do + post_stateless( + %{jsonrpc: "2.0", id: 20, method: method, params: %{"taskId" => "task-1"}}, + [{"mcp-method", method}] + ) + + assert_receive {:conn, conn} + assert conn.status == 400 + error = JSON.decode!(conn.resp_body) + assert error["error"]["code"] == -32020 + assert error["error"]["message"] =~ "Mcp-Name" + end + end + + test "tasks methods reject an Mcp-Name that is not the taskId" do + post_stateless( + %{jsonrpc: "2.0", id: 21, method: "tasks/get", params: %{"taskId" => "task-1"}}, + [{"mcp-method", "tasks/get"}, {"mcp-name", "task-2"}] + ) + + assert_receive {:conn, conn} + assert conn.status == 400 + assert JSON.decode!(conn.resp_body)["error"]["code"] == -32020 + end + + test "tasks methods with a matching Mcp-Name reach the router" do + meta = %{ + "io.modelcontextprotocol/clientCapabilities" => %{ + "extensions" => %{"io.modelcontextprotocol/tasks" => %{}} + } + } + + post_stateless( + %{ + jsonrpc: "2.0", + id: 22, + method: "tasks/get", + params: %{"taskId" => "task-1", "_meta" => meta} + }, + [{"mcp-method", "tasks/get"}, {"mcp-name", "task-1"}] + ) + + assert_connected(_conn) + + assert_receive {:response, 22, "message", + %{result: %{taskId: "task-1", status: "working", resultType: "complete"}}} + end + test "missing Mcp-Method rejects with 400 + -32020 HeaderMismatch" do post_stateless( %{ diff --git a/test/phantom/progress_ref_test.exs b/test/phantom/progress_ref_test.exs new file mode 100644 index 0000000..d251cfd --- /dev/null +++ b/test/phantom/progress_ref_test.exs @@ -0,0 +1,69 @@ +defmodule Phantom.ProgressRefTest do + use ExUnit.Case, async: true + + alias Phantom.Request + alias Phantom.Session + + @pubsub Phantom.ProgressRefTest.PubSub + + setup do + start_supervised!({Phoenix.PubSub, name: @pubsub}) + :ok + end + + defp session(params) do + request = %Request{id: "req-1", method: "tools/call", params: params} + Session.new("session-1", pubsub: @pubsub, pid: self(), request: request) + end + + describe "progress_ref/1" do + test "is a JSON-safe reference to the request" do + ref = Session.progress_ref(session(%{"_meta" => %{"progressToken" => "tok"}})) + + assert ref == %{ + "pubsub" => "Elixir.Phantom.ProgressRefTest.PubSub", + "session_id" => "session-1", + "request_id" => "req-1", + "progress_token" => "tok" + } + + assert ref |> JSON.encode!() |> JSON.decode!() == ref + end + + test "is nil when the client did not ask for progress" do + assert Session.progress_ref(session(%{})) == nil + end + + test "is nil without PubSub" do + session = %{session(%{"_meta" => %{"progressToken" => "tok"}}) | pubsub: nil} + assert Session.progress_ref(session) == nil + end + end + + describe "notify_progress/4 with a progress ref" do + test "reaches the request's stream through PubSub" do + ref = Session.progress_ref(session(%{"_meta" => %{"progressToken" => "tok"}})) + Phantom.Tracker.subscribe_request(@pubsub, "session-1", "req-1") + + assert :ok = Session.notify_progress(ref, 50, 100, "Halfway") + assert_receive {:"$gen_cast", {:progress, "tok", 50, 100, "Halfway"}} + end + + test "works after a JSON round trip" do + ref = + session(%{"_meta" => %{"progressToken" => 7}}) + |> Session.progress_ref() + |> JSON.encode!() + |> JSON.decode!() + + Phantom.Tracker.subscribe_request(@pubsub, "session-1", "req-1") + + assert :ok = Session.notify_progress(ref, 1) + assert_receive {:"$gen_cast", {:progress, 7, 1, nil, nil}} + end + + test "does nothing for a nil ref" do + assert :ok = Session.notify_progress(nil, 50, 100, "Halfway") + end + end +end diff --git a/test/phantom/request_test.exs b/test/phantom/request_test.exs index ed4f099..1bcb939 100644 --- a/test/phantom/request_test.exs +++ b/test/phantom/request_test.exs @@ -159,6 +159,18 @@ defmodule Phantom.RequestTest do end end + describe "missing_task_capability/0" do + test "names the Tasks extension" do + assert Request.missing_task_capability() == %{ + code: -32021, + message: "Missing required client capability", + data: %{ + requiredCapabilities: %{extensions: %{"io.modelcontextprotocol/tasks" => %{}}} + } + } + end + end + describe "trace_context/1" do test "extracts the three W3C fields when present" do {:ok, request} = diff --git a/test/phantom/task_subscription_test.exs b/test/phantom/task_subscription_test.exs new file mode 100644 index 0000000..51e00c6 --- /dev/null +++ b/test/phantom/task_subscription_test.exs @@ -0,0 +1,259 @@ +defmodule Phantom.TaskSubscriptionTest.Router do + use Phantom.Router, name: "Task subscription test" + + def connect(session, %Plug.Conn{} = conn) do + user_id = conn |> Plug.Conn.get_req_header("x-user-id") |> List.first() + {:ok, Phantom.Session.assign(session, user_id: user_id)} + end + + def connect(session, _context), do: {:ok, session} + + def get_task(task_id, session) do + owners = :persistent_term.get({__MODULE__, :owners}, %{}) + + status = :persistent_term.get({__MODULE__, :status}, :working) + + if Map.get(owners, task_id) == session.assigns[:user_id], + do: {:ok, task(task_id, status)}, + else: {:error, :not_found} + end + + def task(id, status \\ :working) do + attrs = [id: id, status: status, created_at: ~U[2025-11-25 10:30:00Z]] + + if status == :completed, + do: Phantom.Tasks.new([result: %{content: [%{type: "text", text: "Done"}]}] ++ attrs), + else: Phantom.Tasks.new(attrs) + end +end + +defmodule Phantom.TaskSubscriptionTest.CustomRouter do + use Phantom.Router, name: "Task subscription custom authorization test" + + def get_task(id, _session), do: {:ok, Phantom.TaskSubscriptionTest.Router.task(id)} + + def authorize_task_subscriptions(task_ids, _session) do + case :persistent_term.get({__MODULE__, :return}, :all) do + :raise -> raise "authorization failed" + :invalid -> %{allowed: task_ids} + :all -> task_ids + allowed -> allowed + end + end +end + +defmodule Phantom.TaskSubscriptionTest do + use ExUnit.Case + + import ExUnit.CaptureLog + import Phantom.TestDispatcher + import Plug.Conn + + alias Phantom.TaskSubscriptionTest.CustomRouter + alias Phantom.TaskSubscriptionTest.Router + alias Phantom.Session + + @tasks_ext "io.modelcontextprotocol/tasks" + + setup do + start_supervised({Phoenix.PubSub, name: Test.PubSub}) + start_supervised({Phantom.Tracker, [name: Phantom.Tracker, pubsub_server: Test.PubSub]}) + Phantom.Cache.register(Router) + Phantom.Cache.register(CustomRouter) + :persistent_term.put({Router, :owners}, %{"alice-1" => "alice", "bob-1" => "bob"}) + + on_exit(fn -> + :persistent_term.erase({Router, :owners}) + :persistent_term.erase({Router, :status}) + :persistent_term.erase({CustomRouter, :return}) + end) + + :ok + end + + describe "authorize_task_subscriptions" do + test "defaults to the tasks get_task/2 returns" do + session = Session.new("s", router: Router) |> Session.assign(user_id: "alice") + + assert Phantom.Router.authorize_task_subscriptions(Router, ["alice-1", "bob-1"], session) == + ["alice-1"] + end + + test "keeps only requested IDs from a custom callback" do + session = Session.new("s", router: CustomRouter) + :persistent_term.put({CustomRouter, :return}, ["a", "not-requested"]) + + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a", "b"], session) == + ["a"] + end + + test "invalid callback results and exceptions fail closed" do + session = Session.new("s", router: CustomRouter) + + :persistent_term.put({CustomRouter, :return}, :invalid) + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a"], session) == [] + + :persistent_term.put({CustomRouter, :return}, nil) + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a"], session) == [] + + :persistent_term.put({CustomRouter, :return}, :raise) + + assert capture_log(fn -> + assert Phantom.Router.authorize_task_subscriptions(CustomRouter, ["a"], session) == + [] + end) =~ "failed closed" + end + end + + describe "subscriptions/listen with taskIds" do + test "acknowledges and tracks only authorized tasks" do + stream_pid = listen(["alice-1", "bob-1"], "alice") + + assert_notify(%{ + method: "notifications/subscriptions/acknowledged", + params: %{notifications: %{"taskIds" => ["alice-1"]}} + }) + + wait_for_task_listeners(["alice-1"]) + tracked = Enum.map(Phantom.Tracker.list_task_listeners(), &elem(&1, 0)) + refute "bob-1" in tracked + + Session.finish(stream_pid) + end + + test "sends notifications/tasks with the stored task" do + stream_pid = listen(["alice-1", "bob-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + :persistent_term.put({Router, :status}, :completed) + task = Router.task("alice-1", :completed) + # A stale copy still sends the stored task. + Phantom.Tracker.notify_task_updated(Router.task("alice-1", :working)) + Phantom.Tracker.notify_task_updated("bob-1") + + expected = Phantom.Tasks.to_json(task) + + assert_notify(%{ + method: "notifications/tasks", + params: %{_meta: %{"io.modelcontextprotocol/subscriptionId" => "listen:0"}} = params + }) + + assert Map.delete(params, :_meta) == expected + + refute_receive {:response, nil, "message", + %{method: "notifications/tasks", params: %{taskId: "bob-1"}}} + + Session.finish(stream_pid) + end + + test "accepts a task ID" do + stream_pid = listen(["alice-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + Phantom.Tracker.notify_task_updated("alice-1") + + assert_notify(%{ + method: "notifications/tasks", + params: %{taskId: "alice-1", status: "working"} + }) + + Session.finish(stream_pid) + end + + test "stops following a task once it finishes" do + stream_pid = listen(["alice-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + :persistent_term.put({Router, :status}, :completed) + Phantom.Tracker.notify_task_updated("alice-1") + assert_notify(%{method: "notifications/tasks", params: %{status: "completed"}}) + + wait_for_no_task_listeners("alice-1") + Session.finish(stream_pid) + end + + test "re-authorizes before each notification" do + stream_pid = listen(["alice-1"], "alice") + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + wait_for_task_listeners(["alice-1"]) + + :persistent_term.put({Router, :owners}, %{"alice-1" => "bob"}) + Phantom.Tracker.notify_task_updated(Router.task("alice-1", :completed)) + + refute_receive {:response, nil, "message", %{method: "notifications/tasks"}} + + Session.finish(stream_pid) + end + + test "ignores an empty taskIds list when the request did not declare the extension" do + stream_pid = listen([], "alice", %{}) + assert_notify(%{method: "notifications/subscriptions/acknowledged"}) + Session.finish(stream_pid) + end + + test "is a missing capability error when the request did not declare the extension" do + listen(["alice-1"], "alice", %{}) + + assert_receive {:conn, conn} + error = JSON.decode!(conn.resp_body)["error"] + assert error["code"] == -32021 + assert error["data"] == %{"requiredCapabilities" => %{"extensions" => %{@tasks_ext => %{}}}} + end + end + + defp listen(task_ids, user_id, capabilities \\ %{"extensions" => %{@tasks_ext => %{}}}) do + :post + |> Plug.Test.conn("/mcp", %{ + "jsonrpc" => "2.0", + "id" => "listen:0", + "method" => "subscriptions/listen", + "params" => %{ + "notifications" => %{"taskIds" => task_ids}, + "_meta" => %{ + "io.modelcontextprotocol/protocolVersion" => "2026-07-28", + "io.modelcontextprotocol/clientCapabilities" => capabilities + } + } + }) + |> put_req_header("content-type", "application/json") + |> put_req_header("mcp-protocol-version", "2026-07-28") + |> put_req_header("mcp-method", "subscriptions/listen") + |> call(%{ + router: Router, + before_call: &put_req_header(&1, "x-user-id", user_id) + }) + end + + defp wait_for_task_listeners(expected, attempts \\ 50) + + defp wait_for_task_listeners(expected, attempts) when attempts > 0 do + actual = Phantom.Tracker.list_task_listeners() |> Enum.map(&elem(&1, 0)) |> MapSet.new() + + if MapSet.subset?(MapSet.new(expected), actual) do + :ok + else + Process.sleep(10) + wait_for_task_listeners(expected, attempts - 1) + end + end + + defp wait_for_no_task_listeners(task_id, attempts \\ 50) + + defp wait_for_no_task_listeners(task_id, attempts) when attempts > 0 do + if Enum.any?(Phantom.Tracker.list_task_listeners(), &match?({^task_id, _}, &1)) do + Process.sleep(10) + wait_for_no_task_listeners(task_id, attempts - 1) + else + :ok + end + end + + defp wait_for_no_task_listeners(task_id, 0), + do: flunk("task listener was still tracked: #{inspect(task_id)}") + + defp wait_for_task_listeners(expected, 0), + do: flunk("task listeners were not tracked: #{inspect(expected)}") +end diff --git a/test/phantom/tasks_dispatch_test.exs b/test/phantom/tasks_dispatch_test.exs new file mode 100644 index 0000000..061983f --- /dev/null +++ b/test/phantom/tasks_dispatch_test.exs @@ -0,0 +1,380 @@ +defmodule Phantom.TasksDispatchTest do + use ExUnit.Case, async: true + + alias Phantom.Session + + @tasks_ext "io.modelcontextprotocol/tasks" + + defmodule Router do + use Phantom.Router, name: "Tasks", vsn: "1.0" + + def get_task("expired", _session), do: {:error, :expired} + def get_task("broken", _session), do: {:error, Phantom.Request.internal_error("broken")} + + def get_task(id, %{assigns: %{owner: "alice"}}) when id in ~w[working input rejecting], + do: {:ok, task(id)} + + def get_task(_task_id, _session), do: {:error, :not_found} + + def update_task(%{id: "rejecting"}, _responses, _session), + do: {:error, Phantom.Request.invalid_params(%{inputResponses: "rejected"})} + + def update_task(task, responses, session) do + send(session.assigns.test_pid, {:update_task, task.id, responses}) + :ok + end + + def cancel_task(task, session) do + send(session.assigns.test_pid, {:cancel_task, task.id}) + :ok + end + + def task("working"), + do: Phantom.Tasks.new(id: "working", status: :working, created_at: ~U[2025-11-25 10:30:00Z]) + + def task(id) when id in ~w[input rejecting] do + Phantom.Tasks.new( + id: id, + status: :input_required, + created_at: ~U[2025-11-25 10:30:00Z], + input_requests: %{ + "name" => %{method: "elicitation/create", params: %{mode: "form", message: "Name?"}}, + "age" => %{method: "elicitation/create", params: %{mode: "form", message: "Age?"}} + } + ) + end + + tool :always_task, description: "Always answers with a task" + tool :input_task, description: "Answers with a task that already needs input" + + def input_task(_params, session), do: {:reply, task("input"), session} + + def always_task(_params, session) do + {:reply, + Phantom.Tasks.new( + id: "task-1", + status: :working, + created_at: ~U[2025-11-25 10:30:00Z], + ttl_ms: 60_000, + poll_interval_ms: 5_000 + ), session} + end + end + + defmodule NoTasksRouter do + use Phantom.Router, name: "NoTasks", vsn: "1.0" + end + + defmodule BadRouter do + use Phantom.Router, name: "Bad", vsn: "1.0" + + def get_task("plain-map", _session), do: {:ok, %{id: "plain-map"}} + def get_task(id, _session), do: {:ok, Phantom.TasksDispatchTest.Router.task(id)} + + def update_task(task, _responses, _session), do: {:ok, task} + end + + defmodule GetOnlyRouter do + use Phantom.Router, name: "GetOnly", vsn: "1.0" + + def get_task(id, _session), do: {:ok, Phantom.TasksDispatchTest.Router.task(id)} + end + + import Phantom.Test + + setup do + Phantom.Test.start(router: Router) + Phantom.Test.start(router: NoTasksRouter) + Phantom.Test.start(router: GetOnlyRouter) + Phantom.Test.start(router: BadRouter) + :ok + end + + defp tasks_capabilities, do: %{extensions: %{@tasks_ext => %{}}} + + defp session(router \\ Router, opts \\ []) do + build_session( + router, + Keyword.merge( + [ + protocol_version: "2026-07-28", + client_capabilities: tasks_capabilities(), + assigns: %{owner: "alice", test_pid: self()} + ], + opts + ) + ) + end + + defp dispatch(session, method, params \\ %{}) do + request = %{build_request(method, params: params) | meta: session.request.meta} + session = %{session | pid: self(), request: request} + session.router.dispatch_method([method, params, request, session]) + end + + describe "Session.tasks_supported?/1" do + test "is true when the request declares the extension" do + assert Session.tasks_supported?(session()) + end + + test "is false when the request does not declare the extension" do + refute Session.tasks_supported?(session(Router, client_capabilities: %{})) + end + + test "is false under a legacy protocol" do + refute Session.tasks_supported?(build_session(Router)) + end + end + + describe "server/discover" do + test "advertises the extension when the router implements get_task/2" do + assert {:reply, %{capabilities: %{extensions: %{@tasks_ext => %{}}}}, _} = + dispatch(session(), "server/discover") + end + + test "does not advertise the extension otherwise" do + assert {:reply, %{capabilities: capabilities}, _} = + dispatch(session(NoTasksRouter), "server/discover") + + refute get_in(capabilities, [:extensions, @tasks_ext]) + end + end + + describe "tools/call returning a task" do + test "responds with a CreateTaskResult" do + assert call_tool(session(), :always_task) == %{ + resultType: "task", + taskId: "task-1", + status: "working", + createdAt: "2025-11-25T10:30:00Z", + lastUpdatedAt: "2025-11-25T10:30:00Z", + ttlMs: 60_000, + pollIntervalMs: 5_000 + } + end + + test "leaves status-specific fields out of the CreateTaskResult" do + result = call_tool(session(), :input_task) + assert %{resultType: "task", taskId: "input", status: "input_required"} = result + refute Map.has_key?(result, :inputRequests) + end + + test "keeps resultType task through result normalization" do + request = %{build_request("tools/call") | meta: session().request.meta} + + assert %{resultType: "task"} = + Phantom.Request.normalize_result(%{resultType: "task", taskId: "t"}, request) + end + + test "is a missing capability error when the request did not declare the extension" do + assert {:jsonrpc_error, error} = + call_tool(session(Router, client_capabilities: %{}), :always_task) + + assert error == %{ + code: -32021, + message: "Missing required client capability", + data: %{requiredCapabilities: %{extensions: %{@tasks_ext => %{}}}} + } + end + + test "is a missing capability error under a legacy protocol" do + assert {:jsonrpc_error, %{code: -32021}} = call_tool(build_session(Router), :always_task) + end + end + + @not_found %{code: -32602, message: "Failed to retrieve task: Task not found"} + + describe "tasks/get" do + test "returns the task" do + assert {:reply, result, _} = dispatch(session(), "tasks/get", %{"taskId" => "working"}) + assert result == Phantom.Tasks.to_json(Router.task("working")) + end + + test "inlines outstanding input requests" do + assert {:reply, %{status: "input_required", inputRequests: %{"name" => _, "age" => _}}, _} = + dispatch(session(), "tasks/get", %{"taskId" => "input"}) + end + + test "is not found when get_task/2 denies the caller" do + session = session(Router, assigns: %{owner: "mallory", test_pid: self()}) + + assert {:error, @not_found, _} = dispatch(session, "tasks/get", %{"taskId" => "working"}) + end + + test "is not found for an unknown task" do + assert {:error, @not_found, _} = dispatch(session(), "tasks/get", %{"taskId" => "nope"}) + end + + test "reports an expired task" do + assert {:error, %{code: -32602, message: "Failed to retrieve task: Task has expired"}, _} = + dispatch(session(), "tasks/get", %{"taskId" => "expired"}) + end + + test "passes through a JSON-RPC error from get_task/2" do + assert {:error, %{code: -32603, message: "broken"}, _} = + dispatch(session(), "tasks/get", %{"taskId" => "broken"}) + end + + test "raises a clear error when get_task/2 returns an unexpected value" do + assert_raise ArgumentError, ~r/get_task\/2/, fn -> + dispatch(session(BadRouter), "tasks/get", %{"taskId" => "plain-map"}) + end + end + + test "is method not found under a legacy protocol" do + request = build_request("tasks/get", params: %{"taskId" => "working"}) + session = %{build_session(Router) | pid: self(), request: request} + + assert {:error, %{code: -32601}, _} = + Router.dispatch_method(["tasks/get", request.params, request, session]) + end + + test "requires a taskId" do + assert {:error, %{code: -32602}, _} = dispatch(session(), "tasks/get", %{}) + end + + test "is a missing capability error when the request did not declare the extension" do + assert {:error, + %{code: -32021, data: %{requiredCapabilities: %{extensions: %{@tasks_ext => %{}}}}}, + _} = + dispatch(session(Router, client_capabilities: %{}), "tasks/get", %{ + "taskId" => "working" + }) + end + + test "is method not found when the router does not implement tasks" do + assert {:error, %{code: -32601}, _} = + dispatch(session(NoTasksRouter), "tasks/get", %{"taskId" => "working"}) + end + + test "other tasks methods are method not found" do + assert {:error, %{code: -32601}, _} = dispatch(session(), "tasks/list", %{}) + end + end + + describe "tasks/update" do + test "passes only outstanding input responses to update_task/3" do + params = %{ + "taskId" => "input", + "inputResponses" => %{ + "name" => %{"action" => "accept", "content" => %{"name" => "Luca"}}, + "unknown" => %{"action" => "accept"} + } + } + + assert {:reply, %{}, _} = dispatch(session(), "tasks/update", params) + + assert_received {:update_task, "input", + %{"name" => %{"action" => "accept", "content" => %{"name" => "Luca"}}} = + responses} + + refute Map.has_key?(responses, "unknown") + end + + test "acknowledges without calling update_task/3 when nothing is outstanding" do + params = %{"taskId" => "working", "inputResponses" => %{"name" => %{"action" => "accept"}}} + + assert {:reply, %{}, _} = dispatch(session(), "tasks/update", params) + refute_received {:update_task, _, _} + end + + test "passes through an error from update_task/3" do + params = %{ + "taskId" => "rejecting", + "inputResponses" => %{"name" => %{"action" => "accept"}} + } + + assert {:error, %{code: -32602, data: %{inputResponses: "rejected"}}, _} = + dispatch(session(), "tasks/update", params) + end + + test "requires inputResponses to be an object" do + assert {:error, %{code: -32602}, _} = + dispatch(session(), "tasks/update", %{"taskId" => "input", "inputResponses" => []}) + end + + test "is not found for a task the caller cannot see" do + params = %{"taskId" => "nope", "inputResponses" => %{"name" => %{"action" => "accept"}}} + assert {:error, @not_found, _} = dispatch(session(), "tasks/update", params) + end + + test "acknowledges when the router does not implement update_task/3" do + params = %{"taskId" => "input", "inputResponses" => %{"name" => %{"action" => "accept"}}} + + assert {:reply, %{}, _} = dispatch(session(GetOnlyRouter), "tasks/update", params) + end + + test "raises a clear error when update_task/3 returns an unexpected value" do + params = %{"taskId" => "input", "inputResponses" => %{"name" => %{"action" => "accept"}}} + + assert_raise ArgumentError, ~r/update_task\/3/, fn -> + dispatch(session(BadRouter), "tasks/update", params) + end + end + end + + describe "tasks/cancel" do + test "calls cancel_task/2 and acknowledges" do + assert {:reply, %{}, _} = dispatch(session(), "tasks/cancel", %{"taskId" => "working"}) + assert_received {:cancel_task, "working"} + end + + test "acknowledges when the router does not implement cancel_task/2" do + assert {:reply, %{}, _} = + dispatch(session(GetOnlyRouter), "tasks/cancel", %{"taskId" => "working"}) + end + + test "is not found for a task the caller cannot see" do + assert {:error, @not_found, _} = dispatch(session(), "tasks/cancel", %{"taskId" => "nope"}) + refute_received {:cancel_task, _} + end + end + + describe "Phantom.Test task helpers" do + test "call_tool/4 declares the extension with tasks: true" do + session(Router, client_capabilities: %{}) + |> call_tool(:always_task, %{}, tasks: true) + |> assert_task(id: "task-1", status: :working) + end + + test "tasks: true requires a stateless session" do + assert_raise ArgumentError, ~r/protocol_version/, fn -> + call_tool(build_session(Router), :always_task, %{}, tasks: true) + end + end + + test "get_task/3 returns the task" do + session(Router, client_capabilities: %{}) + |> get_task("input") + |> assert_task(status: :input_required) + end + + test "get_task/3 returns a JSON-RPC error for an unknown task" do + session() + |> get_task("nope") + |> assert_jsonrpc_error(code: -32602, message: "Failed to retrieve task: Task not found") + end + + test "update_task/4 sends input responses" do + assert update_task(session(), "input", %{name: %{"action" => "accept"}}) == %{} + assert_received {:update_task, "input", %{"name" => %{"action" => "accept"}}} + end + + test "cancel_task/3 cancels" do + assert cancel_task(session(), "working") == %{} + assert_received {:cancel_task, "working"} + end + + test "assert_task/2 fails on a mismatch" do + task = get_task(session(), "working") + + assert_raise ExUnit.AssertionError, ~r/status/, fn -> + assert_task(task, status: :completed) + end + + assert_raise ExUnit.AssertionError, ~r/expected a task/, fn -> + assert_task(%{content: []}) + end + end + end +end diff --git a/test/phantom/tasks_test.exs b/test/phantom/tasks_test.exs new file mode 100644 index 0000000..7e4b4b0 --- /dev/null +++ b/test/phantom/tasks_test.exs @@ -0,0 +1,199 @@ +defmodule Phantom.TasksTest do + use ExUnit.Case, async: true + + alias Phantom.Tasks + + @created_at ~U[2025-11-25 10:30:00Z] + @updated_at ~U[2025-11-25 10:50:00Z] + + defp attrs(overrides \\ []) do + Keyword.merge( + [ + id: "786512e2-9e0d-44bd-8f29-789f320fe840", + status: :working, + created_at: @created_at, + last_updated_at: @updated_at, + ttl_ms: 3_600_000, + poll_interval_ms: 5_000 + ], + overrides + ) + end + + describe "new/1" do + test "builds a working task" do + assert %Tasks{id: "786512e2-9e0d-44bd-8f29-789f320fe840", status: :working} = + Tasks.new(attrs()) + end + + test "accepts a map and a status string, as loaded from a database" do + task = + Tasks.new( + attrs() + |> Map.new() + |> Map.put(:status, "input_required") + |> Map.put(:input_requests, %{"a" => %{method: "elicitation/create", params: %{}}}) + ) + + assert task.status == :input_required + end + + test "defaults last_updated_at to created_at and ttl_ms to unlimited" do + task = Tasks.new(id: "t", status: :working, created_at: @created_at) + assert task.last_updated_at == @created_at + assert task.ttl_ms == nil + end + + test "rejects an unknown status" do + assert_raise ArgumentError, ~r/status/, fn -> Tasks.new(attrs(status: :done)) end + assert_raise ArgumentError, ~r/status/, fn -> Tasks.new(attrs(status: "done")) end + end + + test "requires an id and created_at" do + assert_raise ArgumentError, ~r/:id/, fn -> Tasks.new(attrs(id: nil)) end + assert_raise ArgumentError, ~r/:id/, fn -> Tasks.new(attrs(id: "")) end + assert_raise ArgumentError, ~r/:created_at/, fn -> Tasks.new(attrs(created_at: nil)) end + + assert_raise ArgumentError, ~r/:last_updated_at/, fn -> + Tasks.new(attrs(last_updated_at: "2025-11-25T10:50:00Z")) + end + end + + test "requires a result when completed" do + assert_raise ArgumentError, ~r/:result/, fn -> Tasks.new(attrs(status: :completed)) end + end + + test "requires a JSON-RPC error when failed" do + assert_raise ArgumentError, ~r/:error/, fn -> Tasks.new(attrs(status: :failed)) end + + assert_raise ArgumentError, ~r/:error/, fn -> + Tasks.new(attrs(status: :failed, error: %{message: "no code"})) + end + end + + test "requires input requests when input_required" do + assert_raise ArgumentError, ~r/:input_requests/, fn -> + Tasks.new(attrs(status: :input_required)) + end + + assert_raise ArgumentError, ~r/:input_requests/, fn -> + Tasks.new(attrs(status: :input_required, input_requests: %{})) + end + end + + test "rejects negative or non-integer durations" do + assert_raise ArgumentError, ~r/:ttl_ms/, fn -> Tasks.new(attrs(ttl_ms: -1)) end + + assert_raise ArgumentError, ~r/:poll_interval_ms/, fn -> + Tasks.new(attrs(poll_interval_ms: 1.5)) + end + end + + test "converts Phantom.Elicit input requests" do + elicit = + Phantom.Elicit.form(%{ + message: "Please enter your name.", + requested_schema: [%{name: "name", type: :string, required: true}] + }) + + task = Tasks.new(attrs(status: :input_required, input_requests: %{"name" => elicit})) + + assert %{"name" => %{method: "elicitation/create", params: %{mode: "form"}}} = + task.input_requests + end + end + + describe "encoding a task changed with struct update syntax" do + test "validates it" do + task = Tasks.new(attrs()) + + assert_raise ArgumentError, ~r/:result/, fn -> + Tasks.to_json(%{task | status: :completed}) + end + end + + test "normalizes a status string" do + task = Tasks.new(attrs()) + assert Tasks.to_json(%{task | status: "cancelled"}).status == "cancelled" + assert Tasks.to_create_result(%{task | status: "cancelled"}).status == "cancelled" + end + end + + describe "generate_id/0" do + test "is URL-safe and unique" do + id = Tasks.generate_id() + assert id =~ ~r/^[A-Za-z0-9_-]{43}$/ + refute id == Tasks.generate_id() + end + end + + describe "to_json/1" do + test "encodes a working task as in the spec" do + assert Tasks.to_json(Tasks.new(attrs(status_message: "The operation is now in progress."))) == + %{ + taskId: "786512e2-9e0d-44bd-8f29-789f320fe840", + status: "working", + statusMessage: "The operation is now in progress.", + createdAt: "2025-11-25T10:30:00Z", + lastUpdatedAt: "2025-11-25T10:50:00Z", + ttlMs: 3_600_000, + pollIntervalMs: 5_000 + } + end + + test "keeps ttlMs as null when unlimited and omits a missing poll interval" do + json = Tasks.to_json(Tasks.new(attrs(ttl_ms: nil, poll_interval_ms: nil))) + assert Map.fetch!(json, :ttlMs) == nil + refute Map.has_key?(json, :pollIntervalMs) + end + + test "inlines the input requests when input_required" do + request = %{ + method: "elicitation/create", + params: %{mode: "form", message: "Please enter your name."} + } + + json = + Tasks.to_json( + Tasks.new(attrs(status: :input_required, input_requests: %{"name" => request})) + ) + + assert json.status == "input_required" + assert json.inputRequests == %{"name" => request} + end + + test "inlines the result when completed" do + result = %{content: [%{type: "text", text: "Hello, Luca!"}], isError: false} + json = Tasks.to_json(Tasks.new(attrs(status: :completed, result: result))) + assert json.status == "completed" + assert json.result == result + refute Map.has_key?(json, :inputRequests) + end + + test "inlines the error when failed" do + error = %{code: -32603, message: "API rate limit exceeded"} + json = Tasks.to_json(Tasks.new(attrs(status: :failed, error: error))) + assert json.status == "failed" + assert json.error == error + end + + test "omits status-specific fields for other statuses" do + json = + Tasks.to_json( + Tasks.new( + attrs( + status: :cancelled, + result: %{content: []}, + error: %{code: -32603, message: "x"}, + input_requests: %{"a" => %{method: "elicitation/create", params: %{}}} + ) + ) + ) + + assert json.status == "cancelled" + refute Map.has_key?(json, :result) + refute Map.has_key?(json, :error) + refute Map.has_key?(json, :inputRequests) + end + end +end diff --git a/test/support/app/mcp/router.ex b/test/support/app/mcp/router.ex index aff5d70..16ae983 100644 --- a/test/support/app/mcp/router.ex +++ b/test/support/app/mcp/router.ex @@ -57,6 +57,11 @@ defmodule Test.MCP.Router do def connect(session, _context), do: {:ok, session} + def get_task("task-" <> _ = id, _session), + do: {:ok, Phantom.Tasks.new(id: id, status: :working, created_at: ~U[2025-11-25 10:30:00Z])} + + def get_task(_id, _session), do: {:error, :not_found} + def terminate(session) do if Test.SessionStore.running?(), do: Test.SessionStore.delete(session.id) {:ok, session} @@ -247,6 +252,7 @@ defmodule Test.MCP.Router do tool :really_long_async_tool, AsyncModule, description: "this will notify of progress" tool :timeout_async_tool, AsyncModule, description: "this will timeout!" tool :hanging_tool, description: "Never responds, so only cancellation ends it" + tool :remote_progress_tool, description: "Reports progress from another node, then responds" @audio File.read!(@base <> "/game-over.wav") def audio_tool(_params, session) do @@ -320,6 +326,20 @@ defmodule Test.MCP.Router do def hanging_tool(_params, session), do: {:noreply, session} + # Another node stands in for a job that only has the progress reference. + def remote_progress_tool(_params, session) do + ref = Session.progress_ref(session) + target = Enum.find(Node.list(), node(), &String.starts_with?(Atom.to_string(&1), "node")) + + Node.spawn(target, fn -> + Session.notify_progress(ref, 50, 100, "Halfway on #{node()}") + Process.sleep(200) + Session.respond(session, Tool.text("Done")) + end) + + {:noreply, session} + end + def client_log_tool(params, session) do message = params["message"] || "client-log-test" Phantom.ClientLogger.log(session, :info, %{message: message}, "test") From 2e956b2b8e5124658ef25f59e3f4df733ee45e3d Mon Sep 17 00:00:00 2001 From: David Bernheisel Date: Sun, 4 Oct 2026 20:57:19 -0400 Subject: [PATCH 2/6] CI: check formatting on the newest Elixir only The formatter output differs between Elixir versions, so only the newest version in the matrix checks it. --- .github/workflows/ci.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bbaaec4..8d374d1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,6 +19,8 @@ jobs: otp: "27" - elixir: "1.20" otp: "29" + # Formatter output differs between Elixir versions. + check_format: true steps: - uses: actions/checkout@v7 @@ -52,6 +54,7 @@ jobs: run: mix deps.unlock --check-unused - name: Check formatting + if: matrix.check_format run: mix format --check-formatted - name: Compile (warnings as errors) From 5a5840f72ae1170c96d1de18a3880d881c436a5b Mon Sep 17 00:00:00 2001 From: David Bernheisel Date: Sun, 4 Oct 2026 21:03:36 -0400 Subject: [PATCH 3/6] Conformance: report each failed check when a run fails The suite's summary names only the scenario, which is not enough to diagnose a failure that occurs only in CI. --- test/support/conformance.ex | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/test/support/conformance.ex b/test/support/conformance.ex index ab5188b..4051abf 100644 --- a/test/support/conformance.ex +++ b/test/support/conformance.ex @@ -58,15 +58,36 @@ defmodule Phantom.Test.Conformance do end defp npx(args) do + results = + Path.join(System.tmp_dir!(), "phantom-conformance-#{System.unique_integer([:positive])}") + {output, status} = - System.cmd("npx", ["conformance", "server", "--timeout", "15000" | args], + System.cmd( + "npx", + ["conformance", "server", "--timeout", "15000", "--output-dir", results | args], stderr_to_stdout: true, env: [{"NO_COLOR", "1"}, {"FORCE_COLOR", "0"}] ) + output = if status == 0, do: output, else: output <> failed_checks(results) + File.rm_rf(results) {status, output} end + # The summary names only the scenario, so the report adds each failed check. + defp failed_checks(results) do + failures = + for path <- Path.wildcard(Path.join(results, "*/checks.json")), + check <- path |> File.read!() |> JSON.decode!(), + check["status"] == "FAILURE" do + scenario = path |> Path.dirname() |> Path.basename() + details = if check["details"], do: "\n " <> JSON.encode!(check["details"]), else: "" + " #{scenario} #{check["id"]}: #{check["errorMessage"]}#{details}" + end + + Enum.join(["\nFailed checks (expected failures included):" | failures], "\n") + end + # The CLI takes one baseline file, so the topology's additions are appended # to the revision's entries in a temporary copy. defp baseline(topology, revision) do From 7008ec7b854c53d222aae82f4c9c8a7164cdb214 Mon Sep 17 00:00:00 2001 From: David Bernheisel Date: Sun, 4 Oct 2026 21:08:54 -0400 Subject: [PATCH 4/6] Route client responses to the waiting session through PubSub A response to a server-initiated request, such as elicitation/create, could reach a node whose Phantom.Tracker had not yet replicated the waiting request. The lookup gave up after 2 seconds and dropped the response, so the elicitation hung. The waiting session process now subscribes to a topic for the request, and the response is broadcast to it. The Tracker lookup remains for servers without PubSub. --- lib/phantom/router.ex | 12 ++++++---- lib/phantom/session.ex | 8 +++++++ lib/phantom/tracker.ex | 32 +++++++++++++++++++++++++++ test/phantom/client_response_test.exs | 27 ++++++++++++++++++++++ 4 files changed, 75 insertions(+), 4 deletions(-) create mode 100644 test/phantom/client_response_test.exs diff --git a/lib/phantom/router.ex b/lib/phantom/router.ex index 927709e..c9d8844 100644 --- a/lib/phantom/router.ex +++ b/lib/phantom/router.ex @@ -678,13 +678,13 @@ defmodule Phantom.Router do def dispatch_method(_method, _params, %{id: request_id, response: %{} = response}, session) when is_binary(request_id) do - Phantom.Router.route_client_response(request_id, response) + Phantom.Router.route_client_response(session.pubsub, request_id, response) {:reply, nil, session} end def dispatch_method(_method, _params, %{id: request_id, error: %{} = error}, session) when is_binary(request_id) do - Phantom.Router.route_client_response(request_id, {:error, error}) + Phantom.Router.route_client_response(session.pubsub, request_id, {:error, error}) {:reply, nil, session} end @@ -1560,8 +1560,9 @@ defmodule Phantom.Router do @doc false # Deliver the client's response to a server-initiated request (a result - # map, or `{:error, error}`) to the process waiting on it. - def route_client_response(request_id, response) do + # map, or `{:error, error}`) to the process waiting on it. PubSub reaches it + # at once on any node; a Tracker lookup can miss an unreplicated request. + def route_client_response(nil, request_id, response) do require Logger case await_request_meta(request_id) do @@ -1576,6 +1577,9 @@ defmodule Phantom.Router do end end + def route_client_response(pubsub, request_id, response), + do: Phantom.Tracker.cast_client_response(pubsub, request_id, response) + @doc false # Wait for a tracked request to become visible via Tracker replication. # In a distributed setup, the request may be tracked on one node but diff --git a/lib/phantom/session.ex b/lib/phantom/session.ex index 3d2c90c..315114c 100644 --- a/lib/phantom/session.ex +++ b/lib/phantom/session.ex @@ -718,6 +718,8 @@ defmodule Phantom.Session do {request, ref} = Phantom.Elicit.prepare_request(state.session.id, tool_call_id, elicitation) + Phantom.Tracker.subscribe_client_response(state.session.pubsub, request.id) + caller = %{from: from, request_id: request.id} state = @@ -1173,6 +1175,12 @@ defmodule Phantom.Session do {:noreply, Map.put(state, :workers, workers)} end + def handle_info({:phantom_client_response, request_id, response}, state) do + Phantom.Tracker.unsubscribe_client_response(state.session.pubsub, request_id) + Phantom.Tracker.untrack_request(request_id) + {:noreply, route_local_elicit_response({request_id, response}, state)} + end + def handle_info({:phantom_elicitation_response, ref, response}, state) do case pop_in(state, [:elicitation_callers, ref]) do {nil, state} -> diff --git a/lib/phantom/tracker.ex b/lib/phantom/tracker.ex index 81b21d4..477d3d6 100644 --- a/lib/phantom/tracker.ex +++ b/lib/phantom/tracker.ex @@ -355,6 +355,8 @@ defmodule Phantom.Tracker do "phantom:request:" <> Base.url_encode64(key, padding: false) end + defp client_response_topic(request_id), do: "phantom:client_response:#{request_id}" + if @available do @doc false def subscribe_session(nil, _session_id), do: :ok @@ -403,6 +405,33 @@ defmodule Phantom.Tracker do ) end + @doc false + def subscribe_client_response(nil, _request_id), do: :ok + + def subscribe_client_response(pubsub, request_id) do + Phoenix.PubSub.subscribe(pubsub, client_response_topic(request_id)) + rescue + ArgumentError -> :ok + end + + @doc false + def unsubscribe_client_response(nil, _request_id), do: :ok + + def unsubscribe_client_response(pubsub, request_id) do + Phoenix.PubSub.unsubscribe(pubsub, client_response_topic(request_id)) + rescue + ArgumentError -> :ok + end + + @doc false + def cast_client_response(pubsub, request_id, response) do + Phoenix.PubSub.broadcast( + pubsub, + client_response_topic(request_id), + {:phantom_client_response, request_id, response} + ) + end + @doc false def subscribe_request(nil, _session_id, _request_id), do: :ok @@ -434,6 +463,9 @@ defmodule Phantom.Tracker do def subscribe_listen(_pubsub, _principal, _request_id), do: :ok def cast_listen(_pubsub, _principal, _request_id, _message), do: :ok def subscribe_request(_pubsub, _session_id, _request_id), do: :ok + def subscribe_client_response(_pubsub, _request_id), do: :ok + def unsubscribe_client_response(_pubsub, _request_id), do: :ok + def cast_client_response(_pubsub, _request_id, _response), do: :ok def cast_request(_pubsub, _session_id, _request_id, _message), do: :ok end diff --git a/test/phantom/client_response_test.exs b/test/phantom/client_response_test.exs new file mode 100644 index 0000000..6f181c6 --- /dev/null +++ b/test/phantom/client_response_test.exs @@ -0,0 +1,27 @@ +defmodule Phantom.ClientResponseTest do + use ExUnit.Case, async: true + + @pubsub Phantom.ClientResponseTest.PubSub + + setup do + start_supervised!({Phoenix.PubSub, name: @pubsub}) + :ok + end + + # No `Phantom.Tracker` runs here, so delivery cannot depend on its replication. + test "routes a client response to the waiting process through PubSub" do + Phantom.Tracker.subscribe_client_response(@pubsub, "elicit-1") + + Phantom.Router.route_client_response(@pubsub, "elicit-1", %{"action" => "accept"}) + assert_receive {:phantom_client_response, "elicit-1", %{"action" => "accept"}} + + Phantom.Router.route_client_response(@pubsub, "elicit-1", {:error, %{"code" => -1}}) + assert_receive {:phantom_client_response, "elicit-1", {:error, %{"code" => -1}}} + end + + test "does not deliver a response for another request" do + Phantom.Tracker.subscribe_client_response(@pubsub, "elicit-1") + Phantom.Router.route_client_response(@pubsub, "elicit-2", %{"action" => "accept"}) + refute_receive {:phantom_client_response, _, _} + end +end From 60e623c5cdb6e11728246bcd732dcaea5a281c6e Mon Sep 17 00:00:00 2001 From: David Bernheisel Date: Sun, 4 Oct 2026 21:17:19 -0400 Subject: [PATCH 5/6] Plug: refuse an SSE GET whose Accept gives text/event-stream q=0 The most specific matching media range decides, as RFC 9110 specifies, so "*/*;q=0.5, text/event-stream;q=0" refuses the stream. --- lib/phantom/plug.ex | 27 +++++++++++++++++++++++---- test/phantom/plug_test.exs | 28 ++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 4 deletions(-) diff --git a/lib/phantom/plug.ex b/lib/phantom/plug.ex index 4ddc199..b2c3f97 100644 --- a/lib/phantom/plug.ex +++ b/lib/phantom/plug.ex @@ -386,13 +386,32 @@ defmodule Phantom.Plug do true accept -> - accept - |> Enum.flat_map(&String.split(&1, ",")) - |> Enum.map(&(&1 |> String.split(";") |> hd() |> String.trim() |> String.downcase())) - |> Enum.any?(&(&1 in ~w[text/event-stream text/* */*])) + ranges = accept |> Enum.flat_map(&String.split(&1, ",")) |> Enum.map(&media_range/1) + + # The most specific matching range decides, and q=0 refuses it (RFC 9110 §12.5.1). + case Enum.find_value(~w[text/event-stream text/* */*], &List.keyfind(ranges, &1, 0)) do + {_type, q} -> q > 0 + nil -> false + end end end + defp media_range(range) do + [type | params] = + range |> String.split(";") |> Enum.map(&(&1 |> String.trim() |> String.downcase())) + + {type, Enum.find_value(params, 1.0, &quality/1)} + end + + defp quality("q=" <> value) do + case Float.parse(value) do + {q, ""} -> q + _invalid -> 1.0 + end + end + + defp quality(_param), do: nil + defp validate_protocol_request(%Plug.Conn{halted: true} = conn), do: conn defp validate_protocol_request(%Plug.Conn{method: "POST"} = conn) do diff --git a/test/phantom/plug_test.exs b/test/phantom/plug_test.exs index bba4bba..b0f9bb2 100644 --- a/test/phantom/plug_test.exs +++ b/test/phantom/plug_test.exs @@ -581,6 +581,34 @@ defmodule Phantom.PlugTest do assert error["error"]["code"] == -32600 end + test "GET request that refuses text/event-stream with q=0 returns 406" do + for accept <- [ + "text/event-stream;q=0", + "application/json, text/event-stream; q=0.0", + "text/*;q=0", + "*/*;q=0.5, text/event-stream;q=0" + ] do + :get + |> conn("/mcp") + |> put_req_header("accept", accept) + |> call() + + assert_receive {:conn, conn} + assert conn.status == 406, "expected 406 for Accept: #{accept}" + end + end + + test "GET request accepting text/event-stream with a nonzero q opens a stream" do + for accept <- ["text/event-stream;q=0.5", "application/json, */*;q=0.1"] do + :get + |> conn("/mcp") + |> put_req_header("accept", accept) + |> call() + + assert_sse_connected() + end + end + test "GET request accepting any media type opens a stream" do :get |> conn("/mcp") From aa628f92fa8237925969f57fe41999a9ab75ee8b Mon Sep 17 00:00:00 2001 From: David Bernheisel Date: Sun, 4 Oct 2026 21:17:19 -0400 Subject: [PATCH 6/6] Tracker: repeat a session call until the session stream answers A client can send logging/setLevel or resources/subscribe before its GET stream subscribes on another node. The single broadcast reached no stream, so the call failed with "Connection closed" or "SSE stream not open". The call now repeats with backoff until a stream answers or the timeout passes. --- lib/phantom/tracker.ex | 20 ++++++++++++-- test/phantom/session_call_test.exs | 44 ++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 3 deletions(-) create mode 100644 test/phantom/session_call_test.exs diff --git a/lib/phantom/tracker.ex b/lib/phantom/tracker.ex index 477d3d6..7ef0e2e 100644 --- a/lib/phantom/tracker.ex +++ b/lib/phantom/tracker.ex @@ -377,17 +377,31 @@ defmodule Phantom.Tracker do # stream answer. reply_to = Process.alias([:reply]) call = {:"$gen_call", {reply_to, reply_to}, message} + deadline = System.monotonic_time(:millisecond) + timeout + await_session_reply(pubsub, session_id, call, reply_to, deadline, 50) + end + + # A client may send a request before its GET stream subscribes, and a + # broadcast reaches only current subscribers, so the call repeats. + defp await_session_reply(pubsub, session_id, call, reply_to, deadline, interval) do Phoenix.PubSub.broadcast_from(pubsub, self(), session_topic(session_id), call) + remaining = deadline - System.monotonic_time(:millisecond) receive do {^reply_to, reply} -> reply after - timeout -> - Process.unalias(reply_to) - :error + max(min(interval, remaining), 0) -> + if remaining > interval, + do: await_session_reply(pubsub, session_id, call, reply_to, deadline, interval * 2), + else: give_up(reply_to) end end + defp give_up(reply_to) do + Process.unalias(reply_to) + :error + end + @doc false def subscribe_listen(nil, _principal, _request_id), do: :ok diff --git a/test/phantom/session_call_test.exs b/test/phantom/session_call_test.exs new file mode 100644 index 0000000..2591f3b --- /dev/null +++ b/test/phantom/session_call_test.exs @@ -0,0 +1,44 @@ +defmodule Phantom.SessionCallTest do + use ExUnit.Case, async: true + + @pubsub Phantom.SessionCallTest.PubSub + + setup do + start_supervised!({Phoenix.PubSub, name: @pubsub}) + :ok + end + + defp start_stream(session_id, delay) do + spawn_link(fn -> + Process.sleep(delay) + Phantom.Tracker.subscribe_session(@pubsub, session_id) + + receive do + {:"$gen_call", from, message} -> GenServer.reply(from, {:handled, message}) + end + + Process.sleep(:infinity) + end) + end + + test "reaches a session stream that subscribes after the call starts" do + start_stream("late", 300) + + assert Phantom.Tracker.call_session(@pubsub, "late", :set_level, 2_000) == + {:handled, :set_level} + end + + test "returns the first reply once" do + start_stream("ready", 0) + Process.sleep(50) + + assert Phantom.Tracker.call_session(@pubsub, "ready", :set_level, 2_000) == + {:handled, :set_level} + + refute_receive _ + end + + test "is an error when no stream answers in time" do + assert Phantom.Tracker.call_session(@pubsub, "missing", :set_level, 300) == :error + end +end