diff --git a/lib/acp_runtime/client.ex b/lib/acp_runtime/client.ex
index 2d4f4b4..9476557 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
 
@@ -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 =
       [
@@ -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}
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/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/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()
 
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