OpenAgents

Commit

Run delegated Box work as durable jobs

260233c5eda2 · Devin AI · · parent 3d1b17905755 

Run delegated Box work as durable jobs

Add authenticated detach-and-poll runs for long-lived Box commands.

Each run records durable provider evidence, polls output and exit status,
and reconciles cancellation, timeout, loss, and restart recovery without
replaying an ambiguous dispatch. Dispatch transport failures receive one
directory probe, so the client never replays a command that may have run.

Keep the box:control scope independent from chat, forge, and deployment
authority while preserving conversation and Box ownership boundaries.

Refs #107

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 INVARIANTS.md
  • modified config/config.exs
  • modified docs/api-authentication.md
  • modified docs/openagents-cli/api.md
  • modified lib/openagents/application.ex
  • modified lib/openagents/box/client.ex
  • added lib/openagents/box/run.ex
  • added lib/openagents/box_run_recovery.ex
  • added lib/openagents/box_run_server.ex
  • added lib/openagents/box_run_supervisor.ex
  • added lib/openagents/box_runs.ex
  • modified lib/openagents_web/api_route_authority.ex
  • modified lib/openagents_web/box_rate_limiter.ex
  • added lib/openagents_web/controllers/box_run_controller.ex
  • modified lib/openagents_web/route_authority.ex
  • modified lib/openagents_web/router.ex
  • modified priv/migration_lineages/prior-2026-08-19.json
  • added priv/repo/migrations/20260823133243_create_box_runs.exs
  • added test/openagents/box_client_runs_test.exs
  • added test/openagents/box_runs_test.exs
  • added test/openagents_web/controllers/box_run_controller_test.exs

Diff

21 files changed, +2110 -3

INVARIANTS.md modified +16

@@ -269,6 +269,21 @@ Evidence: `OpenAgentsWeb.Plugs.BoxControlAuth`, `OpenAgents.Box`,

269 269 
`OpenAgentsWeb.BoxController`, and
270 270 
`test/openagents_web/controllers/box_controller_test.exs`.
271 271
272 
### WORK-002 — Detached Box runs reconcile from durable evidence
273
274 
Status: Current
275
276 
Long-running Box work is admitted before dispatch and executes from a
277 
run-specific directory containing its script, combined log, process ID, and
278 
exit sentinel. The application polls that directory, stores bounded redacted
279 
output, and treats a missing process without an exit sentinel as `lost`.
280 
Dispatch ambiguity receives one probe and never an automatic second dispatch.
281 
Cancellation and timeout record requested and effective timestamps separately.
282
283 
Evidence: `OpenAgents.BoxRuns`, `OpenAgents.BoxRunServer`,
284 
`OpenAgents.Box.Client`, and
285 
`test/openagents/box_runs_test.exs`.
286
272 287 
## Data authority and synchronization
273 288
274 289 
### DATA-001 — PostgreSQL is authoritative

@@ -2300,6 +2315,7 @@ contract; the invariant prose above defines the assertion, not the filename.

2300 2315 
| IDENTITY-003 | `test/openagents/memory_portability_test.exs` |
2301 2316 
| IDENTITY-004 | `test/openagents/agents_test.exs`, `test/openagents_web/controllers/agent_controller_test.exs` |
2302 2317 
| IDENTITY-005 | `test/openagents_web/controllers/box_controller_test.exs` |
2318 
| WORK-002 | `test/openagents/box_runs_test.exs` |
2303 2319 
| PROMISE-001 | `test/openagents/promise_registry_test.exs`, `test/openagents_web/controllers/project_controller_test.exs` |
2304 2320 
| PROMISE-002 | `test/openagents/promise_registry_test.exs` |
2305 2321 
| DATA-001 | `test/openagents/conversations_test.exs` |
config/config.exs modified +4

@@ -109,6 +109,10 @@ config :openagents,

109 109 
    ttl_seconds: 3_600,
110 110 
    poll_interval_ms: 1_000,
111 111 
    poll_attempts: 60,
112 
    run_poll_interval_ms: 1_000,
113 
    run_max_duration_seconds: 1_800,
114 
    run_create_rate_limit: 10,
115 
    run_command_rate_limit: 30,
112 116 
    rate_limit_window_seconds: 60,
113 117 
    create_rate_limit: 10,
114 118 
    command_rate_limit: 30
docs/api-authentication.md modified +20

@@ -57,6 +57,26 @@ provider request. Agent participation credentials receive

57 57 
`{"error":{"code":"agent_box_control_forbidden"}}`; linked-agent Box control is
58 58 
deferred to the linked-principal contract.
59 59
60 
### Durable Box runs
61
62 
Use the same `box:control` token to start and inspect detached runs:
63
64 
```sh
65 
openagents api -X POST \
66 
  conversations/CONVERSATION_ID/boxes/BOX_ID/runs \
67 
  -H 'Idempotency-Key: RUN_KEY' \
68 
  -d '{"command":"opencode run --non-interactive ..."}'
69 
openagents api conversations/CONVERSATION_ID/boxes/BOX_ID/runs
70 
openagents api \
71 
  conversations/CONVERSATION_ID/boxes/BOX_ID/runs/RUN_ID/output?offset=0
72 
openagents api -X POST \
73 
  conversations/CONVERSATION_ID/boxes/BOX_ID/runs/RUN_ID/cancel
74 
```
75
76 
Runs return `202 Accepted` when admitted. Their state and bounded output remain
77 
available after the creating request ends. A run is reconciled as `lost` when
78 
its process disappears without an exit sentinel.
79
60 80 
### Agent participation credentials
61 81
62 82 
An agent can register without GitHub by sending its handle and display name to
docs/openagents-cli/api.md modified +19

@@ -162,6 +162,25 @@ credentials are refused with

162 162 
`{"error":{"code":"agent_box_control_forbidden"}}` until linked-agent Box
163 163 
control is implemented.
164 164
165 
Long-running work uses durable runs instead of the synchronous command route:
166
167 
```sh
168 
openagents api -X POST \
169 
  conversations/CONVERSATION_ID/boxes/BOX_ID/runs \
170 
  -H 'Idempotency-Key: RUN_KEY' \
171 
  -d '{"command":"opencode run --non-interactive ..."}'
172 
openagents api conversations/CONVERSATION_ID/boxes/BOX_ID/runs
173 
openagents api \
174 
  conversations/CONVERSATION_ID/boxes/BOX_ID/runs/RUN_ID/output?offset=0
175 
openagents api -X POST \
176 
  conversations/CONVERSATION_ID/boxes/BOX_ID/runs/RUN_ID/cancel
177 
```
178
179 
The run directory, process ID, combined log, and exit sentinel let the server
180 
reconcile work after the creating request ends. A missing process without an
181 
exit sentinel becomes `lost`; the server never reports that state as
182 
`completed`.
183
165 184 
## Work with issues
166 185
167 186 
List open issues. The API returns an object with an `issues` array:
lib/openagents/application.ex modified +4 -1

@@ -54,7 +54,10 @@ defmodule OpenAgents.Application do

54 54 
        {DNSCluster, query: Application.get_env(:openagents, :dns_cluster_query) || :ignore},
55 55 
        {Phoenix.PubSub, name: OpenAgents.PubSub},
56 56 
        OpenAgents.RuntimeSupervisor,
57 
        OpenAgentsWeb.BoxRateLimiter
57 
        OpenAgentsWeb.BoxRateLimiter,
58 
        {Registry, keys: :unique, name: OpenAgents.BoxRunRegistry},
59 
        OpenAgents.BoxRunSupervisor,
60 
        OpenAgents.BoxRunRecovery
58 61 
      ] ++ analytics_children() ++ [OpenAgentsWeb.Endpoint]
59 62
60 63 
    # See https://elixir.hexdocs.pm/Supervisor.html
lib/openagents/box/client.ex modified +260

@@ -43,6 +43,81 @@ defmodule OpenAgents.Box.Client do

43 43 
    end
44 44 
  end
45 45
46 
  @doc "Dispatches one detached, idempotent run directory on a box."
47 
  @spec dispatch_run(String.t(), String.t(), String.t()) :: {:ok, integer()} | {:error, term()}
48 
  def dispatch_run(box_id, run_id, command)
49 
      when is_binary(box_id) and is_binary(run_id) and is_binary(command) do
50 
    dispatch_run(box_id, run_id, command, nil)
51 
  end
52
53 
  @spec dispatch_run(String.t(), String.t(), String.t(), String.t() | nil) ::
54 
          {:ok, integer()} | {:error, term()}
55 
  def dispatch_run(box_id, run_id, command, run_directory)
56 
      when is_binary(box_id) and is_binary(run_id) and is_binary(command) do
57 
    with {:ok, body} <-
58 
           command(box_id, %{
59 
             "command" => dispatch_command(run_id, command, run_directory),
60 
             "timeoutSeconds" => 30
61 
           }),
62 
         {:ok, pid} <- dispatch_pid(body) do
63 
      {:ok, pid}
64 
    end
65 
  end
66
67 
  @doc "Polls one detached run for output and its exit sentinel."
68 
  @spec poll_run(String.t(), String.t(), non_neg_integer()) :: {:ok, map()} | {:error, term()}
69 
  def poll_run(box_id, run_id, offset)
70 
      when is_binary(box_id) and is_binary(run_id) and is_integer(offset) and offset >= 0 do
71 
    poll_run(box_id, run_id, offset, nil)
72 
  end
73
74 
  @spec poll_run(String.t(), String.t(), non_neg_integer(), String.t() | nil) ::
75 
          {:ok, map()} | {:error, term()}
76 
  def poll_run(box_id, run_id, offset, run_directory)
77 
      when is_binary(box_id) and is_binary(run_id) and is_integer(offset) and offset >= 0 do
78 
    with {:ok, body} <-
79 
           command(box_id, %{
80 
             "command" => poll_command(run_id, offset, run_directory),
81 
             "timeoutSeconds" => 30
82 
           }) do
83 
      {:ok, parse_poll(body)}
84 
    end
85 
  end
86
87 
  @doc "Probes a detached run directory after an ambiguous dispatch."
88 
  @spec probe_run(String.t(), String.t()) :: {:ok, map()} | {:error, term()}
89 
  def probe_run(box_id, run_id) when is_binary(box_id) and is_binary(run_id) do
90 
    probe_run(box_id, run_id, nil)
91 
  end
92
93 
  @spec probe_run(String.t(), String.t(), String.t() | nil) :: {:ok, map()} | {:error, term()}
94 
  def probe_run(box_id, run_id, run_directory) when is_binary(box_id) and is_binary(run_id) do
95 
    with {:ok, body} <-
96 
           command(box_id, %{
97 
             "command" => probe_command(run_id, run_directory),
98 
             "timeoutSeconds" => 30
99 
           }) do
100 
      {:ok, parse_probe(body)}
101 
    end
102 
  end
103
104 
  @doc "Kills the recorded process group for a detached run."
105 
  @spec cancel_run(String.t(), String.t()) :: {:ok, body()} | {:error, term()}
106 
  def cancel_run(box_id, run_id) when is_binary(box_id) and is_binary(run_id) do
107 
    cancel_run(box_id, run_id, nil)
108 
  end
109
110 
  @spec cancel_run(String.t(), String.t(), String.t() | nil) :: {:ok, body()} | {:error, term()}
111 
  def cancel_run(box_id, run_id, run_directory) when is_binary(box_id) and is_binary(run_id) do
112 
    with {:ok, body} <-
113 
           command(box_id, %{
114 
             "command" => cancel_command(run_id, run_directory),
115 
             "timeoutSeconds" => 30
116 
           }) do
117 
      {:ok, parse_cancel(body)}
118 
    end
119 
  end
120
46 121 
  @doc "Whether a string is a well-formed box id."
47 122 
  @spec valid_box_id?(String.t()) :: boolean()
48 123 
  def valid_box_id?(box_id) when is_binary(box_id), do: Regex.match?(@box_id_pattern, box_id)

@@ -52,6 +127,191 @@ defmodule OpenAgents.Box.Client do

52 127 
    if valid_box_id?(box_id), do: :ok, else: {:error, :box_not_found}
53 128 
  end
54 129
130 
  defp dispatch_command(run_id, command, run_directory) do
131 
    root = run_root(run_id, run_directory)
132 
    encoded = Base.encode64(command)
133
134 
    """
135 
    set -eu
136 
    root=#{root}
137 
    mkdir -p "$(dirname "$root")"
138 
    if ! mkdir "$root" 2>/dev/null; then
139 
      printf 'OPENAGENTS_RUN_EXISTS\\n'
140 
      exit 73
141 
    fi
142 
    printf '%s' '#{encoded}' | base64 -d > "$root/script.sh"
143 
    chmod 700 "$root/script.sh"
144 
    : > "$root/output.log"
145 
    nohup setsid sh -c 'set +e; sh "$1" > "$2/output.log" 2>&1; status=$?; printf "%s\\n" "$status" > "$2/exit-code"; exit "$status"' _ "$root/script.sh" "$root" </dev/null >/dev/null 2>&1 &
146 
    pid=$!
147 
    printf '%s\\n' "$pid" > "$root/pid"
148 
    printf '%s\\n' "$pid"
149 
    """
150 
  end
151
152 
  defp poll_command(run_id, offset, run_directory) do
153 
    root = run_root(run_id, run_directory)
154
155 
    """
156 
    set -eu
157 
    root=#{root}
158 
    if [ ! -d "$root" ]; then
159 
      printf 'OA_PRESENT=0\\n'
160 
      exit 0
161 
    fi
162 
    size=$(wc -c < "$root/output.log" | tr -d ' ')
163 
    data=$(if [ "$size" -gt #{offset} ]; then dd if="$root/output.log" bs=1 skip=#{offset} 2>/dev/null | base64 -w0; fi)
164 
    printf 'OA_PRESENT=1\\nOA_SIZE=%s\\nOA_DATA=%s\\n' "$size" "$data"
165 
    if [ -f "$root/exit-code" ]; then
166 
      printf 'OA_EXIT=%s\\n' "$(cat "$root/exit-code")"
167 
    fi
168 
    if [ -f "$root/pid" ] && kill -0 "$(cat "$root/pid")" 2>/dev/null; then
169 
      printf 'OA_ALIVE=1\\n'
170 
    else
171 
      printf 'OA_ALIVE=0\\n'
172 
    fi
173 
    """
174 
  end
175
176 
  defp probe_command(run_id, run_directory) do
177 
    root = run_root(run_id, run_directory)
178
179 
    """
180 
    set -eu
181 
    root=#{root}
182 
    if [ ! -d "$root" ]; then
183 
      printf 'OA_PRESENT=0\\n'
184 
      exit 0
185 
    fi
186 
    printf 'OA_PRESENT=1\\n'
187 
    if [ -f "$root/pid" ]; then
188 
      printf 'OA_PID=%s\\n' "$(cat "$root/pid")"
189 
    fi
190 
    """
191 
  end
192
193 
  defp cancel_command(run_id, run_directory) do
194 
    root = run_root(run_id, run_directory)
195
196 
    """
197 
    set -eu
198 
    root=#{root}
199 
    if [ -f "$root/pid" ]; then
200 
      pid=$(cat "$root/pid")
201 
      kill -TERM -- "-$pid" 2>/dev/null || kill -TERM "$pid" 2>/dev/null || true
202 
      printf 'OA_CANCELLED=1\\n'
203 
    elif pkill -TERM -f -- "$root/script.sh" 2>/dev/null; then
204 
      printf 'OA_CANCELLED=1\\n'
205 
    else
206 
      printf 'OA_CANCELLED=0\\n'
207 
    fi
208 
    """
209 
  end
210
211 
  defp run_root(run_id, nil), do: "$HOME/.openagents/box-runs/#{run_id}"
212 
  defp run_root(_run_id, run_directory), do: run_directory
213
214 
  defp dispatch_pid(%{"stdout" => output}) when is_binary(output) do
215 
    case Regex.run(~r/(?:\A|\n)(\d+)\s*\z/, output) do
216 
      [_, pid] -> {:ok, String.to_integer(pid)}
217 
      _ -> {:error, :box_response_invalid}
218 
    end
219 
  end
220
221 
  defp dispatch_pid(%{"pid" => pid}) when is_integer(pid), do: {:ok, pid}
222 
  defp dispatch_pid(%{"pid" => pid}) when is_binary(pid), do: integer_or_nil(pid) |> pid_result()
223 
  defp dispatch_pid(_body), do: {:error, :box_response_invalid}
224
225 
  defp parse_poll(%{"log_size" => size} = body) when is_integer(size) do
226 
    %{
227 
      present: Map.get(body, "present", true),
228 
      log_size: size,
229 
      output: body["output"] || "",
230 
      exit_status: body["exit_status"],
231 
      alive: Map.get(body, "alive", true)
232 
    }
233 
  end
234
235 
  defp parse_poll(%{"logSize" => size} = body) when is_integer(size) do
236 
    parse_poll(%{
237 
      "log_size" => size,
238 
      "present" => Map.get(body, "present", true),
239 
      "output" => body["output"] || "",
240 
      "exit_status" => body["exitStatus"] || body["exitCode"],
241 
      "alive" => Map.get(body, "alive", true)
242 
    })
243 
  end
244
245 
  defp parse_poll(%{"stdout" => output}) when is_binary(output) do
246 
    %{
247 
      present: marker(output, "OA_PRESENT", "0") == "1",
248 
      log_size: marker(output, "OA_SIZE", "0") |> integer_marker(),
249 
      output: output_marker(output),
250 
      exit_status: integer_or_nil(marker(output, "OA_EXIT", nil)),
251 
      alive: marker(output, "OA_ALIVE", "0") == "1"
252 
    }
253 
  end
254
255 
  defp parse_poll(_body),
256 
    do: %{present: false, log_size: 0, output: "", exit_status: nil, alive: false}
257
258 
  defp parse_probe(%{"present" => present} = body) do
259 
    %{present: present == true, pid: body["pid"]}
260 
  end
261
262 
  defp parse_probe(%{"stdout" => output}) when is_binary(output) do
263 
    %{
264 
      present: marker(output, "OA_PRESENT", "0") == "1",
265 
      pid: integer_or_nil(marker(output, "OA_PID", nil))
266 
    }
267 
  end
268
269 
  defp parse_probe(_body), do: %{present: false, pid: nil}
270
271 
  defp parse_cancel(%{"stdout" => output} = body) when is_binary(output) do
272 
    if String.contains?(output, "OA_CANCELLED=0") do
273 
      Map.put(body, "cancelled", false)
274 
    else
275 
      Map.put(body, "cancelled", true)
276 
    end
277 
  end
278
279 
  defp parse_cancel(body), do: body
280
281 
  defp output_marker(output) do
282 
    case Regex.run(~r/OA_DATA=([A-Za-z0-9+\/=]*)/, output) do
283 
      [_, encoded] ->
284 
        case Base.decode64(encoded) do
285 
          {:ok, decoded} -> decoded
286 
          :error -> ""
287 
        end
288
289 
      _missing ->
290 
        ""
291 
    end
292 
  end
293
294 
  defp marker(output, key, default) do
295 
    case Regex.run(~r/#{key}=([^\n]*)/, output) do
296 
      [_, value] -> value
297 
      _missing -> default
298 
    end
299 
  end
300
301 
  defp integer_marker(value), do: String.to_integer(value || "0")
302
303 
  defp integer_or_nil(nil), do: nil
304
305 
  defp integer_or_nil(value) do
306 
    case Integer.parse(value) do
307 
      {integer, ""} -> integer
308 
      _invalid -> nil
309 
    end
310 
  end
311
312 
  defp pid_result(nil), do: {:error, :box_response_invalid}
313 
  defp pid_result(pid), do: {:ok, pid}
314
55 315 
  defp request(method, api_path, options) do
56 316 
    with {:ok, api_key} <- api_key() do
57 317 
      settings = Application.get_env(:openagents, :box_api, [])
lib/openagents/box/run.ex added +115

@@ -0,0 +1,115 @@

1 
defmodule OpenAgents.Box.Run do
2 
  @moduledoc "One durable detached command run on a Box computer."
3
4 
  use Ecto.Schema
5 
  import Ecto.Changeset
6
7 
  @states ~w(admitted dispatched running completed failed cancelled timed_out lost)
8 
  @terminal_states ~w(completed failed cancelled timed_out lost)
9 
  @maximum_command_bytes 8_000
10 
  @maximum_idempotency_key_bytes 256
11
12 
  @primary_key {:id, :binary_id, autogenerate: true}
13 
  @foreign_key_type :binary_id
14 
  @timestamps_opts [type: :utc_datetime_usec]
15
16 
  schema "box_runs" do
17 
    belongs_to :conversation, OpenAgents.Conversations.Conversation
18 
    belongs_to :conversation_box, OpenAgents.Box.ConversationBox
19 
    field :requesting_principal, :map
20 
    field :command, :string
21 
    field :idempotency_key, :string
22 
    field :state, :string, default: "admitted"
23 
    field :exit_status, :integer
24 
    field :timed_out, :boolean, default: false
25 
    field :output, :string, default: ""
26 
    field :output_base_offset, :integer, default: 0
27 
    field :last_output_offset, :integer, default: 0
28 
    field :pid, :integer
29 
    field :run_directory, :string
30 
    field :failure_reason, :string
31 
    field :dispatch_attempted_at, :utc_datetime_usec
32 
    field :probe_attempted_at, :utc_datetime_usec
33 
    field :admitted_at, :utc_datetime_usec
34 
    field :dispatched_at, :utc_datetime_usec
35 
    field :started_at, :utc_datetime_usec
36 
    field :finished_at, :utc_datetime_usec
37 
    field :cancellation_requested_at, :utc_datetime_usec
38 
    field :cancellation_effective_at, :utc_datetime_usec
39 
    field :deadline_at, :utc_datetime_usec
40 
    timestamps()
41 
  end
42
43 
  @type t :: %__MODULE__{}
44
45 
  @spec states() :: [String.t()]
46 
  def states, do: @states
47
48 
  @spec terminal_states() :: [String.t()]
49 
  def terminal_states, do: @terminal_states
50
51 
  @spec terminal?(t()) :: boolean()
52 
  def terminal?(%__MODULE__{state: state}), do: state in @terminal_states
53
54 
  @spec maximum_command_bytes() :: pos_integer()
55 
  def maximum_command_bytes, do: @maximum_command_bytes
56
57 
  @spec maximum_idempotency_key_bytes() :: pos_integer()
58 
  def maximum_idempotency_key_bytes, do: @maximum_idempotency_key_bytes
59
60 
  @spec changeset(t(), map()) :: Ecto.Changeset.t()
61 
  def changeset(run, attributes) do
62 
    run
63 
    |> cast(attributes, [
64 
      :requesting_principal,
65 
      :command,
66 
      :idempotency_key,
67 
      :state,
68 
      :exit_status,
69 
      :timed_out,
70 
      :output,
71 
      :output_base_offset,
72 
      :last_output_offset,
73 
      :pid,
74 
      :run_directory,
75 
      :failure_reason,
76 
      :dispatch_attempted_at,
77 
      :probe_attempted_at,
78 
      :admitted_at,
79 
      :dispatched_at,
80 
      :started_at,
81 
      :finished_at,
82 
      :cancellation_requested_at,
83 
      :cancellation_effective_at,
84 
      :deadline_at
85 
    ])
86 
    |> put_programmatic(attributes, :conversation_id)
87 
    |> put_programmatic(attributes, :conversation_box_id)
88 
    |> validate_required([
89 
      :conversation_id,
90 
      :conversation_box_id,
91 
      :requesting_principal,
92 
      :command,
93 
      :idempotency_key,
94 
      :run_directory,
95 
      :admitted_at,
96 
      :deadline_at
97 
    ])
98 
    |> validate_inclusion(:state, @states)
99 
    |> validate_length(:command, max: @maximum_command_bytes)
100 
    |> validate_length(:idempotency_key, max: @maximum_idempotency_key_bytes)
101 
    |> validate_number(:output_base_offset, greater_than_or_equal_to: 0)
102 
    |> validate_number(:last_output_offset, greater_than_or_equal_to: 0)
103 
    |> foreign_key_constraint(:conversation_id)
104 
    |> foreign_key_constraint(:conversation_box_id)
105 
    |> unique_constraint(:idempotency_key)
106 
    |> unique_constraint(:conversation_box_id, name: :box_runs_one_active_per_box_index)
107 
  end
108
109 
  defp put_programmatic(changeset, attributes, field) do
110 
    case Map.fetch(attributes, field) do
111 
      {:ok, value} -> put_change(changeset, field, value)
112 
      :error -> changeset
113 
    end
114 
  end
115 
end
lib/openagents/box_run_recovery.ex added +15

@@ -0,0 +1,15 @@

1 
defmodule OpenAgents.BoxRunRecovery do
2 
  @moduledoc false
3
4 
  use GenServer, restart: :temporary
5
6 
  def start_link(options \\ []) do
7 
    GenServer.start_link(__MODULE__, options, name: __MODULE__)
8 
  end
9
10 
  @impl true
11 
  def init(_options) do
12 
    :ok = OpenAgents.BoxRuns.reconcile_non_terminal()
13 
    {:ok, %{}}
14 
  end
15 
end
lib/openagents/box_run_server.ex added +247

@@ -0,0 +1,247 @@

1 
defmodule OpenAgents.BoxRunServer do
2 
  @moduledoc false
3
4 
  use GenServer
5
6 
  alias OpenAgents.Box.Client
7 
  alias OpenAgents.Box.Run
8 
  alias OpenAgents.BoxRuns
9 
  alias OpenAgents.Repo
10
11 
  @spec start_link(String.t()) :: GenServer.on_start()
12 
  def start_link(run_id) do
13 
    GenServer.start_link(__MODULE__, run_id, name: via(run_id))
14 
  end
15
16 
  @impl true
17 
  def init(run_id) do
18 
    case Repo.get(Run, run_id) do
19 
      %Run{} = run -> {:ok, %{run: run}, {:continue, :drive}}
20 
      nil -> :ignore
21 
    end
22 
  end
23
24 
  @impl true
25 
  def handle_continue(:drive, state) do
26 
    next_state = drive(state)
27
28 
    if terminal_state?(next_state.run) do
29 
      {:stop, :normal, next_state}
30 
    else
31 
      {:noreply, next_state}
32 
    end
33 
  end
34
35 
  @impl true
36 
  def handle_cast(:cancel, %{run: %Run{} = run} = state) do
37 
    if Run.terminal?(run) do
38 
      {:stop, :normal, state}
39 
    else
40 
      result =
41 
        case Client.cancel_run(box_id(run), run.id, run.run_directory) do
42 
          {:ok, %{"cancelled" => false}} -> {:error, :cancellation_pid_missing}
43 
          {:ok, _body} -> BoxRuns.mark_cancellation_effective(run.id)
44 
          {:error, reason} -> {:error, reason}
45 
        end
46
47 
      case result do
48 
        {:ok, updated} -> {:stop, :normal, %{state | run: updated}}
49 
        {:error, _reason} -> {:noreply, schedule_poll(state)}
50 
      end
51 
    end
52 
  end
53
54 
  @impl true
55 
  def handle_info(:poll, %{run: %Run{} = run} = state) do
56 
    case Repo.get(Run, run.id) do
57 
      %Run{} = refreshed ->
58 
        if Run.terminal?(refreshed) do
59 
          {:stop, :normal, %{state | run: refreshed}}
60 
        else
61 
          next_state = poll(%{state | run: refreshed})
62
63 
          if terminal_state?(next_state.run) do
64 
            {:stop, :normal, next_state}
65 
          else
66 
            {:noreply, next_state}
67 
          end
68 
        end
69
70 
      nil ->
71 
        {:stop, :normal, state}
72 
    end
73 
  end
74
75 
  defp drive(%{run: %Run{} = run} = state) do
76 
    cond do
77 
      Run.terminal?(run) ->
78 
        state
79
80 
      DateTime.compare(DateTime.utc_now(), run.deadline_at) != :lt ->
81 
        _ = BoxRuns.mark_timeout(run.id)
82 
        _ = cancel_for_timeout(run)
83 
        schedule_poll(state)
84
85 
      is_nil(run.dispatch_attempted_at) ->
86 
        dispatch(state)
87
88 
      is_nil(run.probe_attempted_at) and run.state == "admitted" ->
89 
        probe(state)
90
91 
      true ->
92 
        schedule_poll(state)
93 
    end
94 
  end
95
96 
  defp dispatch(%{run: run} = state) do
97 
    case BoxRuns.claim_dispatch(run.id) do
98 
      {:ok, claimed} ->
99 
        case Client.dispatch_run(
100 
               box_id(claimed),
101 
               claimed.id,
102 
               claimed.command,
103 
               claimed.run_directory
104 
             ) do
105 
          {:ok, pid} ->
106 
            {:ok, updated} = BoxRuns.mark_dispatched(claimed.id, pid)
107 
            schedule_poll(%{state | run: updated})
108
109 
          {:error, :box_unreachable} ->
110 
            probe(%{state | run: claimed})
111
112 
          {:error, reason} when reason in [:box_not_found, :box_stopped] ->
113 
            _ = BoxRuns.mark_lost(claimed.id, Atom.to_string(reason))
114 
            %{state | run: %{claimed | state: "lost"}}
115
116 
          {:error, reason} ->
117 
            _ = BoxRuns.finish(claimed.id, "failed", nil, dispatch_failure_reason(reason))
118 
            %{state | run: %{claimed | state: "failed"}}
119 
        end
120
121 
      {:error, _reason} ->
122 
        schedule_poll(state)
123 
    end
124 
  end
125
126 
  defp probe(%{run: run} = state) do
127 
    {:ok, _run} = BoxRuns.mark_probe_attempted(run.id)
128
129 
    case Client.probe_run(box_id(run), run.id, run.run_directory) do
130 
      {:ok, %{present: true, pid: pid}} ->
131 
        {:ok, updated} = BoxRuns.mark_dispatched(run.id, pid || run.pid)
132 
        schedule_poll(%{state | run: updated})
133
134 
      _missing ->
135 
        _ = BoxRuns.mark_lost(run.id, "dispatch_ambiguous")
136 
        %{state | run: %{run | state: "lost"}}
137 
    end
138 
  end
139
140 
  defp poll(%{run: run} = state) do
141 
    case retry_cancellation(run) do
142 
      {:ok, updated} ->
143 
        Process.send_after(self(), :poll, 0)
144 
        %{state | run: updated}
145
146 
      :retry_poll ->
147 
        poll_provider(state)
148 
    end
149 
  end
150
151 
  defp poll_provider(%{run: run} = state) do
152 
    case Client.poll_run(box_id(run), run.id, run.last_output_offset, run.run_directory) do
153 
      {:ok, %{present: false}} ->
154 
        _ = BoxRuns.mark_lost(run.id, "run_directory_missing")
155 
        %{state | run: %{run | state: "lost"}}
156
157 
      {:ok, %{alive: false, exit_status: nil}} ->
158 
        _ = BoxRuns.mark_lost(run.id, "process_missing_without_exit_sentinel")
159 
        %{state | run: %{run | state: "lost"}}
160
161 
      {:ok, result} ->
162 
        {:ok, updated} = BoxRuns.record_poll(run.id, result)
163
164 
        cond do
165 
          result.exit_status != nil ->
166 
            terminal = if result.exit_status == 0, do: "completed", else: "failed"
167 
            {:ok, updated} = BoxRuns.finish(run.id, terminal, result.exit_status)
168 
            %{state | run: updated}
169
170 
          DateTime.compare(DateTime.utc_now(), updated.deadline_at) != :lt ->
171 
            _ = BoxRuns.mark_timeout(updated.id)
172 
            _ = cancel_for_timeout(updated)
173 
            schedule_poll(%{state | run: %{updated | timed_out: true}})
174
175 
          true ->
176 
            schedule_poll(%{state | run: updated})
177 
        end
178
179 
      {:error, reason} ->
180 
        if reason in [:box_not_found, :box_stopped] or box_gone_error?(reason) do
181 
          _ = BoxRuns.mark_lost(run.id, loss_reason(reason))
182 
          %{state | run: %{run | state: "lost"}}
183 
        else
184 
          schedule_poll(state)
185 
        end
186 
    end
187 
  end
188
189 
  defp retry_cancellation(
190 
         %Run{cancellation_requested_at: requested, cancellation_effective_at: nil} = run
191 
       )
192 
       when not is_nil(requested) do
193 
    case Client.cancel_run(box_id(run), run.id, run.run_directory) do
194 
      {:ok, %{"cancelled" => false}} -> :retry_poll
195 
      {:ok, _body} -> BoxRuns.mark_cancellation_effective(run.id)
196 
      {:error, _reason} -> :retry_poll
197 
    end
198 
  end
199
200 
  defp retry_cancellation(_run), do: :retry_poll
201
202 
  defp cancel_for_timeout(run) do
203 
    case Client.cancel_run(box_id(run), run.id, run.run_directory) do
204 
      {:ok, _body} ->
205 
        _ = BoxRuns.finish(run.id, "timed_out", nil, "run_duration_exceeded")
206 
        :ok
207
208 
      {:error, _reason} ->
209 
        _ = BoxRuns.finish(run.id, "timed_out", nil, "run_duration_exceeded")
210 
        :ok
211 
    end
212 
  end
213
214 
  defp dispatch_failure_reason(reason) when is_atom(reason), do: Atom.to_string(reason)
215 
  defp dispatch_failure_reason(_reason), do: "dispatch_failed"
216
217 
  defp box_gone_error?({:box_request_refused, _status, code})
218 
       when code in ["box_stopped", "box_not_found", "box_gone"],
219 
       do: true
220
221 
  defp box_gone_error?(_reason), do: false
222
223 
  defp loss_reason({:box_request_refused, _status, code}), do: code
224 
  defp loss_reason(reason), do: Atom.to_string(reason)
225
226 
  defp schedule_poll(%{run: run} = state) do
227 
    Process.send_after(self(), :poll, poll_interval())
228 
    %{state | run: run}
229 
  end
230
231 
  defp poll_interval do
232 
    :openagents
233 
    |> Application.get_env(:box_api, [])
234 
    |> Keyword.get(:run_poll_interval_ms, 1_000)
235 
    |> max(1)
236 
  end
237
238 
  defp box_id(%Run{conversation_box: %{box_id: box_id}}), do: box_id
239
240 
  defp box_id(%Run{conversation_box_id: id}) do
241 
    Repo.get!(OpenAgents.Box.ConversationBox, id).box_id
242 
  end
243
244 
  defp terminal_state?(%Run{} = run), do: Run.terminal?(run)
245
246 
  defp via(run_id), do: {:via, Registry, {OpenAgents.BoxRunRegistry, run_id}}
247 
end
lib/openagents/box_run_supervisor.ex added +12

@@ -0,0 +1,12 @@

1 
defmodule OpenAgents.BoxRunSupervisor do
2 
  @moduledoc false
3
4 
  use DynamicSupervisor
5
6 
  def start_link(options \\ []) do
7 
    DynamicSupervisor.start_link(__MODULE__, options, name: __MODULE__)
8 
  end
9
10 
  @impl true
11 
  def init(_options), do: DynamicSupervisor.init(strategy: :one_for_one)
12 
end
lib/openagents/box_runs.ex added +335

@@ -0,0 +1,335 @@

1 
defmodule OpenAgents.BoxRuns do
2 
  @moduledoc "Durable detach-and-poll runs for long Box commands."
3
4 
  import Ecto.Query
5
6 
  alias OpenAgents.Box
7 
  alias OpenAgents.Box.ConversationBox
8 
  alias OpenAgents.Box.Run
9 
  alias OpenAgents.Repo
10 
  alias OpenAgents.Tools.BoxOutput
11
12 
  @terminal_states Run.terminal_states()
13
14 
  @spec start_run(String.t(), String.t(), map(), String.t(), String.t()) ::
15 
          {:ok, Run.t()} | {:error, term()}
16 
  def start_run(conversation_id, box_id, principal, command, idempotency_key)
17 
      when is_binary(conversation_id) and is_binary(box_id) and is_map(principal) and
18 
             is_binary(command) and is_binary(idempotency_key) do
19 
    with {:ok, _box} <- Box.get_box(conversation_id, box_id),
20 
         {:ok, box_record} <- box_record(conversation_id, box_id),
21 
         {:ok, result} <- admit(box_record, principal, command, idempotency_key) do
22 
      case result do
23 
        {:existing, run} ->
24 
          {:ok, Repo.preload(run, :conversation_box)}
25
26 
        {:new, run} ->
27 
          case start_worker(run.id) do
28 
            {:ok, _pid} -> {:ok, Repo.preload(run, :conversation_box)}
29 
            {:error, reason} -> {:error, reason}
30 
          end
31 
      end
32 
    end
33 
  end
34
35 
  @spec list_runs(String.t(), String.t()) :: [Run.t()]
36 
  def list_runs(conversation_id, box_id) do
37 
    Repo.all(
38 
      from run in Run,
39 
        join: box in ConversationBox,
40 
        on: box.id == run.conversation_box_id,
41 
        where: run.conversation_id == ^conversation_id and box.box_id == ^box_id,
42 
        order_by: [desc: run.inserted_at],
43 
        preload: [conversation_box: box]
44 
    )
45 
  end
46
47 
  @spec get_run(String.t(), String.t(), String.t()) :: {:ok, Run.t()} | {:error, :not_found}
48 
  def get_run(conversation_id, box_id, run_id) do
49 
    with {:ok, _uuid} <- Ecto.UUID.cast(run_id) do
50 
      case Repo.one(
51 
             from run in Run,
52 
               join: box in ConversationBox,
53 
               on: box.id == run.conversation_box_id,
54 
               where:
55 
                 run.id == ^run_id and run.conversation_id == ^conversation_id and
56 
                   box.box_id == ^box_id,
57 
               preload: [conversation_box: box]
58 
           ) do
59 
        %Run{} = run -> {:ok, run}
60 
        nil -> {:error, :not_found}
61 
      end
62 
    else
63 
      :error -> {:error, :not_found}
64 
    end
65 
  end
66
67 
  @spec cancel(Run.t()) :: {:ok, Run.t()} | {:error, term()}
68 
  def cancel(%Run{} = run) do
69 
    case Repo.transaction(fn ->
70 
           locked = Repo.one!(from r in Run, where: r.id == ^run.id, lock: "FOR UPDATE")
71
72 
           if Run.terminal?(locked) do
73 
             locked
74 
           else
75 
             locked
76 
             |> Run.changeset(%{
77 
               cancellation_requested_at: locked.cancellation_requested_at || now()
78 
             })
79 
             |> Repo.update!()
80 
           end
81 
         end) do
82 
      {:ok, cancelled_run} ->
83 
        if Run.terminal?(cancelled_run),
84 
          do: {:ok, cancelled_run},
85 
          else: cancel_worker(cancelled_run)
86
87 
      {:error, reason} ->
88 
        {:error, reason}
89 
    end
90 
  rescue
91 
    Ecto.NoResultsError -> {:error, :not_found}
92 
  end
93
94 
  @spec read_output(Run.t(), non_neg_integer()) :: {:ok, map()} | {:error, term()}
95 
  def read_output(%Run{} = run, offset) when is_integer(offset) and offset >= 0 do
96 
    start_offset = max(offset, run.output_base_offset)
97 
    relative = min(start_offset - run.output_base_offset, byte_size(run.output))
98 
    output = binary_part(run.output, relative, byte_size(run.output) - relative)
99
100 
    {:ok,
101 
     %{
102 
       "output" => output,
103 
       "offset" => offset,
104 
       "next_offset" => run.last_output_offset,
105 
       "output_base_offset" => run.output_base_offset,
106 
       "truncated" => offset < run.output_base_offset
107 
     }}
108 
  end
109
110 
  @spec reconcile_non_terminal() :: :ok
111 
  def reconcile_non_terminal do
112 
    Repo.all(from run in Run, where: run.state not in ^@terminal_states)
113 
    |> Enum.each(fn run -> _ = start_worker(run.id) end)
114
115 
    :ok
116 
  end
117
118 
  @spec start_worker(String.t()) :: DynamicSupervisor.on_start_child()
119 
  def start_worker(run_id) do
120 
    DynamicSupervisor.start_child(OpenAgents.BoxRunSupervisor, {OpenAgents.BoxRunServer, run_id})
121 
  end
122
123 
  @spec claim_dispatch(String.t()) :: {:ok, Run.t()} | {:error, term()}
124 
  def claim_dispatch(run_id) do
125 
    Repo.transaction(fn ->
126 
      run = Repo.one!(from r in Run, where: r.id == ^run_id, lock: "FOR UPDATE")
127
128 
      if is_nil(run.dispatch_attempted_at) and run.state == "admitted" do
129 
        run
130 
        |> Run.changeset(%{dispatch_attempted_at: now()})
131 
        |> Repo.update!()
132 
      else
133 
        run
134 
      end
135 
    end)
136 
  rescue
137 
    Ecto.NoResultsError -> {:error, :not_found}
138 
  end
139
140 
  @spec mark_dispatched(String.t(), integer() | nil) :: {:ok, Run.t()}
141 
  def mark_dispatched(run_id, pid) do
142 
    update_run(run_id, %{
143 
      state: "dispatched",
144 
      pid: pid,
145 
      dispatched_at: now(),
146 
      started_at: now()
147 
    })
148 
  end
149
150 
  @spec mark_probe_attempted(String.t()) :: {:ok, Run.t()}
151 
  def mark_probe_attempted(run_id) do
152 
    update_run(run_id, %{probe_attempted_at: now()})
153 
  end
154
155 
  @spec mark_lost(String.t(), String.t()) :: {:ok, Run.t()}
156 
  def mark_lost(run_id, reason), do: update_run(run_id, terminal_attrs("lost", reason))
157
158 
  @spec record_poll(String.t(), map()) :: {:ok, Run.t()}
159 
  def record_poll(run_id, %{present: true, log_size: log_size, output: chunk} = result) do
160 
    Repo.transaction(fn ->
161 
      run = Repo.one!(from r in Run, where: r.id == ^run_id, lock: "FOR UPDATE")
162
163 
      if Run.terminal?(run) do
164 
        run
165 
      else
166 
        {chunk, _truncated_chunk} =
167 
          if log_size > run.last_output_offset do
168 
            BoxOutput.bounded(chunk)
169 
          else
170 
            {"", false}
171 
          end
172
173 
        accepted_offset = max(run.last_output_offset, log_size)
174 
        combined = run.output <> chunk
175 
        {bounded, truncated} = BoxOutput.bounded(combined)
176
177 
        base_offset =
178 
          if truncated, do: accepted_offset - byte_size(bounded), else: run.output_base_offset
179
180 
        attrs = %{
181 
          output: bounded,
182 
          output_base_offset: max(base_offset, 0),
183 
          last_output_offset: accepted_offset,
184 
          state: if(result[:exit_status] == nil, do: "running", else: run.state)
185 
        }
186
187 
        run
188 
        |> Run.changeset(attrs)
189 
        |> Repo.update!()
190 
      end
191 
    end)
192 
  end
193
194 
  def record_poll(run_id, _result), do: mark_lost(run_id, "run_directory_missing")
195
196 
  @spec finish(String.t(), String.t(), integer() | nil, String.t() | nil) ::
197 
          {:ok, Run.t()}
198 
  def finish(run_id, state, exit_status \\ nil, reason \\ nil)
199 
      when state in @terminal_states do
200 
    attrs = %{state: state, exit_status: exit_status, finished_at: now()}
201 
    attrs = if state == "timed_out", do: Map.put(attrs, :timed_out, true), else: attrs
202 
    attrs = if reason, do: Map.put(attrs, :failure_reason, reason), else: attrs
203 
    update_run(run_id, attrs)
204 
  end
205
206 
  @spec mark_cancellation_effective(String.t()) :: {:ok, Run.t()}
207 
  def mark_cancellation_effective(run_id) do
208 
    update_run(run_id, %{
209 
      state: "cancelled",
210 
      cancellation_effective_at: now(),
211 
      finished_at: now()
212 
    })
213 
  end
214
215 
  @spec mark_timeout(String.t()) :: {:ok, Run.t()}
216 
  def mark_timeout(run_id), do: update_run(run_id, %{timed_out: true})
217
218 
  defp admit(box_record, principal, command, idempotency_key) do
219 
    Repo.transaction(fn ->
220 
      existing =
221 
        Repo.one(
222 
          from run in Run,
223 
            where:
224 
              run.conversation_id == ^box_record.conversation_id and
225 
                run.idempotency_key == ^idempotency_key
226 
        )
227
228 
      cond do
229 
        existing && existing.conversation_box_id == box_record.id && existing.command == command ->
230 
          {:existing, existing}
231
232 
        existing ->
233 
          Repo.rollback(:box_run_idempotency_conflict)
234
235 
        Repo.exists?(
236 
          from run in Run,
237 
            where:
238 
              run.conversation_box_id == ^box_record.id and
239 
                  run.state not in ^@terminal_states
240 
        ) ->
241 
          Repo.rollback(:box_run_in_progress)
242
243 
        true ->
244 
          admitted_at = now()
245 
          deadline_at = DateTime.add(admitted_at, maximum_duration_seconds(), :second)
246 
          run_id = Ecto.UUID.generate()
247
248 
          attrs = %{
249 
            conversation_id: box_record.conversation_id,
250 
            conversation_box_id: box_record.id,
251 
            requesting_principal: principal,
252 
            command: command,
253 
            idempotency_key: idempotency_key,
254 
            id: run_id,
255 
            run_directory: run_directory(principal, run_id),
256 
            admitted_at: admitted_at,
257 
            deadline_at: deadline_at
258 
          }
259
260 
          run =
261 
            %Run{}
262 
            |> Run.changeset(attrs)
263 
            |> Repo.insert!()
264
265 
          {:new, run}
266 
      end
267 
    end)
268 
  end
269
270 
  defp box_record(conversation_id, box_id) do
271 
    case Repo.one(
272 
           from box in ConversationBox,
273 
             where: box.conversation_id == ^conversation_id and box.box_id == ^box_id
274 
         ) do
275 
      %ConversationBox{} = box -> {:ok, box}
276 
      nil -> {:error, :box_not_owned}
277 
    end
278 
  end
279
280 
  defp cancel_worker(run) do
281 
    case Registry.lookup(OpenAgents.BoxRunRegistry, run.id) do
282 
      [{pid, _value}] ->
283 
        GenServer.cast(pid, :cancel)
284 
        {:ok, run}
285
286 
      [] ->
287 
        case start_worker(run.id) do
288 
          {:ok, pid} ->
289 
            GenServer.cast(pid, :cancel)
290 
            {:ok, run}
291
292 
          {:error, {:already_started, _pid}} ->
293 
            {:ok, run}
294
295 
          {:error, reason} ->
296 
            {:error, reason}
297 
        end
298 
    end
299 
  end
300
301 
  defp update_run(run_id, attrs) do
302 
    with {:ok, _uuid} <- Ecto.UUID.cast(run_id) do
303 
      case Repo.get(Run, run_id) do
304 
        %Run{} = run -> {:ok, run |> Run.changeset(attrs) |> Repo.update!()}
305 
        nil -> {:error, :not_found}
306 
      end
307 
    else
308 
      :error -> {:error, :not_found}
309 
    end
310 
  end
311
312 
  defp terminal_attrs(state, reason),
313 
    do: %{state: state, failure_reason: reason, finished_at: now()}
314
315 
  defp now, do: DateTime.utc_now() |> DateTime.truncate(:microsecond)
316
317 
  defp maximum_duration_seconds do
318 
    settings = Application.get_env(:openagents, :box_api, [])
319 
    configured = Keyword.get(settings, :run_max_duration_seconds, 1_800)
320 
    ttl = Keyword.get(settings, :ttl_seconds, 3_600)
321 
    configured |> max(1) |> min(max(ttl, 1))
322 
  end
323
324 
  defp run_directory(principal, run_id) do
325 
    principal_id = Map.get(principal, "id") || Map.get(principal, :id) || "unknown"
326 
    principal_id = principal_path_segment(principal_id)
327 
    "$HOME/.openagents/box-runs/users/#{principal_id}/#{run_id}"
328 
  end
329
330 
  defp principal_path_segment(principal_id) when is_binary(principal_id) do
331 
    Regex.replace(~r/[^A-Za-z0-9_-]/, principal_id, "-")
332 
  end
333
334 
  defp principal_path_segment(_principal_id), do: "unknown"
335 
end
lib/openagents_web/api_route_authority.ex modified +7

@@ -93,6 +93,13 @@ defmodule OpenAgentsWeb.ApiRouteAuthority do

93 93 
      "get /api/v3/conversations/:conversation_id/boxes/:box_id" => :required_bearer,
94 94 
      "post /api/v3/conversations/:conversation_id/boxes/:box_id/commands" => :required_bearer,
95 95 
      "post /api/v3/conversations/:conversation_id/boxes/:box_id/stop" => :required_bearer,
96 
      "post /api/v3/conversations/:conversation_id/boxes/:box_id/runs" => :required_bearer,
97 
      "get /api/v3/conversations/:conversation_id/boxes/:box_id/runs" => :required_bearer,
98 
      "get /api/v3/conversations/:conversation_id/boxes/:box_id/runs/:run_id" => :required_bearer,
99 
      "get /api/v3/conversations/:conversation_id/boxes/:box_id/runs/:run_id/output" =>
100 
        :required_bearer,
101 
      "post /api/v3/conversations/:conversation_id/boxes/:box_id/runs/:run_id/cancel" =>
102 
        :required_bearer,
96 103 
      "delete /api/v3/repos/:owner/:repo" => :required_bearer,
97 104 
      "delete /api/v3/repos/:owner/:repo/issues/:issue_number/assignees" => :required_bearer,
98 105 
      "delete /api/v3/repos/:owner/:repo/issues/:issue_number/labels/:name" => :required_bearer,
lib/openagents_web/box_rate_limiter.ex modified +6 -1

@@ -16,7 +16,8 @@ defmodule OpenAgentsWeb.BoxRateLimiter do

16 16 
  end
17 17
18 18 
  @spec allow?(term(), :create | :command) :: :ok | {:error, :rate_limited}
19 
  def allow?(principal, operation) when operation in [:create, :command] do
19 
  def allow?(principal, operation)
20 
      when operation in [:create, :command, :run_create, :run_command] do
20 21 
    GenServer.call(__MODULE__, {:allow, principal, operation})
21 22 
  end
22 23

@@ -76,9 +77,13 @@ defmodule OpenAgentsWeb.BoxRateLimiter do

76 77
77 78 
  defp rate_limit_key(:create), do: :create_rate_limit
78 79 
  defp rate_limit_key(:command), do: :command_rate_limit
80 
  defp rate_limit_key(:run_create), do: :run_create_rate_limit
81 
  defp rate_limit_key(:run_command), do: :run_command_rate_limit
79 82
80 83 
  defp default_limit(:create), do: @default_create_limit
81 84 
  defp default_limit(:command), do: @default_command_limit
85 
  defp default_limit(:run_create), do: @default_create_limit
86 
  defp default_limit(:run_command), do: @default_command_limit
82 87
83 88 
  defp setting(key, default) do
84 89 
    case Keyword.get(Application.get_env(:openagents, :box_api, []), key, default) do
lib/openagents_web/controllers/box_run_controller.ex added +215

@@ -0,0 +1,215 @@

1 
defmodule OpenAgentsWeb.BoxRunController do
2 
  @moduledoc "Authenticated API access to durable detached Box runs."
3
4 
  use OpenAgentsWeb, :controller
5
6 
  alias OpenAgents.Box
7 
  alias OpenAgents.Box.Run
8 
  alias OpenAgents.BoxRuns
9 
  alias OpenAgents.Conversations
10 
  alias OpenAgents.Conversations.Conversation
11 
  alias OpenAgentsWeb.BoxRateLimiter
12
13 
  @maximum_command_bytes 8_000
14 
  @maximum_idempotency_key_bytes 256
15
16 
  def create(conn, %{
17 
        "conversation_id" => conversation_id,
18 
        "box_id" => box_id,
19 
        "command" => command
20 
      })
21 
      when is_binary(command) do
22 
    idempotency_key =
23 
      Map.get(conn.params, "idempotency_key") ||
24 
        List.first(Plug.Conn.get_req_header(conn, "idempotency-key"))
25
26 
    with :ok <- validate_command(command),
27 
         :ok <- validate_idempotency_key(idempotency_key),
28 
         {:ok, _conversation} <- owned_conversation(conn, conversation_id),
29 
         :ok <- BoxRateLimiter.allow?(conn.assigns.current_user.id, :run_create),
30 
         {:ok, run} <-
31 
           BoxRuns.start_run(
32 
             conversation_id,
33 
             box_id,
34 
             %{"type" => "user", "id" => conn.assigns.current_user.id},
35 
             command,
36 
             idempotency_key
37 
           ) do
38 
      conn
39 
      |> put_status(:accepted)
40 
      |> json(%{"run" => projection(run)})
41 
    else
42 
      {:error, :invalid_command} ->
43 
        refusal(conn, :unprocessable_entity, "invalid_command")
44
45 
      {:error, :invalid_idempotency_key} ->
46 
        refusal(conn, :unprocessable_entity, "invalid_idempotency_key")
47
48 
      {:error, :not_found} ->
49 
        refusal(conn, :not_found, "conversation_not_found")
50
51 
      {:error, :rate_limited} ->
52 
        refusal(conn, :too_many_requests, "box_api_rate_limited")
53
54 
      {:error, reason} ->
55 
        run_error(conn, reason)
56 
    end
57 
  end
58
59 
  def create(conn, _params), do: refusal(conn, :unprocessable_entity, "invalid_command")
60
61 
  def index(conn, %{"conversation_id" => conversation_id, "box_id" => box_id}) do
62 
    with {:ok, _conversation} <- owned_conversation(conn, conversation_id),
63 
         {:ok, _box} <- Box.get_box(conversation_id, box_id) do
64 
      json(conn, %{"runs" => Enum.map(BoxRuns.list_runs(conversation_id, box_id), &projection/1)})
65 
    else
66 
      {:error, :not_found} -> refusal(conn, :not_found, "conversation_not_found")
67 
      {:error, reason} -> run_error(conn, reason)
68 
    end
69 
  end
70
71 
  def show(conn, params) do
72 
    with {:ok, _conversation} <- owned_conversation(conn, params["conversation_id"]),
73 
         {:ok, run} <- owned_run(params) do
74 
      json(conn, %{"run" => projection(run)})
75 
    else
76 
      {:error, :not_found} -> refusal(conn, :not_found, "conversation_not_found")
77 
      {:error, :run_not_found} -> refusal(conn, :not_found, "box_run_not_found")
78 
    end
79 
  end
80
81 
  def output(conn, params) do
82 
    with {:ok, _conversation} <- owned_conversation(conn, params["conversation_id"]),
83 
         {:ok, run} <- owned_run(params),
84 
         {:ok, offset} <- parse_offset(params["offset"]),
85 
         {:ok, result} <- BoxRuns.read_output(run, offset) do
86 
      json(conn, %{"run_id" => run.id, "output" => result})
87 
    else
88 
      {:error, :not_found} -> refusal(conn, :not_found, "conversation_not_found")
89 
      {:error, :run_not_found} -> refusal(conn, :not_found, "box_run_not_found")
90 
      {:error, :invalid_offset} -> refusal(conn, :unprocessable_entity, "invalid_output_offset")
91 
    end
92 
  end
93
94 
  def cancel(conn, params) do
95 
    with {:ok, _conversation} <- owned_conversation(conn, params["conversation_id"]),
96 
         {:ok, run} <- owned_run(params),
97 
         {:ok, run} <- BoxRuns.cancel(run) do
98 
      conn
99 
      |> put_status(:accepted)
100 
      |> json(%{"run" => projection(run)})
101 
    else
102 
      {:error, :not_found} -> refusal(conn, :not_found, "conversation_not_found")
103 
      {:error, :run_not_found} -> refusal(conn, :not_found, "box_run_not_found")
104 
      {:error, reason} -> run_error(conn, reason)
105 
    end
106 
  end
107
108 
  defp owned_conversation(conn, conversation_id) do
109 
    case Conversations.get_conversation_for_user(conn.assigns.current_user, conversation_id) do
110 
      %Conversation{} = conversation -> {:ok, conversation}
111 
      nil -> {:error, :not_found}
112 
    end
113 
  end
114
115 
  defp owned_run(params) do
116 
    case BoxRuns.get_run(params["conversation_id"], params["box_id"], params["run_id"]) do
117 
      {:ok, run} -> {:ok, run}
118 
      {:error, :not_found} -> {:error, :run_not_found}
119 
    end
120 
  end
121
122 
  defp validate_command(command) do
123 
    if String.trim(command) != "" and String.valid?(command) and
124 
         not String.contains?(command, "\0") and byte_size(command) <= @maximum_command_bytes do
125 
      :ok
126 
    else
127 
      {:error, :invalid_command}
128 
    end
129 
  end
130
131 
  defp validate_idempotency_key(key)
132 
       when is_binary(key) and byte_size(key) > 0 and
133 
              byte_size(key) <= @maximum_idempotency_key_bytes,
134 
       do: :ok
135
136 
  defp validate_idempotency_key(_key), do: {:error, :invalid_idempotency_key}
137
138 
  defp parse_offset(nil), do: {:ok, 0}
139 
  defp parse_offset(offset) when is_integer(offset) and offset >= 0, do: {:ok, offset}
140
141 
  defp parse_offset(offset) when is_binary(offset) do
142 
    case Integer.parse(offset) do
143 
      {value, ""} when value >= 0 -> {:ok, value}
144 
      _invalid -> {:error, :invalid_offset}
145 
    end
146 
  end
147
148 
  defp parse_offset(_offset), do: {:error, :invalid_offset}
149
150 
  defp projection(%Run{} = run) do
151 
    %{
152 
      "id" => run.id,
153 
      "box_id" => run.conversation_box.box_id,
154 
      "command" => run.command,
155 
      "state" => run.state,
156 
      "exit_status" => run.exit_status,
157 
      "timed_out" => run.timed_out,
158 
      "output_offset" => run.last_output_offset,
159 
      "output_base_offset" => run.output_base_offset,
160 
      "failure_reason" => run.failure_reason,
161 
      "admitted_at" => iso8601(run.admitted_at),
162 
      "dispatched_at" => iso8601(run.dispatched_at),
163 
      "started_at" => iso8601(run.started_at),
164 
      "finished_at" => iso8601(run.finished_at),
165 
      "deadline_at" => iso8601(run.deadline_at),
166 
      "cancellation_requested_at" => iso8601(run.cancellation_requested_at),
167 
      "cancellation_effective_at" => iso8601(run.cancellation_effective_at)
168 
    }
169 
  end
170
171 
  defp iso8601(nil), do: nil
172 
  defp iso8601(%DateTime{} = value), do: DateTime.to_iso8601(value)
173
174 
  defp run_error(conn, :box_run_in_progress),
175 
    do: refusal(conn, :conflict, "box_run_in_progress")
176
177 
  defp run_error(conn, :box_run_idempotency_conflict),
178 
    do: refusal(conn, :conflict, "box_run_idempotency_conflict")
179
180 
  defp run_error(conn, :box_quota_reached),
181 
    do: refusal(conn, :conflict, "box_quota_reached")
182
183 
  defp run_error(conn, reason) when reason in [:box_not_owned, :box_not_found],
184 
    do: refusal(conn, :not_found, "box_not_found")
185
186 
  defp run_error(conn, :box_billing_required),
187 
    do: refusal(conn, :payment_required, "box_billing_required")
188
189 
  defp run_error(conn, :box_rate_limited),
190 
    do: refusal(conn, :too_many_requests, "box_provider_rate_limited")
191
192 
  defp run_error(conn, reason) when reason in [:box_not_configured, :box_unreachable],
193 
    do: refusal(conn, :service_unavailable, Atom.to_string(reason))
194
195 
  defp run_error(conn, :box_response_invalid),
196 
    do: refusal(conn, :bad_gateway, "box_provider_response_invalid")
197
198 
  defp run_error(conn, {:box_request_refused, _status, _code}),
199 
    do: refusal(conn, :bad_gateway, "box_provider_request_refused")
200
201 
  defp run_error(conn, :box_unauthorized),
202 
    do: refusal(conn, :bad_gateway, "box_provider_unauthorized")
203
204 
  defp run_error(conn, :box_stopped),
205 
    do: refusal(conn, :conflict, "box_stopped")
206
207 
  defp run_error(conn, :box_not_ready),
208 
    do: refusal(conn, :conflict, "box_not_ready")
209
210 
  defp run_error(conn, reason), do: refusal(conn, :bad_gateway, Atom.to_string(reason))
211
212 
  defp refusal(conn, status, code) do
213 
    conn |> put_status(status) |> json(%{"error" => %{"code" => code}})
214 
  end
215 
end
lib/openagents_web/route_authority.ex modified +31

@@ -261,6 +261,37 @@ defmodule OpenAgentsWeb.RouteAuthority do

261 261 
  defp policy(%{path: "/api/v3/conversations/:conversation_id/boxes/:box_id/stop", verb: :post}),
262 262 
    do: declaration(:authenticated_api, "human account bearer token", "box:control", true)
263 263
264 
  defp policy(%{
265 
         path: "/api/v3/conversations/:conversation_id/boxes/:box_id/runs",
266 
         verb: verb
267 
       })
268 
       when verb in [:get, :post],
269 
       do:
270 
         declaration(
271 
           :authenticated_api,
272 
           "human account bearer token",
273 
           "box:control",
274 
           verb == :post
275 
         )
276
277 
  defp policy(%{
278 
         path: "/api/v3/conversations/:conversation_id/boxes/:box_id/runs/:run_id",
279 
         verb: :get
280 
       }),
281 
       do: declaration(:authenticated_api, "human account bearer token", "box:control", false)
282
283 
  defp policy(%{
284 
         path: "/api/v3/conversations/:conversation_id/boxes/:box_id/runs/:run_id/output",
285 
         verb: :get
286 
       }),
287 
       do: declaration(:authenticated_api, "human account bearer token", "box:control", false)
288
289 
  defp policy(%{
290 
         path: "/api/v3/conversations/:conversation_id/boxes/:box_id/runs/:run_id/cancel",
291 
         verb: :post
292 
       }),
293 
       do: declaration(:authenticated_api, "human account bearer token", "box:control", true)
294
264 295 
  defp policy(%{path: "/api/operator/artifact-listings" <> _path, verb: verb}),
265 296 
    do:
266 297 
      declaration(
lib/openagents_web/router.ex modified +5

@@ -421,6 +421,11 @@ defmodule OpenAgentsWeb.Router do

421 421 
    get "/:box_id", BoxController, :show
422 422 
    post "/:box_id/commands", BoxController, :command
423 423 
    post "/:box_id/stop", BoxController, :stop
424 
    post "/:box_id/runs", BoxRunController, :create
425 
    get "/:box_id/runs", BoxRunController, :index
426 
    get "/:box_id/runs/:run_id", BoxRunController, :show
427 
    get "/:box_id/runs/:run_id/output", BoxRunController, :output
428 
    post "/:box_id/runs/:run_id/cancel", BoxRunController, :cancel
424 429 
  end
425 430
426 431 
  scope "/api/v3", OpenAgentsWeb do
priv/migration_lineages/prior-2026-08-19.json modified +2 -1

@@ -255,7 +255,8 @@

255 255 
    20260823122038,
256 256 
    20260823122039,
257 257 
    20260823122040,
258 
    20260823131739
258 
    20260823131739,
259 
    20260823133243
259 260 
  ],
260 261 
  "required_tables": [
261 262 
    "users",
priv/repo/migrations/20260823133243_create_box_runs.exs added +60

@@ -0,0 +1,60 @@

1 
defmodule OpenAgents.Repo.Migrations.CreateBoxRuns do
2 
  use Ecto.Migration
3
4 
  def change do
5 
    create table(:box_runs, primary_key: false) do
6 
      add :id, :binary_id, primary_key: true
7
8 
      add :conversation_id,
9 
          references(:conversations, type: :binary_id, on_delete: :delete_all),
10 
          null: false
11
12 
      add :conversation_box_id,
13 
          references(:conversation_boxes, type: :binary_id, on_delete: :restrict),
14 
          null: false
15
16 
      add :requesting_principal, :map, null: false
17 
      add :command, :text, null: false
18 
      add :idempotency_key, :string, null: false
19 
      add :state, :string, null: false, default: "admitted"
20 
      add :exit_status, :integer
21 
      add :timed_out, :boolean, null: false, default: false
22 
      add :output, :text, null: false, default: ""
23 
      add :output_base_offset, :integer, null: false, default: 0
24 
      add :last_output_offset, :integer, null: false, default: 0
25 
      add :pid, :integer
26 
      add :run_directory, :string, null: false
27 
      add :failure_reason, :string
28 
      add :dispatch_attempted_at, :utc_datetime_usec
29 
      add :probe_attempted_at, :utc_datetime_usec
30 
      add :admitted_at, :utc_datetime_usec, null: false
31 
      add :dispatched_at, :utc_datetime_usec
32 
      add :started_at, :utc_datetime_usec
33 
      add :finished_at, :utc_datetime_usec
34 
      add :cancellation_requested_at, :utc_datetime_usec
35 
      add :cancellation_effective_at, :utc_datetime_usec
36 
      add :deadline_at, :utc_datetime_usec, null: false
37
38 
      timestamps(type: :utc_datetime_usec)
39 
    end
40
41 
    create unique_index(:box_runs, [:conversation_id, :idempotency_key])
42
43 
    create unique_index(:box_runs, [:conversation_box_id],
44 
             name: :box_runs_one_active_per_box_index,
45 
             where: "state IN ('admitted', 'dispatched', 'running')"
46 
           )
47
48 
    create index(:box_runs, [:conversation_id, :inserted_at])
49 
    create index(:box_runs, [:state])
50
51 
    create constraint(:box_runs, :box_runs_state_check,
52 
             check:
53 
               "state IN ('admitted', 'dispatched', 'running', 'completed', 'failed', 'cancelled', 'timed_out', 'lost')"
54 
           )
55
56 
    create constraint(:box_runs, :box_runs_offsets_check,
57 
             check: "output_base_offset >= 0 AND last_output_offset >= output_base_offset"
58 
           )
59 
  end
60 
end
test/openagents/box_client_runs_test.exs added +80

@@ -0,0 +1,80 @@

1 
defmodule OpenAgents.BoxClientRunsTest do
2 
  use ExUnit.Case, async: false
3
4 
  alias OpenAgents.Box.Client
5
6 
  setup {Req.Test, :verify_on_exit!}
7
8 
  setup do
9 
    original_api = Application.get_env(:openagents, :box_api)
10 
    original_key = Application.get_env(:openagents, :box_api_key)
11
12 
    Application.put_env(:openagents, :box_api,
13 
      base_url: "https://box-api.internal",
14 
      request_options: [plug: {Req.Test, __MODULE__}]
15 
    )
16
17 
    Application.put_env(:openagents, :box_api_key, "box-client-test")
18
19 
    on_exit(fn ->
20 
      restore_env(:box_api, original_api)
21 
      restore_env(:box_api_key, original_key)
22 
    end)
23 
  end
24
25 
  test "dispatch uses one detached mkdir-and-launch command", do: begin_dispatch()
26
27 
  test "poll decodes bounded output from a recorded offset" do
28 
    encoded = Base.encode64("hello")
29
30 
    Req.Test.expect(__MODULE__, fn request ->
31 
      assert request.method == "POST"
32 
      assert request.request_path == "/boxes/bx_8bhkse3n/commands"
33 
      body = request.body_params
34 
      assert body["command"] =~ "OA_SIZE"
35 
      assert body["command"] =~ "dd"
36
37 
      Req.Test.json(request, %{
38 
        "stdout" => "OA_PRESENT=1\nOA_SIZE=5\nOA_DATA=#{encoded}\nOA_ALIVE=1\n"
39 
      })
40 
    end)
41
42 
    assert {:ok, %{present: true, log_size: 5, output: "hello", exit_status: nil, alive: true}} =
43 
             Client.poll_run("bx_8bhkse3n", "11111111-1111-4111-8111-111111111111", 0)
44 
  end
45
46 
  test "an exit sentinel is parsed as an integer status" do
47 
    Req.Test.expect(__MODULE__, fn request ->
48 
      Req.Test.json(request, %{
49 
        "stdout" => "OA_PRESENT=1\nOA_SIZE=0\nOA_DATA=\nOA_EXIT=17\nOA_ALIVE=0\n"
50 
      })
51 
    end)
52
53 
    assert {:ok, %{exit_status: 17, alive: false}} =
54 
             Client.poll_run("bx_8bhkse3n", "11111111-1111-4111-8111-111111111111", 0)
55 
  end
56
57 
  defp begin_dispatch do
58 
    Req.Test.expect(__MODULE__, fn request ->
59 
      assert request.method == "POST"
60 
      assert request.request_path == "/boxes/bx_8bhkse3n/commands"
61 
      body = request.body_params
62 
      assert body["timeoutSeconds"] == 30
63 
      assert body["command"] =~ "mkdir"
64 
      assert body["command"] =~ "setsid"
65 
      assert body["command"] =~ "output.log"
66 
      assert body["command"] =~ "exit-code"
67 
      Req.Test.json(request, %{"stdout" => "4242\n"})
68 
    end)
69
70 
    assert {:ok, 4242} =
71 
             Client.dispatch_run(
72 
               "bx_8bhkse3n",
73 
               "11111111-1111-4111-8111-111111111111",
74 
               "echo detached"
75 
             )
76 
  end
77
78 
  defp restore_env(key, nil), do: Application.delete_env(:openagents, key)
79 
  defp restore_env(key, value), do: Application.put_env(:openagents, key, value)
80 
end
test/openagents/box_runs_test.exs added +490

@@ -0,0 +1,490 @@

1 
defmodule OpenAgents.BoxRunsTest do
2 
  use OpenAgents.DataCase, async: false
3
4 
  alias OpenAgents.Box.ConversationBox
5 
  alias OpenAgents.Box.Run
6 
  alias OpenAgents.BoxRuns
7 
  alias OpenAgents.Conversations
8 
  alias OpenAgents.Repo
9
10 
  setup {Req.Test, :verify_on_exit!}
11
12 
  setup do
13 
    Req.Test.set_req_test_to_shared()
14
15 
    original_api = Application.get_env(:openagents, :box_api)
16 
    original_key = Application.get_env(:openagents, :box_api_key)
17
18 
    Application.put_env(:openagents, :box_api,
19 
      base_url: "https://box-api.internal",
20 
      run_poll_interval_ms: 60_000,
21 
      request_options: [plug: {Req.Test, __MODULE__}, retry_delay: 0]
22 
    )
23
24 
    Application.put_env(:openagents, :box_api_key, "box-runs-test-key")
25
26 
    on_exit(fn ->
27 
      restore_env(:box_api, original_api)
28 
      restore_env(:box_api_key, original_key)
29 
    end)
30
31 
    :ok
32 
  end
33
34 
  test "run states and bounded offset reads are durable" do
35 
    {:ok, conversation} = Conversations.ensure_conversation("box-runs-state")
36
37 
    box =
38 
      %ConversationBox{}
39 
      |> ConversationBox.changeset(%{
40 
        conversation_id: conversation.id,
41 
        box_id: "bx_run_state",
42 
        state: "ready",
43 
        setup_status: "done"
44 
      })
45 
      |> Repo.insert!()
46
47 
    now = DateTime.utc_now() |> DateTime.truncate(:microsecond)
48
49 
    run =
50 
      %Run{}
51 
      |> Run.changeset(%{
52 
        conversation_id: conversation.id,
53 
        conversation_box_id: box.id,
54 
        requesting_principal: %{"type" => "user", "id" => "user"},
55 
        command: "printf hello",
56 
        idempotency_key: "state-key",
57 
        run_directory: "/tmp/openagents-box-runs/state",
58 
        admitted_at: now,
59 
        deadline_at: DateTime.add(now, 60, :second)
60 
      })
61 
      |> Repo.insert!()
62
63 
    assert {:ok, run} =
64 
             BoxRuns.record_poll(run.id, %{present: true, log_size: 5, output: "hello"})
65
66 
    assert run.state == "running"
67 
    assert run.last_output_offset == 5
68
69 
    assert {:ok, run} = BoxRuns.finish(run.id, "completed", 0)
70 
    assert run.state == "completed"
71 
    assert Run.terminal?(run)
72
73 
    assert {:ok, output} = BoxRuns.read_output(run, 0)
74 
    assert output["output"] == "hello"
75 
    assert output["next_offset"] == 5
76 
    assert output["truncated"] == false
77 
  end
78
79 
  test "one active run is allowed per box" do
80 
    {:ok, conversation} = Conversations.ensure_conversation("box-runs-lane")
81
82 
    box =
83 
      %ConversationBox{}
84 
      |> ConversationBox.changeset(%{
85 
        conversation_id: conversation.id,
86 
        box_id: "bx_run_lane",
87 
        state: "ready",
88 
        setup_status: "done"
89 
      })
90 
      |> Repo.insert!()
91
92 
    now = DateTime.utc_now() |> DateTime.truncate(:microsecond)
93
94 
    attrs = %{
95 
      conversation_id: conversation.id,
96 
      conversation_box_id: box.id,
97 
      requesting_principal: %{"type" => "user", "id" => "user"},
98 
      command: "true",
99 
      run_directory: "/tmp/openagents-box-runs/lane",
100 
      admitted_at: now,
101 
      deadline_at: DateTime.add(now, 60, :second)
102 
    }
103
104 
    assert {:ok, _run} =
105 
             %Run{}
106 
             |> Run.changeset(Map.put(attrs, :idempotency_key, "lane-one"))
107 
             |> Repo.insert()
108
109 
    assert {:error, changeset} =
110 
             %Run{}
111 
             |> Run.changeset(Map.put(attrs, :idempotency_key, "lane-two"))
112 
             |> Repo.insert()
113
114 
    assert %{conversation_box_id: ["has already been taken"]} = errors_on(changeset)
115 
  end
116
117 
  test "every terminal state is durable" do
118 
    {:ok, conversation} = Conversations.ensure_conversation("box-runs-terminal")
119
120 
    box =
121 
      %ConversationBox{}
122 
      |> ConversationBox.changeset(%{
123 
        conversation_id: conversation.id,
124 
        box_id: "bx_run_terminal",
125 
        state: "ready",
126 
        setup_status: "done"
127 
      })
128 
      |> Repo.insert!()
129
130 
    for state <- Run.terminal_states() do
131 
      run = insert_run(conversation.id, box.id, "terminal-#{state}")
132 
      exit_status = if state == "failed", do: 17
133 
      reason = if state in ["lost", "timed_out"], do: "test_reason"
134
135 
      assert {:ok, finished} = BoxRuns.finish(run.id, state, exit_status, reason)
136 
      assert finished.state == state
137 
      assert Run.terminal?(finished)
138 
      assert finished.finished_at
139 
      assert finished.timed_out == (state == "timed_out")
140 
    end
141 
  end
142
143 
  test "poll output is bounded, redacted, and does not duplicate" do
144 
    {:ok, conversation} = Conversations.ensure_conversation("box-runs-output")
145
146 
    box =
147 
      %ConversationBox{}
148 
      |> ConversationBox.changeset(%{
149 
        conversation_id: conversation.id,
150 
        box_id: "bx_run_output",
151 
        state: "ready",
152 
        setup_status: "done"
153 
      })
154 
      |> Repo.insert!()
155
156 
    run = insert_run(conversation.id, box.id, "output-key")
157 
    payload = String.duplicate("x", 30_000) <> " https://openagents.com/clone"
158
159 
    assert {:ok, first} =
160 
             BoxRuns.record_poll(run.id, %{
161 
               present: true,
162 
               log_size: byte_size(payload),
163 
               output: payload
164 
             })
165
166 
    assert {:ok, second} =
167 
             BoxRuns.record_poll(run.id, %{
168 
               present: true,
169 
               log_size: byte_size(payload),
170 
               output: payload
171 
             })
172
173 
    assert second.output == first.output
174 
    assert byte_size(second.output) <= 24 * 1_024
175 
    assert second.output =~ "https://openagents.com/clone"
176 
    assert second.last_output_offset == byte_size(payload)
177 
  end
178
179 
  test "worker drives a detached run to completed" do
180 
    run = insert_worker_run("worker-completed", state: "admitted")
181
182 
    Req.Test.expect(__MODULE__, fn request ->
183 
      assert request.method == "POST"
184 
      assert request.request_path == "/boxes/bx_8bhkse3n/commands"
185 
      assert request.body_params["command"] =~ run.run_directory
186 
      Req.Test.json(request, %{"stdout" => "4242\n"})
187 
    end)
188
189 
    Req.Test.expect(__MODULE__, fn request ->
190 
      assert request.body_params["command"] =~ "OA_EXIT"
191
192 
      Req.Test.json(request, %{
193 
        "stdout" => "OA_PRESENT=1\nOA_SIZE=0\nOA_DATA=\nOA_EXIT=0\nOA_ALIVE=0\n"
194 
      })
195 
    end)
196
197 
    {pid, ref} = start_worker(run.id)
198 
    _ = :sys.get_state(pid)
199 
    send(pid, :poll)
200 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
201 
    assert Repo.get!(Run, run.id).state == "completed"
202 
  end
203
204 
  test "worker records a nonzero exit as failed" do
205 
    run = insert_worker_run("worker-failed", state: "admitted")
206
207 
    Req.Test.expect(__MODULE__, fn request ->
208 
      Req.Test.json(request, %{"stdout" => "4242\n"})
209 
    end)
210
211 
    Req.Test.expect(__MODULE__, fn request ->
212 
      assert request.body_params["command"] =~ "OA_EXIT"
213
214 
      Req.Test.json(request, %{
215 
        "stdout" => "OA_PRESENT=1\nOA_SIZE=0\nOA_DATA=\nOA_EXIT=17\nOA_ALIVE=0\n"
216 
      })
217 
    end)
218
219 
    {pid, ref} = start_worker(run.id)
220 
    _ = :sys.get_state(pid)
221 
    send(pid, :poll)
222 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
223
224 
    failed = Repo.get!(Run, run.id)
225 
    assert failed.state == "failed"
226 
    assert failed.exit_status == 17
227 
  end
228
229 
  test "ambiguous dispatch probes once and becomes lost without redispatch" do
230 
    run = insert_worker_run("worker-ambiguous", state: "admitted")
231
232 
    Req.Test.expect(__MODULE__, fn request ->
233 
      assert request.body_params["command"] =~ "mkdir"
234 
      Req.Test.transport_error(request, :econnrefused)
235 
    end)
236
237 
    Req.Test.expect(__MODULE__, fn request ->
238 
      assert request.body_params["command"] =~ "OA_PRESENT"
239 
      Req.Test.json(request, %{"stdout" => "OA_PRESENT=0\n"})
240 
    end)
241
242 
    {pid, ref} = start_worker(run.id)
243 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
244
245 
    lost = Repo.get!(Run, run.id)
246 
    assert lost.state == "lost"
247 
    assert lost.failure_reason == "dispatch_ambiguous"
248 
  end
249
250 
  test "a present directory after ambiguous dispatch recovers without a fake pid" do
251 
    run = insert_worker_run("worker-probe", state: "admitted")
252
253 
    Req.Test.expect(__MODULE__, fn request ->
254 
      Req.Test.transport_error(request, :closed)
255 
    end)
256
257 
    Req.Test.expect(__MODULE__, fn request ->
258 
      assert request.body_params["command"] =~ "OA_PID"
259 
      Req.Test.json(request, %{"stdout" => "OA_PRESENT=1\n"})
260 
    end)
261
262 
    Req.Test.expect(__MODULE__, fn request ->
263 
      Req.Test.json(request, %{
264 
        "stdout" => "OA_PRESENT=1\nOA_SIZE=0\nOA_DATA=\nOA_ALIVE=1\n"
265 
      })
266 
    end)
267
268 
    {pid, _ref} = start_worker(run.id)
269 
    _ = :sys.get_state(pid)
270 
    send(pid, :poll)
271 
    _ = :sys.get_state(pid)
272
273 
    recovered = Repo.get!(Run, run.id)
274 
    assert recovered.state == "running"
275 
    assert recovered.pid == nil
276 
  end
277
278 
  test "cancellation retries the kill and remains terminal on later read" do
279 
    run = insert_worker_run("worker-cancel", state: "running", pid: 4242)
280
281 
    {pid, ref} = start_worker(run.id)
282 
    _ = :sys.get_state(pid)
283
284 
    assert {:ok, requested} = BoxRuns.cancel(run)
285 
    assert requested.cancellation_requested_at
286
287 
    Req.Test.expect(__MODULE__, fn request ->
288 
      Req.Test.transport_error(request, :econnrefused)
289 
    end)
290
291 
    _ = :sys.get_state(pid)
292
293 
    Req.Test.expect(__MODULE__, fn request ->
294 
      assert request.body_params["command"] =~ "kill"
295 
      Req.Test.json(request, %{"stdout" => "OA_CANCELLED=1\n"})
296 
    end)
297
298 
    send(pid, :poll)
299 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
300
301 
    cancelled = Repo.get!(Run, run.id)
302 
    assert cancelled.state == "cancelled"
303 
    assert cancelled.cancellation_requested_at
304 
    assert cancelled.cancellation_effective_at
305 
    assert cancelled.cancellation_effective_at >= cancelled.cancellation_requested_at
306 
    assert {:ok, later} = BoxRuns.get_run(run.conversation_id, "bx_8bhkse3n", run.id)
307 
    assert later.state == "cancelled"
308 
  end
309
310 
  test "deadline expiration kills the run and reaches timed out" do
311 
    run = insert_worker_run("worker-timeout", state: "running", pid: 4343, deadline_at: past())
312
313 
    Req.Test.expect(__MODULE__, fn request ->
314 
      assert request.body_params["command"] =~ "kill"
315 
      Req.Test.json(request, %{"stdout" => "OA_CANCELLED=1\n"})
316 
    end)
317
318 
    {pid, ref} = start_worker(run.id)
319 
    _ = :sys.get_state(pid)
320 
    send(pid, :poll)
321 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
322 
    timed_out = Repo.get!(Run, run.id)
323 
    assert timed_out.state == "timed_out"
324 
    assert timed_out.timed_out
325 
  end
326
327 
  test "a missing run directory becomes lost with a reason" do
328 
    run = insert_worker_run("worker-missing", state: "running", pid: 4444)
329
330 
    Req.Test.expect(__MODULE__, fn request ->
331 
      Req.Test.json(request, %{"stdout" => "OA_PRESENT=0\n"})
332 
    end)
333
334 
    {pid, ref} = start_worker(run.id)
335 
    _ = :sys.get_state(pid)
336 
    send(pid, :poll)
337 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
338 
    lost = Repo.get!(Run, run.id)
339 
    assert lost.state == "lost"
340 
    assert lost.failure_reason == "run_directory_missing"
341 
  end
342
343 
  test "a stopped box becomes lost with the provider reason" do
344 
    run = insert_worker_run("worker-stopped", state: "running", pid: 4646)
345
346 
    Req.Test.expect(__MODULE__, fn request ->
347 
      request
348 
      |> Plug.Conn.put_status(409)
349 
      |> Req.Test.json(%{"code" => "box_stopped"})
350 
    end)
351
352 
    {pid, ref} = start_worker(run.id)
353 
    _ = :sys.get_state(pid)
354 
    send(pid, :poll)
355 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
356 
    lost = Repo.get!(Run, run.id)
357 
    assert lost.state == "lost"
358 
    assert lost.failure_reason == "box_stopped"
359 
  end
360
361 
  test "startup reconciliation restarts a persisted nonterminal run" do
362 
    run = insert_worker_run("worker-recovery", state: "running", pid: 4545)
363
364 
    Req.Test.expect(__MODULE__, fn request ->
365 
      Req.Test.json(request, %{
366 
        "stdout" => "OA_PRESENT=1\nOA_SIZE=0\nOA_DATA=\nOA_EXIT=0\nOA_ALIVE=0\n"
367 
      })
368 
    end)
369
370 
    assert :ok = BoxRuns.reconcile_non_terminal()
371 
    assert [{pid, _}] = Registry.lookup(OpenAgents.BoxRunRegistry, run.id)
372 
    ref = Process.monitor(pid)
373 
    _ = :sys.get_state(pid)
374 
    send(pid, :poll)
375 
    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
376 
    assert Repo.get!(Run, run.id).state == "completed"
377 
  end
378
379 
  test "idempotency conflicts refuse a different command" do
380 
    {:ok, conversation} = Conversations.ensure_conversation("worker-idempotency")
381 
    insert_box(conversation.id, "bx_8bhkse3n")
382
383 
    Req.Test.expect(__MODULE__, fn request ->
384 
      assert request.method == "GET"
385
386 
      Req.Test.json(request, %{
387 
        "box" => %{"id" => "bx_8bhkse3n", "state" => "ready", "setupStatus" => "done"}
388 
      })
389 
    end)
390
391 
    Req.Test.expect(__MODULE__, fn request ->
392 
      assert request.body_params["command"] =~ "mkdir"
393 
      Req.Test.json(request, %{"stdout" => "4242\n"})
394 
    end)
395
396 
    assert {:ok, first} =
397 
             BoxRuns.start_run(
398 
               conversation.id,
399 
               "bx_8bhkse3n",
400 
               %{"type" => "user", "id" => "worker-idempotency"},
401 
               "echo one",
402 
               "same-key"
403 
             )
404
405 
    Req.Test.expect(__MODULE__, fn request ->
406 
      assert request.method == "GET"
407
408 
      Req.Test.json(request, %{
409 
        "box" => %{"id" => "bx_8bhkse3n", "state" => "ready", "setupStatus" => "done"}
410 
      })
411 
    end)
412
413 
    assert {:error, :box_run_idempotency_conflict} =
414 
             BoxRuns.start_run(
415 
               conversation.id,
416 
               "bx_8bhkse3n",
417 
               %{"type" => "user", "id" => "worker-idempotency"},
418 
               "echo two",
419 
               "same-key"
420 
             )
421
422 
    assert Repo.aggregate(Run, :count, :id) == 1
423 
    assert first.command == "echo one"
424 
  end
425
426 
  defp insert_run(conversation_id, conversation_box_id, idempotency_key) do
427 
    now = DateTime.utc_now() |> DateTime.truncate(:microsecond)
428
429 
    %Run{}
430 
    |> Run.changeset(%{
431 
      conversation_id: conversation_id,
432 
      conversation_box_id: conversation_box_id,
433 
      requesting_principal: %{"type" => "user", "id" => "user"},
434 
      command: "true",
435 
      idempotency_key: idempotency_key,
436 
      run_directory: "/tmp/openagents-box-runs/#{Ecto.UUID.generate()}",
437 
      admitted_at: now,
438 
      deadline_at: DateTime.add(now, 60, :second)
439 
    })
440 
    |> Repo.insert!()
441 
  end
442
443 
  defp insert_worker_run(key, options) do
444 
    {:ok, conversation} = Conversations.ensure_conversation("worker-" <> key)
445 
    box = insert_box(conversation.id, "bx_8bhkse3n")
446 
    id = Ecto.UUID.generate()
447 
    now = DateTime.utc_now() |> DateTime.truncate(:microsecond)
448 
    state = Keyword.get(options, :state, "running")
449
450 
    %Run{id: id}
451 
    |> Run.changeset(%{
452 
      conversation_id: conversation.id,
453 
      conversation_box_id: box.id,
454 
      requesting_principal: %{"type" => "user", "id" => key},
455 
      command: "echo worker",
456 
      idempotency_key: key,
457 
      state: state,
458 
      pid: Keyword.get(options, :pid),
459 
      run_directory: "$HOME/.openagents/box-runs/users/#{key}/#{id}",
460 
      admitted_at: now,
461 
      deadline_at: Keyword.get(options, :deadline_at, DateTime.add(now, 60, :second)),
462 
      dispatch_attempted_at: if(state == "admitted", do: nil, else: now)
463 
    })
464 
    |> Repo.insert!()
465 
    |> Repo.preload(:conversation_box)
466 
  end
467
468 
  defp insert_box(conversation_id, box_id) do
469 
    %ConversationBox{}
470 
    |> ConversationBox.changeset(%{
471 
      conversation_id: conversation_id,
472 
      box_id: box_id,
473 
      state: "ready",
474 
      setup_status: "done"
475 
    })
476 
    |> Repo.insert!()
477 
  end
478
479 
  defp start_worker(run_id) do
480 
    pid = start_supervised!({OpenAgents.BoxRunServer, run_id})
481 
    {pid, Process.monitor(pid)}
482 
  end
483
484 
  defp past do
485 
    DateTime.utc_now() |> DateTime.add(-60, :second) |> DateTime.truncate(:microsecond)
486 
  end
487
488 
  defp restore_env(key, nil), do: Application.delete_env(:openagents, key)
489 
  defp restore_env(key, value), do: Application.put_env(:openagents, key, value)
490 
end
test/openagents_web/controllers/box_run_controller_test.exs added +167

@@ -0,0 +1,167 @@

1 
defmodule OpenAgentsWeb.BoxRunControllerTest do
2 
  use OpenAgentsWeb.ConnCase, async: false
3
4 
  alias OpenAgents.Box.ConversationBox
5 
  alias OpenAgents.Box.Run
6 
  alias OpenAgents.Conversations
7 
  alias OpenAgents.Repo
8
9 
  setup {Req.Test, :verify_on_exit!}
10
11 
  setup do
12 
    Req.Test.set_req_test_to_shared()
13
14 
    original_api = Application.get_env(:openagents, :box_api)
15 
    original_key = Application.get_env(:openagents, :box_api_key)
16
17 
    Application.put_env(:openagents, :box_api,
18 
      base_url: "https://box-api.internal",
19 
      run_poll_interval_ms: 60_000,
20 
      request_options: [plug: {Req.Test, __MODULE__}, retry_delay: 0]
21 
    )
22
23 
    Application.put_env(:openagents, :box_api_key, "box-run-controller-test")
24
25 
    on_exit(fn ->
26 
      restore_env(:box_api, original_api)
27 
      restore_env(:box_api_key, original_key)
28 
    end)
29
30 
    :ok
31 
  end
32
33 
  test "reads a durable run and output from an offset", %{conn: conn} do
34 
    user = github_user("api-token-box-run-controller")
35 
    {:ok, conversation} = Conversations.ensure_conversation(user)
36
37 
    box =
38 
      %ConversationBox{}
39 
      |> ConversationBox.changeset(%{
40 
        conversation_id: conversation.id,
41 
        box_id: "bx_controller_run",
42 
        state: "ready",
43 
        setup_status: "done"
44 
      })
45 
      |> Repo.insert!()
46
47 
    now = DateTime.utc_now() |> DateTime.truncate(:microsecond)
48
49 
    run =
50 
      %Run{}
51 
      |> Run.changeset(%{
52 
        conversation_id: conversation.id,
53 
        conversation_box_id: box.id,
54 
        requesting_principal: %{"type" => "user", "id" => user.id},
55 
        command: "printf hello",
56 
        idempotency_key: "controller-key",
57 
        run_directory: "/tmp/openagents-box-runs/controller",
58 
        state: "completed",
59 
        exit_status: 0,
60 
        output: "hello",
61 
        last_output_offset: 5,
62 
        admitted_at: now,
63 
        started_at: now,
64 
        finished_at: now,
65 
        deadline_at: DateTime.add(now, 60, :second)
66 
      })
67 
      |> Repo.insert!()
68
69 
    response =
70 
      conn
71 
      |> put_box_api_token("box-run-controller")
72 
      |> get("/api/v3/conversations/#{conversation.id}/boxes/#{box.box_id}/runs/#{run.id}")
73 
      |> json_response(200)
74
75 
    assert response["run"]["id"] == run.id
76 
    assert response["run"]["state"] == "completed"
77 
    refute Map.has_key?(response["run"], "output")
78
79 
    output =
80 
      conn
81 
      |> put_box_api_token("box-run-controller")
82 
      |> get(
83 
        "/api/v3/conversations/#{conversation.id}/boxes/#{box.box_id}/runs/#{run.id}/output?offset=2"
84 
      )
85 
      |> json_response(200)
86
87 
    assert output["output"]["output"] == "llo"
88 
    assert output["output"]["offset"] == 2
89 
    assert output["output"]["next_offset"] == 5
90 
  end
91
92 
  test "a foreign conversation is indistinguishable from a missing one", %{conn: conn} do
93 
    {:ok, conversation} =
94 
      Conversations.ensure_conversation(github_user("api-token-box-run-owner"))
95
96 
    {:ok, foreign} = Conversations.ensure_conversation(github_user("box-run-foreign"))
97
98 
    path =
99 
      "/api/v3/conversations/#{foreign.id}/boxes/bx_foreign/runs/missing/output?offset=0"
100
101 
    response =
102 
      conn
103 
      |> put_box_api_token("box-run-owner")
104 
      |> get(path)
105
106 
    assert json_response(response, 404) == %{"error" => %{"code" => "conversation_not_found"}}
107 
    assert conversation.id != foreign.id
108 
  end
109
110 
  test "idempotency rejects a different command through the API", %{conn: conn} do
111 
    user = github_user("api-token-box-run-idempotency")
112 
    {:ok, conversation} = Conversations.ensure_conversation(user)
113
114 
    box =
115 
      %ConversationBox{}
116 
      |> ConversationBox.changeset(%{
117 
        conversation_id: conversation.id,
118 
        box_id: "bx_8bhkse3n",
119 
        state: "ready",
120 
        setup_status: "done"
121 
      })
122 
      |> Repo.insert!()
123
124 
    Req.Test.expect(__MODULE__, fn request ->
125 
      assert request.method == "GET"
126
127 
      Req.Test.json(request, %{
128 
        "box" => %{"id" => box.box_id, "state" => "ready", "setupStatus" => "done"}
129 
      })
130 
    end)
131
132 
    Req.Test.expect(__MODULE__, fn request ->
133 
      assert request.method == "POST"
134 
      Req.Test.json(request, %{"stdout" => "4242\n"})
135 
    end)
136
137 
    path = "/api/v3/conversations/#{conversation.id}/boxes/#{box.box_id}/runs"
138
139 
    first =
140 
      conn
141 
      |> put_box_api_token("box-run-idempotency")
142 
      |> post(path, %{"command" => "echo one", "idempotency_key" => "same-key"})
143 
      |> json_response(202)
144
145 
    assert first["run"]["command"] == "echo one"
146
147 
    Req.Test.expect(__MODULE__, fn request ->
148 
      assert request.method == "GET"
149
150 
      Req.Test.json(request, %{
151 
        "box" => %{"id" => box.box_id, "state" => "ready", "setupStatus" => "done"}
152 
      })
153 
    end)
154
155 
    response =
156 
      conn
157 
      |> put_box_api_token("box-run-idempotency")
158 
      |> post(path, %{"command" => "echo two", "idempotency_key" => "same-key"})
159
160 
    assert json_response(response, 409) == %{
161 
             "error" => %{"code" => "box_run_idempotency_conflict"}
162 
           }
163 
  end
164
165 
  defp restore_env(key, nil), do: Application.delete_env(:openagents, key)
166 
  defp restore_env(key, value), do: Application.put_env(:openagents, key, value)
167 
end

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