fix(agent): stabilize async runs

Start agent runs in the background so slow providers do not keep POST /agent-runs open. Surface run failures in the UI, update live run status from SSE, and extend provider/runtime timeouts for slow OpenRouter models.

Author: outlndrr

.env.example

@@ -16,6 +16,8 @@ LUMMY_DEFAULT_MODEL_SELECTION=basic-local
 # LUMMY_OPENROUTER_BASE_URL=https://openrouter.ai/api/v1
 # LUMMY_OPENROUTER_SITE_URL=https://your-app.example.com
 # LUMMY_OPENROUTER_APP_NAME=lummy_agent
+# LUMMY_OPENROUTER_TIMEOUT_MS=300000
+# LUMMY_AGENT_MAX_DURATION_MS=600000
 
 # OpenAI-compatible example
 # LUMMY_DEFAULT_PROVIDER_ID=openai-compatible

src/ffi/lummy_agent_provider_http_client_ffi.erl

@@ -48,33 +48,46 @@ stream_json_sse_cancellable(Url, Headers, Body, Timeout, CancelSubject, OnData)
     RequestOptions = [{sync, false}, {stream, self}, {body_format, binary}],
     case httpc:request(post, Request, HttpOptions, RequestOptions) of
         {ok, RequestId} ->
-            await_stream(RequestId, Timeout, CancelSubject, OnData, <<>>);
+            Deadline = deadline(Timeout),
+            await_stream(RequestId, Deadline, CancelSubject, OnData, <<>>);
         {error, Reason} ->
             {error, reason_to_binary(Reason)}
     end.
 
-await_stream(RequestId, Timeout, CancelSubject = {subject, _Owner, Tag}, OnData, Buffer) ->
+await_stream(RequestId, Deadline, CancelSubject = {subject, _Owner, Tag}, OnData, Buffer) ->
     receive
         {Tag, _} ->
             _ = httpc:cancel_request(RequestId),
             {error, <<"cancelled">>};
         {http, {RequestId, stream_start, _Headers}} ->
-            await_stream(RequestId, Timeout, CancelSubject, OnData, Buffer);
+            await_stream(RequestId, Deadline, CancelSubject, OnData, Buffer);
         {http, {RequestId, stream, Chunk}} ->
             NextBuffer = emit_stream_events(<<Buffer/binary, Chunk/binary>>, OnData),
-            await_stream(RequestId, Timeout, CancelSubject, OnData, NextBuffer);
+            await_stream(RequestId, Deadline, CancelSubject, OnData, NextBuffer);
         {http, {RequestId, stream_end, _Headers}} ->
             _ = emit_stream_events(Buffer, OnData),
             {ok, {200, <<>>}};
         {http, {RequestId, {{_Version, Status, _Reason}, _Headers, ResponseBody}}} ->
             {ok, {Status, unicode:characters_to_binary(ResponseBody)}};
         {http, {RequestId, {error, Reason}}} ->
             {error, reason_to_binary(Reason)}
-    after Timeout ->
+    after remaining_timeout(Deadline) ->
         _ = httpc:cancel_request(RequestId),
         {error, <<"timeout">>}
     end.
 
+deadline(Timeout) when Timeout =< 0 ->
+    erlang:monotonic_time(millisecond);
+deadline(Timeout) ->
+    erlang:monotonic_time(millisecond) + Timeout.
+
+remaining_timeout(Deadline) ->
+    Remaining = Deadline - erlang:monotonic_time(millisecond),
+    case Remaining > 0 of
+        true -> Remaining;
+        false -> 0
+    end.
+
 emit_stream_events(Buffer, OnData) ->
     Normalized = binary:replace(Buffer, <<"\r\n">>, <<"\n">>, [global]),
     case binary:match(Normalized, <<"\n\n">>) of

src/lummy_agent/agent/basic_agent.gleam

@@ -29,6 +29,13 @@ type ToolInvocationMessage {
   ToolInvocationExited(process.Down)
 }
 
+type BackgroundRunMessage {
+  BackgroundRunFinished(
+    Result(session_actor.Snapshot, domain_error.DomainError),
+  )
+  BackgroundRunExited(process.Down)
+}
+
 pub fn execute(
   config: app_config.Config,
   _repositories: repository.Repositories,
@@ -42,6 +49,122 @@ pub fn execute(
 
   use _ <- result.try(session_manager.start_run(manager, session_id, run))
 
+  execute_started_run(
+    config,
+    manager,
+    session_id,
+    run,
+    tools,
+    initial_usage,
+    started_ms,
+  )
+}
+
+pub fn start_background(
+  config: app_config.Config,
+  manager: session_manager.Handle,
+  session_id: domain_id.SessionId,
+  run: domain_run.AgentRun,
+) -> Result(session_actor.Snapshot, domain_error.DomainError) {
+  use tools <- result.try(runtime_builder.for_run(config, run))
+  let assert Ok(initial_usage) = domain_run.usage(0, 0, 0, 0)
+  let started_ms = clock.monotonic_ms()
+
+  use snapshot <- result.try(session_manager.start_run(manager, session_id, run))
+
+  spawn_background_run_worker(
+    config,
+    manager,
+    session_id,
+    run,
+    tools,
+    initial_usage,
+    started_ms,
+  )
+
+  Ok(snapshot)
+}
+
+fn spawn_background_run_worker(
+  config: app_config.Config,
+  manager: session_manager.Handle,
+  session_id: domain_id.SessionId,
+  run: domain_run.AgentRun,
+  tools: tool_runtime.Runtime,
+  initial_usage: domain_run.Usage,
+  started_ms: Int,
+) -> Nil {
+  let _ =
+    process.spawn_unlinked(fn() {
+      let result_subject = process.new_subject()
+      let worker =
+        process.spawn_unlinked(fn() {
+          process.send(
+            result_subject,
+            execute_started_run(
+              config,
+              manager,
+              session_id,
+              run,
+              tools,
+              initial_usage,
+              started_ms,
+            ),
+          )
+        })
+      let monitor = process.monitor(worker)
+      let selector =
+        process.new_selector()
+        |> process.select_map(result_subject, BackgroundRunFinished)
+        |> process.select_specific_monitor(monitor, BackgroundRunExited)
+
+      case process.selector_receive_forever(selector) {
+        BackgroundRunFinished(Ok(_)) -> process.demonitor_process(monitor)
+
+        BackgroundRunFinished(Error(error)) -> {
+          process.demonitor_process(monitor)
+          let _ = fail_started_run(manager, session_id, run.run_id, error)
+          Nil
+        }
+
+        BackgroundRunExited(down) -> {
+          case process.receive(result_subject, within: 0) {
+            Ok(Ok(_)) -> Nil
+            Ok(Error(error)) -> {
+              let _ = fail_started_run(manager, session_id, run.run_id, error)
+              Nil
+            }
+            Error(_) -> {
+              let _ =
+                fail_started_run(
+                  manager,
+                  session_id,
+                  run.run_id,
+                  domain_error.internal(
+                    "basic_agent_worker_crashed",
+                    "basic agent worker exited before finishing: "
+                      <> string.inspect(down),
+                  ),
+                )
+              Nil
+            }
+          }
+        }
+      }
+    })
+
+  Nil
+}
+
+fn execute_started_run(
+  config: app_config.Config,
+  manager: session_manager.Handle,
+  session_id: domain_id.SessionId,
+  run: domain_run.AgentRun,
+  tools: tool_runtime.Runtime,
+  initial_usage: domain_run.Usage,
+  started_ms: Int,
+) -> Result(session_actor.Snapshot, domain_error.DomainError) {
   execute_step(
     config,
     manager,

src/lummy_agent/app/config.gleam

@@ -108,7 +108,7 @@ pub fn defaults() -> Config {
       openrouter_base_url: "https://openrouter.ai/api/v1",
       openrouter_site_url: "http://localhost",
       openrouter_app_name: "lummy_agent",
-      openrouter_timeout_ms: 30_000,
+      openrouter_timeout_ms: 300_000,
       openai_compatible_api_key: "",
       openai_compatible_base_url: "https://api.openai.com/v1",
       openai_compatible_timeout_ms: 30_000,
@@ -122,7 +122,7 @@ pub fn defaults() -> Config {
       max_steps: 4,
       max_total_tokens: 16_000,
       max_cost_micros: 5_000_000,
-      max_duration_ms: 60_000,
+      max_duration_ms: 600_000,
     ),
     tools: ToolsConfig(
       timeout_ms: 5000,

src/lummy_agent/session/session_actor.gleam

@@ -175,11 +175,11 @@ type CollectorMessage {
   CollectorDone(Result(provider.CompletionResponse, domain_error.DomainError))
 }
 
-const default_actor_timeout_ms = 5000
+const default_actor_timeout_ms = 600_000
 
 const model_adjacent_actor_timeout_ms = 30_000
 
-const model_execution_timeout_ms = 60_000
+const model_execution_timeout_ms = 600_000
 
 const chunk_flush_count_threshold = 8
 

src/lummy_agent/session/session_manager.gleam

@@ -203,11 +203,11 @@ type State {
   )
 }
 
-const default_actor_timeout_ms = 5000
+const default_actor_timeout_ms = 600_000
 
 const model_adjacent_actor_timeout_ms = 30_000
 
-const model_execution_timeout_ms = 60_000
+const model_execution_timeout_ms = 600_000
 
 pub fn start(config: Config, name: Name) -> actor.StartResult(Handle) {
   actor.new_with_initialiser(1000, fn(subject) {

src/lummy_agent/transport/http_server.gleam

@@ -339,13 +339,7 @@ pub fn handle(
       handle_start_run(config, request, manager, session_id)
 
     http.Post, ["api", "v1", "sessions", session_id, "agent-runs"] ->
-      handle_start_basic_agent_run(
-        config,
-        request,
-        dependencies.repositories,
-        manager,
-        session_id,
-      )
+      handle_start_basic_agent_run(config, request, manager, session_id)
 
     http.Get, ["api", "v1", "runs", run_id, "event-log"] ->
       handle_get_run_event_log(request, dependencies.repositories, run_id)
@@ -1055,7 +1049,6 @@ fn handle_start_run(
 fn handle_start_basic_agent_run(
   config: app_config.Config,
   request: wisp.Request,
-  repositories: repository.Repositories,
   manager: session_manager.Handle,
   raw_session_id: String,
 ) -> wisp.Response {
@@ -1074,13 +1067,7 @@ fn handle_start_basic_agent_run(
 
             Ok(run) ->
               case
-                basic_agent.execute(
-                  config,
-                  repositories,
-                  manager,
-                  session_id,
-                  run,
-                )
+                basic_agent.start_background(config, manager, session_id, run)
               {
                 Ok(snapshot) ->
                   json_response(api_json.snapshot_json(snapshot), 202)