feat(sessions): persist immutable execution routes

lib/kodo/agent/execution_route_change.ex

@@ -0,0 +1,72 @@
+defmodule Kodo.Agent.ExecutionRouteChange do
+  @moduledoc "Validates atomic execution-route changes against approved compatibility records."
+
+  alias Kodo.Agent.ModelMapping
+
+  @routes ~w(openai openai_codex)
+  @approved_records []
+
+  @doc "Returns records backed by successful exact-identity compatibility evidence."
+  def approved_records, do: @approved_records
+
+  def change(mapping, destination, records \\ @approved_records)
+
+  def change(mapping, destination, records) when destination in @routes and is_list(records) do
+    mapping = ModelMapping.snapshot(mapping)
+
+    affected =
+      Enum.filter(mapping["roles"], fn {_role, role_mapping} ->
+        role_mapping["execution_route"] in @routes and
+          role_mapping["execution_route"] != destination
+      end)
+
+    with false <- affected == [],
+         [] <- incompatible_roles(affected, destination, records) do
+      {:ok, apply_route(mapping, affected, destination)}
+    else
+      true -> {:error, :execution_route_unchanged}
+      roles when is_list(roles) -> {:error, {:incompatible_execution_route, roles}}
+    end
+  end
+
+  def change(_mapping, _destination, _records), do: {:error, :unsupported_execution_route}
+
+  defp incompatible_roles(affected, destination, records) do
+    for {role, role_mapping} <- affected,
+        not compatible?(role, role_mapping, destination, records),
+        do: role
+  end
+
+  defp compatible?(role, role_mapping, destination, records) do
+    Enum.any?(records, fn record ->
+      record.role == role and
+        record.source == role_mapping["execution_route"] and
+        record.destination == destination and
+        record.selector == role_mapping["model_selector"] and
+        record.role_contract == role_mapping["role_contract"] and
+        record.status == :approved
+    end)
+  end
+
+  defp apply_route(mapping, affected, destination) do
+    affected_roles = MapSet.new(affected, &elem(&1, 0))
+
+    roles =
+      Map.new(mapping["roles"], fn {role, role_mapping} ->
+        if MapSet.member?(affected_roles, role) do
+          selector = role_mapping["model_selector"]
+
+          {role,
+           role_mapping
+           |> Map.put("provider", destination)
+           |> Map.put("execution_route", destination)
+           |> Map.put("model", "#{destination}:#{selector}")
+           |> put_in(["sources", "model"], "session_route_change")}
+        else
+          {role, role_mapping}
+        end
+      end)
+
+    %{mapping | "roles" => roles}
+  end
+end

lib/kodo/agent/loop.ex

@@ -89,7 +89,7 @@ defmodule Kodo.Agent.Loop do
            response,
            calls,
            adapter,
-           projection.model_mapping || legacy_mapping(projection.model),
+           turn_mapping(projection),
            budgets,
            ownership
          ) do
@@ -125,8 +125,7 @@ defmodule Kodo.Agent.Loop do
   end
 
   defp review_final_answer(text, response, context) do
-    mapping =
-      context.projection.model_mapping || legacy_mapping(context.projection.model)
+    mapping = turn_mapping(context.projection)
 
     primary_invocation_id = response.payload["invocation_id"]
 
@@ -165,10 +164,11 @@ defmodule Kodo.Agent.Loop do
 
   defp run_final_review(text, primary_invocation_id, context) do
     review = ModelMapping.role!(context.mapping, :review)
+    review_model = ModelMapping.request_model(review)
     contract = Roles.fetch!(:review, review["role_contract"])
 
     with {:ok, capability_validation} <-
-           context.adapter.validate_model(review["model"], review, contract),
+           context.adapter.validate_model(review_model, review, contract),
          {:ok, request} <- resolve_request(context.session_id, review),
          :ok <-
            Phoenix.PubSub.subscribe(
@@ -458,9 +458,14 @@ defmodule Kodo.Agent.Loop do
     end
   end
 
+  defp turn_mapping(projection) do
+    projection.turn_model_mapping || projection.model_mapping || legacy_mapping(projection.model)
+  end
+
   defp infer(session_id, projection, events, adapter, budgets, invocation, ownership) do
-    mapping = projection.model_mapping || legacy_mapping(projection.model)
+    mapping = turn_mapping(projection)
     primary = ModelMapping.role!(mapping, :primary)
+    primary_model = ModelMapping.request_model(primary)
     contract = Roles.fetch!(:primary, primary["role_contract"])
 
     tools =
@@ -472,7 +477,7 @@ defmodule Kodo.Agent.Loop do
 
     with :ok <- within_budget(invocation, usage(current_turn(events)), budgets),
          {:ok, capability_validation} <-
-           adapter.validate_model(primary["model"], primary, contract),
+           adapter.validate_model(primary_model, primary, contract),
          {:ok, request} <- resolve_request(session_id, primary),
          {:ok, invocation_id} <-
            start_invocation(
@@ -487,7 +492,11 @@ defmodule Kodo.Agent.Loop do
          :ok <- rehoming_boundary(),
          {:ok, response} <-
            Sessions.dispatch_if_owner(ownership, fn ->
-             generate(adapter, request, transcript(events, contract), tools,
+             generate(
+               adapter,
+               request,
+               transcript(events, contract, primary["execution_route"] || primary["provider"]),
+               tools,
                timeout: budgets[:model_timeout],
                reasoning: primary["reasoning"]
              )
@@ -506,32 +515,62 @@ defmodule Kodo.Agent.Loop do
     end
   end
 
-  defp transcript(events, contract) do
+  defp transcript(events, contract, current_route) do
     system = %{"role" => "system", "content" => contract.prompt}
 
     primary_invocations =
       events
       |> Enum.filter(&(&1.type == "model_invocation_started"))
       |> MapSet.new(& &1.payload["invocation_id"])
 
+    invocation_routes =
+      events
+      |> Enum.filter(&(&1.type == "model_invocation_started"))
+      |> Map.new(&{&1.payload["invocation_id"], &1.payload["provider"]})
+
     Enum.reduce(events, [system], fn event, messages ->
-      case transcript_message(event, primary_invocations) do
+      case transcript_message(event, primary_invocations, invocation_routes, current_route) do
         nil -> messages
         message -> messages ++ [message]
       end
     end)
   end
 
-  defp transcript_message(%{type: "user_message", payload: payload}, _primary_invocations),
-    do: %{"role" => "user", "content" => payload["content"]}
+  defp transcript_message(
+         %{type: "user_message", payload: payload},
+         _primary_invocations,
+         _invocation_routes,
+         _current_route
+       ),
+       do: %{"role" => "user", "content" => payload["content"]}
 
-  defp transcript_message(%{type: "model_response", payload: payload}, _primary_invocations),
-    do: assistant_message(payload)
+  defp transcript_message(
+         %{type: "model_response", payload: payload},
+         _primary_invocations,
+         invocation_routes,
+         current_route
+       ) do
+    if invocation_routes[payload["invocation_id"]] == current_route do
+      assistant_message(payload)
+    else
+      visible_assistant_message(payload)
+    end
+  end
 
-  defp transcript_message(%{type: "review_feedback", payload: payload}, _primary_invocations),
-    do: %{"role" => "user", "content" => payload["content"]}
+  defp transcript_message(
+         %{type: "review_feedback", payload: payload},
+         _primary_invocations,
+         _invocation_routes,
+         _current_route
+       ),
+       do: %{"role" => "user", "content" => payload["content"]}
 
-  defp transcript_message(%{type: type, payload: payload}, primary_invocations)
+  defp transcript_message(
+         %{type: type, payload: payload},
+         primary_invocations,
+         _invocation_routes,
+         _current_route
+       )
        when type in ["tool_completed", "tool_failed"] do
     if MapSet.member?(primary_invocations, payload["invocation_id"]) do
       content =
@@ -546,19 +585,24 @@ defmodule Kodo.Agent.Loop do
     end
   end
 
-  defp transcript_message(_event, _primary_invocations), do: nil
+  defp transcript_message(_event, _primary_invocations, _invocation_routes, _current_route),
+    do: nil
 
   defp assistant_message(%{"assistant" => nil} = payload) do
+    visible_assistant_message(payload)
+  end
+
+  defp assistant_message(%{"assistant" => provider_state}),
+    do: %{"role" => "assistant", "provider_state" => provider_state}
+
+  defp visible_assistant_message(payload) do
     %{
       "role" => "assistant",
       "content" => payload["text"] || "",
       "tool_calls" => payload["tool_calls"] || []
     }
   end
 
-  defp assistant_message(%{"assistant" => provider_state}),
-    do: %{"role" => "assistant", "provider_state" => provider_state}
-
   defp start_invocation(
          session_id,
          continuation,
@@ -809,10 +853,11 @@ defmodule Kodo.Agent.Loop do
 
   defp run_search(question, parent_call, context) do
     search = ModelMapping.role!(context.mapping, :search)
+    search_model = ModelMapping.request_model(search)
     contract = Roles.fetch!(:search, search["role_contract"])
 
     with {:ok, capability_validation} <-
-           context.adapter.validate_model(search["model"], search, contract) do
+           context.adapter.validate_model(search_model, search, contract) do
       state = %{
         parent_call: parent_call,
         search: search,
@@ -1634,7 +1679,8 @@ defmodule Kodo.Agent.Loop do
 
   defp resolve_request(session_id, role) do
     with {:ok, scope} <- Sessions.owner_scope(session_id),
-         {:ok, model, reference} <- LLM.resolve_integration(scope, role["model"]) do
+         {:ok, model, reference} <-
+           LLM.resolve_integration(scope, ModelMapping.request_model(role)) do
       {:ok, %{scope: scope, model: model, reference: reference}}
     end
   end

lib/kodo/agent/model_mapping.ex

@@ -60,6 +60,53 @@ defmodule Kodo.Agent.ModelMapping do
 
   def normalize(_mapping), do: balanced()
 
+  @doc "Freezes explicit route and selector identity into every role for durable turn replay."
+  def snapshot(mapping) do
+    mapping = normalize(mapping)
+
+    roles =
+      Map.new(mapping["roles"], fn {role, role_mapping} ->
+        contract = Roles.fetch!(role_atom(role), role_mapping["role_contract"])
+
+        {role,
+         role_mapping
+         |> Map.put("execution_route", role_mapping["provider"])
+         |> Map.put("model_selector", model_selector(role_mapping["model"]))
+         |> Map.put("capability_contract", capability_contract(contract))}
+      end)
+
+    %{mapping | "roles" => roles}
+  end
+
+  @doc "Builds the provider-specific request model from an immutable route snapshot."
+  def request_model(%{"execution_route" => route, "model_selector" => selector})
+      when is_binary(route) and route != "" and is_binary(selector) and selector != "" do
+    "#{route}:#{selector}"
+  end
+
+  def request_model(%{"model" => model}), do: model
+
+  defp capability_contract(contract) do
+    %{
+      "id" => contract.id,
+      "toolset_version" => contract.toolset_version,
+      "requirements" => %{
+        "tools" => contract.capabilities.tools,
+        "structured_output" => stringify(contract.capabilities.structured_output),
+        "min_context" => contract.capabilities.min_context,
+        "input_modalities" => Enum.map(contract.capabilities.input_modalities, &to_string/1)
+      }
+    }
+  end
+
+  defp role_atom("primary"), do: :primary
+  defp role_atom("search"), do: :search
+  defp role_atom("review"), do: :review
+
+  defp stringify(value) when is_boolean(value), do: value
+  defp stringify(value) when is_atom(value), do: Atom.to_string(value)
+  defp stringify(value), do: value
+
   defp normalize_role(current, persisted) do
     normalized =
       Enum.reduce(["model", "reasoning"], current, fn field, mapping ->
@@ -124,4 +171,11 @@ defmodule Kodo.Agent.ModelMapping do
       {:error, _reason} -> nil
     end
   end
+
+  defp model_selector(model) do
+    case String.split(model, ":", parts: 2) do
+      [_provider, selector] -> selector
+      [selector] -> selector
+    end
+  end
 end

lib/kodo/integrations.ex

@@ -90,6 +90,45 @@ defmodule Kodo.Integrations do
     end
   end
 
+  @doc false
+  def admit_execution_route_change(%Scope{user: user}, provider) do
+    integration =
+      Integration
+      |> where(
+        [integration],
+        integration.user_id == ^user.id and integration.provider == ^provider and
+          integration.active
+      )
+      |> lock("FOR UPDATE")
+      |> Repo.one()
+
+    case integration do
+      %Integration{
+        connection_status: "connected",
+        validation_status: status,
+        encrypted_credentials: encrypted
+      }
+      when status != "invalid" and not is_nil(encrypted) ->
+        {:ok, integration}
+
+      %Integration{} ->
+        {:error, :destination_integration_unavailable}
+
+      nil ->
+        {:error, :destination_integration_required}
+    end
+  end
+
+  @doc false
+  def audit_execution_route_change(
+        %Scope{user: user},
+        %Integration{user_id: user_id} = integration
+      )
+      when user.id == user_id do
+    audit!(user.id, integration, "execution_route_changed")
+    :ok
+  end
+
   def list_audit_events(%Scope{user: user}) do
     AuditEvent
     |> where([event], event.actor_user_id == ^user.id)

lib/kodo/integrations/audit_event.ex

@@ -21,6 +21,7 @@ defmodule Kodo.Integrations.AuditEvent do
     refresh_invalid_grant
     refresh_reauthorization_required
     refresh_failed
+    execution_route_changed
   )
 
   @primary_key {:id, :binary_id, autogenerate: true}