OpenAgents

Commit

Make the credit race actually race

574f15ebecc3 · AtlantisPleb · · parent 118708f411b8 

Make the credit race actually race

THREAD-001 cites this file as proving the admission cap holds under
concurrency. It proved nothing of the kind: the case ran through
DataCase with async false, which starts the sandbox owner in shared
mode, so every Task.async_stream worker serialized on one connection.
The concurrent opens were savepoints inside a single transaction, not
competing transactions — and both FOR UPDATE locks in
OpenAgents.Threads could be deleted with the suite still green.

Each worker now checks out a real connection outside the sandbox and
cleans up after itself on exit. The locks are what hold the cap again,
and the mutation runs say so: deleting the admission-cap lock admits
both concurrent opens instead of one, deleting the mint-side lock
records two events instead of three, and restoring both returns the
suite to green. Ran twice to confirm the unsandboxed rows leave
nothing behind.

Production is unchanged; this is the proof catching up to the claim.

Built by a Devin child through the openagents coder's delegate tool;
44 thread tests green on two consecutive runs before landing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GoYpb8FEmdxVErsv7ABCYi
Co-Authored-By
Claude Fable 5 <noreply@anthropic.com>

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

pushed
by user · WAL seq 342 · 2026-08-25T05:34:11.874444Z

Changed files

  • modified test/openagents/threads/credit_race_test.exs

Diff

1 file changed, +113 -53

test/openagents/threads/credit_race_test.exs modified +113 -53

@@ -10,62 +10,64 @@ defmodule OpenAgents.Threads.CreditRaceTest do

10 10
11 11 
  What is deliberately not asserted is that overlapping live ceilings sum to
12 12 
  the remainder. A live grant's unspent headroom does not reserve credit,
13 
  because a delegated child thread must be mintable while its parent holds the
14 
  whole remaining balance as its ceiling — reserving headroom would refuse
13 
  because a delegated child thread must be mintable while its parent holds
14 
  the whole remaining balance as its ceiling — reserving headroom would refuse
15 15 
  every such child with `:credit_exhausted`. The joint exposure is instead
16 16 
  bounded by the admission cap, which is why the cap has to hold under
17 17 
  concurrency (THREAD-001).
18
19 
  These tests do not use the SQL sandbox. Each concurrent worker checks out a
20 
  real database connection, so the `FOR UPDATE` locks are exercised against
21 
  PostgreSQL and not flattened by a single shared sandbox connection.
18 22 
  """
19 23
20 
  use OpenAgents.DataCase, async: false
24 
  use ExUnit.Case, async: false
21 25
22 26 
  alias OpenAgents.Conversations
23 27 
  alias OpenAgents.Conversations.Visitor
24 28 
  alias OpenAgents.Inference
25 29 
  alias OpenAgents.Inference.Credit
26 30 
  alias OpenAgents.Inference.Grant
31 
  alias OpenAgents.Repo
27 32 
  alias OpenAgents.Threads
33 
  alias OpenAgents.Threads.Thread
34
35 
  import Ecto.Query
28 36
29 37 
  test "simultaneous opens at the admission boundary admit exactly one thread" do
30 38 
    cap(2)
31 
    visitor = visitor("cap-race")
39 
    visitor_id = fresh_visitor("cap-race")
32 40
33 
    {:ok, _first} = Threads.open(%Visitor{id: visitor}, "Already open")
41 
    unboxed(fn ->
42 
      {:ok, _first} = Threads.open(%Visitor{id: visitor_id}, "Already open")
43 
    end)
34 44
35 45 
    results =
36 
      ["Second", "Third"]
37 
      |> Task.async_stream(
38 
        fn objective -> Threads.open(%Visitor{id: visitor}, objective) end,
39 
        max_concurrency: 2,
40 
        ordered: false,
41 
        timeout: :infinity
42 
      )
43 
      |> Enum.map(fn {:ok, result} -> result end)
46 
      concurrent(["Second", "Third"], fn objective ->
47 
        Threads.open(%Visitor{id: visitor_id}, objective)
48 
      end)
44 49
45 50 
    assert Enum.count(results, &match?({:ok, _thread}, &1)) == 1
46 51 
    assert Enum.count(results, &match?({:error, :thread_quota_reached}, &1)) == 1
47 
    assert Threads.open_count(visitor) == 2
52 
    assert unboxed(fn -> Threads.open_count(visitor_id) end) == 2
48 53 
  end
49 54
50 55 
  test "simultaneous opens against an exhausted account jointly mint nothing" do
51 
    visitor = visitor("exhausted-race")
56 
    visitor_id = fresh_visitor("exhausted-race")
52 57
53 
    {:ok, _spent} =
54 
      Inference.record_usage(minted(visitor), %{
55 
        "output_tokens" => output_tokens_costing(Credit.allowance(visitor))
56 
      })
58 
    unboxed(fn ->
59 
      {:ok, _spent} =
60 
        Inference.record_usage(minted(visitor_id), %{
61 
          "output_tokens" => output_tokens_costing(Credit.allowance(visitor_id))
62 
        })
63 
    end)
57 64
58 
    assert Credit.remaining(visitor) == 0
65 
    assert unboxed(fn -> Credit.remaining(visitor_id) end) == 0
59 66
60 67 
    results =
61 
      ["First racer", "Second racer"]
62 
      |> Task.async_stream(
63 
        fn objective -> Threads.open_and_mint(%Visitor{id: visitor}, objective) end,
64 
        max_concurrency: 2,
65 
        ordered: false,
66 
        timeout: :infinity
67 
      )
68 
      |> Enum.map(fn {:ok, result} -> result end)
68 
      concurrent(["First racer", "Second racer"], fn objective ->
69 
        Threads.open_and_mint(%Visitor{id: visitor_id}, objective)
70 
      end)
69 71
70 72 
    assert Enum.all?(results, &match?({:error, :credit_exhausted}, &1))
71 73

@@ -73,29 +75,34 @@ defmodule OpenAgents.Threads.CreditRaceTest do

73 75 
    # The racers leave the account exactly one grant — the exhausted one that
74 76 
    # spent the credit — and no open thread: exhausting the grant abandoned
75 77 
    # its thread, and admission reaped it.
76 
    assert Threads.open_count(visitor) == 0
77 
    assert Repo.aggregate(from(g in Grant, where: g.owner_visitor_id == ^visitor), :count) == 1
78 
    assert unboxed(fn -> Threads.open_count(visitor_id) end) == 0
79
80 
    assert unboxed(fn ->
81 
             Repo.aggregate(from(g in Grant, where: g.owner_visitor_id == ^visitor_id), :count)
82 
           end) == 1
78 83 
  end
79 84
80 85 
  test "simultaneous opens mint the serialized remainder, the figure sequential opens mint" do
81 
    visitor = visitor("remainder-race")
86 
    visitor_id = fresh_visitor("remainder-race")
82 87 
    spent = 250_000
83 88
84 
    {:ok, _spent} =
85 
      Inference.record_usage(minted(visitor), %{"output_tokens" => output_tokens_costing(spent)})
89 
    unboxed(fn ->
90 
      {:ok, _spent} =
91 
        Inference.record_usage(minted(visitor_id), %{
92 
          "output_tokens" => output_tokens_costing(spent)
93 
        })
94 
    end)
86 95
87 
    remainder = Credit.allowance(visitor) - spent
88 
    assert Credit.remaining(visitor) == remainder
96 
    remainder = unboxed(fn -> Credit.allowance(visitor_id) - spent end)
97 
    assert unboxed(fn -> Credit.remaining(visitor_id) end) == remainder
89 98
90 99 
    grants =
91 
      ["First racer", "Second racer"]
92 
      |> Task.async_stream(
93 
        fn objective -> Threads.open_and_mint(%Visitor{id: visitor}, objective) end,
94 
        max_concurrency: 2,
95 
        ordered: false,
96 
        timeout: :infinity
97 
      )
98 
      |> Enum.map(fn {:ok, {:ok, _thread, grant, _token}} -> grant end)
100 
      concurrent(["First racer", "Second racer"], fn objective ->
101 
        case Threads.open_and_mint(%Visitor{id: visitor_id}, objective) do
102 
          {:ok, _thread, grant, _token} -> grant
103 
          other -> other
104 
        end
105 
      end)
99 106
100 107 
    # Each racer is ceiled at the metered remainder — the same ceiling the two
101 108 
    # opens mint in either serial order, because a mint spends nothing and a

@@ -106,23 +113,57 @@ defmodule OpenAgents.Threads.CreditRaceTest do

106 113 
  end
107 114
108 115 
  test "serial behavior is unchanged: the next thread is ceiled at what spend left behind" do
109 
    visitor = visitor("serial")
116 
    visitor_id = fresh_visitor("serial")
110 117 
    spent = 250_000
111 118
112 
    {:ok, first} = Threads.open(%Visitor{id: visitor}, "Spend a little")
113 
    {:ok, first, grant, _token} = Threads.mint_grant(first)
119 
    unboxed(fn ->
120 
      {:ok, first} = Threads.open(%Visitor{id: visitor_id}, "Spend a little")
121 
      {:ok, first, grant, _token} = Threads.mint_grant(first)
122 
      assert grant.max_cost_microusd == Credit.allowance(visitor_id)
123
124 
      {:ok, _spent} =
125 
        Inference.record_usage(grant, %{
126 
          "output_tokens" => output_tokens_costing(spent)
127 
        })
128
129 
      {:ok, _finished} = Threads.finish(first, %{report: "Done."})
130 
      {:ok, second} = Threads.open(%Visitor{id: visitor_id}, "Spend the rest")
131 
      {:ok, _second, next_grant, _token} = Threads.mint_grant(second)
132 
      assert next_grant.max_cost_microusd == Credit.allowance(visitor_id) - spent
133 
      :ok
134 
    end)
135 
  end
114 136
115 
    assert grant.max_cost_microusd == Credit.allowance(visitor)
137 
  test "concurrent appends on one thread count every event" do
138 
    visitor_id = fresh_visitor("event-race")
116 139
117 
    {:ok, _spent} =
118 
      Inference.record_usage(grant, %{"output_tokens" => output_tokens_costing(spent)})
140 
    thread =
141 
      unboxed(fn ->
142 
        {:ok, thread} = Threads.open(%Visitor{id: visitor_id}, "Event race")
143 
        thread
144 
      end)
145
146 
    results =
147 
      concurrent([%{"i" => "a"}, %{"i" => "b"}], fn payload ->
148 
        Threads.record_event(thread, "thread.race", payload)
149 
      end)
119 150
120 
    {:ok, _finished} = Threads.finish(first, %{report: "Done."})
151 
    assert Enum.all?(results, &match?({:ok, _thread}, &1))
121 152
122 
    {:ok, second} = Threads.open(%Visitor{id: visitor}, "Spend the rest")
123 
    {:ok, _second, next_grant, _token} = Threads.mint_grant(second)
153 
    {event_count, events} =
154 
      unboxed(fn ->
155 
        current = Repo.get!(Thread, thread.id)
156 
        {current.event_count, Threads.list_events(current)}
157 
      end)
124 158
125 
    assert next_grant.max_cost_microusd == Credit.allowance(visitor) - spent
159 
    assert event_count == 3
160 
    assert length(events) == 3
161 
  end
162
163 
  defp fresh_visitor(key) do
164 
    visitor_id = unboxed(fn -> visitor(key) end)
165 
    on_exit(fn -> unboxed(fn -> cleanup(visitor_id) end) end)
166 
    visitor_id
126 167 
  end
127 168
128 169 
  defp visitor(key) do

@@ -155,4 +196,23 @@ defmodule OpenAgents.Threads.CreditRaceTest do

155 196 
      Application.put_env(:openagents, :maximum_open_threads_per_account, previous)
156 197 
    end)
157 198 
  end
199
200 
  defp concurrent(inputs, fun) do
201 
    inputs
202 
    |> Task.async_stream(
203 
      fn item -> unboxed(fn -> fun.(item) end) end,
204 
      max_concurrency: 2,
205 
      ordered: false,
206 
      timeout: :infinity
207 
    )
208 
    |> Enum.map(fn {:ok, result} -> result end)
209 
  end
210
211 
  defp unboxed(fun) do
212 
    Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fun)
213 
  end
214
215 
  defp cleanup(visitor_id) do
216 
    Repo.delete_all(from(v in Visitor, where: v.id == ^visitor_id))
217 
  end
158 218 
end

This page updates live while a promote is in flight · changelog