Skip to content

refactor(conversations): Launch, part c: attach_conversation and its checks #2245

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jhgaylor merged 1 commit into from
Sep 15, 2026
Merged

refactor(conversations): Launch, part c: attach_conversation and its checks #2245

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
250 changes: 1 addition & 249 deletions apps/fountain/lib/fountain/conversations.ex
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2919,7 +2919,7 @@ defmodule Fountain.Conversations do
def resolve_sandbox_api_access(_access, _mode), do: {:error, :invalid_sandbox_api_access}

# ADR 0045's machine isolation is a claim about a machine, not about a row:
# a `none` conversation must be alone on a fresh one. `check_sandbox_api_attach/2`
# a `none` conversation must be alone on a fresh one. `Launch.check_sandbox_api_attach/2`
# answers that by asking which conversations point at `sandbox.id`, so it can
# only see machines Fountain knows it is sharing. A caller-supplied name can
# name a machine that already exists — the provider adopts it rather than
Expand All @@ -2932,21 +2932,6 @@ defmodule Fountain.Conversations do

def check_sandbox_api_name(_access, _name), do: :ok

defp check_sandbox_api_attach(sandbox, access) do
# A fresh none launch must never inherit another conversation's credential,
# and attaching an owner conversation must not inject one into its machine.
has_none =
Repo.exists?(
from(c in Conversation,
where: c.sandbox_id == ^sandbox.id and c.sandbox_api_access == "none"
)
)

if access in [nil, "owner"] and not has_none,
do: :ok,
else: {:error, :invalid_sandbox_api_access}
end

# The launch's sandbox mode: the agent's default unless the launch names
# one (ADR 0023). Not an allowlisted override like `environment_id` — the
# mode is not a security boundary; the tenant scope on the sandbox is.
Expand Down Expand Up @@ -3596,164 +3581,6 @@ defmodule Fountain.Conversations do

defp reset_message(_owner), do: "The sandbox was reset by its owner. " <> @reset_tail

# A conversation on a machine the caller already has (ADR 0023 gate 3).
#
# The launch is resolved exactly as a fresh one — agent, vault, environment,
# permission policy, parent, the account and billing gates — and then the
# sandbox is fetched tenant-scoped and checked instead of created: it must
# be `ready` or `suspended`, it must have been built for the same agent,
# environment and vault, and the runtime that shaped its disk must be the
# agent's runtime still. No quota reservation: nothing new is provisioned,
# and waking a `suspended` machine goes through the quota gate on the first
# prompt as every wake does. The conversation is opened `idle` with no
# server; a prompt supplied here is delivered through the ordinary wake
# path, so a `ready` machine reattaches and a `suspended` one resumes, and
# if that delivery is refused the row is removed again so a refused request
# creates nothing.
# A door for `Fountain.Conversations.Launch` (#2217), whose fresh-start
# `start_conversation/2` clause calls it for the "sandbox_id" attach; not
# part of the context's public surface.
@doc false
def attach_conversation(
sandbox_id,
%{"agent_id" => agent_id, "user_id" => user_id} = attrs,
opts
)
when is_binary(user_id) do
with :ok <- require_provider_commit_boundary(),
:ok <- Fountain.Conversations.PromptInput.validate_initial(attrs),
%Agents.Agent{} = agent <- Agents.get_agent(agent_id, user_id) || {:error, :not_found},
:ok <- check_execution_limits(user_id, attrs["execution_limits"]),
{:ok, _runtime_module} <- Fountain.RuntimeDispatch.for_agent(agent),
{:ok, vault_id} <- resolve_vault_id(attrs["vault_id"], user_id, agent),
{:ok, env_id} <- resolve_environment_id(attrs["environment_id"], user_id, agent),
{:ok, cred_set_id} <-
resolve_inference_credential_id(attrs["inference_credential_id"], user_id, agent),
{:ok, perm_policy} <- resolve_permission_policy(attrs["permission_policy"], agent),
{:ok, parent_id} <- resolve_parent_id(attrs["parent_conversation_id"], user_id),
:ok <- Fountain.Accounts.check_not_suspended(user_id),
:ok <- Fountain.Billing.check_spend(user_id),
{:ok, inference_source} <-
Launch.resolve_admission_inference(user_id, agent, env_id, vault_id, cred_set_id),
%Sandbox{} = sandbox <- get_sandbox(sandbox_id, user_id) || {:error, :sandbox_not_found},
:ok <- check_sandbox_api_attach(sandbox, attrs["sandbox_api_access"]),
:ok <- check_attachable(sandbox, agent, vault_id, env_id),
:ok <- check_attach_capacity(sandbox, agent, attrs["prompt"]),
{:ok, conv} <-
create_attached_conversation(
%{
sandbox_id: sandbox.id,
agent_id: agent.id,
# Ownership: agent came from the scoped get_agent above.
agent_version_id: Agents._unsafe_current_version_id(agent.id),
vault_id: vault_id,
environment_id: env_id,
inference_credential_id: inference_source.set_id,
inference_source: Source.dump(inference_source),
user_id: user_id,
runtime: agent.runtime,
status: "idle",
source: attrs["source"] || "api",
parent_conversation_id: parent_id,
channel_id: attrs["channel_id"],
title: attrs["title"],
permission_policy: perm_policy,
# The bridge's tools (#1202) ride on both create paths: this
# one is what a home sandbox's second conversation takes.
caller_tools: attrs["caller_tools"] || [],
labels: attrs["labels"] || %{}
},
attrs["execution_limits"],
opts
) do
Audit.record(%{
user_id: user_id,
action: "conversation.created",
resource_type: "conversation",
resource_id: conv.id,
actor: Keyword.get(opts, :actor, "self"),
request_ip: Keyword.get(opts, :request_ip),
metadata: %{
"agent_id" => agent.id,
"agent_name" => agent.name,
"source" => conv.source,
"with_prompt" => is_binary(attrs["prompt"]) and attrs["prompt"] != "",
"parent_conversation_id" => parent_id,
"sandbox_attached" => sandbox.id
}
})

broadcast_sidebar_update(user_id)
deliver_attach_prompt(conv, attrs, opts)
else
nil -> {:error, :not_found}
{:error, _} = err -> err
end
end

# Commit policy with the new conversation, before analytics, audit or prompt
# delivery. Lock its owners and recheck ceilings after the early preflight.
defp create_attached_conversation(attrs, request, opts) do
result =
Repo.transaction(fn ->
:ok = InferenceCredentials.lock_source(attrs.user_id)
# Reservation re-reads the parent before the Codex sandbox row. Take
# the same machine lock before this path's earlier sandbox row lock.
Repo.query!("SELECT pg_advisory_xact_lock($1, $2)", [
@sandbox_lock_namespace,
:erlang.phash2(attrs.sandbox_id)
])

# Deliberately unlocked. `users` is the row every credit posting takes
# `FOR UPDATE` (`Credits.insert_and_move/3` holds it across a ledger
# insert, lot consumption and the balance move), so locking it here
# would park admission behind an unrelated billing transaction. This
# read is an ownership recheck; the ceiling below is read the same way,
# and the insert's foreign keys are what actually enforce integrity.
Repo.one(
from u in Fountain.Accounts.User,
where: u.id == ^attrs.user_id,
select: u.id
) || Repo.rollback(:not_found)

agent =
Repo.one(
from a in Agents.Agent,
where: a.id == ^attrs.agent_id and a.user_id == ^attrs.user_id,
lock: "FOR SHARE"
) || Repo.rollback(:not_found)

case Launch.unbind_rotated_channel(attrs, opts) do
:ok -> :ok
{:error, reason} -> Repo.rollback(reason)
end

sandbox =
Repo.one(
from s in Sandbox,
where: s.id == ^attrs.sandbox_id and s.user_id == ^attrs.user_id,
lock: "FOR NO KEY UPDATE"
) || Repo.rollback(:not_found)

with :ok <- check_attachable(sandbox, agent, attrs.vault_id, attrs.environment_id),
{:ok, limits} <- resolve_admission_limits(attrs.user_id, request),
{:ok, conv} <- insert_conversation_row(attrs),
:ok <- reserve_inference(conv),
{:ok, allowance} <-
conv.id |> ExecutionAllowance.new_changeset(limits) |> Repo.insert() do
{conv, allowance}
else
{:error, reason} -> Repo.rollback(reason)
end
end)

with {:ok, {conv, allowance}} <- result do
after_conversation_created(conv)
record_execution_allowance_created(allowance, conv.user_id, opts)
{:ok, conv}
end
end

# A nested transaction does not commit: workers and provider calls must not
# escape a caller's transaction that can still roll back the accepted rows.
# Public (door for `Fountain.Conversations.Wake`, #2211, and
Expand All @@ -3765,81 +3592,6 @@ defmodule Fountain.Conversations do
if Repo.in_transaction?(), do: {:error, :provider_transaction_open}, else: :ok
end

defp deliver_attach_prompt(conv, attrs, opts) do
prompt = attrs["prompt"]

if is_binary(prompt) and prompt != "" do
case ConversationServer.send_prompt(conv.id, prompt, attrs["images"] || [], opts) do
:ok ->
{:ok, _unsafe_get_conversation!(conv.id)}

{:error, _} = err ->
# Nothing ran. Take the row back so a refused request created
# nothing, exactly like a refused fresh launch.
Launch.restore_rotated_channel(conv, opts)
_ = Repo.delete(conv)
broadcast_sidebar_update(conv.user_id)
err
end
else
{:ok, _unsafe_get_conversation!(conv.id)}
end
end

defp check_attachable(%Sandbox{reset_requested_at: at}, _agent, _vault_id, _env_id)
when not is_nil(at),
do: {:error, :sandbox_reset_pending}

defp check_attachable(%Sandbox{status: status}, _agent, _vault_id, _env_id)
when status not in ["ready", "suspended"],
do: {:error, {:sandbox_not_attachable, status}}

defp check_attachable(%Sandbox{} = sandbox, %Agents.Agent{} = agent, vault_id, env_id) do
cond do
sandbox.agent_id != agent.id ->
{:error, :sandbox_identity_mismatch}

sandbox.vault_id != vault_id ->
{:error, :sandbox_identity_mismatch}

sandbox.environment_id != (env_id || agent.environment_id) ->
{:error, :sandbox_identity_mismatch}

# The disk was shaped by the runtime that first ran on it; an agent
# whose runtime changed since gets a new machine, not this one.
_unsafe_sandbox_runtime(sandbox.id) not in [nil, agent.runtime] ->
{:error, :sandbox_runtime_mismatch}

true ->
:ok
end
end

# With a prompt, the attach is a turn start too, so the capacity rule of
# step 4 applies at the door; without one, the later prompt is gated by
# `ConversationServer` as any prompt is.
defp check_attach_capacity(%Sandbox{} = sandbox, %Agents.Agent{runtime: runtime}, prompt)
when is_binary(prompt) and prompt != "" do
capacity = Fountain.RuntimeDispatch.concurrency(runtime)

if _unsafe_sandbox_at_capacity?(sandbox.id, nil, capacity),
do: {:error, :sandbox_at_capacity},
else: :ok
end

defp check_attach_capacity(_sandbox, _agent, _prompt), do: :ok

@doc "The runtime of the newest conversation on `sandbox_id`, or nil when it has none."
def _unsafe_sandbox_runtime(sandbox_id) when is_binary(sandbox_id) do
Repo.one(
from c in Conversation,
where: c.sandbox_id == ^sandbox_id,
order_by: [desc: c.inserted_at, desc: c.id],
limit: 1,
select: c.runtime
)
end

@doc "One of the caller's sandboxes, or nil. A foreign or malformed id reads as nil."
def get_sandbox(id, user_id) when is_binary(id) and is_binary(user_id) do
case Ecto.UUID.dump(id) do
Expand Down
Loading
Loading