- Waiting for OpenAI authorization + {if(authorization.stage == :polling, + do: "Waiting for OpenAI authorization", + else: "Completing OpenAI authorization" + )}
-Open this verification page
<.link @@ -855,6 +990,13 @@ defmodule KodoWeb.IntegrationsLive do+ Authorization was approved. Securely exchanging credentials… +
<.link - :if={!integration_connected?(integration)} + :if={integration.connection_status == "disconnected"} id={dom_id(integration, "reconnect")} patch={action_path(integration, "connect")} phx-click={JS.push_focus()} @@ -942,7 +1084,10 @@ defmodule KodoWeb.IntegrationsLive do Replace key <.link - :if={integration_connected?(integration) and integration.provider == "openai_codex"} + :if={ + integration.provider == "openai_codex" and + integration.connection_status in ~w(connected reauthorization_required) + } id={dom_id(integration, "reauthorize")} patch={action_path(integration, "reauthorize")} phx-click={JS.push_focus()} @@ -951,7 +1096,7 @@ defmodule KodoWeb.IntegrationsLive do Reauthorize <.link - :if={integration_connected?(integration)} + :if={integration.connection_status in ~w(connected reauthorization_required)} id={dom_id(integration, "disconnect")} patch={action_path(integration, "disconnect")} phx-click={JS.push_focus()} diff --git a/priv/repo/migrations/20260912195208_add_refresh_source_generation_to_provider_integrations.exs b/priv/repo/migrations/20260912195208_add_refresh_source_generation_to_provider_integrations.exs new file mode 100644 index 0000000..09f600e --- /dev/null +++ b/priv/repo/migrations/20260912195208_add_refresh_source_generation_to_provider_integrations.exs @@ -0,0 +1,15 @@ +defmodule Kodo.Repo.Migrations.AddRefreshSourceGenerationToProviderIntegrations do + use Ecto.Migration + + def change do + alter table(:provider_integrations) do + add :refresh_source_generation, :bigint + end + + create constraint( + :provider_integrations, + :provider_integrations_refresh_source_generation_valid, + check: "refresh_source_generation IS NULL OR refresh_source_generation >= 0" + ) + end +end diff --git a/priv/repo/migrations/20260912200217_add_activate_on_completion_to_device_authorization_attempts.exs b/priv/repo/migrations/20260912200217_add_activate_on_completion_to_device_authorization_attempts.exs new file mode 100644 index 0000000..28f85d9 --- /dev/null +++ b/priv/repo/migrations/20260912200217_add_activate_on_completion_to_device_authorization_attempts.exs @@ -0,0 +1,9 @@ +defmodule Kodo.Repo.Migrations.AddActivateOnCompletionToDeviceAuthorizationAttempts do + use Ecto.Migration + + def change do + alter table(:device_authorization_attempts) do + add :activate_on_completion, :boolean, null: false, default: false + end + end +end diff --git a/priv/repo/migrations/20260912213842_add_refresh_restore_active_to_provider_integrations.exs b/priv/repo/migrations/20260912213842_add_refresh_restore_active_to_provider_integrations.exs new file mode 100644 index 0000000..15ce076 --- /dev/null +++ b/priv/repo/migrations/20260912213842_add_refresh_restore_active_to_provider_integrations.exs @@ -0,0 +1,9 @@ +defmodule Kodo.Repo.Migrations.AddRefreshRestoreActiveToProviderIntegrations do + use Ecto.Migration + + def change do + alter table(:provider_integrations) do + add :refresh_restore_active, :boolean, null: false, default: false + end + end +end diff --git a/priv/repo/structure.sql b/priv/repo/structure.sql index bef1f4c..ab75fb7 100644 --- a/priv/repo/structure.sql +++ b/priv/repo/structure.sql @@ -2,7 +2,7 @@ -- PostgreSQL database dump -- -\restrict VoWETYGZi4vEwyliUNWQG3BVneag5OunvKzcS2NOBFzVtuTzyBWpDfhYIneu2ek +\restrict bhNtvgo2sLbQAwP0xzeErVvadl8Hb8oQ9uHPQt7qDfIjayod6OKCW93lzqbuYlP -- Dumped from database version 15.19 (Debian 15.19-0+deb12u1) -- Dumped by pg_dump version 15.19 (Debian 15.19-0+deb12u1) @@ -135,6 +135,7 @@ CREATE TABLE public.device_authorization_attempts ( terminal_error_code character varying(64), inserted_at timestamp without time zone NOT NULL, updated_at timestamp without time zone NOT NULL, + activate_on_completion boolean DEFAULT false NOT NULL, CONSTRAINT device_attempts_claim_epoch_valid CHECK ((claim_epoch >= 0)), CONSTRAINT device_attempts_claim_valid CHECK (((((state)::text = 'active'::text) AND (terminal_error_code IS NULL) AND (((claim_owner_id IS NULL) AND (claim_lease_expires_at IS NULL)) OR ((claim_owner_id IS NOT NULL) AND (claim_lease_expires_at IS NOT NULL)))) OR (((state)::text <> 'active'::text) AND (claim_owner_id IS NULL) AND (claim_lease_expires_at IS NULL)))), CONSTRAINT device_attempts_generations_valid CHECK (((attempt_generation > 0) AND (expected_integration_generation >= 0))), @@ -187,6 +188,8 @@ CREATE TABLE public.provider_integrations ( refresh_claim_epoch bigint DEFAULT 0 NOT NULL, refresh_claim_generation bigint, refresh_lease_expires_at timestamp without time zone, + refresh_source_generation bigint, + refresh_restore_active boolean DEFAULT false NOT NULL, CONSTRAINT provider_integrations_active_connected CHECK (((NOT active) OR ((connection_status)::text = 'connected'::text))), CONSTRAINT provider_integrations_authentication_type_valid CHECK (((authentication_type)::text = ANY ((ARRAY['api_key'::character varying, 'oauth'::character varying])::text[]))), CONSTRAINT provider_integrations_credential_generation_valid CHECK ((credential_generation >= 0)), @@ -194,6 +197,7 @@ CREATE TABLE public.provider_integrations ( CONSTRAINT provider_integrations_provider_valid CHECK (((provider)::text = ANY ((ARRAY['openai'::character varying, 'openai_codex'::character varying, 'anthropic'::character varying, 'openrouter'::character varying])::text[]))), CONSTRAINT provider_integrations_refresh_claim_epoch_valid CHECK ((refresh_claim_epoch >= 0)), CONSTRAINT provider_integrations_refresh_claim_valid CHECK ((((refresh_claim_owner_id IS NULL) AND (refresh_claim_generation IS NULL) AND (refresh_lease_expires_at IS NULL)) OR ((refresh_claim_owner_id IS NOT NULL) AND (refresh_claim_generation IS NOT NULL) AND (refresh_claim_generation >= 0) AND (refresh_lease_expires_at IS NOT NULL)))), + CONSTRAINT provider_integrations_refresh_source_generation_valid CHECK (((refresh_source_generation IS NULL) OR (refresh_source_generation >= 0))), CONSTRAINT provider_integrations_state_valid CHECK (((((connection_status)::text = 'disconnected'::text) AND ((validation_status)::text = 'unverified'::text) AND (encrypted_credentials IS NULL) AND (encryption_key_version IS NULL) AND (credential_format_version IS NULL) AND (expires_at IS NULL) AND (validated_at IS NULL) AND (refreshed_at IS NULL) AND (validation_error_code IS NULL)) OR (((connection_status)::text = 'connected'::text) AND ((validation_status)::text = ANY ((ARRAY['unverified'::character varying, 'valid'::character varying, 'invalid'::character varying, 'unavailable'::character varying])::text[])) AND (encrypted_credentials IS NOT NULL) AND (encryption_key_version IS NOT NULL) AND (credential_format_version IS NOT NULL)) OR (((connection_status)::text = 'reauthorization_required'::text) AND ((validation_status)::text = 'unverified'::text) AND (encrypted_credentials IS NOT NULL) AND (encryption_key_version IS NOT NULL) AND (credential_format_version IS NOT NULL)))) ); @@ -780,7 +784,7 @@ ALTER TABLE ONLY public.users_tokens -- PostgreSQL database dump complete -- -\unrestrict VoWETYGZi4vEwyliUNWQG3BVneag5OunvKzcS2NOBFzVtuTzyBWpDfhYIneu2ek +\unrestrict bhNtvgo2sLbQAwP0xzeErVvadl8Hb8oQ9uHPQt7qDfIjayod6OKCW93lzqbuYlP INSERT INTO public."schema_migrations" (version) VALUES (20260807073017); INSERT INTO public."schema_migrations" (version) VALUES (20260808062115); @@ -800,3 +804,6 @@ INSERT INTO public."schema_migrations" (version) VALUES (20260904051620); INSERT INTO public."schema_migrations" (version) VALUES (20260907083117); INSERT INTO public."schema_migrations" (version) VALUES (20260907161929); INSERT INTO public."schema_migrations" (version) VALUES (20260910044734); +INSERT INTO public."schema_migrations" (version) VALUES (20260912195208); +INSERT INTO public."schema_migrations" (version) VALUES (20260912200217); +INSERT INTO public."schema_migrations" (version) VALUES (20260912213842); diff --git a/rfd/0005/IMPLEMENTATION.org b/rfd/0005/IMPLEMENTATION.org index 10f72c5..3780fcd 100644 --- a/rfd/0005/IMPLEMENTATION.org +++ b/rfd/0005/IMPLEMENTATION.org @@ -430,7 +430,7 @@ manufactured after that no-go decision. cross-route replay, no automatic route fallback, same-origin and cross-origin redirect rejection, malicious verification URLs, absence of a credential-bearing follow-up request, and secret redaction. -- [X] Run the Phase 0 live compatibility suite against the production path and +- [ ] Run the Phase 0 live compatibility suite against the production path and do not declare ChatGPT/Codex implementation complete while a required capability fails; classify later contract failures as regressions and never fall back to Platform billing. diff --git a/test/kodo/agent/execution_route_change_test.exs b/test/kodo/agent/execution_route_change_test.exs index c35a952..2f4a5d8 100644 --- a/test/kodo/agent/execution_route_change_test.exs +++ b/test/kodo/agent/execution_route_change_test.exs @@ -40,6 +40,50 @@ defmodule Kodo.Agent.ExecutionRouteChangeTest do assert ModelMapping.role!(mapping, :primary)["provider"] == "openai" end + test "rejects evidence for a different reasoning or capability envelope" do + mapping = mapping_for_selector("gpt-4o-mini") + records = records_for(mapping, "openai_codex") + + reasoning_mismatch = + Enum.map(records, &Map.put(&1, :reasoning, "medium")) + + assert {:error, {:incompatible_execution_route, ["primary", "review", "search"]}} = + ExecutionRouteChange.change(mapping, "openai_codex", reasoning_mismatch) + + capability_mismatch = + Enum.map(records, fn record -> + put_in(record, [:capability_contract, "requirements", "tools"], false) + end) + + assert {:error, {:incompatible_execution_route, ["primary", "review", "search"]}} = + ExecutionRouteChange.change(mapping, "openai_codex", capability_mismatch) + end + + test "preserves and verifies an explicit mapping capability envelope" do + mapping = mapping_for_selector("gpt-4o-mini") |> ModelMapping.snapshot() + records = records_for(mapping, "openai_codex") + + changed_requirement = + put_in( + mapping, + ["roles", "primary", "capability_contract", "requirements", "min_context"], + 200_000 + ) + + assert {:error, {:incompatible_execution_route, ["primary"]}} = + ExecutionRouteChange.change(changed_requirement, "openai_codex", records) + end + + test "returns a bounded error for a partially explicit malformed snapshot" do + malformed = + mapping_for_selector("gpt-4o-mini") + |> ModelMapping.snapshot() + |> update_in(["roles", "primary"], &Map.delete(&1, "capability_contract")) + + assert {:error, :invalid_model_mapping_snapshot} = + ExecutionRouteChange.change(malformed, "openai_codex", []) + end + defp mapping_for_selector(selector) do ModelMapping.balanced([ {"test", @@ -52,13 +96,18 @@ defmodule Kodo.Agent.ExecutionRouteChangeTest do end defp records_for(mapping, destination) do - Enum.map(mapping["roles"], fn {role, role_mapping} -> + mapping + |> ModelMapping.snapshot() + |> Map.fetch!("roles") + |> Enum.map(fn {role, role_mapping} -> %{ role: role, source: "openai", destination: destination, selector: role_mapping["model"] |> String.split(":", parts: 2) |> List.last(), role_contract: role_mapping["role_contract"], + reasoning: role_mapping["reasoning"], + capability_contract: role_mapping["capability_contract"], status: :approved } end) diff --git a/test/kodo/agent/loop_test.exs b/test/kodo/agent/loop_test.exs index dee42bf..183a91a 100644 --- a/test/kodo/agent/loop_test.exs +++ b/test/kodo/agent/loop_test.exs @@ -2,6 +2,7 @@ defmodule Kodo.Agent.LoopTest do use Kodo.DataCase alias Kodo.Agent.Loop + alias Kodo.Agent.ModelMapping alias Kodo.Agent.ModelSettings alias Kodo.Agent.Tools alias Kodo.Integrations @@ -99,10 +100,14 @@ defmodule Kodo.Agent.LoopTest do [created] = Sessions.events_after(session.id) mapping = - update_in(created.payload["model_mapping"], ["roles"], fn roles -> + created.payload["model_mapping"] + |> Kodo.Agent.ModelMapping.snapshot() + |> update_in(["roles"], fn roles -> Map.new(roles, fn {role, role_mapping} -> {role, role_mapping + |> Map.put("provider", "openai_codex") + |> Map.put("model", "openai_codex:gpt-5.4") |> Map.put("execution_route", "openai_codex") |> Map.put("model_selector", "gpt-5.4")} end) @@ -138,6 +143,44 @@ defmodule Kodo.Agent.LoopTest do refute Enum.any?(Sessions.events_after(session.id), &(&1.type == "model_invocation_started")) end + test "rejects a malformed persisted turn capability envelope before dispatch", %{ + session: session, + ownership: ownership + } do + [created] = Sessions.events_after(session.id) + + baseline = ModelMapping.snapshot(created.payload["model_mapping"]) + assert {:ok, ^baseline} = ModelMapping.validate_snapshot(baseline) + + malformed = update_in(baseline, ["roles", "primary"], &Map.delete(&1, "capability_contract")) + + {:ok, _snapshot} = + Sessions.append_event( + session.id, + "turn_route_snapshot", + %{"route_revision" => 1, "model_mapping" => malformed}, + source: "system", + ownership: ownership + ) + + {:ok, _message} = + Sessions.append_event( + session.id, + "user_message", + %{"role" => "user", "content" => "do not repair this snapshot"}, + ownership: ownership + ) + + assert {:error, :invalid_model_mapping_snapshot} = + Loop.run(session.id, + adapter: Kodo.Test.FakeLLM, + budgets: budgets([]), + ownership: ownership + ) + + refute_received {:llm_request, _model, _system, _tools, _opts} + end + test "drops provider-private assistant state when replay crosses execution routes", %{ runner: runner, session: session, diff --git a/test/kodo/agent/model_mapping_test.exs b/test/kodo/agent/model_mapping_test.exs index 8ca5a66..7f28d43 100644 --- a/test/kodo/agent/model_mapping_test.exs +++ b/test/kodo/agent/model_mapping_test.exs @@ -58,4 +58,55 @@ defmodule Kodo.Agent.ModelMappingTest do assert mapping["roles"]["search"]["model"] == "not-a-model" assert mapping["roles"]["search"]["provider"] == nil end + + test "rejects malformed immutable snapshots instead of filling current defaults" do + snapshot = ModelMapping.snapshot(ModelMapping.balanced()) + + assert {:ok, ^snapshot} = ModelMapping.validate_snapshot(snapshot) + + malformed = + update_in(snapshot, ["roles", "primary"], &Map.delete(&1, "capability_contract")) + + assert {:error, :invalid_model_mapping_snapshot} = + ModelMapping.validate_snapshot(malformed) + end + + test "rejects malformed capability requirements instead of accepting a partial envelope" do + snapshot = ModelMapping.snapshot(ModelMapping.balanced()) + + malformed = + put_in( + snapshot, + ["roles", "primary", "capability_contract", "requirements", "min_context"], + "current default" + ) + + assert {:error, :invalid_model_mapping_snapshot} = + ModelMapping.validate_snapshot(malformed) + end + + test "rejects capability variants and toolsets with no executable implementation" do + snapshot = ModelMapping.snapshot(ModelMapping.balanced()) + + unsupported_output = + put_in( + snapshot, + ["roles", "primary", "capability_contract", "requirements", "structured_output"], + true + ) + + assert {:error, :invalid_model_mapping_snapshot} = + ModelMapping.validate_snapshot(unsupported_output) + + unsupported_toolset = + snapshot + |> put_in(["roles", "primary", "toolset_version"], "workspace-v999") + |> put_in( + ["roles", "primary", "capability_contract", "toolset_version"], + "workspace-v999" + ) + + assert {:error, :invalid_model_mapping_snapshot} = + ModelMapping.validate_snapshot(unsupported_toolset) + end end diff --git a/test/kodo/integrations/device_authorization_test.exs b/test/kodo/integrations/device_authorization_test.exs index 7003fd9..32e3fd7 100644 --- a/test/kodo/integrations/device_authorization_test.exs +++ b/test/kodo/integrations/device_authorization_test.exs @@ -118,6 +118,46 @@ defmodule Kodo.Integrations.DeviceAuthorizationTest do assert calls(client) == [] end + test "terminalizes an attempt whose deadline expires while its worker is paused", %{ + scope: scope, + integration: integration + } do + client = fake_client(poll: [{:ok, exchange_payload()}]) + {claim, payload} = claimed_attempt(scope, integration) + + Repo.update!(change(claim, provider_deadline: DateTime.add(DateTime.utc_now(), -1, :second))) + + assert {:error, :stale_device_authorization_claim} = + DeviceAuthorization.run(scope, claim, payload, client_opts(client)) + + assert %{state: "expired", encrypted_payload: nil} = Repo.reload!(claim) + assert calls(client) == [] + end + + test "a healthy claim cannot be replaced by another task using the same node owner", %{ + scope: scope, + integration: integration + } do + owner = Ecto.UUID.generate() + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization(scope, integration.id, owner) + + assert {:error, :device_authorization_not_claimable} = + Integrations.claim_device_authorization(scope, integration.id, owner) + + assert Repo.reload!(claim).claim_epoch == claim.claim_epoch + end + test "a poll admitted before takeover may finish but cannot persist its result", %{ scope: scope, integration: integration @@ -162,6 +202,59 @@ defmodule Kodo.Integrations.DeviceAuthorizationTest do assert calls(client) == [{:poll, payload}] end + test "waits a full provider interval after a pending response", %{ + scope: scope, + integration: integration + } do + owner = self() + + blocking_poll = fn :poll, _payload -> + send(owner, {:pending_poll_started, self()}) + + receive do + :finish_pending -> :pending + end + end + + second_poll = fn :poll, _payload -> + send(owner, :second_poll_started) + {:error, :device_authorization_rejected} + end + + client = fake_client(poll: [blocking_poll, second_poll]) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 100 + ) + + assert {:ok, {claim, payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + Repo.update!(change(claim, next_poll_at: DateTime.add(DateTime.utc_now(), -1, :second))) + claim = Repo.reload!(claim) + supervisor = start_supervised!(Task.Supervisor) + + task = + Task.Supervisor.async_nolink(supervisor, fn -> + DeviceAuthorization.run(scope, claim, payload, client_opts(client)) + end) + + assert_receive {:pending_poll_started, poller} + send(poller, :finish_pending) + refute_receive :second_poll_started, 50 + assert_receive :second_poll_started, 200 + assert {:error, :device_authorization_rejected} = Task.await(task) + end + test "begin persists the provider response and launches a supervised finite task", %{ scope: scope, integration: integration @@ -195,6 +288,55 @@ defmodule Kodo.Integrations.DeviceAuthorizationTest do assert Repo.reload!(started.attempt).state == "failed" end + test "begin claims the new attempt before another worker can take it", %{ + scope: scope, + integration: integration + } do + owner = self() + supervisor = start_supervised!(Task.Supervisor) + + blocking_poll = fn :poll, _payload -> + send(owner, {:begin_worker_claimed, self()}) + + receive do + :finish_begin_worker -> {:error, :device_authorization_rejected} + end + end + + client = + fake_client( + create: [ + {:ok, + %{ + payload: %{"device_auth_id" => "device", "user_code" => "CODE"}, + polling_interval_ms: 0, + verification_url: "https://auth.openai.com/codex/device" + }} + ], + poll: [blocking_poll] + ) + + assert {:ok, started} = + DeviceAuthorization.begin( + scope, + integration.id, + integration.credential_generation, + client_opts(client, supervisor: supervisor, claim_owner_id: Ecto.UUID.generate()) + ) + + assert_receive {:begin_worker_claimed, worker} + + assert {:error, :device_authorization_not_claimable} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + send(worker, :finish_begin_worker) + assert {:error, :device_authorization_rejected} = Task.await(started.task) + end + test "begin rejects a client-selected verification destination before persistence", %{ scope: scope, integration: integration diff --git a/test/kodo/integrations/oauth_refresh_test.exs b/test/kodo/integrations/oauth_refresh_test.exs index ba2aff2..2b97e81 100644 --- a/test/kodo/integrations/oauth_refresh_test.exs +++ b/test/kodo/integrations/oauth_refresh_test.exs @@ -5,6 +5,7 @@ defmodule Kodo.Integrations.OAuthRefreshTest do alias Kodo.Integrations alias Kodo.Integrations.CredentialEncryption alias Kodo.Integrations.DeviceAuthorizationAttempt + alias Kodo.Integrations.DeviceAuthorizationEncryption alias Kodo.Integrations.Integration alias Kodo.Integrations.OAuthAccessValidation alias Kodo.Integrations.OAuthRefresh @@ -164,7 +165,222 @@ defmodule Kodo.Integrations.OAuthRefreshTest do assert persisted["access_token"] == second_tokens["access_token"] end - test "user deletion cascades OAuth state and fences an in-flight refresh", context do + test "a delayed success restores its active selection after provisional invalid grant", + context do + owner = self() + + first_response = fn _refresh_token -> + send(owner, {:successful_refresh_started, self()}) + + receive do + :finish_successful_refresh -> {:ok, rotated_tokens("account", "recovered")} + end + end + + client = fake_client([first_response, {:error, :invalid_grant}]) + first = Task.async(fn -> refresh(context, client) end) + assert_receive {:successful_refresh_started, first_worker} + + context.integration + |> Repo.reload!() + |> change(refresh_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second)) + |> Repo.update!() + + assert {:error, :integration_reauthorization_required} = refresh(context, client) + refute Repo.reload!(context.integration).active + + send(first_worker, :finish_successful_refresh) + assert {:ok, recovered} = Task.await(first) + assert recovered.active + + assert {:ok, active} = + Integrations.get_active_integration_by_provider(context.scope, "openai_codex") + + assert active.id == context.integration.id + end + + test "a delayed success does not replace a newer explicit active selection", context do + owner = self() + + first_response = fn _refresh_token -> + send(owner, {:selectable_refresh_started, self()}) + + receive do + :finish_selectable_refresh -> {:ok, rotated_tokens("account", "late")} + end + end + + client = fake_client([first_response, {:error, :invalid_grant}]) + first = Task.async(fn -> refresh(context, client) end) + assert_receive {:selectable_refresh_started, first_worker} + + context.integration + |> Repo.reload!() + |> change(refresh_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second)) + |> Repo.update!() + + assert {:error, :integration_reauthorization_required} = refresh(context, client) + + sibling = oauth_integration(context.scope, "Second subscription") + + assert {:ok, sibling} = + Integrations.oauth_succeeded(context.scope, sibling.id, 0, current_tokens(), + expires_at: DateTime.add(DateTime.utc_now(), 3_600, :second) + ) + + assert {:ok, sibling} = + Integrations.activate( + context.scope, + sibling.id, + sibling.credential_generation + ) + + assert {:ok, _disconnected} = + Integrations.disconnect( + context.scope, + sibling.id, + sibling.credential_generation + ) + + send(first_worker, :finish_selectable_refresh) + assert {:ok, recovered} = Task.await(first) + refute recovered.active + + assert {:error, :integration_not_found} = + Integrations.get_active_integration_by_provider(context.scope, "openai_codex") + end + + test "an ordinary delayed refresh does not undo a later decision to leave no active account", + context do + owner = self() + + response = fn _refresh_token -> + send(owner, {:ordinary_refresh_started, self()}) + + receive do + :finish_ordinary_refresh -> {:ok, rotated_tokens("account", "late")} + end + end + + client = fake_client([response]) + refresh = Task.async(fn -> refresh(context, client) end) + assert_receive {:ordinary_refresh_started, worker} + + sibling = oauth_integration(context.scope, "Temporary selection") + + assert {:ok, sibling} = + Integrations.oauth_succeeded(context.scope, sibling.id, 0, current_tokens(), + expires_at: DateTime.add(DateTime.utc_now(), 3_600, :second) + ) + + assert {:ok, sibling} = + Integrations.activate(context.scope, sibling.id, sibling.credential_generation) + + assert {:ok, _disconnected} = + Integrations.disconnect(context.scope, sibling.id, sibling.credential_generation) + + send(worker, :finish_ordinary_refresh) + assert {:ok, recovered} = Task.await(refresh) + refute recovered.active + + assert {:error, :integration_not_found} = + Integrations.get_active_integration_by_provider(context.scope, "openai_codex") + end + + test "persists a rotated response that expires during a worker pause and refreshes it again", + context do + admitted_at = DateTime.add(DateTime.utc_now(), -2, :second) + + expired_rotation = + rotated_tokens("account", "short") + |> put_in( + ["access_token"], + jwt(%{"exp" => DateTime.utc_now() |> DateTime.to_unix() |> Kernel.-(1)}) + ) + + client = fake_client([{:ok, expired_rotation}, {:ok, rotated_tokens("account", "usable")}]) + + assert {:ok, refreshed} = + OAuthRefresh.ensure_fresh(context.scope, context.integration, + force: true, + now: admitted_at, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate() + ) + + assert refreshed.credential_generation == context.integration.credential_generation + 2 + assert Agent.get(client, &Enum.reverse(&1.refresh_tokens)) == ["old-refresh", "short-refresh"] + + assert {:ok, credentials} = CredentialEncryption.decrypt(refreshed) + assert credentials["refresh_token"] == "usable-refresh" + end + + test "bounds automatic recovery when refreshed access is already expired", context do + admitted_at = DateTime.add(DateTime.utc_now(), -10, :second) + + expired = + rotated_tokens("account", "expired") + |> put_in( + ["access_token"], + jwt(%{"exp" => DateTime.utc_now() |> DateTime.to_unix() |> Kernel.-(1)}) + ) + + client = fake_client([{:ok, expired}, {:ok, expired}]) + + assert {:error, :provider_unavailable} = + OAuthRefresh.ensure_fresh(context.scope, context.integration, + force: true, + now: admitted_at, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate() + ) + + _ = :sys.get_state(client) + assert length(Agent.get(client, & &1.refresh_tokens)) == 2 + end + + test "persists a rotated response after its caller stops waiting", context do + owner = self() + + blocked = fn _refresh_token -> + send(owner, {:rotated_response_ready, self()}) + + receive do + :persist_rotated_response -> {:ok, rotated_tokens("account", "detached")} + end + end + + client = fake_client([blocked]) + + caller = + Task.async(fn -> + OAuthRefresh.ensure_fresh(context.scope, context.integration, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate(), + wait_timeout: 1 + ) + end) + + assert_receive {:rotated_response_ready, worker} + worker_ref = Process.monitor(worker) + assert {:error, :provider_unavailable} = Task.await(caller) + send(worker, :persist_rotated_response) + assert_receive {:DOWN, ^worker_ref, :process, ^worker, :normal} + + refreshed = Repo.reload!(context.integration) + assert refreshed.refresh_source_generation == context.integration.credential_generation + assert refreshed.credential_generation == context.integration.credential_generation + 1 + assert {:ok, credentials} = CredentialEncryption.decrypt(refreshed) + assert credentials["refresh_token"] == "detached-refresh" + end + + test "an active reauthorization attempt blocks refresh without using its credential", context do assert {:ok, _attempt} = Integrations.begin_device_authorization( context.scope, @@ -174,7 +390,93 @@ defmodule Kodo.Integrations.OAuthRefreshTest do 0 ) - integration = Repo.reload!(context.integration) + reauthorizing = Repo.reload!(context.integration) + client = fake_client([]) + + assert {:error, :integration_reauthorization_required} = + OAuthRefresh.ensure_fresh(context.scope, reauthorizing, + force: true, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate() + ) + + assert Agent.get(client, & &1.refresh_tokens) == [] + + assert Repo.reload!(reauthorizing).credential_generation == + reauthorizing.credential_generation + end + + test "an expired abandoned authorization attempt is terminalized before refresh", context do + assert {:ok, attempt} = + Integrations.begin_device_authorization( + context.scope, + context.integration.id, + context.integration.credential_generation, + %{"device_auth_id" => "abandoned", "user_code" => "OLD"}, + 0 + ) + + Repo.update!( + change(attempt, provider_deadline: DateTime.add(DateTime.utc_now(), -1, :second)) + ) + + reauthorizing = Repo.reload!(context.integration) + client = fake_client([{:ok, rotated_tokens("account")}]) + + assert {:ok, _refreshed} = + OAuthRefresh.ensure_fresh(context.scope, reauthorizing, + force: true, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate() + ) + + assert %{state: "expired", encrypted_payload: nil} = Repo.reload!(attempt) + assert Agent.get(client, &Enum.reverse(&1.refresh_tokens)) == ["old-refresh"] + end + + test "an attempt expiring after refresh claim does not veto admission", context do + owner_id = Ecto.UUID.generate() + + assert {:ok, claim} = + Integrations.claim_refresh( + context.scope, + context.integration.id, + context.integration.credential_generation, + owner_id + ) + + insert_device_attempt(context, DateTime.add(DateTime.utc_now(), -1, :second)) + + assert {:ok, admitted} = Integrations.admit_refresh_claim(context.scope, claim) + assert admitted.refresh_claim_owner_id == owner_id + assert :ok = Integrations.release_refresh_claim(context.scope, claim) + end + + test "a pre-network admission failure can release its exact refresh claim", context do + owner_id = Ecto.UUID.generate() + + assert {:ok, claim} = + Integrations.claim_refresh( + context.scope, + context.integration.id, + context.integration.credential_generation, + owner_id + ) + + insert_device_attempt(context, DateTime.add(DateTime.utc_now(), 60, :second)) + + assert {:error, :stale_refresh_claim} = + Integrations.admit_refresh_claim(context.scope, claim) + + assert :ok = Integrations.release_refresh_claim(context.scope, claim) + assert is_nil(Repo.reload!(context.integration).refresh_claim_owner_id) + end + + test "user deletion cascades OAuth state and fences an in-flight refresh", context do owner = self() blocked = fn _refresh_token -> @@ -186,10 +488,20 @@ defmodule Kodo.Integrations.OAuthRefreshTest do end client = fake_client([blocked]) - refresh = Task.async(fn -> refresh(%{context | integration: integration}, client) end) + refresh = Task.async(fn -> refresh(context, client) end) assert_receive {:deletion_refresh_started, worker} - assert Repo.reload!(integration).refresh_claim_owner_id + assert Repo.reload!(context.integration).refresh_claim_owner_id + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + context.scope, + context.integration.id, + context.integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + Repo.delete!(context.scope.user) refute Repo.exists?(Integration) refute Repo.exists?(DeviceAuthorizationAttempt) @@ -216,6 +528,49 @@ defmodule Kodo.Integrations.OAuthRefreshTest do ) end + defp oauth_integration(scope, display_name) do + %Integration{user_id: scope.user.id} + |> Integration.create_changeset(%{ + provider: "openai_codex", + authentication_type: "oauth", + display_name: display_name + }) + |> Repo.insert!() + end + + defp insert_device_attempt(context, deadline) do + timestamp = DateTime.utc_now() + + attempt = %DeviceAuthorizationAttempt{ + id: Ecto.UUID.generate(), + user_id: context.scope.user.id, + integration_id: context.integration.id, + provider: "openai_codex", + attempt_generation: 1, + expected_integration_generation: context.integration.credential_generation, + activate_on_completion: false + } + + assert {:ok, encrypted} = + DeviceAuthorizationEncryption.encrypt(attempt, %{ + "device_auth_id" => "device", + "user_code" => "CODE" + }) + + attempt + |> Map.merge(encrypted) + |> DeviceAuthorizationAttempt.create_changeset(%{ + provider: "openai_codex", + attempt_generation: 1, + expected_integration_generation: context.integration.credential_generation, + activate_on_completion: false, + provider_deadline: deadline, + polling_interval_ms: 0, + next_poll_at: timestamp + }) + |> Repo.insert!() + end + defp current_tokens do %{ "access_token" => "old-access", diff --git a/test/kodo/integrations/req_device_authorization_client_test.exs b/test/kodo/integrations/req_device_authorization_client_test.exs index c8fca92..53f5938 100644 --- a/test/kodo/integrations/req_device_authorization_client_test.exs +++ b/test/kodo/integrations/req_device_authorization_client_test.exs @@ -175,7 +175,10 @@ defmodule Kodo.Integrations.ReqDeviceAuthorizationClientTest do plug = fn conn -> Agent.update(counter, &(&1 + 1)) - Plug.Conn.send_resp(conn, 302, "redirect without a location") + + conn + |> Plug.Conn.put_resp_header("location", "https://attacker.example/collect") + |> Plug.Conn.send_resp(302, "redirect") end payloads = [ diff --git a/test/kodo/integrations/req_refresh_client_test.exs b/test/kodo/integrations/req_refresh_client_test.exs index 53a7b58..71ce119 100644 --- a/test/kodo/integrations/req_refresh_client_test.exs +++ b/test/kodo/integrations/req_refresh_client_test.exs @@ -29,7 +29,7 @@ defmodule Kodo.Integrations.ReqRefreshClientTest do ReqRefreshClient.refresh("refresh-secret", plug: plug) end - test "classifies only an exact invalid_grant response as invalid credentials" do + test "classifies the pinned permanent refresh failures as invalid credentials" do invalid_grant = fn conn -> conn |> Plug.Conn.put_status(400) |> Req.Test.json(%{"error" => "invalid_grant"}) end @@ -37,6 +37,18 @@ defmodule Kodo.Integrations.ReqRefreshClientTest do assert {:error, :invalid_grant} = ReqRefreshClient.refresh("refresh-secret", plug: invalid_grant) + for {status, body} <- [ + {401, %{"code" => "token_expired"}}, + {400, %{"code" => "refresh_token_expired"}}, + {400, %{"code" => "refresh_token_reused"}}, + {400, %{"code" => "refresh_token_invalidated"}} + ] do + plug = fn conn -> conn |> Plug.Conn.put_status(status) |> Req.Test.json(body) end + + assert {:error, :invalid_grant} = + ReqRefreshClient.refresh("refresh-secret", plug: plug) + end + for body <- [%{"error" => "temporarily_unavailable"}, %{"message" => "invalid_grant"}] do plug = fn conn -> conn |> Plug.Conn.put_status(400) |> Req.Test.json(body) end @@ -46,10 +58,20 @@ defmodule Kodo.Integrations.ReqRefreshClientTest do end test "rejects redirects, malformed success, and unsafe caller input with bounded errors" do - redirect = fn conn -> Plug.Conn.send_resp(conn, 302, "private") end + counter = start_supervised!({Agent, fn -> 0 end}) + + redirect = fn conn -> + Agent.update(counter, &(&1 + 1)) + + conn + |> Plug.Conn.put_resp_header("location", "https://auth.openai.com/oauth/redirected") + |> Plug.Conn.send_resp(302, "private") + end + malformed = fn conn -> Req.Test.json(conn, %{"refresh_token" => "private"}) end assert {:error, :redirect} = ReqRefreshClient.refresh("refresh-secret", plug: redirect) + assert Agent.get(counter, & &1) == 1 assert {:error, :oauth_refresh_response_invalid} = ReqRefreshClient.refresh("refresh-secret", plug: malformed) diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs index bb9022b..148fc67 100644 --- a/test/kodo/integrations_test.exs +++ b/test/kodo/integrations_test.exs @@ -823,6 +823,7 @@ defmodule Kodo.IntegrationsTest do assert after_start.connection_status == "connected" assert after_start.validation_status == "valid" assert after_start.active + assert is_nil(after_start.refresh_source_generation) assert {:ok, _cancelled} = Integrations.cancel_device_authorization( @@ -856,6 +857,77 @@ defmodule Kodo.IntegrationsTest do Repo.reload!(valid) end + test "a later authorization completion preserves its inactive account decision", %{ + scope: scope + } do + first = oauth_integration(scope) + + assert {:ok, active} = + Integrations.oauth_succeeded(scope, first.id, 0, %{"access_token" => "first"}) + + second = oauth_integration(scope) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + second.id, + second.credential_generation, + %{"device_auth_id" => "second", "user_code" => "SECOND"}, + 0 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization( + scope, + second.id, + Ecto.UUID.generate() + ) + + assert {:ok, _disconnected} = + Integrations.disconnect(scope, active.id, active.credential_generation) + + assert {:ok, completed} = + Integrations.complete_device_authorization( + scope, + claim, + %{ + "access_token" => "second-access", + "refresh_token" => "second-refresh", + "id_token" => "second-id", + "account_id" => "second-account" + }, + DateTime.add(DateTime.utc_now(), 3_600, :second) + ) + + refute completed.active + end + + test "an inactive account does not become active merely because its sibling was disconnected", + %{ + scope: scope + } do + first = oauth_integration(scope) + + assert {:ok, active} = + Integrations.oauth_succeeded(scope, first.id, 0, %{"access_token" => "first"}) + + second = oauth_integration(scope) + + assert {:ok, _disconnected} = + Integrations.disconnect(scope, active.id, active.credential_generation) + + assert {:ok, attempt} = + Integrations.begin_device_authorization( + scope, + second.id, + second.credential_generation, + %{"device_auth_id" => "second", "user_code" => "SECOND"}, + 0 + ) + + refute attempt.activate_on_completion + end + test "attempt operations suppress identifiers and ciphertext from query observability", %{ scope: scope } do @@ -981,6 +1053,15 @@ defmodule Kodo.IntegrationsTest do assert {:error, :device_authorization_not_claimable} = Integrations.claim_device_authorization(scope, integration.id, second_owner) + assert {:error, :device_authorization_not_claimable} = + Integrations.claim_device_authorization(scope, integration.id, first_owner) + + Repo.update!( + change(first_claim, + claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second) + ) + ) + assert {:ok, {replacement_claim, ^payload}} = Integrations.claim_device_authorization(scope, integration.id, first_owner) @@ -1308,6 +1389,120 @@ defmodule Kodo.IntegrationsTest do assert Repo.reload!(integration).connection_status == "disconnected" end + test "completion accepts an exchange admitted before its lease expires", %{scope: scope} do + integration = oauth_integration(scope) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + assert {:ok, claim} = + Integrations.store_device_authorization_exchange(scope, claim, %{ + "authorization_code" => "code", + "code_challenge" => "challenge", + "code_verifier" => "verifier" + }) + + assert {:ok, {admission, _payload}} = + Integrations.admit_device_authorization_exchange(scope, claim) + + Repo.update!( + change(admission, claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second)) + ) + + credentials = %{ + "access_token" => "access", + "refresh_token" => "refresh", + "id_token" => "identity", + "account_id" => "account" + } + + assert {:ok, completed} = + Integrations.complete_device_authorization( + scope, + admission, + credentials, + DateTime.add(DateTime.utc_now(), 3_600, :second) + ) + + assert completed.connection_status == "connected" + assert Repo.reload!(admission).state == "completed" + end + + test "device completion does not undo a later decision to leave no active account", %{ + scope: scope + } do + integration = oauth_integration(scope) + + assert {:ok, integration} = + Integrations.oauth_succeeded(scope, integration.id, 0, %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh", + "account_id" => "old-account" + }) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + sibling = oauth_integration(scope) + + assert {:ok, sibling} = + Integrations.oauth_succeeded(scope, sibling.id, 0, %{ + "access_token" => "sibling-access", + "refresh_token" => "sibling-refresh", + "account_id" => "sibling-account" + }) + + assert {:ok, sibling} = + Integrations.activate(scope, sibling.id, sibling.credential_generation) + + assert {:ok, _disconnected} = + Integrations.disconnect(scope, sibling.id, sibling.credential_generation) + + assert {:ok, completed} = + Integrations.complete_device_authorization( + scope, + Repo.reload!(claim), + %{ + "access_token" => "new-access", + "refresh_token" => "new-refresh", + "id_token" => "new-identity", + "account_id" => "old-account" + }, + DateTime.add(DateTime.utc_now(), 3_600, :second) + ) + + refute completed.active + + assert {:error, :integration_not_found} = + Integrations.get_active_integration_by_provider(scope, "openai_codex") + end + test "cancellation is owned and generation fenced and removes the one-time code", %{ scope: scope } do diff --git a/test/kodo/llm/credential_resolver_test.exs b/test/kodo/llm/credential_resolver_test.exs index ef0500c..8fa7e9b 100644 --- a/test/kodo/llm/credential_resolver_test.exs +++ b/test/kodo/llm/credential_resolver_test.exs @@ -44,7 +44,7 @@ defmodule Kodo.LLM.CredentialResolverTest do {:finch_request, fn _request -> :forged end} ] do assert {:error, :credential_options_not_allowed} = - LLM.generate( + LLM.generate_evaluation( context.scope, resolved_model, reference, @@ -57,6 +57,17 @@ defmodule Kodo.LLM.CredentialResolverTest do ) end + assert {:ok, credential} = LLM.admit_credential(context.scope, resolved_model, reference) + + assert {:error, :invocation_not_admitted} = + LLM.generate_admitted( + credential, + [%{"role" => "user", "content" => "final answer"}], + [], + adapter: Kodo.Test.FakeLLM, + timeout: 1_000 + ) + assert {:ok, _replaced} = Integrations.replace_credentials( context.scope, @@ -66,7 +77,7 @@ defmodule Kodo.LLM.CredentialResolverTest do ) assert {:error, %Kodo.LLM.ProviderError{} = error} = - LLM.generate( + LLM.generate_evaluation( context.scope, resolved_model, reference, @@ -209,7 +220,7 @@ defmodule Kodo.LLM.CredentialResolverTest do assert invalid.credential_generation == reference.credential_generation end - test "keeps admitted references pinned while new preflight selects the activated account", + test "rejects a selected reference when another account becomes active", context do assert {:ok, second} = Integrations.connect(context.scope, "openai", "api_key", %{ @@ -219,9 +230,7 @@ defmodule Kodo.LLM.CredentialResolverTest do assert {:ok, _activated} = Integrations.activate(context.scope, second.id, second.credential_generation) - assert {:ok, admitted} = resolve(context) - assert admitted.integration_id == context.integration.id - assert admitted.token == "scoped-secret" + assert {:error, :stale_credential_generation} = resolve(context) assert {:ok, next_reference} = CredentialResolver.reference(context.scope, model()) assert next_reference.integration_id == second.id diff --git a/test/kodo/llm/req_llm_test.exs b/test/kodo/llm/req_llm_test.exs index b960079..1901267 100644 --- a/test/kodo/llm/req_llm_test.exs +++ b/test/kodo/llm/req_llm_test.exs @@ -93,6 +93,7 @@ defmodule Kodo.LLM.ReqLLMTest do assert options[:auth_mode] == :oauth assert options[:access_token] == "request-local-access" assert options[:chatgpt_account_id] == "request-local-account" + assert options[:provider_options] == [codex_originator: "kodo"] refute Keyword.has_key?(options, :refresh_token) end diff --git a/test/kodo/llm/safe_req_adapter_test.exs b/test/kodo/llm/safe_req_adapter_test.exs index c95827e..26d5d4e 100644 --- a/test/kodo/llm/safe_req_adapter_test.exs +++ b/test/kodo/llm/safe_req_adapter_test.exs @@ -5,6 +5,13 @@ defmodule Kodo.LLM.SafeReqAdapterTest do alias Kodo.LLM.SafeReqAdapter + test "uses an origin-form request target for absolute provider URLs" do + url = URI.parse("https://chatgpt.com/backend-api/codex/responses?stream=true") + + assert SafeReqAdapter.request_target(url) == + "/backend-api/codex/responses?stream=true" + end + test "removes operation secrets and untrusted details before error steps" do request = Req.new( diff --git a/test/kodo/sessions_test.exs b/test/kodo/sessions_test.exs index ca98e7c..50b840e 100644 --- a/test/kodo/sessions_test.exs +++ b/test/kodo/sessions_test.exs @@ -2,8 +2,12 @@ defmodule Kodo.SessionsTest do use Kodo.DataCase alias Kodo.Agent.ExecutionRouteChange + alias Kodo.Agent.ModelMapping + alias Kodo.Agent.ModelSettings alias Kodo.Cluster.Instances alias Kodo.Integrations + alias Kodo.LLM + alias Kodo.LLM.ProviderError alias Kodo.Runners alias Kodo.Sessions alias Kodo.Sessions.Session @@ -311,6 +315,9 @@ defmodule Kodo.SessionsTest do runner: runner, scope: scope } do + assert {:ok, _integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "turn-key"}) + {:ok, session} = Sessions.create_session(scope, %{ runner_id: runner.id, @@ -358,6 +365,209 @@ defmodule Kodo.SessionsTest do assert Enum.count(Sessions.events_after(session.id), &(&1.type == "user_message")) == 1 end + test "rejects a turn before acceptance when any configured role lacks access", %{ + runner: runner, + scope: scope + } do + assert {:ok, _override} = + ModelSettings.put_user_override(scope, :review, %{ + model: "anthropic:claude-3-5-haiku-latest" + }) + + assert {:ok, _integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "primary-key"}) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Missing review provider", + model: "openai:gpt-4o-mini" + }) + + assert {:error, %ProviderError{provider: "anthropic", kind: :integration_required}} = + Sessions.start_turn(scope, session.id, "Do not accept", Ecto.UUID.generate()) + + assert Enum.map(Sessions.events_after(session.id), & &1.type) == ["session_created"] + end + + test "rechecks every role inside the transaction that accepts the turn", %{ + runner: runner, + scope: scope + } do + assert {:ok, _override} = + ModelSettings.put_user_override(scope, :review, %{ + model: "anthropic:claude-3-5-haiku-latest" + }) + + assert {:ok, _primary} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "primary-key"}) + + assert {:ok, review} = + Integrations.connect(scope, "anthropic", "api_key", %{"api_key" => "review-key"}) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Atomic role preflight", + model: "openai:gpt-4o-mini" + }) + + assert {:ok, _disconnected} = + Integrations.disconnect(scope, review.id, review.credential_generation) + + assert {:error, %ProviderError{provider: "anthropic", kind: :integration_required}} = + Sessions.begin_turn(session.id, "Do not accept") + + assert Enum.map(Sessions.events_after(session.id), & &1.type) == ["session_created"] + end + + test "holds provider selection locks through turn acceptance", %{ + runner: runner, + scope: scope + } do + assert {:ok, _integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "locked-key"}) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Atomic provider preflight", + model: "openai:gpt-4o-mini" + }) + + assert {:ok, _events} = Sessions.begin_turn(session.id, "Accept atomically") + + key = "provider-integration:#{scope.user.id}:openai" + + lock_available? = + Task.async(fn -> + Ecto.Adapters.SQL.Sandbox.unboxed_run(Repo, fn -> + %{rows: [[available?]]} = + Ecto.Adapters.SQL.query!( + Repo, + "SELECT pg_try_advisory_xact_lock(hashtextextended($1, 0))", + [key] + ) + + available? + end) + end) + |> Task.await() + + refute lock_available? + end + + test "atomically records invocation admission for the exact active credential generation", %{ + runner: runner, + scope: scope + } do + assert {:ok, integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "admitted-key"}) + + assert {:ok, model, reference} = LLM.resolve_integration(scope, "openai:gpt-4o-mini") + assert {:ok, credential} = LLM.admit_credential(scope, model, reference) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Admitted generation", + model: "openai:gpt-4o-mini" + }) + + payload = %{ + "invocation_id" => Ecto.UUID.generate(), + "provider" => "forged-provider", + "authentication_type" => "forged-authentication", + "billing_path" => "forged-billing" + } + + assert {:error, :stale_credential_generation} = + Sessions.append_admitted_model_event( + session.id, + "model_invocation_started", + payload, + model, + %{credential | billing_path: :aggregator} + ) + + assert {:ok, {event, admission}} = + Sessions.append_admitted_model_event( + session.id, + "model_invocation_started", + payload, + model, + credential + ) + + assert event.payload == %{ + payload + | "provider" => credential.provider, + "authentication_type" => credential.authentication_type, + "billing_path" => Atom.to_string(credential.billing_path) + } + + refute inspect(event) =~ "admitted-key" + assert integration.credential_generation == credential.credential_generation + + assert {:error, :credential_options_not_allowed} = + LLM.generate_admitted(admission, [], [], + adapter: Kodo.Test.FakeLLM, + timeout: 1_000, + api_key: "forged" + ) + + assert {:ok, _disconnected} = + Integrations.disconnect(scope, integration.id, integration.credential_generation) + + assert {:ok, %{type: :final_answer}} = + LLM.generate_admitted( + admission, + [%{"role" => "user", "content" => "final answer"}], + [], + adapter: Kodo.Test.FakeLLM, + timeout: 1_000 + ) + end + + test "rejects a decrypted credential replaced before invocation admission", %{ + runner: runner, + scope: scope + } do + assert {:ok, integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "stale-key"}) + + assert {:ok, model, reference} = LLM.resolve_integration(scope, "openai:gpt-4o-mini") + assert {:ok, credential} = LLM.admit_credential(scope, model, reference) + + assert {:ok, _replacement} = + Integrations.replace_credentials( + scope, + integration.id, + integration.credential_generation, + %{"api_key" => "replacement-key"} + ) + + assert {:ok, session} = + Sessions.create_session(scope, %{ + runner_id: runner.id, + title: "Stale generation", + model: "openai:gpt-4o-mini" + }) + + events_before = Sessions.events_after(session.id) + + assert {:error, :stale_credential_generation} = + Sessions.append_admitted_model_event( + session.id, + "model_invocation_started", + %{"invocation_id" => Ecto.UUID.generate()}, + model, + credential + ) + + assert Sessions.events_after(session.id) == events_before + end + test "rejects route changes without approved exact-identity evidence", %{ runner: runner, scope: scope @@ -406,6 +616,21 @@ defmodule Kodo.SessionsTest do runner: runner, scope: scope } do + assert {:ok, codex} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Route subscription" + ) + + assert {:ok, _codex} = + Integrations.oauth_succeeded(scope, codex.id, codex.credential_generation, %{ + "access_token" => "route-access", + "refresh_token" => "route-refresh", + "account_id" => "route-account" + }) + + assert {:ok, _platform} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "route-key"}) + {:ok, session} = Sessions.create_session(scope, %{ runner_id: runner.id, @@ -415,6 +640,11 @@ defmodule Kodo.SessionsTest do {:ok, [first_snapshot, _message, _status]} = Sessions.begin_turn(session.id, "First") + [created | _events] = Sessions.events_after(session.id) + + primary = + created.payload["model_mapping"] |> ModelMapping.snapshot() |> ModelMapping.role!(:primary) + records = [ %{ role: "primary", @@ -422,12 +652,12 @@ defmodule Kodo.SessionsTest do destination: "openai", selector: "gpt-5.4", role_contract: "alpha-v1", + reasoning: primary["reasoning"], + capability_contract: primary["capability_contract"], status: :approved } ] - [created | _events] = Sessions.events_after(session.id) - assert {:ok, changed_mapping} = ExecutionRouteChange.change( created.payload["model_mapping"], @@ -435,6 +665,13 @@ defmodule Kodo.SessionsTest do records ) + changed_mapping = + put_in( + changed_mapping, + ["roles", "primary", "capability_contract", "requirements", "min_context"], + 200_000 + ) + assert {:ok, route_event} = Sessions.append_event( session.id, @@ -465,6 +702,15 @@ defmodule Kodo.SessionsTest do "primary", "execution_route" ]) == "openai" + + assert get_in(second_snapshot.payload, [ + "model_mapping", + "roles", + "primary", + "capability_contract", + "requirements", + "min_context" + ]) == 200_000 end test "advances ownership epochs and fences every stale state mutation", %{ diff --git a/test/kodo_web/live/integrations_live_test.exs b/test/kodo_web/live/integrations_live_test.exs index 7248721..2f5bcb4 100644 --- a/test/kodo_web/live/integrations_live_test.exs +++ b/test/kodo_web/live/integrations_live_test.exs @@ -163,7 +163,7 @@ defmodule KodoWeb.IntegrationsLiveTest do refute inspect(:sys.get_state(view.pid)) =~ "private-device" render_click(view, "cancel_device_authorization", %{ - "attempt" => Ecto.UUID.generate(), + "device_authorization_attempt_id" => Ecto.UUID.generate(), "generation" => "1" }) @@ -254,6 +254,235 @@ defmodule KodoWeb.IntegrationsLiveTest do assert Repo.reload!(attempt).state == "expired" end + test "resumes an exchange-stage authorization after its worker lease expires", %{ + conn: conn, + scope: scope + } do + owner = self() + + blocking_exchange = fn :exchange, _payload -> + send(owner, {:exchange_resumed, self()}) + + receive do + :finish_exchange -> {:ok, device_tokens()} + end + end + + _client = configure_device_client(exchange: [blocking_exchange]) + Phoenix.PubSub.subscribe(Kodo.PubSub, "integration:#{scope.user.id}") + + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Recovering subscription" + ) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization(scope, integration.id, Ecto.UUID.generate()) + + assert {:ok, persisted} = + Integrations.store_device_authorization_exchange(scope, claim, %{ + "authorization_code" => "authorization", + "code_challenge" => "challenge", + "code_verifier" => "verifier" + }) + + Repo.update!( + Ecto.Changeset.change(persisted, + claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second) + ) + ) + + {:ok, view, _html} = live(conn, ~p"/integrations") + assert_receive {:exchange_resumed, worker} + + panel = "#integration-#{integration.id}-device-authorization" + assert has_element?(view, panel, "Completing OpenAI authorization") + assert has_element?(view, panel, "Securely exchanging credentials") + refute has_element?(view, "#integration-#{integration.id}-copy-device-code") + + send(worker, :finish_exchange) + integration_id = integration.id + assert_receive message = {:integration_changed, ^integration_id, _generation} + send(view.pid, message) + _ = :sys.get_state(view.pid) + + assert {:ok, connected} = Integrations.get_integration(scope, integration.id) + assert connected.connection_status == "connected" + refute has_element?(view, panel) + end + + test "cancelling authorization removes its code from another open tab", %{ + conn: conn, + scope: scope, + user: user + } do + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Shared subscription" + ) + + assert {:ok, attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "SHARED"}, + 30_000 + ) + + {:ok, first_view, _html} = live(conn, ~p"/integrations") + {:ok, second_view, _html} = live(log_in_user(build_conn(), user), ~p"/integrations") + panel = "#integration-#{integration.id}-device-authorization" + assert has_element?(first_view, panel, "SHARED") + assert has_element?(second_view, panel, "SHARED") + + first_view + |> element("#integration-#{integration.id}-cancel-device-authorization") + |> render_click() + + _ = :sys.get_state(second_view.pid) + refute has_element?(second_view, panel) + assert Repo.reload!(attempt).state == "cancelled" + end + + test "stops remount retries when the exact authorization attempt becomes terminal", %{ + conn: conn, + scope: scope + } do + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Contended subscription" + ) + + assert {:ok, attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CONTENDED"}, + 30_000 + ) + + assert {:ok, {_claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + {:ok, view, _html} = live(conn, ~p"/integrations") + retry_key = {integration.id, attempt.attempt_generation} + assert Map.has_key?(live_assign(view, :device_authorization_retries), retry_key) + + assert {:ok, _cancelled} = + Integrations.cancel_device_authorization( + scope, + attempt.id, + attempt.attempt_generation + ) + + _ = :sys.get_state(view.pid) + assert live_assign(view, :device_authorization_retries) == %{} + + send(view.pid, {:retry_device_authorization, integration.id, attempt.attempt_generation}) + _ = :sys.get_state(view.pid) + assert live_assign(view, :device_authorization_retries) == %{} + end + + test "drops an old retry immediately when a new authorization supersedes it", %{ + conn: conn, + scope: scope + } do + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Superseded subscription" + ) + + assert {:ok, first} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "first", "user_code" => "FIRST"}, + 30_000 + ) + + assert {:ok, {_claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + {:ok, view, _html} = live(conn, ~p"/integrations") + retry_key = {integration.id, first.attempt_generation} + assert Map.has_key?(live_assign(view, :device_authorization_retries), retry_key) + + assert {:ok, second} = + Integrations.begin_device_authorization( + scope, + integration.id, + first.expected_integration_generation, + %{"device_auth_id" => "second", "user_code" => "SECOND"}, + 30_000 + ) + + _ = :sys.get_state(view.pid) + assert second.attempt_generation > first.attempt_generation + assert live_assign(view, :device_authorization_retries) == %{} + end + + test "updates an open authorization panel when polling advances to exchange", %{ + conn: conn, + scope: scope + } do + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Advancing subscription" + ) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + integration.id, + integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "ADVANCE"}, + 30_000 + ) + + assert {:ok, {claim, _payload}} = + Integrations.claim_device_authorization( + scope, + integration.id, + Ecto.UUID.generate() + ) + + {:ok, view, _html} = live(conn, ~p"/integrations") + panel = "#integration-#{integration.id}-device-authorization" + assert has_element?(view, panel, "Waiting for OpenAI authorization") + assert has_element?(view, "#integration-#{integration.id}-copy-device-code") + + assert {:ok, _exchange} = + Integrations.store_device_authorization_exchange(scope, claim, %{ + "authorization_code" => "code", + "code_challenge" => "challenge", + "code_verifier" => "verifier" + }) + + _ = :sys.get_state(view.pid) + assert has_element?(view, panel, "Completing OpenAI authorization") + refute has_element?(view, "#integration-#{integration.id}-copy-device-code") + end + test "shows completion and reauthorization after the device flow succeeds", %{ conn: conn, scope: scope @@ -312,6 +541,37 @@ defmodule KodoWeb.IntegrationsLiveTest do assert has_element?(view, "#{status(integration)} dd.text-green-700", "Valid") end + test "offers recovery and disconnect actions when ChatGPT authorization is required", %{ + conn: conn, + scope: scope + } do + assert {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Expired subscription" + ) + + assert {:ok, connected} = + Integrations.oauth_succeeded(scope, integration.id, 0, %{ + "access_token" => "access", + "refresh_token" => "refresh" + }) + + assert {:ok, required} = + Integrations.refresh_invalid_grant( + scope, + connected.id, + connected.credential_generation + ) + + {:ok, view, _html} = live(conn, ~p"/integrations") + assert has_element?(view, "#integration-#{required.id}-reauthorize", "Reauthorize") + assert has_element?(view, "#integration-#{required.id}-disconnect", "Disconnect") + refute has_element?(view, "#integration-#{required.id}-reconnect") + + view |> element("#integration-#{required.id}-reauthorize") |> render_click() + assert has_element?(view, "#device-authorization-form") + end + test "adds multiple accounts for one provider without exposing secrets", %{ conn: conn, scope: scope