From f77954c92e1cad84969421c8eb2b124eacb92af6 Mon Sep 17 00:00:00 2001 From: Tim Mecklem Date: Thu, 27 Aug 2026 01:27:32 +0000 Subject: [PATCH 1/2] Resume a persisted ACP session instead of always starting a new one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Session restarted from its persisted row — after the local runtime crashed, or was lost to a restart — always called session/new. The agent still held the conversation, so the user was handed an empty transcript for work that was still there, and the old session was left stranded in the agent. start_session/7 now takes a :resume_session_id, which Session fills in from the persisted acp_session_id. When present, the client calls session/load so the agent replays the conversation as session/update notifications. The load is best-effort: an agent that doesn't advertise loadSession, or one that has since expired the session, falls back to session/new rather than failing the connect. A resume that can't be honoured still has to leave the user with a working agent. --- lib/acp_runtime/client.ex | 25 +++++- lib/acp_runtime/session.ex | 7 +- test/acp_runtime/client_test.exs | 128 ++++++++++++++++++++++++++++++ test/acp_runtime/session_test.exs | 26 ++++++ test/support/session_fakes.ex | 5 +- 5 files changed, 186 insertions(+), 5 deletions(-) diff --git a/lib/acp_runtime/client.ex b/lib/acp_runtime/client.ex index 2d4f4b4..444a9dc 100644 --- a/lib/acp_runtime/client.ex +++ b/lib/acp_runtime/client.ex @@ -13,7 +13,7 @@ defmodule ACPRuntime.Client do with {:ok, client} <- start_client(context, target, cwd, owner, agent, env, opts) do {agent_info, agent_capabilities} = metadata(client) send(owner, {:acp_client_initialized, client, agent_info, agent_capabilities}) - open_session(client, cwd, agent_info, agent_capabilities) + open_session(client, cwd, agent_info, agent_capabilities, opts[:resume_session_id]) end end @@ -74,7 +74,28 @@ defmodule ACPRuntime.Client do DynamicSupervisor.start_child(supervisor, child_spec) end - defp open_session(client, cwd, agent_info, agent_capabilities) do + # A session that already carries an ACP session id is being resumed rather + # than started: the agent still holds that conversation, so `session/load` + # restores it (replaying the history as `session/update` notifications) + # instead of handing the user a fresh, empty transcript for work that is + # still there. + # + # The load is best-effort. An agent that doesn't advertise `loadSession`, or + # one that has since expired the session, must still leave the user with a + # working agent — so those fall back to opening a new session rather than + # failing the connect outright. + defp open_session(client, cwd, agent_info, agent_capabilities, resume_session_id) + when is_binary(resume_session_id) and resume_session_id != "" do + case ExMCPClient.load_session(client, resume_session_id, cwd, timeout: @new_session_timeout) do + {:ok, _result} -> + {:ok, client, resume_session_id, agent_info, agent_capabilities} + + {:error, _reason} -> + open_session(client, cwd, agent_info, agent_capabilities, nil) + end + end + + defp open_session(client, cwd, agent_info, agent_capabilities, _resume_session_id) do case ExMCPClient.new_session(client, cwd, timeout: @new_session_timeout) do {:ok, %{"sessionId" => session_id}} when is_binary(session_id) -> {:ok, client, session_id, agent_info, agent_capabilities} diff --git a/lib/acp_runtime/session.ex b/lib/acp_runtime/session.ex index 093ec2e..5bbe878 100644 --- a/lib/acp_runtime/session.ex +++ b/lib/acp_runtime/session.ex @@ -376,6 +376,11 @@ defmodule ACPRuntime.Session do defp connect(state, owner) do with {:ok, environment} <- state.credential_provider.credential_env(state.context, state.agent) do + # A persisted `acp_session_id` means this session already ran: the local + # runtime died and is being restarted from its row. Hand the id down so + # the client reclaims that conversation instead of opening a new one. + opts = Keyword.put(state.client_options, :resume_session_id, state.acp_session_id) + state.client_module.start_session( state.context, state.environment, @@ -383,7 +388,7 @@ defmodule ACPRuntime.Session do state.agent, state.working_directory, environment, - state.client_options + opts ) end end diff --git a/test/acp_runtime/client_test.exs b/test/acp_runtime/client_test.exs index a438886..6b6206f 100644 --- a/test/acp_runtime/client_test.exs +++ b/test/acp_runtime/client_test.exs @@ -68,6 +68,123 @@ defmodule ACPRuntime.ClientTest do assert :ok = Client.stop(client, :ready) end + test "loads the existing session when one is being resumed" do + test_pid = self() + environment = %{id: unique_id()} + agent = %{executable: "agent-acp", args: ["serve"]} + + task = + start_supervised!({ + Task, + fn -> + result = + Client.start_session( + test_pid, + environment, + test_pid, + agent, + "/workspace/repository", + %{}, + supervisor: ACPRuntime.TestSessionSupervisor, + process_runner: TestProcessRunner, + resume_session_id: "acp-session-1" + ) + + send(test_pid, {:client_start_result, result}) + end + }) + + monitor = Process.monitor(task) + + assert_receive {:acp_process_started, broker, _environment_id, command, _opts, process} + send(broker, {:stdout, process, Enum.at(command, 4) <> "\n"}) + + assert_receive {:acp_process_write, initialize_json} + + send_response(broker, process, Jason.decode!(initialize_json)["id"], %{ + "protocolVersion" => 1, + "agentCapabilities" => %{"loadSession" => true}, + "agentInfo" => %{"name" => "fake", "title" => "Fake Agent"} + }) + + assert_receive {:acp_client_initialized, client, _info, _caps} + + # The agent still holds this conversation, so resuming has to load it rather + # than starting the user over with an empty transcript. + assert_receive {:acp_process_write, load_session_json} + load_session = Jason.decode!(load_session_json) + + assert load_session["method"] == "session/load" + assert load_session["params"]["sessionId"] == "acp-session-1" + assert load_session["params"]["cwd"] == "/workspace/repository" + + send_response(broker, process, load_session["id"], %{}) + + assert_receive {:client_start_result, {:ok, ^client, "acp-session-1", _info, _caps}} + assert_receive {:DOWN, ^monitor, :process, ^task, :normal} + assert :ok = Client.stop(client, :ready) + end + + test "falls back to a new session when the agent will not load the old one" do + test_pid = self() + environment = %{id: unique_id()} + agent = %{executable: "agent-acp", args: ["serve"]} + + task = + start_supervised!({ + Task, + fn -> + result = + Client.start_session( + test_pid, + environment, + test_pid, + agent, + "/workspace/repository", + %{}, + supervisor: ACPRuntime.TestSessionSupervisor, + process_runner: TestProcessRunner, + resume_session_id: "expired-session" + ) + + send(test_pid, {:client_start_result, result}) + end + }) + + monitor = Process.monitor(task) + + assert_receive {:acp_process_started, broker, _environment_id, command, _opts, process} + send(broker, {:stdout, process, Enum.at(command, 4) <> "\n"}) + + assert_receive {:acp_process_write, initialize_json} + + send_response(broker, process, Jason.decode!(initialize_json)["id"], %{ + "protocolVersion" => 1, + "agentCapabilities" => %{"loadSession" => true}, + "agentInfo" => %{"name" => "fake", "title" => "Fake Agent"} + }) + + assert_receive {:acp_client_initialized, client, _info, _caps} + + assert_receive {:acp_process_write, load_session_json} + load_session = Jason.decode!(load_session_json) + assert load_session["method"] == "session/load" + + # The agent has since expired the session. A resume it can't honour still + # has to leave the user with a working agent, not a failed connect. + send_error_response(broker, process, load_session["id"], -32_602, "Session not found") + + assert_receive {:acp_process_write, new_session_json} + new_session = Jason.decode!(new_session_json) + assert new_session["method"] == "session/new" + + send_response(broker, process, new_session["id"], %{"sessionId" => "acp-session-2"}) + + assert_receive {:client_start_result, {:ok, ^client, "acp-session-2", _info, _caps}} + assert_receive {:DOWN, ^monitor, :process, ^task, :normal} + assert :ok = Client.stop(client, :ready) + end + test "normalizes client exits and stops clients during initialization" do dead_client = start_supervised!({ @@ -108,5 +225,16 @@ defmodule ACPRuntime.ClientTest do send(broker, {:stdout, process, response <> "\n"}) end + defp send_error_response(broker, process, id, code, message) do + response = + Jason.encode!(%{ + "jsonrpc" => "2.0", + "id" => id, + "error" => %{"code" => code, "message" => message} + }) + + send(broker, {:stdout, process, response <> "\n"}) + end + defp unique_id, do: "test-#{System.unique_integer([:positive])}" end diff --git a/test/acp_runtime/session_test.exs b/test/acp_runtime/session_test.exs index f559af5..1439cb3 100644 --- a/test/acp_runtime/session_test.exs +++ b/test/acp_runtime/session_test.exs @@ -161,6 +161,32 @@ defmodule ACPRuntime.SessionTest do assert_receive {:acp_process_killed, ^environment_id, "remote-session-1"} end + test "resumes a persisted ACP session instead of starting a fresh one" do + # A session whose local runtime died — crashed, or lost to a restart — is + # restarted from its persisted row. The agent still holds the conversation, + # so reconnecting has to reclaim it rather than hand the user an empty + # transcript for work that is still there. + session = %{session_fixture() | acp_session_id: "acp-session-earlier", status: "ready"} + + pid = start_supervised!({Session, session_options(session)}) + + assert_receive {:acp_fake_client_options, opts} + assert opts[:resume_session_id] == "acp-session-earlier" + + assert_receive {:acp_status, %{status: :ready, acp_session_id: "acp-session-earlier"}} + assert %{status: :ready} = Session.session_state(pid) + end + + test "starts a new ACP session when there is nothing to resume" do + session = session_fixture() + + start_supervised!({Session, session_options(session)}) + + assert_receive {:acp_fake_client_options, opts} + assert opts[:resume_session_id] == nil + assert_receive {:acp_status, %{status: :ready, acp_session_id: "acp-session-1"}} + end + test "cancel notifies the client and cancels pending permission calls" do session = session_fixture() diff --git a/test/support/session_fakes.ex b/test/support/session_fakes.ex index 9df827d..f6ab650 100644 --- a/test/support/session_fakes.ex +++ b/test/support/session_fakes.ex @@ -1,13 +1,14 @@ defmodule ACPRuntime.ClientFake do @moduledoc false - def start_session(context, target, owner, agent, cwd, env, _opts) do + def start_session(context, target, owner, agent, cwd, env, opts) do test_pid = test_pid(context) send(test_pid, {:acp_fake_client_started, target, owner, agent, cwd, env}) + send(test_pid, {:acp_fake_client_options, opts}) send(owner, {:acp_transport_connected, %{ref: make_ref()}, test_pid}) send(owner, {:acp_transport_session_id, "remote-session-1"}) send(owner, {:acp_client_initialized, test_pid, %{"title" => "Fake Agent"}, %{}}) - {:ok, test_pid, "acp-session-1", %{"title" => "Fake Agent"}, %{}} + {:ok, test_pid, opts[:resume_session_id] || "acp-session-1", %{"title" => "Fake Agent"}, %{}} end def prompt(test_pid, session_id, prompt) do From d8e65d74bde9bf728cddd821b7a5991c2b28ba4f Mon Sep 17 00:00:00 2001 From: Tim Mecklem Date: Thu, 27 Aug 2026 02:41:45 +0000 Subject: [PATCH 2/2] Let the host label and address the agent's process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A host whose only record of a running agent is the environment itself needs that process addressable and tagged from the moment it spawns — not once the runner happens to report back, which leaves a window where a live agent is untagged and therefore undiscoverable. start_session/7 forwards :process_session_id, :process_name and :process_metadata to the process runner. All optional; runners that ignore them behave exactly as before. --- lib/acp_runtime/client.ex | 7 +++++- lib/acp_runtime/process_transport_broker.ex | 8 ++++++- test/acp_runtime/process_transport_test.exs | 24 +++++++++++++++++++++ 3 files changed, 37 insertions(+), 2 deletions(-) diff --git a/lib/acp_runtime/client.ex b/lib/acp_runtime/client.ex index 444a9dc..9476557 100644 --- a/lib/acp_runtime/client.ex +++ b/lib/acp_runtime/client.ex @@ -45,7 +45,12 @@ defmodule ACPRuntime.Client do process_runner = Keyword.fetch!(client_opts, :process_runner) transport_opts = - Keyword.take(client_opts, [:shell_startup_timeout]) + Keyword.take(client_opts, [ + :shell_startup_timeout, + :process_session_id, + :process_name, + :process_metadata + ]) opts = [ diff --git a/lib/acp_runtime/process_transport_broker.ex b/lib/acp_runtime/process_transport_broker.ex index f73e189..d984c9b 100644 --- a/lib/acp_runtime/process_transport_broker.ex +++ b/lib/acp_runtime/process_transport_broker.ex @@ -46,7 +46,13 @@ defmodule ACPRuntime.ProcessTransportBroker do detachable: true, owner: self(), working_directory: cwd, - environment: environment + environment: environment, + # The host may want the agent's process addressable and labelled + # on the environment from the moment it spawns, rather than only + # once the runner reports back. + session_id: Keyword.get(opts, :process_session_id), + name: Keyword.get(opts, :process_name), + metadata: Keyword.get(opts, :process_metadata) ) do Process.monitor(owner) shell_timer = Process.send_after(self(), :shell_startup_timeout, shell_timeout) diff --git a/test/acp_runtime/process_transport_test.exs b/test/acp_runtime/process_transport_test.exs index 2a64b1c..ddf6be0 100644 --- a/test/acp_runtime/process_transport_test.exs +++ b/test/acp_runtime/process_transport_test.exs @@ -105,6 +105,30 @@ defmodule ACPRuntime.ProcessTransportTest do refute_receive {:acp_process_started, _broker, _environment_id, _command, _opts, _process} end + test "hands the runner the session identity the caller chose for the agent" do + test_pid = self() + + start_supervised!({ + Task, + fn -> + ProcessTransport.connect( + transport_opts(test_pid, + process_session_id: "chosen-id", + process_name: "Claude Code", + process_metadata: %{"kind" => "acp"} + ) + ) + end + }) + + # The host has to be able to label and address the agent's process on the + # environment from the moment it spawns — that is the only record of it. + assert_receive {:acp_process_started, _broker, _environment_id, _command, opts, _process} + assert opts[:session_id] == "chosen-id" + assert opts[:name] == "Claude Code" + assert opts[:metadata] == %{"kind" => "acp"} + end + defp connect_transport do test_pid = self()