One-time code
-
+
{authorization.user_code}
+
@@ -1311,9 +1377,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/rfd/0005/IMPLEMENTATION.org b/rfd/0005/IMPLEMENTATION.org
index 3780fcd..c6030f2 100644
--- a/rfd/0005/IMPLEMENTATION.org
+++ b/rfd/0005/IMPLEMENTATION.org
@@ -327,7 +327,8 @@ manufactured after that no-go decision.
- [X] Render the fixed allowlisted HTTPS verification destination and one-time
code separately, reject a provider-returned verification URL that does not
match the exact allowlist, and never automatically navigate to that URL.
-- [X] Trigger claims only from attempt creation or LiveView mount/remount, use
+- [X] Trigger claims from connection-driven attempt creation, LiveView
+ mount/remount, generation-specific retry, and task-completion recovery; use
one conditional update against the exact attempt row that increments and
returns its claim epoch, and run a successful claim as a finite task under
~Kodo.ControlPlaneTaskSupervisor~ that survives the originating LiveView
@@ -346,9 +347,9 @@ manufactured after that no-go decision.
generation. Starting reauthorization advances the integration generation and
the new attempt captures that post-advance value; install OAuth credentials
only when both attempt and expected integration generations still match.
-- [X] Store access tokens, refresh tokens, account identifiers, expiries, and
- credential generations in the encrypted integration payload without logging
- device or token material.
+- [X] Store access tokens, refresh tokens, and account identifiers in the
+ encrypted integration payload. Keep expiries and credential generations as
+ queryable lifecycle metadata, without logging device or token material.
- [X] Test user deletion with OAuth credentials, an active authorization
attempt, a held claim, and in-flight refresh; all rows cascade and every late
completion becomes a no-op.
@@ -388,8 +389,10 @@ manufactured after that no-go decision.
has an accessible Copy action, is retained only while needed to present the
active attempt, and disappears after every terminal state without entering
logs or telemetry.
-- [X] Extend onboarding and missing-provider feedback for ChatGPT Subscription,
- and emit credential-free audit events for authorization start/completion and
+- [ ] Extend onboarding for ChatGPT Subscription. The baseline onboarding and
+ confirmed model-preset interaction remain Phase 1B and Phase 2 work.
+- [X] Extend missing-provider feedback for ChatGPT Subscription and emit
+ credential-free audit events for authorization start/completion and
cancellation, refresh outcomes, reauthorization, and disconnection.
- [ ] Add an explicit per-session execution-route change between OpenAI
Platform and ChatGPT Subscription only when the destination supports every
@@ -411,6 +414,25 @@ manufactured after that no-go decision.
- [X] Route requests whose persisted turn snapshot selects ~openai_codex~ only
through the user's ChatGPT integration and prohibit silent fallback to OpenAI
API billing.
+
+Production generation entry points consume a ~Kodo.LLM.InvocationAdmission~
+returned by the session transaction that records invocation start and canonical
+billing provenance. Scoped credential resolution and decryption precede that
+transaction; the transaction rechecks session ownership, integration state, and
+credential generation before permitting inference. The admission struct makes
+that internal boundary explicit, but is not an unforgeable bearer capability
+because BEAM modules can construct structs. The separately named evaluation
+entry points are out-of-band harness operations and do not represent production
+session admission. Turn acceptance holds deterministic provider-selection and
+active-row locks while validating the complete role set; those locks end at
+commit and are never held across inference.
+
+Device and refresh recovery also preserve already-admitted provider results
+while bounding automatic expired-token recovery. Explicit activation or
+disconnection invalidates older selection-restoration intent, expired device
+attempts cannot poison refresh admission, and open LiveViews reconcile polling,
+exchange, terminal, and superseded attempt state from durable storage.
+
- [ ] Test authorization ownership and expiry, polling backoff, two-node refresh
contention, multi-node and concurrent-tab device claims, claim epoch takeover
and conditional renewal, stale claimant polling and completion, pauses before
diff --git a/test/kodo/agent/loop_test.exs b/test/kodo/agent/loop_test.exs
index 183a91a..e9602c6 100644
--- a/test/kodo/agent/loop_test.exs
+++ b/test/kodo/agent/loop_test.exs
@@ -306,8 +306,17 @@ defmodule Kodo.Agent.LoopTest do
[created] = Sessions.events_after(session.id)
+ session
+ |> 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} =
@@ -914,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",
@@ -951,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,
diff --git a/test/kodo/agent/model_mapping_test.exs b/test/kodo/agent/model_mapping_test.exs
index 7f28d43..eb8befe 100644
--- a/test/kodo/agent/model_mapping_test.exs
+++ b/test/kodo/agent/model_mapping_test.exs
@@ -71,6 +71,42 @@ defmodule Kodo.Agent.ModelMappingTest do
ModelMapping.validate_snapshot(malformed)
end
+ test "canonicalizes every supported string model specification in snapshots" 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"] == "#{route}:#{selector}"
+ assert primary["execution_route"] == route
+ assert primary["model_selector"] == selector
+ assert {:ok, ^snapshot} = ModelMapping.validate_snapshot(snapshot)
+ 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())
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..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
@@ -1503,6 +1503,141 @@ 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 "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
@@ -1640,6 +1775,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
diff --git a/test/kodo/sessions/active_session_test.exs b/test/kodo/sessions/active_session_test.exs
index 237da55..0acc434 100644
--- a/test/kodo/sessions/active_session_test.exs
+++ b/test/kodo/sessions/active_session_test.exs
@@ -278,6 +278,83 @@ 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}
+
+ 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", %{
+ 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)
diff --git a/test/kodo/sessions_test.exs b/test/kodo/sessions_test.exs
index 50b840e..473aee8 100644
--- a/test/kodo/sessions_test.exs
+++ b/test/kodo/sessions_test.exs
@@ -1,11 +1,14 @@
defmodule Kodo.SessionsTest do
use Kodo.DataCase
+ alias Kodo.Accounts.User
alias Kodo.Agent.ExecutionRouteChange
alias Kodo.Agent.ModelMapping
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 +368,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 +447,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, %{
@@ -457,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
@@ -929,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
diff --git a/test/kodo_web/live/integrations_live_test.exs b/test/kodo_web/live/integrations_live_test.exs
index 2f5bcb4..e4df513 100644
--- a/test/kodo_web/live/integrations_live_test.exs
+++ b/test/kodo_web/live/integrations_live_test.exs
@@ -153,13 +153,23 @@ defmodule KodoWeb.IntegrationsLiveTest do
panel = "#integration-#{integration.id}-device-authorization"
assert has_element?(view, panel, "https://auth.openai.com/codex/device")
- assert has_element?(view, panel, "ABCD-EFGH")
+
+ assert has_element?(
+ view,
+ "#{panel} code.whitespace-nowrap",
+ "ABCD-EFGH"
+ )
assert has_element?(
view,
"#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']:not(.sr-only)"
+ )
+
refute inspect(:sys.get_state(view.pid)) =~ "private-device"
render_click(view, "cancel_device_authorization", %{
@@ -252,6 +262,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", %{
@@ -398,7 +415,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
@@ -438,7 +455,119 @@ 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")
+
+ 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", %{
+ conn: conn,
+ scope: scope
+ } do
+ assert {:ok, integration} =
+ Integrations.create_oauth_integration(scope, "openai_codex",
+ display_name: "Failed subscription"
+ )
+
+ 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
+ )
+
+ 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")
+
+ 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']",
+ "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", %{