From 063911bc00027450a147d0ef30680dce56cb7782 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 01/17] fix(integrations): order recovery transition locks --- lib/kodo/integrations.ex | 9 +- test/kodo/integrations/oauth_refresh_test.exs | 17 ++-- test/kodo/integrations_test.exs | 95 +++++++++++++++++++ 3 files changed, 114 insertions(+), 7 deletions(-) diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex index 346118a..9dbfaec 100644 --- a/lib/kodo/integrations.ex +++ b/lib/kodo/integrations.ex @@ -1655,7 +1655,14 @@ defmodule Kodo.Integrations do case Repo.get_by(Integration, id: integration_id, user_id: user_id) do %Integration{provider: provider} -> lock_provider_identity(user_id, provider) - invalidate_provider_selection_recovery(user_id, provider) + + case lock_provider_rows(user_id, provider, integration_id) do + {:ok, _integration, _provider_integrations} -> + invalidate_provider_selection_recovery(user_id, provider) + + {:error, reason} -> + Repo.rollback(reason) + end nil -> :ok diff --git a/test/kodo/integrations/oauth_refresh_test.exs b/test/kodo/integrations/oauth_refresh_test.exs index 2b97e81..91f417f 100644 --- a/test/kodo/integrations/oauth_refresh_test.exs +++ b/test/kodo/integrations/oauth_refresh_test.exs @@ -35,7 +35,8 @@ defmodule Kodo.Integrations.OAuthRefreshTest do end test "refreshes shortly before expiry and persists rotated token state", context do - client = fake_client([{:ok, rotated_tokens("account")}]) + rotated = rotated_tokens("account") + client = fake_client([{:ok, rotated}]) assert {:ok, refreshed} = refresh(context, client) assert refreshed.credential_generation == context.integration.credential_generation + 1 @@ -45,7 +46,7 @@ defmodule Kodo.Integrations.OAuthRefreshTest do assert is_nil(refreshed.refresh_claim_owner_id) assert {:ok, credentials} = CredentialEncryption.decrypt(refreshed) - assert credentials["access_token"] == rotated_tokens("account")["access_token"] + assert credentials["access_token"] == rotated["access_token"] assert credentials["refresh_token"] == "rotated-refresh" assert credentials["account_id"] == "account" assert Agent.get(client, &Enum.reverse(&1.refresh_tokens)) == ["old-refresh"] @@ -317,7 +318,7 @@ defmodule Kodo.Integrations.OAuthRefreshTest do assert credentials["refresh_token"] == "usable-refresh" end - test "bounds automatic recovery when refreshed access is already expired", context do + test "does not start another refresh after expired-success recovery is exhausted", context do admitted_at = DateTime.add(DateTime.utc_now(), -10, :second) expired = @@ -327,7 +328,7 @@ defmodule Kodo.Integrations.OAuthRefreshTest do jwt(%{"exp" => DateTime.utc_now() |> DateTime.to_unix() |> Kernel.-(1)}) ) - client = fake_client([{:ok, expired}, {:ok, expired}]) + client = fake_client([{:ok, expired}, {:ok, rotated_tokens("account", "unused")}]) assert {:error, :provider_unavailable} = OAuthRefresh.ensure_fresh(context.scope, context.integration, @@ -336,11 +337,15 @@ defmodule Kodo.Integrations.OAuthRefreshTest do client: FakeRefreshClient, client_options: [agent: client], supervisor: context.supervisor, - claim_owner_id: Ecto.UUID.generate() + claim_owner_id: Ecto.UUID.generate(), + expired_success_retries: 0 ) _ = :sys.get_state(client) - assert length(Agent.get(client, & &1.refresh_tokens)) == 2 + assert Agent.get(client, &Enum.reverse(&1.refresh_tokens)) == ["old-refresh"] + + persisted = Repo.reload!(context.integration) + assert persisted.credential_generation == context.integration.credential_generation + 1 end test "persists a rotated response after its caller stops waiting", context do diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs index 148fc67..8275161 100644 --- a/test/kodo/integrations_test.exs +++ b/test/kodo/integrations_test.exs @@ -1503,6 +1503,86 @@ defmodule Kodo.IntegrationsTest do Integrations.get_active_integration_by_provider(scope, "openai_codex") end + test "disconnect does not invert integration and device-attempt lock order" do + parent = self() + + {scope, integration, attempt} = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + scope = AccountsFixtures.user_scope_fixture() + integration = oauth_integration(scope) + + {:ok, attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 30_000 + ) + + {scope, Repo.reload!(integration), attempt} + end) + end) + |> Task.await() + + on_exit(fn -> + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> Repo.delete(scope.user) end) + end) + |> Task.await() + end) + + integration_holder = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + Repo.transaction(fn -> + Ecto.Adapters.SQL.query!( + Repo, + "SELECT id FROM provider_integrations WHERE id = $1 FOR UPDATE", + [Ecto.UUID.dump!(integration.id)] + ) + + send(parent, {:integration_locked, self()}) + + receive do + :lock_attempt -> + Ecto.Adapters.SQL.query!( + Repo, + "SELECT id FROM device_authorization_attempts WHERE id = $1 FOR UPDATE", + [Ecto.UUID.dump!(attempt.id)] + ) + end + end) + end) + end) + + assert_receive {:integration_locked, holder} + + disconnect = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + Repo.transaction(fn -> + %{rows: [[backend_pid]]} = Ecto.Adapters.SQL.query!(Repo, "SELECT pg_backend_pid()") + send(parent, {:disconnect_backend, backend_pid}) + + Integrations.disconnect( + scope, + integration.id, + integration.credential_generation + ) + end) + end) + end) + + assert_receive {:disconnect_backend, backend_pid} + assert_query_blocked!(backend_pid) + send(holder, :lock_attempt) + + assert {:ok, {:ok, _integration}} = Task.await(disconnect) + assert {:ok, _result} = Task.await(integration_holder) + end + test "cancellation is owned and generation fenced and removes the one-time code", %{ scope: scope } do @@ -1640,6 +1720,21 @@ defmodule Kodo.IntegrationsTest do task end + defp assert_query_blocked!(backend_pid, attempts \\ 100) + + defp assert_query_blocked!(_backend_pid, 0), do: flunk("query did not block") + + defp assert_query_blocked!(backend_pid, attempts) do + %{rows: [[blocked?]]} = + Ecto.Adapters.SQL.query!( + Repo, + "SELECT cardinality(pg_blocking_pids($1)) > 0", + [backend_pid] + ) + + if blocked?, do: :ok, else: assert_query_blocked!(backend_pid, attempts - 1) + end + defp blocking_value(owner, ref, value) do %BlockingJSONValue{owner: owner, ref: ref, value: value} end From 322b85bcbf97e3567be2ab094299b6a61f5bd5ba Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 02/17] fix(agent): canonicalize model snapshot syntax --- lib/kodo/agent/model_mapping.ex | 23 ++++++++---- test/kodo/agent/loop_test.exs | 4 +++ test/kodo/agent/model_mapping_test.exs | 13 +++++++ test/kodo/sessions_test.exs | 49 ++++++++++++++++++++++++-- 4 files changed, 80 insertions(+), 9 deletions(-) diff --git a/lib/kodo/agent/model_mapping.ex b/lib/kodo/agent/model_mapping.ex index c7fe6b3..f2e53cb 100644 --- a/lib/kodo/agent/model_mapping.ex +++ b/lib/kodo/agent/model_mapping.ex @@ -67,11 +67,14 @@ defmodule Kodo.Agent.ModelMapping do roles = Map.new(mapping["roles"], fn {role, role_mapping} -> contract = Roles.fetch!(role_atom(role), role_mapping["role_contract"]) + {route, selector} = model_route_and_selector(role_mapping) {role, role_mapping - |> Map.put("execution_route", role_mapping["provider"]) - |> Map.put("model_selector", model_selector(role_mapping["model"])) + |> Map.put("provider", route) + |> Map.put("execution_route", route) + |> Map.put("model", canonical_model(route, selector, role_mapping["model"])) + |> Map.put("model_selector", selector) |> Map.put_new("capability_contract", capability_contract(contract))} end) @@ -249,10 +252,18 @@ defmodule Kodo.Agent.ModelMapping do end end - defp model_selector(model) do - case String.split(model, ":", parts: 2) do - [_provider, selector] -> selector - [selector] -> selector + defp model_route_and_selector(%{"model" => model, "provider" => provider}) do + case String.split(model, [":", "@"], parts: 2) do + [^provider, selector] -> {provider, selector} + [selector, ^provider] -> {provider, selector} + [_other, selector] -> {provider, selector} + [selector] -> {provider, selector} end end + + defp canonical_model(route, selector, _original) + when is_binary(route) and route != "" and is_binary(selector) and selector != "", + do: "#{route}:#{selector}" + + defp canonical_model(_route, _selector, original), do: original end diff --git a/test/kodo/agent/loop_test.exs b/test/kodo/agent/loop_test.exs index 183a91a..ea1189b 100644 --- a/test/kodo/agent/loop_test.exs +++ b/test/kodo/agent/loop_test.exs @@ -306,6 +306,10 @@ defmodule Kodo.Agent.LoopTest do [created] = Sessions.events_after(session.id) + session + |> Ecto.Changeset.change(model: "gpt-4o-mini@openai") + |> Repo.update!() + created |> Ecto.Changeset.change(payload: Map.delete(created.payload, "model_mapping")) |> Repo.update!() diff --git a/test/kodo/agent/model_mapping_test.exs b/test/kodo/agent/model_mapping_test.exs index 7f28d43..293fcba 100644 --- a/test/kodo/agent/model_mapping_test.exs +++ b/test/kodo/agent/model_mapping_test.exs @@ -71,6 +71,19 @@ defmodule Kodo.Agent.ModelMappingTest do ModelMapping.validate_snapshot(malformed) end + test "canonicalizes every supported string model specification in snapshots" do + for specification <- ["openai:gpt-4o-mini", "gpt-4o-mini@openai"] do + mapping = ModelMapping.balanced([{"session", %{primary: %{model: specification}}}]) + snapshot = ModelMapping.snapshot(mapping) + primary = snapshot["roles"]["primary"] + + assert primary["model"] == "openai:gpt-4o-mini" + assert primary["execution_route"] == "openai" + assert primary["model_selector"] == "gpt-4o-mini" + assert {:ok, ^snapshot} = ModelMapping.validate_snapshot(snapshot) + end + end + test "rejects malformed capability requirements instead of accepting a partial envelope" do snapshot = ModelMapping.snapshot(ModelMapping.balanced()) diff --git a/test/kodo/sessions_test.exs b/test/kodo/sessions_test.exs index 50b840e..ffb8790 100644 --- a/test/kodo/sessions_test.exs +++ b/test/kodo/sessions_test.exs @@ -6,6 +6,8 @@ defmodule Kodo.SessionsTest do alias Kodo.Agent.ModelSettings alias Kodo.Cluster.Instances alias Kodo.Integrations + alias Kodo.Integrations.CredentialEncryption + alias Kodo.Integrations.Integration alias Kodo.LLM alias Kodo.LLM.ProviderError alias Kodo.Runners @@ -365,6 +367,29 @@ defmodule Kodo.SessionsTest do assert Enum.count(Sessions.events_after(session.id), &(&1.type == "user_message")) == 1 end + test "accepts alternate provider-qualified model syntax into a canonical snapshot", %{ + runner: runner, + scope: scope + } do + assert {:ok, _integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "alternate-key"}) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Alternate model syntax", + model: "gpt-4o-mini@openai" + }) + + assert {:ok, [snapshot, _message, _status]} = + Sessions.begin_turn(session.id, "Accept supported syntax") + + primary = snapshot.payload["model_mapping"]["roles"]["primary"] + assert primary["model"] == "openai:gpt-4o-mini" + assert primary["model_selector"] == "gpt-4o-mini" + assert primary["execution_route"] == "openai" + end + test "rejects a turn before acceptance when any configured role lacks access", %{ runner: runner, scope: scope @@ -421,12 +446,30 @@ defmodule Kodo.SessionsTest do assert Enum.map(Sessions.events_after(session.id), & &1.type) == ["session_created"] end - test "holds provider selection locks through turn acceptance", %{ + test "holds the provider selection lock acquired by turn acceptance", %{ runner: runner, scope: scope } do - assert {:ok, _integration} = - Integrations.connect(scope, "openai", "api_key", %{"api_key" => "locked-key"}) + integration = %Integration{ + id: Ecto.UUID.generate(), + user_id: scope.user.id, + provider: "openai", + authentication_type: "api_key" + } + + assert {:ok, encrypted} = + CredentialEncryption.encrypt(integration, %{"api_key" => "locked-key"}) + + integration + |> change( + Map.merge(encrypted, %{ + display_name: "Lock fixture", + active: true, + connection_status: "connected", + validation_status: "unverified" + }) + ) + |> Repo.insert!() assert {:ok, session} = Sessions.create_session(scope, %{ From 87ce0ffb0659a1dc7acaf1365db36eea69150e91 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 03/17] fix(integrations): order audited lifecycle locks --- lib/kodo/integrations.ex | 5 +++ test/kodo/integrations_test.exs | 55 +++++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex index 9dbfaec..a9b780c 100644 --- a/lib/kodo/integrations.ex +++ b/lib/kodo/integrations.ex @@ -441,6 +441,7 @@ defmodule Kodo.Integrations do do: {:error, :authentication_type_mismatch} defp insert_oauth_integration(user_id, changeset) do + lock_user!(user_id) lock_provider_identity(user_id, "openai_codex") case Repo.insert(changeset) do @@ -782,6 +783,8 @@ defmodule Kodo.Integrations do defp activate_transaction(user_id, id, generation, audit?) do Repo.transaction(fn -> + lock_user!(user_id) + case lock_provider_integrations(user_id, id) do {:ok, integration, provider_integrations} -> activate_locked(user_id, integration, provider_integrations, generation, audit?) @@ -1561,6 +1564,7 @@ defmodule Kodo.Integrations do case Ecto.UUID.cast(id) do {:ok, id} -> Repo.transaction(fn -> + lock_user!(user.id) maybe_lock_transition(user.id, id, audit_event_type) {restore_active, changes} = Map.pop(changes, :restore_active, false) @@ -1617,6 +1621,7 @@ defmodule Kodo.Integrations do defp insert_connected(changeset, actor_user_id, provider, event_type) do Repo.transaction(fn -> + lock_user!(actor_user_id) lock_provider_identity(actor_user_id, provider) first_account? = !provider_account_exists?(actor_user_id, provider) diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs index 8275161..51c34eb 100644 --- a/test/kodo/integrations_test.exs +++ b/test/kodo/integrations_test.exs @@ -1583,6 +1583,61 @@ defmodule Kodo.IntegrationsTest do assert {:ok, _result} = Task.await(integration_holder) end + test "audited lifecycle transitions lock the user before integrations" do + parent = self() + + {scope, integration} = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + scope = AccountsFixtures.user_scope_fixture() + {:ok, integration} = connect(scope, "lifecycle-key") + {scope, integration} + end) + end) + |> Task.await() + + user_holder = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + Repo.transaction(fn -> + Ecto.Adapters.SQL.query!(Repo, "SELECT id FROM users WHERE id = $1 FOR UPDATE", [ + scope.user.id + ]) + + send(parent, {:user_locked, self()}) + + receive do + :delete_user -> Repo.delete!(scope.user) + end + end) + end) + end) + + assert_receive {:user_locked, holder} + + disconnect = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + %{rows: [[backend_pid]]} = Ecto.Adapters.SQL.query!(Repo, "SELECT pg_backend_pid()") + send(parent, {:lifecycle_backend, backend_pid}) + + Integrations.disconnect( + scope, + integration.id, + integration.credential_generation + ) + end) + end) + + assert_receive {:lifecycle_backend, backend_pid} + assert_query_blocked!(backend_pid) + send(holder, :delete_user) + + assert {:ok, _deleted} = Task.await(user_holder) + assert {:error, :integration_not_found} = Task.await(disconnect) + assert is_nil(Repo.get(Integration, integration.id)) + end + test "cancellation is owned and generation fenced and removes the one-time code", %{ scope: scope } do From 4850bd84cc37dd63cc16d24e2c508a8c35439099 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 04/17] fix(agent): parse qualified model snapshots --- lib/kodo/agent/model_mapping.ex | 8 +++----- test/kodo/agent/loop_test.exs | 7 ++++++- test/kodo/agent/model_mapping_test.exs | 17 +++++++++++++---- 3 files changed, 22 insertions(+), 10 deletions(-) diff --git a/lib/kodo/agent/model_mapping.ex b/lib/kodo/agent/model_mapping.ex index f2e53cb..31a834b 100644 --- a/lib/kodo/agent/model_mapping.ex +++ b/lib/kodo/agent/model_mapping.ex @@ -253,11 +253,9 @@ defmodule Kodo.Agent.ModelMapping do end defp model_route_and_selector(%{"model" => model, "provider" => provider}) do - case String.split(model, [":", "@"], parts: 2) do - [^provider, selector] -> {provider, selector} - [selector, ^provider] -> {provider, selector} - [_other, selector] -> {provider, selector} - [selector] -> {provider, selector} + case LLMDB.Spec.parse_spec(model) do + {:ok, {parsed_provider, selector}} -> {Atom.to_string(parsed_provider), selector} + {:error, _reason} -> {provider, model} end end diff --git a/test/kodo/agent/loop_test.exs b/test/kodo/agent/loop_test.exs index ea1189b..fc630ce 100644 --- a/test/kodo/agent/loop_test.exs +++ b/test/kodo/agent/loop_test.exs @@ -310,8 +310,13 @@ defmodule Kodo.Agent.LoopTest do |> Ecto.Changeset.change(model: "gpt-4o-mini@openai") |> Repo.update!() + legacy_payload = + created.payload + |> Map.delete("model_mapping") + |> Map.put("model", "gpt-4o-mini@openai") + created - |> Ecto.Changeset.change(payload: Map.delete(created.payload, "model_mapping")) + |> Ecto.Changeset.change(payload: legacy_payload) |> Repo.update!() {:ok, _event} = diff --git a/test/kodo/agent/model_mapping_test.exs b/test/kodo/agent/model_mapping_test.exs index 293fcba..9d61d94 100644 --- a/test/kodo/agent/model_mapping_test.exs +++ b/test/kodo/agent/model_mapping_test.exs @@ -72,14 +72,23 @@ defmodule Kodo.Agent.ModelMappingTest do end test "canonicalizes every supported string model specification in snapshots" do - for specification <- ["openai:gpt-4o-mini", "gpt-4o-mini@openai"] do + specifications = [ + {"openai:gpt-4o-mini", "openai", "gpt-4o-mini"}, + {"gpt-4o-mini@openai", "openai", "gpt-4o-mini"}, + {"openrouter:nvidia/nemotron-3-super-120b-a12b:free", "openrouter", + "nvidia/nemotron-3-super-120b-a12b:free"}, + {"nvidia/nemotron-3-super-120b-a12b:free@openrouter", "openrouter", + "nvidia/nemotron-3-super-120b-a12b:free"} + ] + + for {specification, route, selector} <- specifications do mapping = ModelMapping.balanced([{"session", %{primary: %{model: specification}}}]) snapshot = ModelMapping.snapshot(mapping) primary = snapshot["roles"]["primary"] - assert primary["model"] == "openai:gpt-4o-mini" - assert primary["execution_route"] == "openai" - assert primary["model_selector"] == "gpt-4o-mini" + assert primary["model"] == "#{route}:#{selector}" + assert primary["execution_route"] == route + assert primary["model_selector"] == selector assert {:ok, ^snapshot} = ModelMapping.validate_snapshot(snapshot) end end From 58cb20c26c564c0e045d1dc2b60f2f513fbc325e Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 05/17] fix(integrations): announce clipboard failures --- lib/kodo_web/live/integrations_live.ex | 19 ++++++++++++++++--- test/kodo_web/live/integrations_live_test.exs | 5 +++++ 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex index 6185d72..269428e 100644 --- a/lib/kodo_web/live/integrations_live.ex +++ b/lib/kodo_web/live/integrations_live.ex @@ -987,6 +987,12 @@ defmodule KodoWeb.IntegrationsLive do > Copy + @@ -1311,9 +1317,16 @@ defmodule KodoWeb.IntegrationsLive do export default { mounted() { this.el.addEventListener("click", async () => { - await navigator.clipboard.writeText(this.el.dataset.copyText) - this.el.textContent = "Copied" - window.setTimeout(() => { this.el.textContent = "Copy" }, 1500) + const status = this.el.nextElementSibling + + try { + await navigator.clipboard.writeText(this.el.dataset.copyText) + this.el.textContent = "Copied" + status.textContent = "One-time code copied." + window.setTimeout(() => { this.el.textContent = "Copy" }, 1500) + } catch (_error) { + status.textContent = "Could not copy. Select and copy the code manually." + } }) } } diff --git a/test/kodo_web/live/integrations_live_test.exs b/test/kodo_web/live/integrations_live_test.exs index 2f5bcb4..5ece48e 100644 --- a/test/kodo_web/live/integrations_live_test.exs +++ b/test/kodo_web/live/integrations_live_test.exs @@ -160,6 +160,11 @@ defmodule KodoWeb.IntegrationsLiveTest do "#integration-#{integration.id}-copy-device-code[data-copy-text='ABCD-EFGH'][aria-label='Copy one-time code']" ) + assert has_element?( + view, + "#integration-#{integration.id}-copy-device-code-status[role='status'][aria-live='polite']" + ) + refute inspect(:sys.get_state(view.pid)) =~ "private-device" render_click(view, "cancel_device_authorization", %{ From 1babfb91985a7ac49d3420a6b775b84acba30abe Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 06/17] fix(agent): canonicalize application model routes --- lib/kodo/agent/model_mapping.ex | 25 +++++++++++++++++++++++++ test/kodo/agent/model_mapping_test.exs | 14 ++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/lib/kodo/agent/model_mapping.ex b/lib/kodo/agent/model_mapping.ex index 31a834b..6d4b0b8 100644 --- a/lib/kodo/agent/model_mapping.ex +++ b/lib/kodo/agent/model_mapping.ex @@ -252,7 +252,32 @@ defmodule Kodo.Agent.ModelMapping do end end + defp model_route_and_selector(%{ + "execution_route" => route, + "model_selector" => selector + }) + when is_binary(route) and route != "" and is_binary(selector) and selector != "", + do: {route, selector} + + defp model_route_and_selector(%{"model" => model, "provider" => provider}) + when is_binary(model) and is_binary(provider) and provider != "" do + accepted_prefix = + [provider, String.replace(provider, "_", "-")] + |> Enum.uniq() + |> Enum.find(&String.starts_with?(model, &1 <> ":")) + + if accepted_prefix do + {provider, String.replace_prefix(model, accepted_prefix <> ":", "")} + else + parse_model_route_and_selector(model, provider) + end + end + defp model_route_and_selector(%{"model" => model, "provider" => provider}) do + parse_model_route_and_selector(model, provider) + end + + defp parse_model_route_and_selector(model, provider) do case LLMDB.Spec.parse_spec(model) do {:ok, {parsed_provider, selector}} -> {Atom.to_string(parsed_provider), selector} {:error, _reason} -> {provider, model} diff --git a/test/kodo/agent/model_mapping_test.exs b/test/kodo/agent/model_mapping_test.exs index 9d61d94..eb8befe 100644 --- a/test/kodo/agent/model_mapping_test.exs +++ b/test/kodo/agent/model_mapping_test.exs @@ -93,6 +93,20 @@ defmodule Kodo.Agent.ModelMappingTest do end end + test "canonicalizes application-defined execution route spellings idempotently" do + for specification <- ["openai_codex:gpt-5.4", "openai-codex:gpt-5.4"] do + mapping = ModelMapping.balanced([{"session", %{primary: %{model: specification}}}]) + + snapshot = ModelMapping.snapshot(mapping) + primary = snapshot["roles"]["primary"] + + assert primary["model"] == "openai_codex:gpt-5.4" + assert primary["model_selector"] == "gpt-5.4" + assert ModelMapping.snapshot(snapshot) == snapshot + assert {:ok, ^snapshot} = ModelMapping.validate_snapshot(snapshot) + end + end + test "rejects malformed capability requirements instead of accepting a partial envelope" do snapshot = ModelMapping.snapshot(ModelMapping.balanced()) From 501164c0a6fc075cfad4c0fa5dbe1b2b9c61ee38 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 07/17] fix(sessions): bound account deletion races --- lib/kodo/sessions.ex | 70 +++++++++--- lib/kodo/sessions/active_session.ex | 13 +++ test/kodo/sessions/active_session_test.exs | 18 +++ test/kodo/sessions_test.exs | 125 +++++++++++++++++++++ 4 files changed, 212 insertions(+), 14 deletions(-) diff --git a/lib/kodo/sessions.ex b/lib/kodo/sessions.ex index 2db0f59..60a22ad 100644 --- a/lib/kodo/sessions.ex +++ b/lib/kodo/sessions.ex @@ -588,13 +588,14 @@ defmodule Kodo.Sessions do end defp preflight_turn(scope, session_id) do - mapping = - session_id - |> get_session!() - |> turn_route_snapshot() - |> Map.fetch!("model_mapping") + case get_session(scope, session_id) do + %Session{} = session -> + mapping = session |> turn_route_snapshot() |> Map.fetch!("model_mapping") + preflight_mapping(scope, mapping) - preflight_mapping(scope, mapping) + nil -> + {:error, :session_not_found} + end end defp preflight_mapping(scope, mapping) do @@ -643,7 +644,7 @@ defmodule Kodo.Sessions do end defp append_admitted_model_event_locked(session_id, type, payload, model, credential, opts) do - session = lock_session!(session_id, opts) + session = lock_session_after_owner!(session_id, opts) integration = admit_model_credential!(session, credential) billing_path = IntegrationRef.from_integration(integration).billing_path @@ -1090,7 +1091,7 @@ defmodule Kodo.Sessions do end defp begin_turn_locked(session_id, content, client_request_id, opts) do - session = lock_session!(session_id, opts) + session = lock_session_after_owner!(session_id, opts) if turn_request_recorded?(session_id, client_request_id) do [] @@ -1149,6 +1150,7 @@ defmodule Kodo.Sessions do end defp change_execution_route_locked(scope, session_id, destination) do + lock_scope_user!(scope) session = lock_user_session!(scope, session_id) projection = session.id |> events_after() |> Projection.from_events() @@ -1498,17 +1500,55 @@ defmodule Kodo.Sessions do Session |> where([session], session.id == ^session_id) |> lock("FOR UPDATE") - |> Repo.one!() + |> Repo.one() ownership = Keyword.get(opts, :ownership) cond do + is_nil(session) -> Repo.rollback(:session_not_found) Keyword.get(opts, :allow_unowned, false) -> session ownership_matches?(session, ownership) and owner_alive?(ownership) -> session true -> Repo.rollback(:stale_ownership) end end + defp lock_session_after_owner!(session_id, opts) do + owner = + Session + |> where([session], session.id == ^session_id) + |> select([session], {session.id, session.user_id}) + |> Repo.one() + + case owner do + {^session_id, user_id} when is_integer(user_id) -> + lock_user!(user_id) + session = lock_session!(session_id, opts) + + if session.user_id == user_id, + do: session, + else: Repo.rollback(:session_not_found) + + {^session_id, nil} -> + lock_session!(session_id, opts) + + nil -> + Repo.rollback(:session_not_found) + end + end + + defp lock_scope_user!(%Scope{user: user}), do: lock_user!(user.id) + + defp lock_user!(user_id) do + case User + |> where([user], user.id == ^user_id) + |> select([user], user.id) + |> lock("FOR KEY SHARE") + |> Repo.one() do + ^user_id -> :ok + nil -> Repo.rollback(:session_not_found) + end + end + defp lock_user_session!(%Scope{user: user}, session_id) do case Ecto.UUID.cast(session_id) do {:ok, session_id} -> @@ -1583,11 +1623,13 @@ defmodule Kodo.Sessions do from session in Session, where: session.id == ^event.session_id, select: session.user_id ) - Phoenix.PubSub.broadcast( - Kodo.PubSub, - session_index_topic(user_id), - {:session_index_changed, event.session_id} - ) + if user_id do + Phoenix.PubSub.broadcast( + Kodo.PubSub, + session_index_topic(user_id), + {:session_index_changed, event.session_id} + ) + end end end diff --git a/lib/kodo/sessions/active_session.ex b/lib/kodo/sessions/active_session.ex index d101ce7..d50eb58 100644 --- a/lib/kodo/sessions/active_session.ex +++ b/lib/kodo/sessions/active_session.ex @@ -122,6 +122,9 @@ defmodule Kodo.Sessions.ActiveSession do {:error, :stale_ownership} = error -> {:stop, :normal, error, stop_task(state)} + {:error, :session_not_found} = error -> + {:stop, :normal, error, state} + {:error, reason} -> {:reply, {:error, reason}, state} end @@ -137,6 +140,9 @@ defmodule Kodo.Sessions.ActiveSession do {:error, :stale_ownership} = error -> {:stop, :normal, error, stop_task(state)} + {:error, :session_not_found} = error -> + {:stop, :normal, error, state} + {:error, reason} -> {:reply, {:error, reason}, state} end @@ -243,6 +249,10 @@ defmodule Kodo.Sessions.ActiveSession do reply_drain_waiters(state, :ok) {:stop, :normal, state |> Map.put(:task, nil) |> Map.delete(:drain_waiters)} + {:error, :session_not_found} -> + reply_drain_waiters(state, :ok) + {:stop, :normal, state |> Map.put(:task, nil) |> Map.delete(:drain_waiters)} + {:error, :stale_ownership} -> reply_drain_waiters(state, {:error, :stale_ownership}) {:stop, :normal, stop_task(state)} @@ -261,6 +271,9 @@ defmodule Kodo.Sessions.ActiveSession do {:error, :session_not_active} -> {:stop, :normal, {:error, :already_finished}, %{state | task: nil}} + {:error, :session_not_found} -> + {:stop, :normal, {:error, :already_finished}, %{state | task: nil}} + {:error, :stale_ownership} = error -> {:stop, :normal, error, stop_task(state)} diff --git a/test/kodo/sessions/active_session_test.exs b/test/kodo/sessions/active_session_test.exs index 237da55..5233e4f 100644 --- a/test/kodo/sessions/active_session_test.exs +++ b/test/kodo/sessions/active_session_test.exs @@ -278,6 +278,24 @@ defmodule Kodo.Sessions.ActiveSessionTest do assert_receive {:DOWN, ^coordinator_ref, :process, ^coordinator, :normal} end + test "stops normally when its user and session disappear during inference", %{ + scope: scope, + session: session + } do + assert {:ok, coordinator} = Sessions.ensure_started(session.id) + assert :ok = ActiveSession.start_turn(coordinator, "wait") + assert_receive :fake_llm_waiting + + task = :sys.get_state(coordinator).task.pid + coordinator_ref = Process.monitor(coordinator) + + Repo.delete!(scope.user) + send(task, :never) + + assert_receive {:DOWN, ^coordinator_ref, :process, ^coordinator, :normal} + assert Registry.lookup(Kodo.SessionRegistry, session.id) == [] + end + test "reconciles a repeated turn request while its task is active", %{session: session} do request_id = Ecto.UUID.generate() assert {:ok, pid} = Sessions.ensure_started(session.id) diff --git a/test/kodo/sessions_test.exs b/test/kodo/sessions_test.exs index ffb8790..473aee8 100644 --- a/test/kodo/sessions_test.exs +++ b/test/kodo/sessions_test.exs @@ -1,6 +1,7 @@ defmodule Kodo.SessionsTest do use Kodo.DataCase + alias Kodo.Accounts.User alias Kodo.Agent.ExecutionRouteChange alias Kodo.Agent.ModelMapping alias Kodo.Agent.ModelSettings @@ -500,6 +501,75 @@ defmodule Kodo.SessionsTest do refute lock_available? end + test "admission locks the owning user before session and integration rows" do + for operation <- [:begin_turn, :append_invocation] do + parent = self() + + {scope, integration, session, model, credential} = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> admission_fixture() end) + end) + |> Task.await() + + on_exit(fn -> delete_external_user(scope.user.id) end) + + deletion = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + Repo.transaction(fn -> + Ecto.Adapters.SQL.query!(Repo, "SELECT id FROM users WHERE id = $1 FOR UPDATE", [ + scope.user.id + ]) + + Ecto.Adapters.SQL.query!( + Repo, + "SELECT id FROM provider_integrations WHERE id = $1 FOR UPDATE", + [Ecto.UUID.dump!(integration.id)] + ) + + send(parent, {:admission_delete_ready, self()}) + + receive do + :delete_user -> Repo.delete!(scope.user) + end + end) + end) + end) + + assert_receive {:admission_delete_ready, deletion_holder} + + admission = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + %{rows: [[backend_pid]]} = Ecto.Adapters.SQL.query!(Repo, "SELECT pg_backend_pid()") + send(parent, {:admission_backend, backend_pid}) + + case operation do + :begin_turn -> + Sessions.begin_turn(session.id, "Race account deletion") + + :append_invocation -> + Sessions.append_admitted_model_event( + session.id, + "model_invocation_started", + %{"invocation_id" => Ecto.UUID.generate()}, + model, + credential + ) + end + end) + end) + + assert_receive {:admission_backend, backend_pid} + assert_query_blocked!(backend_pid) + send(deletion_holder, :delete_user) + + assert {:ok, _deleted} = Task.await(deletion) + assert {:error, :session_not_found} = Task.await(admission) + assert is_nil(Repo.get(Session, session.id)) + end + end + test "atomically records invocation admission for the exact active credential generation", %{ runner: runner, scope: scope @@ -972,4 +1042,59 @@ defmodule Kodo.SessionsTest do ] } end + + defp admission_fixture do + scope = user_scope_fixture() + + {:ok, runner} = + Runners.register(scope, %{ + workspace_root: "/work/#{Ecto.UUID.generate()}", + platform: "linux", + architecture: "x86_64", + runner_version: "0.1.0", + protocol_version: 5, + capabilities: [] + }) + + {:ok, integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "admission-lock-key"}) + + {:ok, model, reference} = LLM.resolve_integration(scope, "openai:gpt-4o-mini") + {:ok, credential} = LLM.admit_credential(scope, model, reference) + + {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Admission deletion lock order", + model: "openai:gpt-4o-mini" + }) + + {scope, integration, session, model, credential} + end + + defp delete_external_user(user_id) do + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> delete_user_if_present(user_id) end) + end) + |> Task.await() + end + + defp delete_user_if_present(user_id) do + if user = Repo.get(User, user_id), do: Repo.delete!(user) + end + + defp assert_query_blocked!(backend_pid, attempts \\ 100) + + defp assert_query_blocked!(_backend_pid, 0), do: flunk("query did not block") + + defp assert_query_blocked!(backend_pid, attempts) do + %{rows: [[blocked?]]} = + Ecto.Adapters.SQL.query!( + Repo, + "SELECT cardinality(pg_blocking_pids($1)) > 0", + [backend_pid] + ) + + if blocked?, do: :ok, else: assert_query_blocked!(backend_pid, attempts - 1) + end end From fe8f7fc8492053479e94d6f594a99aab0e6d9dbf Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 08/17] fix(integrations): recover observed authorization attempts --- lib/kodo/integrations.ex | 24 ++++ lib/kodo_web/live/integrations_live.ex | 136 ++++++++++++++++-- test/kodo_web/live/integrations_live_test.exs | 66 ++++++++- 3 files changed, 211 insertions(+), 15 deletions(-) diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex index a9b780c..6e882ab 100644 --- a/lib/kodo/integrations.ex +++ b/lib/kodo/integrations.ex @@ -251,6 +251,30 @@ defmodule Kodo.Integrations do end end + @doc "Returns bounded terminal authorization feedback without loading encrypted payloads." + def get_device_authorization_outcome(%Scope{user: user}, integration_id) do + with {:ok, integration_id} <- cast_uuid(integration_id) do + DeviceAuthorizationAttempt + |> join(:inner, [attempt], integration in assoc(attempt, :integration)) + |> where( + [attempt, integration], + attempt.integration_id == ^integration_id and integration.user_id == ^user.id + ) + |> order_by([attempt], desc: attempt.attempt_generation) + |> limit(1) + |> select([attempt], %{ + attempt_generation: attempt.attempt_generation, + state: attempt.state, + terminal_error_code: attempt.terminal_error_code + }) + |> Repo.one() + |> case do + %{state: state} = outcome when state in ["failed", "expired"] -> {:ok, outcome} + _missing_or_non_failure -> {:error, :device_authorization_outcome_not_found} + end + end + end + @doc false def expire_device_authorization(%Scope{user: user}, integration_id) do with {:ok, integration_id} <- cast_uuid(integration_id) do diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex index 269428e..7474f37 100644 --- a/lib/kodo_web/live/integrations_live.ex +++ b/lib/kodo_web/live/integrations_live.ex @@ -65,6 +65,7 @@ defmodule KodoWeb.IntegrationsLive do |> assign(:validation_tasks, %{}) |> assign(:device_authorization_tasks, %{}) |> assign(:device_authorization_retries, %{}) + |> assign(:device_authorization_outcomes, %{}) |> load_integrations() socket = @@ -234,7 +235,11 @@ defmodule KodoWeb.IntegrationsLive do ) do {:noreply, socket - |> track_device_authorization_task(started.task, integration.id) + |> track_device_authorization_task( + started.task, + integration.id, + started.attempt.attempt_generation + ) |> put_flash(:info, "Authorization started. Enter the one-time code with OpenAI.") |> push_patch(to: ~p"/integrations")} else @@ -372,19 +377,21 @@ defmodule KodoWeb.IntegrationsLive do end def handle_info( - {:device_authorization_changed, integration_id, attempt_generation, state, _error}, + {:device_authorization_changed, integration_id, attempt_generation, state, error}, socket ) do socket = socket |> maybe_clear_device_authorization_retry(integration_id, attempt_generation, state) |> load_integrations() + |> record_device_authorization_outcome( + integration_id, + attempt_generation, + state, + error + ) - if state == "failed" do - {:noreply, put_flash(socket, :error, "Authorization failed. Try again.")} - else - {:noreply, socket} - end + {:noreply, socket} end def handle_info({:retry_device_authorization, integration_id, attempt_generation}, socket) do @@ -479,9 +486,13 @@ defmodule KodoWeb.IntegrationsLive do device_authorizations = load_device_authorizations(socket.assigns.current_scope, integrations) + device_authorization_outcomes = + load_device_authorization_outcomes(socket.assigns.current_scope, integrations) + socket |> assign(:integrations, integrations) |> assign(:device_authorizations, device_authorizations) + |> assign(:device_authorization_outcomes, device_authorization_outcomes) |> reconcile_device_authorization_retries() end @@ -501,7 +512,19 @@ defmodule KodoWeb.IntegrationsLive do end end) - assign(socket, :device_authorization_retries, retries) + socket = assign(socket, :device_authorization_retries, retries) + + tracked_keys = MapSet.new(socket.assigns.device_authorization_tasks, &elem(&1, 1)) + + if connected?(socket) do + Enum.reduce(active_keys, socket, fn {integration_id, generation} = key, acc -> + if Map.has_key?(retries, key) or MapSet.member?(tracked_keys, key), + do: acc, + else: schedule_device_authorization_retry(acc, integration_id, generation) + end) + else + socket + end end # Browser-facing state needs lifecycle metadata only. In particular, keeping @@ -538,9 +561,9 @@ defmodule KodoWeb.IntegrationsLive do update(socket, :validation_tasks, &Map.delete(&1, reference)) end - defp track_device_authorization_task(socket, task, integration_id) do + defp track_device_authorization_task(socket, task, integration_id, attempt_generation) do update(socket, :device_authorization_tasks, fn tasks -> - Map.put(tasks, task.ref, integration_id) + Map.put(tasks, task.ref, {integration_id, attempt_generation}) end) end @@ -559,7 +582,7 @@ defmodule KodoWeb.IntegrationsLive do {:ok, task} -> socket |> clear_device_authorization_retry(integration_id, attempt_generation) - |> track_device_authorization_task(task, integration_id) + |> track_device_authorization_task(task, integration_id, attempt_generation) {:error, :device_authorization_not_claimable} when schedule_retry? -> schedule_device_authorization_retry(socket, integration_id, attempt_generation) @@ -605,6 +628,59 @@ defmodule KodoWeb.IntegrationsLive do end end + defp record_device_authorization_outcome( + socket, + integration_id, + attempt_generation, + state, + _error + ) do + current_generation = + case socket.assigns.device_authorizations[integration_id] do + %{attempt_generation: generation} -> generation + nil -> nil + end + + cond do + is_integer(current_generation) and current_generation > attempt_generation -> + socket + + state == "failed" -> + put_device_authorization_outcome( + socket, + integration_id, + attempt_generation, + "OpenAI could not complete authorization. Try again." + ) + + state == "expired" -> + put_device_authorization_outcome( + socket, + integration_id, + attempt_generation, + "Authorization expired. Start again to get a new one-time code." + ) + + true -> + clear_device_authorization_outcome(socket, integration_id, attempt_generation) + end + end + + defp put_device_authorization_outcome(socket, integration_id, generation, message) do + update(socket, :device_authorization_outcomes, fn outcomes -> + Map.put(outcomes, integration_id, %{attempt_generation: generation, message: message}) + end) + end + + defp clear_device_authorization_outcome(socket, integration_id, attempt_generation) do + update(socket, :device_authorization_outcomes, fn outcomes -> + case outcomes[integration_id] do + %{attempt_generation: generation} when generation > attempt_generation -> outcomes + _older_or_missing -> Map.delete(outcomes, integration_id) + end + end) + end + defp load_device_authorizations(scope, integrations) do integrations |> Enum.filter(&(&1.provider == "openai_codex")) @@ -634,6 +710,34 @@ defmodule KodoWeb.IntegrationsLive do end) end + defp load_device_authorization_outcomes(scope, integrations) do + integrations + |> Enum.filter(&(&1.provider == "openai_codex")) + |> Map.new(fn integration -> + outcome = + case Integrations.get_device_authorization_outcome(scope, integration.id) do + {:ok, %{attempt_generation: generation, state: "failed"}} -> + %{ + attempt_generation: generation, + message: "OpenAI could not complete authorization. Try again." + } + + {:ok, %{attempt_generation: generation, state: "expired"}} -> + %{ + attempt_generation: generation, + message: "Authorization expired. Start again to get a new one-time code." + } + + {:error, :device_authorization_outcome_not_found} -> + nil + end + + {integration.id, outcome} + end) + |> Enum.reject(fn {_integration_id, outcome} -> is_nil(outcome) end) + |> Map.new() + end + defp find_device_authorization(authorizations, attempt_id) do Enum.find_value(authorizations, fn {_integration_id, attempt} -> if attempt.attempt_id == attempt_id, do: attempt @@ -945,6 +1049,14 @@ defmodule KodoWeb.IntegrationsLive do > {validation_detail(integration)}

+
diff --git a/test/kodo_web/live/integrations_live_test.exs b/test/kodo_web/live/integrations_live_test.exs index 5ece48e..12e0ead 100644 --- a/test/kodo_web/live/integrations_live_test.exs +++ b/test/kodo_web/live/integrations_live_test.exs @@ -162,7 +162,7 @@ defmodule KodoWeb.IntegrationsLiveTest do assert has_element?( view, - "#integration-#{integration.id}-copy-device-code-status[role='status'][aria-live='polite']" + "#integration-#{integration.id}-copy-device-code-status[role='status'][aria-live='polite']:not(.sr-only)" ) refute inspect(:sys.get_state(view.pid)) =~ "private-device" @@ -257,6 +257,13 @@ defmodule KodoWeb.IntegrationsLiveTest do {:ok, view, _html} = live(conn, ~p"/integrations") refute has_element?(view, "#integration-#{integration.id}-device-authorization") assert Repo.reload!(attempt).state == "expired" + _ = :sys.get_state(view.pid) + + assert has_element?( + view, + "#integration-#{integration.id}-device-authorization-outcome[role='alert']", + "Authorization expired" + ) end test "resumes an exchange-stage authorization after its worker lease expires", %{ @@ -403,7 +410,7 @@ defmodule KodoWeb.IntegrationsLiveTest do assert live_assign(view, :device_authorization_retries) == %{} end - test "drops an old retry immediately when a new authorization supersedes it", %{ + test "replaces an old retry when another tab supersedes an authorization", %{ conn: conn, scope: scope } do @@ -443,7 +450,60 @@ defmodule KodoWeb.IntegrationsLiveTest do _ = :sys.get_state(view.pid) assert second.attempt_generation > first.attempt_generation - assert live_assign(view, :device_authorization_retries) == %{} + + retries = live_assign(view, :device_authorization_retries) + refute Map.has_key?(retries, retry_key) + assert Map.has_key?(retries, {integration.id, second.attempt_generation}) + + send( + view.pid, + {:device_authorization_changed, integration.id, first.attempt_generation, "failed", + "provider_rejected"} + ) + + _ = :sys.get_state(view.pid) + refute has_element?(view, "#integration-#{integration.id}-device-authorization-outcome") + end + + test "shows failed authorization feedback on the affected account", %{ + conn: conn, + scope: scope + } do + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Failed subscription" + ) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "failed-device", "user_code" => "FAILED"}, + 30_000 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + {:ok, view, _html} = live(conn, ~p"/integrations") + + assert {:ok, _failed} = + Integrations.fail_device_authorization(scope, claim, "provider_rejected") + + _ = :sys.get_state(view.pid) + + assert has_element?( + view, + "#integration-#{integration.id}-device-authorization-outcome[role='alert']", + "OpenAI could not complete authorization" + ) + + refute has_element?(view, "#integration-#{integration.id}-device-authorization") end test "updates an open authorization panel when polling advances to exchange", %{ From 3facee5d935e09bf6530325215f36787dc4a24b1 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 09/17] fix(sessions): finish deletion lifecycle handling --- lib/kodo/sessions.ex | 13 ++++- lib/kodo/sessions/active_session.ex | 9 ++-- test/kodo/sessions/active_session_test.exs | 56 ++++++++++++++++++++++ 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/lib/kodo/sessions.ex b/lib/kodo/sessions.ex index 60a22ad..52b4fb4 100644 --- a/lib/kodo/sessions.ex +++ b/lib/kodo/sessions.ex @@ -733,8 +733,17 @@ defmodule Kodo.Sessions do end defp reconcile_cancel({:error, reason}, scope, session_id) - when reason in [:not_running, :already_finished, :coordinator_unavailable] do - if get_session(scope, session_id).status == "cancelled", do: :ok, else: {:error, reason} + when reason in [ + :not_running, + :already_finished, + :coordinator_unavailable, + :session_not_found + ] do + case get_session(scope, session_id) do + %Session{status: "cancelled"} -> :ok + %Session{} -> {:error, reason} + nil -> {:error, :not_found} + end end defp reconcile_cancel(result, _scope, _session_id), do: result diff --git a/lib/kodo/sessions/active_session.ex b/lib/kodo/sessions/active_session.ex index d50eb58..cfb2621 100644 --- a/lib/kodo/sessions/active_session.ex +++ b/lib/kodo/sessions/active_session.ex @@ -123,7 +123,7 @@ defmodule Kodo.Sessions.ActiveSession do {:stop, :normal, error, stop_task(state)} {:error, :session_not_found} = error -> - {:stop, :normal, error, state} + {:stop, :normal, error, stop_task(state)} {:error, reason} -> {:reply, {:error, reason}, state} @@ -141,7 +141,7 @@ defmodule Kodo.Sessions.ActiveSession do {:stop, :normal, error, stop_task(state)} {:error, :session_not_found} = error -> - {:stop, :normal, error, state} + {:stop, :normal, error, stop_task(state)} {:error, reason} -> {:reply, {:error, reason}, state} @@ -167,6 +167,9 @@ defmodule Kodo.Sessions.ActiveSession do {:error, :stale_ownership} = error -> {:stop, :normal, error, stop_task(state)} + {:error, :session_not_found} = error -> + {:stop, :normal, error, %{state | task: nil}} + {:error, reason} -> {:stop, {:cancellation_persistence_failed, reason}, {:error, reason}, state} end @@ -272,7 +275,7 @@ defmodule Kodo.Sessions.ActiveSession do {:stop, :normal, {:error, :already_finished}, %{state | task: nil}} {:error, :session_not_found} -> - {:stop, :normal, {:error, :already_finished}, %{state | task: nil}} + {:stop, :normal, {:error, :session_not_found}, %{state | task: nil}} {:error, :stale_ownership} = error -> {:stop, :normal, error, stop_task(state)} diff --git a/test/kodo/sessions/active_session_test.exs b/test/kodo/sessions/active_session_test.exs index 5233e4f..efcb615 100644 --- a/test/kodo/sessions/active_session_test.exs +++ b/test/kodo/sessions/active_session_test.exs @@ -296,6 +296,62 @@ defmodule Kodo.Sessions.ActiveSessionTest do assert Registry.lookup(Kodo.SessionRegistry, session.id) == [] end + test "stops its active task when a queued turn observes session deletion", %{ + scope: scope, + session: session + } do + assert {:ok, coordinator} = Sessions.ensure_started(session.id) + assert :ok = ActiveSession.start_turn(coordinator, "wait") + assert_receive :fake_llm_waiting + + task = :sys.get_state(coordinator).task.pid + task_ref = Process.monitor(task) + coordinator_ref = Process.monitor(coordinator) + Repo.delete!(scope.user) + + assert {:error, :session_not_found} = ActiveSession.start_turn(coordinator, "queued turn") + assert_receive {:DOWN, ^task_ref, :process, ^task, :killed} + assert_receive {:DOWN, ^coordinator_ref, :process, ^coordinator, :normal} + end + + test "cancellation stops normally when deletion wins before task shutdown", %{ + scope: scope, + session: session + } do + assert {:ok, coordinator} = Sessions.ensure_started(session.id) + assert :ok = ActiveSession.start_turn(coordinator, "wait") + assert_receive :fake_llm_waiting + coordinator_ref = Process.monitor(coordinator) + + Repo.delete!(scope.user) + + assert {:error, :session_not_found} = ActiveSession.cancel(coordinator) + assert_receive {:DOWN, ^coordinator_ref, :process, ^coordinator, :normal} + end + + test "cancellation preserves missing-session outcome from a completed task", %{ + scope: scope, + session: session + } do + assert {:ok, coordinator} = Sessions.ensure_started(session.id) + assert :ok = ActiveSession.start_turn(coordinator, "wait") + assert_receive :fake_llm_waiting + + task = :sys.get_state(coordinator).task.pid + task_ref = Process.monitor(task) + coordinator_ref = Process.monitor(coordinator) + :ok = :sys.suspend(coordinator) + reply_ref = make_ref() + send(coordinator, {:"$gen_call", {self(), reply_ref}, :cancel}) + send(task, :never) + assert_receive {:DOWN, ^task_ref, :process, ^task, :normal} + Repo.delete!(scope.user) + :ok = :sys.resume(coordinator) + + assert_receive {^reply_ref, {:error, :session_not_found}} + assert_receive {:DOWN, ^coordinator_ref, :process, ^coordinator, :normal} + end + test "reconciles a repeated turn request while its task is active", %{session: session} do request_id = Ecto.UUID.generate() assert {:ok, pid} = Sessions.ensure_started(session.id) From 57b72e1e417ab5863a4381a2b6ae28bcf180b32c Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 10/17] fix(agent): defer final review admission --- lib/kodo/agent/loop.ex | 31 +++++++-------- test/kodo/agent/loop_test.exs | 72 +++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 15 deletions(-) diff --git a/lib/kodo/agent/loop.ex b/lib/kodo/agent/loop.ex index 8569011..26788f2 100644 --- a/lib/kodo/agent/loop.ex +++ b/lib/kodo/agent/loop.ex @@ -159,10 +159,10 @@ defmodule Kodo.Agent.Loop do review = ModelMapping.role!(context.mapping, :review) review_model = ModelMapping.request_model(review) contract = Roles.fetch!(:review, review["role_contract"]) + review_invocation_id = Ecto.UUID.generate() with {:ok, capability_validation} <- context.adapter.validate_model(review_model, review, contract), - {:ok, request} <- resolve_request(context.session_id, review), :ok <- Phoenix.PubSub.subscribe( Kodo.PubSub, @@ -170,17 +170,6 @@ defmodule Kodo.Agent.Loop do ), :ok <- Phoenix.PubSub.subscribe(Kodo.PubSub, "session:#{context.session_id}"), :ok <- Runners.subscribe_lifecycle(), - {:ok, {review_invocation_id, admission}} <- - start_review_invocation( - context.session_id, - primary_invocation_id, - context.mapping, - review, - contract, - capability_validation, - request, - context.ownership - ), {:ok, diff} <- execute_tool( context.session_id, @@ -192,6 +181,19 @@ defmodule Kodo.Agent.Loop do context.ownership ), :ok <- rehoming_boundary(), + {:ok, request} <- resolve_request(context.session_id, review), + {:ok, admission} <- + start_review_invocation( + context.session_id, + primary_invocation_id, + review_invocation_id, + context.mapping, + review, + contract, + capability_validation, + request, + context.ownership + ), {:ok, generated} <- Sessions.dispatch_if_owner(context.ownership, fn -> task = original_task(context.events) @@ -239,6 +241,7 @@ defmodule Kodo.Agent.Loop do defp start_review_invocation( session_id, primary_invocation_id, + invocation_id, mapping, review, contract, @@ -246,8 +249,6 @@ defmodule Kodo.Agent.Loop do request, ownership ) do - invocation_id = Ecto.UUID.generate() - case Sessions.append_admitted_model_event( session_id, "review_invocation_started", @@ -272,7 +273,7 @@ defmodule Kodo.Agent.Loop do parent_id: primary_invocation_id, ownership: ownership ) do - {:ok, {_event, admission}} -> {:ok, {invocation_id, admission}} + {:ok, {_event, admission}} -> {:ok, admission} error -> normalize_admission_error(error, request) end end diff --git a/test/kodo/agent/loop_test.exs b/test/kodo/agent/loop_test.exs index fc630ce..e9602c6 100644 --- a/test/kodo/agent/loop_test.exs +++ b/test/kodo/agent/loop_test.exs @@ -923,6 +923,11 @@ defmodule Kodo.Agent.LoopTest do assert_receive {:tool_request, request} assert request["request"] == %{"tool" => "git_diff", "paths" => []} + refute Enum.any?( + Sessions.events_after(session.id), + &(&1.type == "review_invocation_started") + ) + broadcast_success(runner, request, %{ "result" => "output", "content" => "clean diff", @@ -960,6 +965,73 @@ defmodule Kodo.Agent.LoopTest do end) end + test "resolves and admits the review credential after the final diff is available", %{ + runner: runner, + scope: scope + } do + {:ok, _registration} = Registry.register(Kodo.RunnerRegistry, runner.id, nil) + + assert {:ok, anthropic} = + Integrations.connect(scope, "anthropic", "api_key", %{"api_key" => "review-key"}) + + assert {:ok, _override} = + ModelSettings.put_user_override(scope, :review, %{ + model: "anthropic:claude-3-5-haiku-latest" + }) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Fresh review credential", + model: "openai:gpt-4o-mini" + }) + + assert {:ok, ownership} = Sessions.claim_ownership(session.id, nil) + + assert {:ok, _event} = + Sessions.append_event( + session.id, + "user_message", + %{"role" => "user", "content" => "final answer"}, + ownership: ownership + ) + + loop = + Task.async(fn -> + Loop.run(session.id, + adapter: Kodo.Test.FakeLLM, + budgets: budgets([]), + ownership: ownership + ) + end) + + assert_receive {:tool_request, request} + assert request["request"]["tool"] == "git_diff" + + refute Enum.any?( + Sessions.events_after(session.id), + &(&1.type == "review_invocation_started") + ) + + assert {:ok, _disconnected} = + Integrations.disconnect(scope, anthropic.id, anthropic.credential_generation) + + broadcast_success(runner, request, %{ + "result" => "output", + "content" => "clean diff", + "truncated" => false + }) + + assert {:error, %Kodo.LLM.ProviderError{} = error} = Task.await(loop) + assert error.provider == "anthropic" + assert error.kind == :integration_required + + refute Enum.any?( + Sessions.events_after(session.id), + &(&1.type == "review_invocation_started") + ) + end + test "feeds supported review findings back to primary and reviews the correction", %{ runner: runner, session: session, From 2ed836532f702df9ba3cf9d3335f17eebb559dac Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 11/17] fix(integrations): trust durable authorization outcomes --- lib/kodo_web/live/integrations_live.ex | 71 ++----------------- test/kodo_web/live/integrations_live_test.exs | 61 +++++++++++++++- 2 files changed, 65 insertions(+), 67 deletions(-) diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex index 7474f37..8750ce5 100644 --- a/lib/kodo_web/live/integrations_live.ex +++ b/lib/kodo_web/live/integrations_live.ex @@ -377,21 +377,13 @@ defmodule KodoWeb.IntegrationsLive do end def handle_info( - {:device_authorization_changed, integration_id, attempt_generation, state, error}, + {:device_authorization_changed, integration_id, attempt_generation, state, _error}, socket ) do - socket = - socket - |> maybe_clear_device_authorization_retry(integration_id, attempt_generation, state) - |> load_integrations() - |> record_device_authorization_outcome( - integration_id, - attempt_generation, - state, - error - ) - - {:noreply, socket} + {:noreply, + socket + |> maybe_clear_device_authorization_retry(integration_id, attempt_generation, state) + |> load_integrations()} end def handle_info({:retry_device_authorization, integration_id, attempt_generation}, socket) do @@ -628,59 +620,6 @@ defmodule KodoWeb.IntegrationsLive do end end - defp record_device_authorization_outcome( - socket, - integration_id, - attempt_generation, - state, - _error - ) do - current_generation = - case socket.assigns.device_authorizations[integration_id] do - %{attempt_generation: generation} -> generation - nil -> nil - end - - cond do - is_integer(current_generation) and current_generation > attempt_generation -> - socket - - state == "failed" -> - put_device_authorization_outcome( - socket, - integration_id, - attempt_generation, - "OpenAI could not complete authorization. Try again." - ) - - state == "expired" -> - put_device_authorization_outcome( - socket, - integration_id, - attempt_generation, - "Authorization expired. Start again to get a new one-time code." - ) - - true -> - clear_device_authorization_outcome(socket, integration_id, attempt_generation) - end - end - - defp put_device_authorization_outcome(socket, integration_id, generation, message) do - update(socket, :device_authorization_outcomes, fn outcomes -> - Map.put(outcomes, integration_id, %{attempt_generation: generation, message: message}) - end) - end - - defp clear_device_authorization_outcome(socket, integration_id, attempt_generation) do - update(socket, :device_authorization_outcomes, fn outcomes -> - case outcomes[integration_id] do - %{attempt_generation: generation} when generation > attempt_generation -> outcomes - _older_or_missing -> Map.delete(outcomes, integration_id) - end - end) - end - defp load_device_authorizations(scope, integrations) do integrations |> Enum.filter(&(&1.provider == "openai_codex")) diff --git a/test/kodo_web/live/integrations_live_test.exs b/test/kodo_web/live/integrations_live_test.exs index 12e0ead..91d231a 100644 --- a/test/kodo_web/live/integrations_live_test.exs +++ b/test/kodo_web/live/integrations_live_test.exs @@ -463,6 +463,40 @@ defmodule KodoWeb.IntegrationsLiveTest do _ = :sys.get_state(view.pid) refute has_element?(view, "#integration-#{integration.id}-device-authorization-outcome") + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + assert {:ok, _completed} = + Integrations.complete_device_authorization( + scope, + claim, + %{ + "access_token" => "new-access", + "refresh_token" => "new-refresh", + "id_token" => "new-identity", + "account_id" => "new-account" + }, + DateTime.add(DateTime.utc_now(), 3_600, :second) + ) + + send( + view.pid, + {:device_authorization_changed, integration.id, first.attempt_generation, "active", nil} + ) + + send( + view.pid, + {:device_authorization_changed, integration.id, first.attempt_generation, "failed", + "provider_rejected"} + ) + + _ = :sys.get_state(view.pid) + refute has_element?(view, "#integration-#{integration.id}-device-authorization-outcome") end test "shows failed authorization feedback on the affected account", %{ @@ -474,11 +508,20 @@ defmodule KodoWeb.IntegrationsLiveTest do display_name: "Failed subscription" ) - assert {:ok, _attempt} = + assert {:ok, first} = Integrations.begin_device_authorization( scope, integration.id, integration.credential_generation, + %{"device_auth_id" => "old-device", "user_code" => "OLD"}, + 30_000 + ) + + assert {:ok, second} = + Integrations.begin_device_authorization( + scope, + integration.id, + first.expected_integration_generation, %{"device_auth_id" => "failed-device", "user_code" => "FAILED"}, 30_000 ) @@ -495,8 +538,24 @@ defmodule KodoWeb.IntegrationsLiveTest do assert {:ok, _failed} = Integrations.fail_device_authorization(scope, claim, "provider_rejected") + send( + view.pid, + {:device_authorization_changed, integration.id, first.attempt_generation, "active", nil} + ) + + send( + view.pid, + {:device_authorization_changed, integration.id, first.attempt_generation, "failed", + "provider_rejected"} + ) + _ = :sys.get_state(view.pid) + assert %{attempt_generation: generation} = + live_assign(view, :device_authorization_outcomes)[integration.id] + + assert generation == second.attempt_generation + assert has_element?( view, "#integration-#{integration.id}-device-authorization-outcome[role='alert']", From 34ec711d6085c71fdaff384741b313e6c0d92868 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 12/17] refactor(agent): simplify review admission context --- lib/kodo/agent/loop.ex | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/lib/kodo/agent/loop.ex b/lib/kodo/agent/loop.ex index 26788f2..7f17caa 100644 --- a/lib/kodo/agent/loop.ex +++ b/lib/kodo/agent/loop.ex @@ -184,15 +184,13 @@ defmodule Kodo.Agent.Loop do {:ok, request} <- resolve_request(context.session_id, review), {:ok, admission} <- start_review_invocation( - context.session_id, + context, primary_invocation_id, review_invocation_id, - context.mapping, review, contract, capability_validation, - request, - context.ownership + request ), {:ok, generated} <- Sessions.dispatch_if_owner(context.ownership, fn -> @@ -239,18 +237,16 @@ defmodule Kodo.Agent.Loop do end defp start_review_invocation( - session_id, + context, primary_invocation_id, invocation_id, - mapping, review, contract, capability_validation, - request, - ownership + request ) do case Sessions.append_admitted_model_event( - session_id, + context.session_id, "review_invocation_started", %{ "invocation_id" => invocation_id, @@ -265,13 +261,13 @@ defmodule Kodo.Agent.Loop do "role_contract" => contract.id, "toolset_version" => contract.toolset_version, "capability_validation" => capability_validation, - "model_mapping" => mapping + "model_mapping" => context.mapping }, request.model, request.credential, version: 2, parent_id: primary_invocation_id, - ownership: ownership + ownership: context.ownership ) do {:ok, {_event, admission}} -> {:ok, admission} error -> normalize_admission_error(error, request) From 45b24d668fb52c4ceea25edab80f3c7b45d19c37 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 13/17] refactor(integrations): simplify retry reconciliation --- lib/kodo_web/live/integrations_live.ex | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex index 8750ce5..5890334 100644 --- a/lib/kodo_web/live/integrations_live.ex +++ b/lib/kodo_web/live/integrations_live.ex @@ -509,16 +509,25 @@ defmodule KodoWeb.IntegrationsLive do tracked_keys = MapSet.new(socket.assigns.device_authorization_tasks, &elem(&1, 1)) if connected?(socket) do - Enum.reduce(active_keys, socket, fn {integration_id, generation} = key, acc -> - if Map.has_key?(retries, key) or MapSet.member?(tracked_keys, key), - do: acc, - else: schedule_device_authorization_retry(acc, integration_id, generation) + Enum.reduce(active_keys, socket, fn key, acc -> + ensure_device_authorization_retry(acc, key, retries, tracked_keys) end) else socket end end + defp ensure_device_authorization_retry( + socket, + {integration_id, generation} = key, + retries, + tracked_keys + ) do + if Map.has_key?(retries, key) or MapSet.member?(tracked_keys, key), + do: socket, + else: schedule_device_authorization_retry(socket, integration_id, generation) + end + # Browser-facing state needs lifecycle metadata only. In particular, keeping # ciphertext in the LiveView would widen credential retention for no benefit. defp integration_metadata(integration) do From d6d7e47c79da8c6cdc91d6649178b2ededb64c5d Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 14/17] test(integrations): align deletion race outcomes --- test/kodo/integrations_test.exs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs index 51c34eb..505f6ee 100644 --- a/test/kodo/integrations_test.exs +++ b/test/kodo/integrations_test.exs @@ -136,7 +136,7 @@ defmodule Kodo.IntegrationsTest do test "returns a bounded error when connection races account deletion", %{scope: scope} do Repo.delete!(scope.user) - assert {:error, :integration_owner_not_found} = connect(scope) + assert {:error, :integration_not_found} = connect(scope) end test "replaces credentials with a new nonce and advances the generation", %{scope: scope} do @@ -655,7 +655,7 @@ defmodule Kodo.IntegrationsTest do Repo.delete!(scope.user) send(task.pid, {:continue_json_encoding, ref}) - assert {:error, :stale_credential_generation} = Task.await(task) + assert {:error, :integration_not_found} = Task.await(task) refute Repo.get(Integration, integration.id) end From f8dda6e4d4c0c2b1d9f7f4f2c26ce54043954dc6 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 15/17] test(sessions): avoid registry cleanup race --- test/kodo/sessions/active_session_test.exs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/kodo/sessions/active_session_test.exs b/test/kodo/sessions/active_session_test.exs index efcb615..0acc434 100644 --- a/test/kodo/sessions/active_session_test.exs +++ b/test/kodo/sessions/active_session_test.exs @@ -293,7 +293,10 @@ defmodule Kodo.Sessions.ActiveSessionTest do send(task, :never) assert_receive {:DOWN, ^coordinator_ref, :process, ^coordinator, :normal} - assert Registry.lookup(Kodo.SessionRegistry, session.id) == [] + + refute Enum.any?(Registry.lookup(Kodo.SessionRegistry, session.id), fn {pid, _value} -> + Process.alive?(pid) + end) end test "stops its active task when a queued turn observes session deletion", %{ From 0cec04f4db78703d7bf36d5dab03a0b16913ed8c Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Sat, 12 Sep 2026 21:13:35 +0000 Subject: [PATCH 16/17] fix(integrations): keep device codes readable --- lib/kodo_web/live/integrations_live.ex | 2 +- test/kodo_web/live/integrations_live_test.exs | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex index 5890334..f01b575 100644 --- a/lib/kodo_web/live/integrations_live.ex +++ b/lib/kodo_web/live/integrations_live.ex @@ -1033,7 +1033,7 @@ defmodule KodoWeb.IntegrationsLive do

One-time code

- + {authorization.user_code}