Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 29 additions & 3 deletions lib/acp_runtime/client.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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 =
[
Expand Down Expand Up @@ -74,7 +79,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}
Expand Down
8 changes: 7 additions & 1 deletion lib/acp_runtime/process_transport_broker.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
7 changes: 6 additions & 1 deletion lib/acp_runtime/session.ex
Original file line number Diff line number Diff line change
Expand Up @@ -376,14 +376,19 @@ 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,
owner,
state.agent,
state.working_directory,
environment,
state.client_options
opts
)
end
end
Expand Down
128 changes: 128 additions & 0 deletions test/acp_runtime/client_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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!({
Expand Down Expand Up @@ -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
24 changes: 24 additions & 0 deletions test/acp_runtime/process_transport_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down
26 changes: 26 additions & 0 deletions test/acp_runtime/session_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down
5 changes: 3 additions & 2 deletions test/support/session_fakes.ex
Original file line number Diff line number Diff line change
@@ -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
Expand Down
Loading