OpenAgents

Commit

Tighten capacity evidence handling

32e1a2ef3f62 · Devin AI · · parent 6cd31854f2d2 

Tighten capacity evidence handling

Co-Authored-By: Christopher David <chris@openagents.com>
Co-Authored-By
Christopher David <chris@openagents.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.

Not deployed through the forge lane

No push, promotion, build, or deploy receipt references this commit (receipts are scanned over a bounded recent window). Changes shipped by full node replacement carry their proof in the release gate receipt instead.

Changed files

  • modified docs/capacity-and-matching-api.md
  • modified lib/openagents/capacity.ex
  • modified lib/openagents/capacity/broker.ex
  • modified lib/openagents/capacity/connected.ex
  • modified lib/openagents/capacity/estimate.ex
  • modified lib/openagents/capacity/evidence.ex
  • modified test/openagents/capacity_test.exs

Diff

7 files changed, +128 -80

docs/capacity-and-matching-api.md modified +1

@@ -213,6 +213,7 @@ A refusal returns schema `openagents.capacity_refusal.v1`:

213 213 
| `explicit_target_required` | `422` | The requirement needs an explicit connected-computer target |
214 214 
| `computer_not_found` | `404` | The named connected computer is not the caller's |
215 215 
| `quantity_unavailable` | `409` | Every otherwise admitted class lacks the requested quantity |
216 
| `incident_drained` | `503` | Every otherwise admitted class is drained for an incident |
216 217 
| `evidence_stale` | `503` | The only otherwise admitted classes carry stale evidence |
217 218 
| `evidence_unavailable` | `503` | No capacity evidence exists for any admitted class |
218 219
lib/openagents/capacity.ex modified +32 -26

@@ -177,11 +177,11 @@ defmodule OpenAgents.Capacity do

177 177 
      "error" => %{"code" => Atom.to_string(code), "detail" => detail}
178 178 
    }
179 179
180 
  defp class_projection(catalog, :unavailable, _config),
181 
    do: unavailable_class(catalog, "evidence_unavailable")
180 
  defp class_projection(catalog, :unavailable, config),
181 
    do: unavailable_class(catalog, "evidence_unavailable", config)
182 182
183 
  defp class_projection(catalog, nil, _config),
184 
    do: unavailable_class(catalog, "evidence_unavailable")
183 
  defp class_projection(catalog, nil, config),
184 
    do: unavailable_class(catalog, "evidence_unavailable", config)
185 185
186 186 
  defp class_projection(catalog, raw, config) when is_map(raw) do
187 187 
    quantities = Math.quantities(catalog, raw, config)

@@ -227,11 +227,30 @@ defmodule OpenAgents.Capacity do

227 227 
        "freshness" => freshness
228 228 
      },
229 229 
      "admits" => freshness == "fresh" and not incident_drained,
230 
      "refusal" => if(refusal_code, do: %{"code" => refusal_code}, else: nil)
230 
      "refusal" =>
231 
        case refusal_code do
232 
          "evidence_unavailable" ->
233 
            %{
234 
              "code" => refusal_code,
235 
              "detail" => "Capacity evidence is unavailable."
236 
            }
237
238 
          "evidence_stale" ->
239 
            %{"code" => refusal_code, "detail" => "Capacity evidence is stale."}
240
241 
          "incident_drained" ->
242 
            %{
243 
              "code" => refusal_code,
244 
              "detail" => "Capacity is temporarily drained for an incident."
245 
            }
246
247 
          nil ->
248 
            nil
249 
        end
231 250 
    }
232 251 
  end
233 252
234 
  defp unavailable_class(catalog, code) do
253 
  defp unavailable_class(catalog, code, config) do
235 254 
    %{
236 255 
      "id" => catalog["id"],
237 256 
      "label" => catalog["label"],

@@ -254,31 +273,21 @@ defmodule OpenAgents.Capacity do

254 273 
        "source" => if(catalog["id"] == "connected", do: "local", else: "broker"),
255 274 
        "observed_at" => nil,
256 275 
        "age_seconds" => nil,
257 
        "maximum_age_seconds" => Keyword.get(config(), :maximum_evidence_age_seconds, 120),
276 
        "maximum_age_seconds" => Keyword.get(config, :maximum_evidence_age_seconds, 120),
258 277 
        "freshness" => "unavailable"
259 278 
      },
260 279 
      "admits" => false,
261 
      "refusal" => %{"code" => code}
280 
      "refusal" => %{
281 
        "code" => code,
282 
        "detail" => "Capacity evidence is unavailable."
283 
      }
262 284 
    }
263 285 
  end
264 286
265 287 
  defp fetch_broker(config, viewer) do
266 288 
    source = Keyword.get(config, :evidence_source, OpenAgents.Capacity.Broker)
267 289
268 
    result =
269 
      try do
270 
        Code.ensure_loaded(source)
271
272 
        cond do
273 
          function_exported?(source, :fetch, 1) -> source.fetch(viewer)
274 
          function_exported?(source, :read, 1) -> source.read(viewer)
275 
          true -> {:error, :evidence_unavailable}
276 
        end
277 
      rescue
278 
        _error -> {:error, :evidence_unavailable}
279 
      end
280
281 
    case result do
290 
    case source.fetch(viewer) do
282 291 
      {:ok, %{"classes" => classes}} when is_list(classes) ->
283 292 
        Enum.reduce(classes, %{}, fn raw, acc ->
284 293 
          if is_map(raw) and is_binary(raw["id"]) do

@@ -288,10 +297,7 @@ defmodule OpenAgents.Capacity do

288 297 
          end
289 298 
        end)
290 299
291 
      classes when is_map(classes) ->
292 
        classes
293
294 
      _error ->
300 
      {:error, _reason} ->
295 301 
        %{}
296 302 
    end
297 303 
  end
lib/openagents/capacity/broker.ex modified +16 -17

@@ -3,6 +3,21 @@ defmodule OpenAgents.Capacity.Broker do

3 3
4 4 
  @behaviour OpenAgents.Capacity.Evidence
5 5
6 
  @allowed_keys [
7 
    "id",
8 
    "logical",
9 
    "active_reservations",
10 
    "reported_free",
11 
    "queued",
12 
    "observed_limit",
13 
    "budget_limit",
14 
    "drain_limit",
15 
    "observed_at",
16 
    "estimated_wait_seconds",
17 
    "private",
18 
    "incident_drained"
19 
  ]
20
6 21 
  @impl true
7 22 
  def fetch(_viewer) do
8 23 
    config = Application.get_env(:openagents, OpenAgents.Capacity, [])

@@ -50,23 +65,7 @@ defmodule OpenAgents.Capacity.Broker do

50 65 
    id = Map.get(class, "id")
51 66
52 67 
    if is_binary(id) do
53 
      allowed = [
54 
        "id",
55 
        "logical",
56 
        "active_reservations",
57 
        "reported_free",
58 
        "queued",
59 
        "observed_limit",
60 
        "budget_limit",
61 
        "drain_limit",
62 
        "observed_at",
63 
        "estimated_wait_seconds",
64 
        "private",
65 
        "incident_drained"
66 
      ]
67
68 
      {Map.take(class, allowed), id}
69 
      |> then(fn {safe, _id} -> [safe] end)
68 
      [Map.take(class, @allowed_keys)]
70 69 
    else
71 70 
      []
72 71 
    end
lib/openagents/capacity/connected.ex modified +15 -19

@@ -3,7 +3,7 @@ defmodule OpenAgents.Capacity.Connected do

3 3
4 4 
  import Ecto.Query
5 5
6 
  alias OpenAgents.Conversations.{Visitor}
6 
  alias OpenAgents.Conversations.Visitor
7 7 
  alias OpenAgents.Machines
8 8 
  alias OpenAgents.Repo
9 9 
  alias OpenAgents.Work.Job

@@ -16,29 +16,25 @@ defmodule OpenAgents.Capacity.Connected do

16 16 
    logical = length(machines)
17 17 
    machine_ids = Enum.map(machines, & &1.id)
18 18
19 
    active_reservations =
20 
      Repo.aggregate(
19 
    counts =
20 
      Repo.one(
21 21 
        from(job in Job,
22 22 
          join: visitor in Visitor,
23 23 
          on: visitor.id == job.owner_visitor_id,
24 
          where:
25 
            visitor.user_id == ^user_id and job.machine_id in ^machine_ids and
26 
              job.status in ["queued", "running"]
27 
        ),
28 
        :count
24 
          where: visitor.user_id == ^user_id and job.machine_id in ^machine_ids,
25 
          select: %{
26 
            active:
27 
              fragment(
28 
                "count(*) FILTER (WHERE ? IN ('queued', 'running'))",
29 
                job.status
30 
              ),
31 
            queued: fragment("count(*) FILTER (WHERE ? = 'queued')", job.status)
32 
          }
33 
        )
29 34 
      )
30 35
31 
    queued =
32 
      Repo.aggregate(
33 
        from(job in Job,
34 
          join: visitor in Visitor,
35 
          on: visitor.id == job.owner_visitor_id,
36 
          where:
37 
            visitor.user_id == ^user_id and job.machine_id in ^machine_ids and
38 
              job.status == "queued"
39 
        ),
40 
        :count
41 
      )
36 
    active_reservations = counts.active
37 
    queued = counts.queued
42 38
43 39 
    observed_at = DateTime.utc_now() |> DateTime.truncate(:second) |> DateTime.to_iso8601()
44 40
lib/openagents/capacity/estimate.ex modified +12 -14

@@ -7,7 +7,8 @@ defmodule OpenAgents.Capacity.Estimate do

7 7 
    wait_low = if is_integer(queued), do: queued * 30, else: 0
8 8 
    wait_high = if is_integer(queued), do: queued * 180, else: 180
9 9 
    buyer = Keyword.get(config, :buyer)
10 
    earnings = earnings(buyer, base, requirement)
10 
    {earnings_value, earnings_reason} = earnings(buyer, base, requirement)
11 
    maximum_age = Keyword.get(config, :maximum_evidence_age_seconds, 120)
11 12
12 13 
    %{
13 14 
      "cost" => %{

@@ -20,14 +21,14 @@ defmodule OpenAgents.Capacity.Estimate do

20 21 
        "low" => 120 + wait_low,
21 22 
        "high" => 960 + wait_high
22 23 
      },
23 
      "confidence" => if(age_seconds && age_seconds <= 120, do: "medium", else: "low"),
24 
      "confidence" => if(age_seconds && age_seconds <= maximum_age, do: "medium", else: "low"),
24 25 
      "evidence_age_seconds" => age_seconds,
25 26 
      "assumptions" => [
26 27 
        "One unit of the class runs the whole job.",
27 28 
        "Queue wait uses the current observed queue depth."
28 29 
      ],
29 
      "earnings" => earnings[:value],
30 
      "earnings_reason" => earnings[:reason]
30 
      "earnings" => earnings_value,
31 
      "earnings_reason" => earnings_reason
31 32 
    }
32 33 
  end
33 34

@@ -47,18 +48,15 @@ defmodule OpenAgents.Capacity.Estimate do

47 48 
  defp earnings(buyer, base, _requirement) when is_map(buyer) do
48 49 
    if is_binary(buyer["name"]) and buyer["name"] != "" and
49 50 
         buyer["verified_payout_policy"] == true do
50 
      %{value: %{"currency" => "usd_cents", "low" => base, "high" => base * 2}, reason: nil}
51 
      {%{"currency" => "usd_cents", "low" => base, "high" => base * 2}, nil}
51 52 
    else
52 
      %{
53 
        value: nil,
54 
        reason:
55 
          if(is_binary(buyer["name"]) and buyer["name"] != "",
56 
            do: "no_verified_payout_policy",
57 
            else: "no_named_buyer"
58 
          )
59 
      }
53 
      {nil,
54 
       if(is_binary(buyer["name"]) and buyer["name"] != "",
55 
         do: "no_verified_payout_policy",
56 
         else: "no_named_buyer"
57 
       )}
60 58 
    end
61 59 
  end
62 60
63 
  defp earnings(_buyer, _base, _requirement), do: %{value: nil, reason: "no_named_buyer"}
61 
  defp earnings(_buyer, _base, _requirement), do: {nil, "no_named_buyer"}
64 62 
end
lib/openagents/capacity/evidence.ex modified -3

@@ -2,7 +2,4 @@ defmodule OpenAgents.Capacity.Evidence do

2 2 
  @moduledoc false
3 3
4 4 
  @callback fetch(term()) :: {:ok, map()} | {:error, term()}
5 
  @callback read(term()) :: {:ok, map()} | {:error, term()}
6
7 
  @optional_callbacks read: 1
8 5 
end
test/openagents/capacity_test.exs modified +52 -1

@@ -2,6 +2,7 @@ defmodule OpenAgents.CapacityTest do

2 2 
  use OpenAgents.DataCase, async: false
3 3
4 4 
  alias OpenAgents.Capacity
5 
  alias OpenAgents.Capacity.Estimate
5 6 
  alias OpenAgents.Capacity.Math
6 7
7 8 
  setup do

@@ -89,7 +90,8 @@ defmodule OpenAgents.CapacityTest do

89 90 
    assert strong["admits"] == false
90 91 
    assert strong["quantities"]["allocatable"] == 0
91 92 
    assert strong["refusal"]["code"] == "evidence_stale"
92 
    refute Enum.any?(projection["classes"], &(&1["id"] == "batch"))
93 
    assert strong["refusal"]["detail"] == "Capacity evidence is stale."
94 
    assert Enum.find(projection["classes"], &(&1["id"] == "batch")) == nil
93 95 
  end
94 96
95 97 
  test "returns a typed refusal and never includes broker-only sensitive fields" do

@@ -124,10 +126,16 @@ defmodule OpenAgents.CapacityTest do

124 126
125 127 
    projection = Capacity.projection(%{id: Ecto.UUID.generate()})
126 128 
    serialized = Jason.encode!(projection)
129 
    unavailable = Enum.find(projection["classes"], &(&1["id"] == "strong"))
127 130
128 131 
    refute serialized =~ secret
129 132 
    refute serialized =~ "project_id"
130 133
134 
    assert unavailable["refusal"] == %{
135 
             "code" => "evidence_unavailable",
136 
             "detail" => "Capacity evidence is unavailable."
137 
           }
138
131 139 
    assert projection["classes"] |> Enum.find(&(&1["id"] == "standard")) |> Map.get("admits") ==
132 140 
             true
133 141

@@ -143,6 +151,48 @@ defmodule OpenAgents.CapacityTest do

143 151 
             )
144 152 
  end
145 153
154 
  test "estimates earnings only with a named buyer and verified payout policy" do
155 
    class = %{"id" => "standard"}
156 
    requirement = %{"quantity" => 1, "duration_seconds" => 3_600}
157 
    base_config = [maximum_evidence_age_seconds: 120]
158
159 
    no_buyer = Estimate.build(class, requirement, 0, 60, base_config)
160 
    assert no_buyer["earnings"] == nil
161 
    assert no_buyer["earnings_reason"] == "no_named_buyer"
162
163 
    unverified_buyer =
164 
      Estimate.build(
165 
        class,
166 
        requirement,
167 
        0,
168 
        60,
169 
        Keyword.put(base_config, :buyer, %{"name" => "Acme"})
170 
      )
171
172 
    assert unverified_buyer["earnings"] == nil
173 
    assert unverified_buyer["earnings_reason"] == "no_verified_payout_policy"
174
175 
    verified_buyer =
176 
      Estimate.build(
177 
        class,
178 
        requirement,
179 
        0,
180 
        60,
181 
        Keyword.put(base_config, :buyer, %{
182 
          "name" => "Acme",
183 
          "verified_payout_policy" => true
184 
        })
185 
      )
186
187 
    assert verified_buyer["earnings"] == %{
188 
             "currency" => "usd_cents",
189 
             "low" => 16,
190 
             "high" => 32
191 
           }
192
193 
    assert verified_buyer["earnings_reason"] == nil
194 
  end
195
146 196 
  test "managed matching never falls back to connected computers" do
147 197 
    Application.put_env(
148 198 
      :openagents,

@@ -295,5 +345,6 @@ defmodule OpenAgents.CapacityTest do

295 345 
    assert standard["admits"] == true
296 346 
    assert batch["admits"] == false
297 347 
    assert batch["refusal"]["code"] == "incident_drained"
348 
    assert batch["refusal"]["detail"] == "Capacity is temporarily drained for an incident."
298 349 
  end
299 350 
end

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