feat: create initial execution allowances by owner

apps/fountain/lib/fountain/conversations.ex

@@ -1151,6 +1151,50 @@ defmodule Fountain.Conversations do
     end
   end
 
+  @doc """
+  Save an initial resolved allowance once, scoped to its conversation owner.
+
+  Internal persistence only: the caller must resolve trusted current ceilings
+  and prove runtime support before admission. This function does not admit work
+  or reset an active turn. No launch or HTTP path calls it yet. A duplicate
+  fails without replacing the saved policy; use `narrow_execution_allowance/3`
+  for subsequent changes. Ownership stays locked through insertion.
+  """
+  def create_execution_allowance(conversation_id, user_id, resolved_limits, opts \\ []) do
+    result =
+      Repo.transaction(fn ->
+        Repo.one(
+          from c in Conversation,
+            where: c.id == ^conversation_id and c.user_id == ^user_id,
+            select: c.id,
+            lock: "FOR SHARE"
+        ) || Repo.rollback(:not_found)
+
+        case conversation_id
+             |> ExecutionAllowance.new_changeset(resolved_limits)
+             |> Repo.insert() do
+          {:ok, allowance} -> allowance
+          {:error, changeset} -> Repo.rollback(changeset)
+        end
+      end)
+
+    with {:ok, allowance} <- result do
+      Audit.record(%{
+        user_id: user_id,
+        action: "conversation.execution_allowance_created",
+        resource_type: "conversation",
+        resource_id: conversation_id,
+        actor: Keyword.get(opts, :actor, "self"),
+        request_ip: Keyword.get(opts, :request_ip),
+        metadata: %{
+          "controls" => Enum.filter(ExecutionLimits.keys(), &Map.has_key?(allowance.limits, &1))
+        }
+      })
+
+      {:ok, allowance}
+    end
+  end
+
   @doc """
   Narrow an existing allowance owned by `user_id`, retaining omitted controls.
 

apps/fountain/lib/fountain/conversations/execution_allowance.ex

@@ -3,7 +3,8 @@ defmodule Fountain.Conversations.ExecutionAllowance do
   Versioned storage for a conversation's resolved execution allowance.
 
   This is a storage primitive, not an admission or enforcement API. Use the
-  owner-scoped `Conversations.narrow_execution_allowance/3` for existing records.
+  owner-scoped `Conversations.create_execution_allowance/3` for initial records
+  and `Conversations.narrow_execution_allowance/3` for subsequent changes.
   Initial admission must establish conversation ownership,
   resolve current ceilings and prove runtime support before saving an allowance.
   Later turns and recovery must consult it before this becomes a usable setting.

apps/fountain/test/fountain/audit_guardrail_test.exs

@@ -57,6 +57,8 @@ defmodule Fountain.AuditGuardrailTest do
     {"inference credential clear", &__MODULE__.do_cred_clear/1, "inference_credential.delete"},
     {"conversation delete", &__MODULE__.do_conv_delete/1, "conversation.deleted"},
     {"conversation caller tools", &__MODULE__.do_caller_tools/1, "conversation.caller_tools_set"},
+    {"allowance creation", &__MODULE__.do_allowance_creation/1,
+     "conversation.execution_allowance_created"},
     {"allowance narrowing", &__MODULE__.do_allowance_narrowing/1,
      "conversation.execution_allowance_narrowed"},
     {"sandbox reset", &__MODULE__.do_sandbox_reset/1, "sandbox.reset"},
@@ -390,6 +392,11 @@ defmodule Fountain.AuditGuardrailTest do
     {:ok, _} = Conversations.delete_conversation(conv)
   end
 
+  def do_allowance_creation(user) do
+    conv = insert_conversation(user_id: user.id)
+    {:ok, _} = Conversations.create_execution_allowance(conv.id, user.id, %{})
+  end
+
   def do_allowance_narrowing(user) do
     conv = insert_conversation(user_id: user.id)
 

apps/fountain/test/fountain/conversations/execution_allowance_test.exs

@@ -231,6 +231,114 @@ defmodule Fountain.Conversations.ExecutionAllowanceTest do
     refute Repo.exists?(from e in Fountain.Billing.UsageEvent, where: e.user_id == ^conv.user_id)
   end
 
+  test "scoped creation hides foreign and missing conversations before validation", %{
+    conversation: conv
+  } do
+    other = insert_conversation()
+
+    for limits <- [%{}, %{"private-field" => "private-value"}],
+        {id, user_id} <- [{conv.id, other.user_id}, {Ecto.UUID.generate(), conv.user_id}] do
+      assert Conversations.create_execution_allowance(id, user_id, limits) == {:error, :not_found}
+    end
+
+    assert Repo.get(Allowance, conv.id) == nil
+    assert Repo.get(Allowance, other.id) == nil
+    assert creation_events(conv) == []
+  end
+
+  test "scoped creation records only control names and leaves active work unchanged", %{
+    conversation: conv
+  } do
+    before = Repo.reload!(conv)
+    turn = insert_turn(conv, status: "running")
+    sandbox = Repo.reload!(conv.sandbox)
+    other = insert_conversation() |> Map.fetch!(:id) |> insert_allowance()
+
+    assert {:ok, allowance} =
+             Conversations.create_execution_allowance(
+               conv.id,
+               conv.user_id,
+               %{max_model_turns: 2, wall_time_seconds: 30, max_estimated_cost_usd: 0.25},
+               actor: "api",
+               request_ip: "192.0.2.1"
+             )
+
+    assert allowance.limits == %{
+             "max_model_turns" => 2,
+             "wall_time_seconds" => 30,
+             "max_estimated_cost_usd" => 0.25
+           }
+
+    assert Repo.reload!(allowance) == allowance
+    assert Repo.reload!(conv) == before
+    assert Repo.reload!(turn) == turn
+    assert Repo.reload!(sandbox) == sandbox
+    assert Repo.reload!(other) == other
+    assert [event] = creation_events(conv)
+    assert event.user_id == conv.user_id
+    assert event.actor == "api"
+    assert event.request_ip == "192.0.2.1"
+
+    assert event.metadata == %{
+             "controls" => ["wall_time_seconds", "max_model_turns", "max_estimated_cost_usd"]
+           }
+  end
+
+  test "scoped creation cannot replace an existing policy even with omission", %{
+    conversation: conv
+  } do
+    allowance = insert_allowance(conv.id)
+
+    for limits <- [nil, %{}, %{max_model_turns: 1}, %{max_model_turns: 20}] do
+      assert {:error, changeset} =
+               Conversations.create_execution_allowance(conv.id, conv.user_id, limits)
+
+      assert errors_on(changeset).conversation_id == ["has already been taken"]
+      assert Repo.reload!(allowance) == allowance
+    end
+
+    assert creation_events(conv) == []
+  end
+
+  test "scoped creation refuses invalid controls without a row or audit", %{conversation: conv} do
+    for limits <- [
+          %{"private-field" => "private-value"},
+          %{max_model_turns: 0},
+          %{wall_time_seconds: nil}
+        ] do
+      assert {:error, changeset} =
+               Conversations.create_execution_allowance(conv.id, conv.user_id, limits)
+
+      errors = Jason.encode!(errors_on(changeset))
+      refute errors =~ "private-field"
+      refute errors =~ "private-value"
+      assert errors_on(changeset).limits != []
+    end
+
+    assert Repo.get(Allowance, conv.id) == nil
+    assert creation_events(conv) == []
+  end
+
+  test "a saved initial allowance does not grant unsupported execution", %{conversation: conv} do
+    assert {:ok, allowance} =
+             Conversations.create_execution_allowance(conv.id, conv.user_id, %{max_model_turns: 2})
+
+    assert {:error, {:execution_limits_unsupported, ["max_model_turns"]}} =
+             Fountain.Conversations.TurnMachine.open(conv.id, conv.sandbox_id, "refused")
+
+    assert Repo.reload!(allowance).limits == %{"max_model_turns" => 2}
+    assert Conversations._unsafe_list_turns(conv.id) == []
+    refute Repo.exists?(from e in Fountain.Billing.UsageEvent, where: e.user_id == ^conv.user_id)
+  end
+
+  defp creation_events(conv),
+    do:
+      Repo.all(
+        from e in Fountain.Audit.Event,
+          where:
+            e.resource_id == ^conv.id and e.action == "conversation.execution_allowance_created"
+      )
+
   defp allowance_events(conv),
     do:
       Repo.all(
@@ -269,6 +377,90 @@ defmodule Fountain.Conversations.ExecutionAllowanceRaceTest do
     race(:disjoint)
   end
 
+  for mode <- [:duplicate, :ownership] do
+    test "initial creation rechecks #{mode} after a PostgreSQL lock wait" do
+      creation_race(unquote(mode))
+    end
+  end
+
+  defp creation_race(mode) do
+    Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn ->
+      users =
+        for _ <- 1..2,
+            do:
+              Repo.insert!(%Fountain.Accounts.User{
+                email: "initial-allowance-race-#{Ecto.UUID.generate()}@example.test"
+              })
+
+      [user, other] = users
+      conv = insert_conversation(user_id: user.id)
+      owner = self()
+
+      winner =
+        independent_writer(fn ->
+          Repo.transaction(fn ->
+            value =
+              case mode do
+                :duplicate -> insert_allowance(conv.id)
+                :ownership -> conv |> Ecto.Changeset.change(user_id: other.id) |> Repo.update!()
+              end
+
+            send(owner, :changed)
+
+            receive do
+              :commit -> value
+            after
+              5_000 -> raise "commit barrier timed out"
+            end
+          end)
+        end)
+
+      try do
+        assert_receive :changed, 5_000
+
+        loser =
+          independent_writer(fn ->
+            Conversations.create_execution_allowance(conv.id, user.id, %{})
+          end)
+
+        try do
+          assert_receive {:backend, winner_pid, winner_backend}, 5_000
+          assert winner_pid == winner.pid
+          assert_receive {:backend, loser_pid, loser_backend}, 5_000
+          assert loser_pid == loser.pid
+          refute winner_backend == loser_backend
+          await_blocked(loser_backend, System.monotonic_time(:millisecond) + 5_000)
+          send(winner.pid, :commit)
+          assert {:ok, saved} = Task.await(winner)
+
+          case {mode, Task.await(loser)} do
+            {:duplicate, {:error, changeset}} ->
+              assert errors_on(changeset).conversation_id == ["has already been taken"]
+              assert Repo.get!(Allowance, conv.id) == saved
+
+            {:ownership, {:error, :not_found}} ->
+              assert Repo.reload!(conv).user_id == other.id
+              assert Repo.get(Allowance, conv.id) == nil
+          end
+
+          refute Repo.exists?(
+                   from e in Fountain.Audit.Event,
+                     where:
+                       e.resource_id == ^conv.id and
+                         e.action == "conversation.execution_allowance_created"
+                 )
+        after
+          Task.shutdown(loser, :brutal_kill)
+        end
+      after
+        Task.shutdown(winner, :brutal_kill)
+        for user <- users, do: Repo.delete!(user)
+        Repo.get!(Sandbox, conv.sandbox_id) |> Repo.delete!()
+        assert Repo.get(Allowance, conv.id) == nil
+      end
+    end)
+  end
+
   defp race(mode) do
     Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn ->
       # These rows must be committed: sharing the test's sandbox connection would