Refuse conversation release while remote execution is unresolved

apps/fountain/lib/fountain/conversations.ex

@@ -1193,6 +1193,23 @@ defmodule Fountain.Conversations do
     |> notify_parent_change()
   end
 
+  @doc "Terminate the owned conversation only when no turn or remote execution remains open."
+  def _unsafe_release_conversation(conversation_id, opts \\ []) do
+    # ownership: the lifecycle client/actor received an already-owned conversation.
+    ExecutionGuard._unsafe_release_parent(
+      conversation_id,
+      fn current ->
+        current |> Conversation.changeset(%{status: "terminated"}) |> Repo.update()
+      end,
+      opts
+    )
+    |> notify_parent_change()
+    |> case do
+      {:ok, _} -> :ok
+      error -> error
+    end
+  end
+
   defp write_turn_parent(turn, mode, attrs) do
     # ownership: the calling actor/recovery path already owns this exact turn.
     result =

apps/fountain/lib/fountain/conversations/conversation_server.ex

@@ -372,22 +372,19 @@ defmodule Fountain.Conversations.ConversationServer do
   takes the `sandbox_id`, and its first prompt reattaches through the
   ordinary wake path — a new runtime session on the same disk.
 
-  `{:error, :busy}` while a turn is running; nothing is interrupted. With no
-  server alive the row alone is marked, the same as `terminate_conversation/2`.
+  `{:error, :busy}` while a turn runs on a **live** server; nothing is
+  interrupted. With no server alive that row is as likely an orphan (see
+  `Conversations.wake_for_interrupt/1`), so release proceeds. Unresolved
+  bounded execution answers `{:error, :execution_fenced}` either way: a
+  durable fact rather than an inference, and bounded (ADR 0046).
+
   Audited as `conversation.released` unless `audit: false`.
   """
   def release_conversation(conv_id, opts \\ []) do
     result =
       case whereis(conv_id) do
         nil ->
-          case Conversations._unsafe_get_conversation(conv_id) do
-            nil ->
-              {:error, :not_running}
-
-            conv ->
-              {:ok, _} = Conversations.update_conversation(conv, %{status: "terminated"})
-              :ok
-          end
+          Conversations._unsafe_release_conversation(conv_id, actor_alive?: false)
 
         pid ->
           call_server(pid, :release_conv)
@@ -1652,11 +1649,15 @@ defmodule Fountain.Conversations.ConversationServer do
   end
 
   def handle_call(:release_conv, _from, state) do
-    state = drop_connection(state, "released")
-    conv = Conversations._unsafe_get_conversation!(state.conversation_id)
-    {:ok, _} = Conversations.update_conversation(conv, %{status: "terminated"})
-    Output.publish_stage(state.conversation_id, "terminate", "done", %{event: "released"})
-    {:stop, :normal, :ok, %{state | handle: nil}}
+    case Conversations._unsafe_release_conversation(state.conversation_id) do
+      :ok ->
+        state = drop_connection(state, "released")
+        Output.publish_stage(state.conversation_id, "terminate", "done", %{event: "released"})
+        {:stop, :normal, :ok, %{state | handle: nil}}
+
+      {:error, _} = error ->
+        {:reply, error, state}
+    end
   end
 
   # A notification for the revision this server already holds is a no-op: it
@@ -2411,8 +2412,7 @@ defmodule Fountain.Conversations.ConversationServer do
     state = %{state | current_turn: nil}
 
     # A bounded turn that never started still holds a journal and a transport.
-    # Closing here is the retirement intent, not a confirmed remote stop — the
-    # coordinator owns the confirmation.
+    # Closing is retirement intent, not a confirmed stop; the coordinator confirms.
     if state.turn_execution, do: close_bounded_connection(state), else: state
   end
 

apps/fountain/lib/fountain/conversations/execution_guard.ex

@@ -117,6 +117,40 @@ defmodule Fountain.Conversations.ExecutionGuard do
     end)
   end
 
+  @doc "Release only a durably idle parent; refusal never retires or interrupts execution."
+  def _unsafe_release_parent(conversation_id, writer, opts \\ []) do
+    transaction(fn ->
+      conv = lock_parent(conversation_id) || Repo.rollback(:not_running)
+
+      # A `running` turn row is evidence of a live turn only when there is a
+      # server to run it. Without one it is as likely an orphan — a deploy, a
+      # Horde rebalance or a plain `{:stop, :normal, _}` left it behind, which
+      # `wake_for_interrupt/1` spells out — and release is what an owner
+      # reaches for in exactly that state. Refusing there took away a release
+      # that always worked. The caller says whether a server is alive.
+      if Keyword.get(opts, :actor_alive?, true) do
+        running? =
+          Repo.exists?(
+            from t in Turn, where: t.conversation_id == ^conversation_id and t.status == "running"
+          )
+
+        if running?, do: Repo.rollback(:busy)
+      end
+
+      # The durable fence is unconditional. An unresolved bounded execution
+      # means a remote command may still be running, and releasing would drop
+      # the row that says so. This one has an age rather than being permanent
+      # — `_unsafe_retire_unresolved/2` writes it off — so the refusal is
+      # bounded, unlike the running-turn check it used to sit beside.
+      if open_execution?(conversation_id), do: Repo.rollback(:execution_fenced)
+
+      case writer.(conv) do
+        {:ok, updated} -> {%{applied: true, conversation: updated}, nil, nil}
+        {:error, reason} -> Repo.rollback(reason)
+      end
+    end)
+  end
+
   @doc "Find the immutable journal for an already-owned actor's turn."
   def _unsafe_for_turn(turn_id), do: Repo.get_by(TurnExecution, turn_id: turn_id)
 

apps/fountain/test/fountain/conversations/release_fence_actor_test.exs

@@ -0,0 +1,38 @@
+defmodule Fountain.Conversations.ReleaseFenceActorTest do
+  use Fountain.ConversationServerCase
+
+  alias Fountain.Conversations.ExecutionGuard
+
+  test "an idle actor refuses unresolved remote work before closing anything" do
+    user = insert_verified_user()
+    sandbox = insert_sandbox(user_id: user.id, status: "ready")
+    conv = insert_conversation(user_id: user.id, sandbox: sandbox, status: "idle")
+    stub_happy_sprite(sandbox.sprite_name)
+    {pid, _monitor, :alive} = start_server(conv)
+    on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end)
+    turn = insert_turn(conv, status: "running")
+
+    {:ok, execution} =
+      ExecutionGuard._unsafe_register(
+        turn.id,
+        Ecto.UUID.generate(),
+        DateTime.add(DateTime.utc_now(), 60)
+      )
+
+    {:ok, _} = ExecutionGuard._unsafe_claim_spawn(execution.id)
+    {:ok, _} = ExecutionGuard._unsafe_interrupt(conv.id)
+    before = :sys.get_state(pid)
+    assert is_nil(before.current_turn)
+    parent = Conversations._unsafe_get_conversation!(conv.id)
+    key = Repo.get!(Fountain.Accounts.ApiKey, parent.callback_api_key_id)
+    events = Conversations._unsafe_list_log_events(conv.id)
+
+    assert {:error, :execution_fenced} = GenServer.call(pid, :release_conv)
+    assert :sys.get_state(pid) == before
+    assert Conversations._unsafe_get_conversation!(conv.id) == parent
+    assert Repo.reload!(key) == key
+    assert Conversations._unsafe_list_log_events(conv.id) == events
+    assert ExecutionGuard._unsafe_for_turn(turn.id).state == "awaiting_identity"
+    assert Repo.reload!(sandbox).status == "ready"
+  end
+end

apps/fountain/test/fountain/conversations/release_fence_test.exs

@@ -0,0 +1,153 @@
+defmodule Fountain.Conversations.ReleaseFenceTest do
+  use Fountain.DataCase, async: true
+
+  alias Fountain.Conversations
+  alias Fountain.Conversations.{ConversationServer, ExecutionGuard, TurnExecution}
+
+  setup do
+    user = insert_verified_user()
+    sandbox = insert_sandbox(user_id: user.id, status: "ready")
+    conv = insert_conversation(user_id: user.id, sandbox: sandbox, status: "idle")
+    %{conv: Repo.reload!(conv), sandbox: sandbox}
+  end
+
+  test "an orphaned running turn does not block release", c do
+    # This test used to assert the opposite, and the opposite took away a
+    # release that always worked. With no server alive a `running` turn row is
+    # as likely an orphan as a live turn — a deploy, a Horde rebalance or a
+    # plain `{:stop, :normal, _}` leaves one behind, which is what
+    # `wake_for_interrupt/1` exists for — and release is what an owner reaches
+    # for in exactly that state. Inferring "busy" from the row there fences
+    # the owner out of their own recovery with nothing to un-fence it.
+    turn = insert_turn(c.conv, status: "running")
+    assert is_nil(ConversationServer.whereis(c.conv.id))
+    assert :ok = ConversationServer.release_conversation(c.conv.id)
+    assert Conversations._unsafe_get_conversation!(c.conv.id).status == "terminated"
+    # Release terminates the parent; it does not rewrite the turn's history.
+    assert Repo.reload!(turn).status == "running"
+  end
+
+  test "a live actor's running turn still refuses release", c do
+    # The row is only authoritative when something is there to run it.
+    turn = insert_turn(c.conv, status: "running")
+
+    assert {:error, :busy} =
+             Conversations._unsafe_release_conversation(c.conv.id, actor_alive?: true)
+
+    assert Conversations._unsafe_get_conversation!(c.conv.id).status == "idle"
+    assert Repo.reload!(turn).status == "running"
+  end
+
+  test "a durable execution fence refuses release with or without an actor", c do
+    execution = execution(c, "ready")
+
+    # Unlike the running-turn inference, this one is a fact the journal holds,
+    # so it refuses either way — and it is bounded, because the coordinator
+    # writes an obligation off once nothing can resolve it.
+    assert {:error, :execution_fenced} =
+             Conversations._unsafe_release_conversation(c.conv.id, actor_alive?: false)
+
+    assert {:error, :execution_fenced} =
+             Conversations._unsafe_release_conversation(c.conv.id, actor_alive?: true)
+
+    assert Repo.get!(TurnExecution, execution.id).state == "ready"
+    assert Conversations._unsafe_get_conversation!(c.conv.id).status == "idle"
+  end
+
+  for state <- ~w(active awaiting_identity ready submitted uncertain) do
+    @state state
+    test "release refuses #{@state} execution without changing it", c do
+      execution = execution(c, @state)
+      turn = Repo.get!(Conversations.Turn, execution.turn_id)
+      assert is_nil(ConversationServer.whereis(c.conv.id))
+      assert {:error, :execution_fenced} = ConversationServer.release_conversation(c.conv.id)
+      assert Repo.reload!(execution) == execution
+      assert Repo.reload!(turn) == turn
+      assert Repo.reload!(c.conv) == c.conv
+      assert Repo.reload!(c.sandbox) == c.sandbox
+    end
+  end
+
+  test "confirmed cleanup permits release and later admission cannot revive the parent", c do
+    execution = execution(c, "submitted")
+
+    # Synthetic cleanup acknowledgment; no provider deletion is claimed.
+    {:ok, _} =
+      ExecutionGuard._unsafe_record_termination(execution.id, execution.attempt_id, :ok)
+
+    assert :ok = ConversationServer.release_conversation(c.conv.id)
+    assert Repo.reload!(c.conv).status == "terminated"
+    assert Repo.reload!(c.sandbox) == c.sandbox
+
+    attrs = %{
+      conversation_id: c.conv.id,
+      turn_number: 2,
+      prompt: "too late",
+      status: "running",
+      started_at: DateTime.utc_now()
+    }
+
+    assert {:error, :not_running} =
+             Conversations._unsafe_create_turn_on_sandbox(attrs, c.sandbox.id, :unbounded)
+  end
+
+  test "an unrelated co-tenant may keep working when this idle conversation releases", c do
+    other =
+      insert_conversation(user_id: c.conv.user_id, sandbox: c.sandbox, status: "running")
+
+    turn = insert_turn(other, status: "running")
+    other = Repo.reload!(other)
+    assert :ok = ConversationServer.release_conversation(c.conv.id)
+    assert Repo.reload!(other) == other
+    assert Repo.reload!(turn) == turn
+    assert Repo.reload!(c.sandbox) == c.sandbox
+  end
+
+  test "a deleted parent returns not_running", c do
+    Repo.delete!(c.conv)
+    assert {:error, :not_running} = ConversationServer.release_conversation(c.conv.id)
+  end
+
+  defp execution(c, state) do
+    turn = insert_turn(c.conv, status: "running")
+
+    {:ok, execution} =
+      ExecutionGuard._unsafe_register(
+        turn.id,
+        Ecto.UUID.generate(),
+        DateTime.add(DateTime.utc_now(), 60)
+      )
+
+    if state != "active" do
+      {:ok, _} = ExecutionGuard._unsafe_claim_spawn(execution.id)
+
+      if state != "awaiting_identity" do
+        {:ok, _} =
+          ExecutionGuard._unsafe_bind_identity(
+            execution.id,
+            execution.connection_id,
+            "synthetic-release-command"
+          )
+      end
+
+      {:ok, _} = ExecutionGuard._unsafe_interrupt(c.conv.id)
+
+      if state in ["submitted", "uncertain"] do
+        {:ok, %{execution: claimed}} = ExecutionGuard._unsafe_claim_termination(execution.id)
+
+        if state == "uncertain" do
+          {:ok, _} =
+            ExecutionGuard._unsafe_record_termination(
+              execution.id,
+              claimed.attempt_id,
+              {:error, :timeout}
+            )
+        end
+      end
+    end
+
+    result = Repo.get!(TurnExecution, execution.id)
+    assert result.state == state
+    result
+  end
+end

decisions/0046-durable-turn-deadlines.md

@@ -164,6 +164,22 @@ explicit policy. Uncertainty must never be erased by transcript deletion — but
 it must not be permanent either, which is what the ageing exit above settles.
 The cutoff itself is the supervisor's to choose and is not fixed here.
 
+### Release refuses on a fact, not an inference
+
+Releasing a conversation refuses while a bounded execution is unresolved,
+because that is a durable fact and the journal row saying so would be dropped
+with the parent. It does **not** refuse on a `running` turn row when no server
+is alive: there, the row is as likely an orphan as a live turn — a deploy, a
+Horde rebalance or a plain `{:stop, :normal, _}` leaves one behind — and
+release is what an owner reaches for in exactly that state. Inferring "busy"
+from the row fenced the owner out of their own recovery with nothing able to
+un-fence it, which is the same failure as the permanent reset refusal above.
+A live server keeps the row authoritative and still refuses.
+
+The two refusals therefore say different things: `:busy` is a turn a live
+actor is running, and `:execution_fenced` is remote work Fountain cannot yet
+account for. The second has an age; the first ends on its own.
+
 ### The fence suppresses; it never rewrites
 
 A retired bounded turn's output events are not written and not broadcast: