diff --git a/apps/fountain/lib/fountain/conversations.ex b/apps/fountain/lib/fountain/conversations.ex index 5dfb9131c..1481d4bbf 100644 --- a/apps/fountain/lib/fountain/conversations.ex +++ b/apps/fountain/lib/fountain/conversations.ex @@ -11,7 +11,18 @@ defmodule Fountain.Conversations do require Logger alias Fountain.Audit - alias Fountain.Conversations.{Blocks, Conversation, Labels, LogEvent, Sandbox, Turn, TurnImage} + + alias Fountain.Conversations.{ + Blocks, + Conversation, + DetachedRequest, + Labels, + LogEvent, + Sandbox, + Turn, + TurnImage + } + alias Fountain.Conversations.Reapply alias Fountain.Conversations.{ExecutionAllowance, ExecutionLimits} alias Fountain.Conversations.Lifecycle @@ -3942,6 +3953,11 @@ defmodule Fountain.Conversations do Audited as a decision about tenant-owned state, per 0013: the tool and the verdict, never the tool's input. + + A request that outlived its turn (#1635) is answered through the same door + and audited the same way. What differs is what the answer does: there is no + peer left to take it, so the request is resolved on the turn row and a new + turn is opened carrying it, which wakes a suspended sandbox on the way. """ @spec answer_permission_request(binary(), binary(), String.t(), String.t(), keyword()) :: :ok | {:error, term()} @@ -3949,38 +3965,234 @@ defmodule Fountain.Conversations do when is_binary(conv_id) and is_binary(user_id) do actor = Keyword.get(opts, :actor, "self") - cond do - actor == "sprite" -> + case {actor, get_conversation(conv_id, user_id)} do + {"sprite", _conv} -> {:error, :sprite_may_not_answer} - is_nil(get_conversation(conv_id, user_id)) -> + {_actor, nil} -> {:error, :not_found} - true -> - do_answer_permission(conv_id, user_id, request_id, option_id, opts) + # A conversation nobody can prompt cannot carry an answer back to the + # agent, so the request is left where it is rather than resolved into + # nothing. + {_actor, %Conversation{status: status}} when status not in ["idle", "running"] -> + {:error, :not_running} + + {_actor, conv} -> + do_answer_permission(conv, user_id, request_id, option_id, opts) + end + end + + # The detached row is looked at first, and deliberately. A turn that ended + # `waiting` (#1635) left the request on its row while the peer that raised + # it may still be idle on the sandbox holding the JSON-RPC id: asking the + # server first would answer a connection whose turn is over and report + # success, and the new turn that actually carries the answer would never + # open. + defp do_answer_permission(conv, user_id, request_id, option_id, opts) do + # Ownership: established by the tenant-scoped `get_conversation/2` in + # `answer_permission_request/5` immediately above this call. + case _unsafe_waiting_turn(conv.id, request_id) do + nil -> answer_held_permission(conv.id, user_id, request_id, option_id, opts) + turn -> answer_detached_permission(conv, turn, user_id, option_id, opts) end end - defp do_answer_permission(conv_id, user_id, request_id, option_id, opts) do + defp answer_held_permission(conv_id, user_id, request_id, option_id, opts) do case ConversationServer.answer_permission(conv_id, request_id, option_id) do :ok -> - Audit.record(%{ - user_id: user_id, - action: "conversation.permission_answered", - resource_type: "conversation", - resource_id: conv_id, - actor: Keyword.get(opts, :actor, "self"), - request_ip: Keyword.get(opts, :request_ip), - metadata: %{"request_id" => request_id, "option_id" => option_id} - }) - - :ok + record_permission_answered(conv_id, user_id, request_id, option_id, opts) {:error, _} = err -> err end end + # A request nobody is holding open any more: resolve the row, then open the + # turn that tells the agent. + # + # Every gate the wake path would apply is applied **first**, before the row + # is touched. Resolving and then failing to deliver loses the answer with + # nothing to retry from, and hands the caller a 409 that says somebody else + # answered — which is a lie about what happened. + defp answer_detached_permission(conv, turn, user_id, option_id, opts) do + request = turn.pending_permission + request_id = request["request_id"] + + if DetachedRequest.offered?(request, option_id) do + with :ok <- _unsafe_resume_gate(conv), + :ok <- _unsafe_resolve_detached_request(turn, "answered", option_id), + :ok <- + record_permission_answered( + turn.conversation_id, + user_id, + request_id, + option_id, + opts + ) do + resume_after_request(turn, request, "answered", option_id, opts) + end + else + {:error, :unknown_option} + end + end + + @doc """ + Whether a resume turn can be opened on this conversation right now (#1635). + + The gates the wake will run, run before the request row is resolved. The + three answers differ in what a caller should do about them: + + * `:ok` — go ahead. + * `{:error, :busy}` — a turn is running, so the resume turn cannot queue + behind it. Retry when the conversation is idle; the sweep does, a minute + later. + * `{:error, :gone}` — the conversation is over, so no turn will ever carry + the answer. + + Anything else is the account's own refusal (suspended, out of credit), and + is retryable once the account is not. + + WARNING: not scoped by owner. The answer door establishes ownership first; + the sweep is a system sweep. + """ + @spec _unsafe_resume_gate(Conversation.t() | binary()) :: :ok | {:error, term()} + def _unsafe_resume_gate(conv_id) when is_binary(conv_id) do + case _unsafe_get_conversation(conv_id) do + nil -> {:error, :gone} + conv -> _unsafe_resume_gate(conv) + end + end + + def _unsafe_resume_gate(%Conversation{} = conv) do + cond do + conv.status in ["terminated", "failed"] -> + {:error, :gone} + + conv.status != "idle" -> + {:error, :busy} + + true -> + with :ok <- Fountain.Accounts.check_not_suspended(conv.user_id) do + Fountain.Billing.check_spend(conv.user_id) + end + end + end + + defp record_permission_answered(conv_id, user_id, request_id, option_id, opts) do + Audit.record(%{ + user_id: user_id, + action: "conversation.permission_answered", + resource_type: "conversation", + resource_id: conv_id, + actor: Keyword.get(opts, :actor, "self"), + request_ip: Keyword.get(opts, :request_ip), + metadata: %{"request_id" => request_id, "option_id" => option_id} + }) + + :ok + end + + @doc """ + The turn a detached request is waiting on, or nil (#1635). + + WARNING: not scoped by owner. Call it after a tenant-scoped fetch of the + conversation, which is what `answer_permission_request/5` does. + """ + @spec _unsafe_waiting_turn(binary(), String.t()) :: Turn.t() | nil + def _unsafe_waiting_turn(conv_id, request_id) do + Turn + |> where([t], t.conversation_id == ^conv_id and t.waiting == true) + |> where([t], fragment("?->>'request_id' = ?", t.pending_permission, ^request_id)) + |> Repo.one() + end + + @doc """ + Every request this conversation is waiting on, oldest turn first (#1635). + + WARNING: not scoped by owner. Call it after a tenant-scoped fetch, which is + what `ConversationController.show/2` does. + """ + @spec _unsafe_list_pending_requests(binary()) :: [map()] + def _unsafe_list_pending_requests(conv_id) do + Turn + |> where([t], t.conversation_id == ^conv_id and t.waiting == true) + |> where([t], not is_nil(t.pending_permission)) + |> order_by([t], asc: t.turn_number) + |> Repo.all() + |> Enum.map(&DetachedRequest.to_json(&1.pending_permission, &1)) + end + + @doc """ + Take a detached request off its turn, once (#1635). + + First answer wins, and here that is enforced by the update itself rather + than by a process holding the request: the `where` names the request id the + caller read, so a second answer, the sweep and a client racing the sweep all + find nothing to update and get `{:error, :no_pending_permission}`. + + The `request`/`done` stage event is published by whoever won, exactly as the + in-turn path publishes it. + + WARNING: not scoped by owner. Both callers establish ownership first — the + answer door by fetching the conversation for the user, the sweep by being a + system sweep. + """ + @spec _unsafe_resolve_detached_request(Turn.t(), String.t(), String.t() | nil) :: + :ok | {:error, :no_pending_permission} + def _unsafe_resolve_detached_request(%Turn{} = turn, outcome, option_id) do + request_id = turn.pending_permission["request_id"] + + {count, _} = + Turn + |> where([t], t.id == ^turn.id and t.waiting == true) + |> where([t], fragment("?->>'request_id' = ?", t.pending_permission, ^request_id)) + |> Repo.update_all(set: [waiting: false, pending_permission: nil, permission_deadline: nil]) + + if count == 1 do + publish_stage(turn.conversation_id, "request", "done", %{ + request_id: request_id, + outcome: outcome, + option_id: option_id, + detached: true + }) + + :ok + else + {:error, :no_pending_permission} + end + end + + # The resolution reaches the agent as a new turn, because the peer that + # raised the request is gone and its JSON-RPC id with it. `send_prompt/4` + # wakes a suspended sandbox on the way, which is the whole point of letting + # the request outlive the turn. + defp resume_after_request(turn, request, outcome, option_id, opts) do + case ConversationServer.send_prompt( + turn.conversation_id, + DetachedRequest.resume_prompt(request, outcome, option_id), + [], + opts + ) do + :ok -> + :ok + + {:error, reason} -> + # The gates above passed and the row is already resolved, so this is a + # race rather than a refusal: something took the conversation between + # the two. Its own error would tell the caller to retry an answer that + # no longer exists, so it becomes one that says what actually + # happened. + Logger.warning( + "conv #{turn.conversation_id}: resolved detached request " <> + "#{request["request_id"]} but could not open the turn that carries " <> + "the answer: #{inspect(reason)}" + ) + + {:error, :answer_not_delivered} + end + end + @doc """ Record that the permission policy withheld a tool from a running agent. diff --git a/apps/fountain/test/fountain/conversations/conversation_server_lifetime_test.exs b/apps/fountain/test/fountain/conversations/conversation_server_lifetime_test.exs index d4625ba07..23bb92673 100644 --- a/apps/fountain/test/fountain/conversations/conversation_server_lifetime_test.exs +++ b/apps/fountain/test/fountain/conversations/conversation_server_lifetime_test.exs @@ -119,6 +119,50 @@ defmodule Fountain.Conversations.ConversationServerLifetimeTest do assert Fountain.Repo.reload(conv).status == "idle" end + test "a request that outlived its turn does not hold the sandbox open" do + # The whole point of #1635. A request held inside a running turn defers + # idle reclaim, which is why its ceiling has to sit under the idle + # bound; a detached one holds nothing, so the machine parks with the + # card still up and the answer wakes it. + {conv, sandbox} = aged_conversation(180) + stub_reattach() + reject(&Managoat.Sandbox.Sprites.destroy/1) + + turn = + insert_turn(conv, %{ + status: "completed", + waiting: true, + pending_permission: %{ + "request_id" => "7.abc", + "tool" => "Bash", + "options" => [%{"optionId" => "yes", "kind" => "allow_once"}] + }, + permission_deadline: + DateTime.utc_now() + |> DateTime.add(2 * 24 * 3600, :second) + |> DateTime.truncate(:second) + }) + + with_bounds([sandbox_idle_timeout_minutes: 60, sandbox_max_lifetime_hours: 24], fn -> + {pid, ref, :alive} = start_server(conv) + + :sys.replace_state(pid, fn state -> + %{state | last_activity_at: DateTime.add(DateTime.utc_now(), -7200, :second)} + end) + + send(pid, :lifecycle_check) + assert :normal = assert_stopped(ref) + end) + + assert Fountain.Repo.reload(sandbox).status == "suspended" + + # And the request survived the park, disk and row alike. + reloaded = Fountain.Repo.reload(turn) + assert reloaded.waiting + assert reloaded.pending_permission["request_id"] == "7.abc" + assert [%{request_id: "7.abc"}] = Conversations._unsafe_list_pending_requests(conv.id) + end + test "a recently active server is left running" do {conv, sandbox} = aged_conversation(180) stub_reattach() diff --git a/apps/fountain/test/fountain/conversations/detached_request_test.exs b/apps/fountain/test/fountain/conversations/detached_request_test.exs index 35bb88791..c7f64c241 100644 --- a/apps/fountain/test/fountain/conversations/detached_request_test.exs +++ b/apps/fountain/test/fountain/conversations/detached_request_test.exs @@ -11,14 +11,73 @@ defmodule Fountain.Conversations.DetachedRequestTest do """ use Fountain.DataCase, async: true + use Mimic - alias Fountain.Conversations.DetachedRequest + alias Fountain.Audit + alias Fountain.Conversations + alias Fountain.Conversations.{ConversationServer, DetachedRequest} @options [ %{"optionId" => "allow", "kind" => "allow_once", "name" => "Apply"}, %{"optionId" => "deny", "kind" => "reject_once", "name" => "Stop"} ] + defp waiting_conversation(opts \\ []) do + user = insert_verified_user() + agent = insert_agent(user_id: user.id, runtime: "claude") + + # Parked, which is where a conversation with a detached request spends + # its wait (0017): the answer has to wake it. + sandbox = insert_sandbox(user_id: user.id, sprite_name: "test-sprite", status: "suspended") + + conv = + insert_conversation(user_id: user.id, agent: agent, sandbox: sandbox, status: "idle") + + request = %{ + "request_id" => Keyword.get(opts, :request_id, "7.abc"), + "tool" => "Bash", + "options" => @options, + "asked_at" => DateTime.utc_now() |> DateTime.to_iso8601(), + "detached_timeout_ms" => 86_400_000 + } + + turn = insert_turn(conv, %{prompt: "apply the plan", status: "completed"}) + + {:ok, turn} = + Conversations._unsafe_update_turn(turn, %{ + waiting: true, + pending_permission: request, + permission_deadline: Keyword.get(opts, :deadline, hours_from_now(24)) + }) + + %{user: user, conv: conv, sandbox: sandbox, turn: turn, request: request} + end + + defp hours_from_now(hours) do + DateTime.utc_now() |> DateTime.add(hours * 3600, :second) |> DateTime.truncate(:second) + end + + # The wake path starts a server through Horde. Only the prompt it is handed + # matters here, so record it and start nothing. + defp record_wake do + test = self() + + stub(Horde.DynamicSupervisor, :start_child, fn _sup, _spec -> + {:ok, spawn(fn -> Process.sleep(:infinity) end)} + end) + + stub(ConversationServer, :queue_initial_prompt, fn _pid, prompt -> + send(test, {:resume_prompt, prompt}) + :ok + end) + + stub(Managoat.Sandbox.Sprites, :get, fn _handle -> + {:ok, %{status: :suspended, raw: %{name: "test-sprite"}}} + end) + + stub(Managoat.Sandbox.Sprites, :resume, fn handle -> {:ok, handle} end) + end + describe "the wire shape of the resume prompt" do test "is one line of JSON under the key a _meta field would use" do request = %{"request_id" => "7.abc", "tool" => "Bash"} @@ -127,4 +186,217 @@ defmodule Fountain.Conversations.DetachedRequestTest do assert DetachedRequest.timeout_ms(params, nil) > idle_ms end end + + describe "listing what a conversation waits on" do + test "GET-shaped data for every request that outlived a turn" do + %{conv: conv, turn: turn} = waiting_conversation() + + assert [request] = Conversations._unsafe_list_pending_requests(conv.id) + assert request.request_id == "7.abc" + assert request.tool == "Bash" + assert Enum.map(request.options, & &1["optionId"]) == ["allow", "deny"] + assert request.turn_id == turn.id + assert request.deadline + end + + test "a turn that ended without one is not listed" do + %{conv: conv, turn: turn} = waiting_conversation() + + {:ok, _} = + Conversations._unsafe_update_turn(turn, %{waiting: false, pending_permission: nil}) + + assert Conversations._unsafe_list_pending_requests(conv.id) == [] + end + end + + describe "answering" do + test "resolves the row, says done on the stream and opens the resume turn" do + %{user: user, conv: conv, turn: turn} = waiting_conversation() + record_wake() + + assert :ok = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow", + actor: "api" + ) + + reloaded = Repo.reload(turn) + refute reloaded.waiting + refute reloaded.pending_permission + refute reloaded.permission_deadline + + assert [event] = request_stages(conv.id, "done") + assert event["outcome"] == "answered" + assert event["option_id"] == "allow" + + assert_receive {:resume_prompt, prompt} + + assert %{"fountain/permission_answer" => %{"request_id" => "7.abc", "option_id" => "allow"}} = + Jason.decode!(prompt) + end + + test "the audit row carries the answerer, exactly as an in-turn answer does" do + %{user: user, conv: conv} = waiting_conversation() + record_wake() + + assert :ok = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow", + actor: "api", + request_ip: "203.0.113.9" + ) + + assert answered = + user.id + |> Audit.list_recent_for_user(50) + |> Enum.find(&(&1.action == "conversation.permission_answered")) + + assert answered.actor == "api" + assert answered.request_ip == "203.0.113.9" + assert answered.metadata["request_id"] == "7.abc" + assert answered.metadata["option_id"] == "allow" + end + + test "the first answer wins and the second is too late" do + %{user: user, conv: conv} = waiting_conversation() + record_wake() + + assert :ok = Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + + # Every "too late" is one 409 at the door, so which of the two reasons + # comes back is not a distinction a client can act on. + assert {:error, reason} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + + assert reason in [:no_pending_permission, :not_running] + end + + test "an option the agent never offered is refused rather than relayed" do + # The fail-closed rule the peer applies in turn, applied here where no + # peer is left to apply it. + %{user: user, conv: conv, turn: turn} = waiting_conversation() + + assert {:error, :unknown_option} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "made-up") + + assert Repo.reload(turn).waiting + end + + test "a sprite may not answer its own prompt, detached or not" do + %{user: user, conv: conv, turn: turn} = waiting_conversation() + + assert {:error, :sprite_may_not_answer} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow", + actor: "sprite" + ) + + assert Repo.reload(turn).waiting + end + + test "another tenant gets not_found rather than a hint" do + %{conv: conv} = waiting_conversation() + other = insert_verified_user() + + assert {:error, :not_found} = + Conversations.answer_permission_request(conv.id, other.id, "7.abc", "allow") + end + + test "a turn already running refuses the answer before the row is touched" do + # The resume turn cannot queue behind one, so refusing beats resolving + # the request into a prompt nobody delivers. + %{user: user, conv: conv, turn: turn} = waiting_conversation() + {:ok, _} = Conversations.update_conversation(conv, %{status: "running"}) + + assert {:error, :busy} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + + assert Repo.reload(turn).waiting + end + + test "a spent balance refuses the answer, and the request survives to be answered again" do + # The wake would refuse this anyway; running the gate first is what keeps + # the answer from being resolved into a prompt nobody delivers, and the + # caller from being told somebody else answered. + %{user: user, conv: conv, turn: turn} = waiting_conversation() + drain_credit(user) + + assert {:error, :insufficient_credits} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + + assert Repo.reload(turn).waiting + refute Repo.reload(turn).pending_permission == nil + + # Topped up, the same answer lands. + {:ok, _} = Fountain.Credits.grant(user.id, 500, "grant_admin", idempotency_key: "topup") + record_wake() + + assert :ok = Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + refute Repo.reload(turn).waiting + end + + test "a delivery that fails after the row is resolved says so in its own words" do + # The gates passed and something took the conversation in between. The + # answer is gone, so the caller must not be told to retry it, and must + # not be told somebody else answered either. + %{user: user, conv: conv, turn: turn} = waiting_conversation() + + stub(ConversationServer, :send_prompt, fn _id, _prompt, _images, _opts -> + {:error, :busy} + end) + + assert {:error, :answer_not_delivered} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + + refute Repo.reload(turn).waiting + end + + test "a terminated conversation cannot carry an answer, so the request is left alone" do + %{user: user, conv: conv, turn: turn} = waiting_conversation() + {:ok, _} = Conversations.update_conversation(conv, %{status: "terminated"}) + + assert {:error, :not_running} = + Conversations.answer_permission_request(conv.id, user.id, "7.abc", "allow") + + assert Repo.reload(turn).waiting + end + end + + describe "the row-level resolution" do + test "a second resolution of the same request finds nothing to update" do + %{turn: turn} = waiting_conversation() + + assert :ok = Conversations._unsafe_resolve_detached_request(turn, "answered", "allow") + + assert {:error, :no_pending_permission} = + Conversations._unsafe_resolve_detached_request(turn, "timeout", "deny") + end + + test "the stage event marks the resolution as a detached one" do + %{conv: conv, turn: turn} = waiting_conversation() + + assert :ok = Conversations._unsafe_resolve_detached_request(turn, "answered", "allow") + assert [event] = request_stages(conv.id, "done") + assert event["detached"] == true + end + end + + # `insert_verified_user/1` holds the $5 opening credit (ADR 0031), so + # refusal has to be arranged rather than assumed. Exactly the balance, so a + # later grant puts the account back above zero rather than into a hole no + # top-up in a test would fill. + defp drain_credit(user) do + balance = Repo.reload!(user).credit_balance_cents + + if balance > 0 do + {:ok, _} = + Fountain.Credits.debit(user.id, balance, "burn_turn", idempotency_key: "drain-#{user.id}") + end + + :ok + end + + defp request_stages(conv_id, state) do + conv_id + |> Conversations._unsafe_list_log_events() + |> Enum.filter(&(&1.kind == "stage" and &1.stage == "request" and &1.state == state)) + |> Enum.map(&Jason.decode!(&1.data)) + end end