defmodule SyntropyWeb.GraphViewModel do @moduledoc false alias Syntropy.{ClusterInfo, HistoryStore} @type graph_node :: %{ id: String.t(), name: String.t(), perspective: String.t(), position: float(), temporary: boolean(), selected: boolean(), active: boolean(), current: boolean() } @type graph_edge :: %{ source: String.t(), target: String.t(), weight: float(), interactions: non_neg_integer() } @type meta :: %{ ranked_agent_ids: [String.t()], task_id: String.t() | nil, proposal_id: String.t() | nil, node_id: String.t(), timestamp: String.t() | nil, trigger_kind: String.t() } @type graph :: %{ mode: String.t(), snapshot_id: String.t() | nil, nodes: [graph_node()], edges: [graph_edge()], meta: meta() } @spec from_live([map()], keyword()) :: graph() def from_live(agents, opts \\ []) do selected_agent_ids = Keyword.get(opts, :selected_agent_ids, []) active_agent_ids = Keyword.get(opts, :active_agent_ids, []) focused_agent_id = Keyword.get(opts, :focused_agent_id) %{ mode: "live", snapshot_id: nil, nodes: build_nodes(agents, selected_agent_ids, active_agent_ids, focused_agent_id), edges: build_edges(agents), meta: %{ ranked_agent_ids: Enum.map(agents, &agent_runtime_id/1), task_id: Keyword.get(opts, :task_id), proposal_id: Keyword.get(opts, :proposal_id), node_id: Syntropy.node_id(), timestamp: nil, trigger_kind: "live" } } end @spec from_replay(HistoryStore.snapshot(), keyword()) :: graph() def from_replay(snapshot, opts \\ []) do selected_agent_ids = Keyword.get(opts, :selected_agent_ids, []) focused_agent_id = Keyword.get(opts, :focused_agent_id) %{ mode: "replay", snapshot_id: snapshot.id, nodes: build_nodes(snapshot.agents, selected_agent_ids, [], focused_agent_id), edges: build_edges(snapshot.agents), meta: %{ ranked_agent_ids: snapshot.ranked_agent_ids, task_id: snapshot.task_id, proposal_id: snapshot.proposal_id, node_id: Map.get(snapshot, :node_id, "local"), timestamp: DateTime.to_iso8601(snapshot.inserted_at), trigger_kind: snapshot.trigger_kind } } end @spec to_json(graph()) :: String.t() def to_json(graph) do Jason.encode!(graph) end @doc """ Derives the runtime ids of the agents a task result selected. Prefers the enriched `selected_agents` export, falls back to the selection trace, and finally to raw `selected_agent_ids`. """ @spec task_selected_runtime_ids(map() | nil) :: [String.t()] def task_selected_runtime_ids(nil), do: [] def task_selected_runtime_ids(task_result) do cond do is_list(Map.get(task_result, :selected_agents)) and task_result.selected_agents != [] -> Enum.map(task_result.selected_agents, &Map.get(&1, :runtime_id, &1.id)) is_list(Map.get(task_result, :selection_trace)) and task_result.selection_trace != [] -> task_result.selection_trace |> Enum.filter(& &1.selected) |> Enum.map(&Map.get(&1, :runtime_id, &1.agent_id)) true -> task_result |> Map.get(:selected_agent_ids, []) |> Enum.map(&runtime_identifier_or_local/1) end end defp build_nodes(agents, selected_agent_ids, active_agent_ids, focused_agent_id) do selected_set = MapSet.new(selected_agent_ids) active_set = MapSet.new(active_agent_ids) Enum.map(agents, fn agent -> runtime_id = agent_runtime_id(agent) %{ id: runtime_id, runtime_id: runtime_id, node_id: agent_node_id(agent), name: agent_name(agent), perspective: agent_perspective(agent), position: agent_position(agent), temporary: agent_temporary?(agent), selected: identifier_selected?(selected_set, agent), active: identifier_selected?(active_set, agent), current: identifier_matches?(focused_agent_id, agent) } end) end defp build_edges(agents) do agents |> Enum.flat_map(fn agent -> source = agent_runtime_id(agent) Enum.map(agent_connections(agent), fn connection -> %{ source: source, target: Map.fetch!(connection, :target_agent_id), weight: Map.fetch!(connection, :weight), interactions: Map.fetch!(connection, :interactions) } end) end) |> Enum.reject(&(&1.source == &1.target)) |> Enum.reduce(%{}, fn edge, acc -> key = Enum.sort([edge.source, edge.target]) |> List.to_tuple() Map.update(acc, key, normalize_edge(edge), fn existing -> %{ source: elem(key, 0), target: elem(key, 1), weight: max(existing.weight, edge.weight), interactions: max(existing.interactions, edge.interactions) } end) end) |> Map.values() |> Enum.sort_by(&{&1.source, &1.target}) end defp normalize_edge(edge) do [source, target] = Enum.sort([edge.source, edge.target]) %{source: source, target: target, weight: edge.weight, interactions: edge.interactions} end defp agent_runtime_id(agent) do case Map.get(agent, :runtime_id) do runtime_id when is_binary(runtime_id) and runtime_id != "" -> runtime_id _other -> Map.fetch!(agent, :id) end end defp agent_node_id(agent) do case Map.get(agent, :node_id, "local") do "" -> "local" nil -> "local" node_id -> node_id end end defp agent_name(agent), do: Map.fetch!(agent, :name) defp agent_perspective(agent), do: Map.fetch!(agent, :perspective) defp agent_position(agent), do: Map.fetch!(agent, :position) defp agent_temporary?(agent), do: Map.fetch!(agent, :temporary) defp agent_connections(%{connections: connections} = agent) when is_map(connections) do connections |> Enum.sort_by(fn {target_id, _value} -> target_id end) |> Enum.map(fn {target_id, value} -> %{ target_agent_id: connection_target_id(agent, target_id), weight: Map.fetch!(value, :weight), interactions: Map.fetch!(value, :interactions) } end) end defp agent_connections(%{connections: connections} = agent) when is_list(connections) do connections |> Enum.map(fn connection -> %{ target_agent_id: connection_target_id(agent, Map.fetch!(connection, :target_agent_id)), weight: Map.fetch!(connection, :weight), interactions: Map.fetch!(connection, :interactions) } end) |> Enum.sort_by(& &1.target_agent_id) end defp identifier_selected?(identifier_set, agent) do runtime_id = agent_runtime_id(agent) local_id = Map.fetch!(agent, :id) MapSet.member?(identifier_set, runtime_id) or MapSet.member?(identifier_set, local_id) end defp identifier_matches?(nil, _agent), do: false defp identifier_matches?(identifier, agent) do identifier in [agent_runtime_id(agent), Map.fetch!(agent, :id)] end defp connection_target_id(agent, target_id) when is_binary(target_id) do cond do String.contains?(target_id, "::") -> target_id explicit_runtime_id?(agent) -> ClusterInfo.runtime_id(target_id, agent_node_id(agent)) true -> target_id end end defp explicit_runtime_id?(agent) do case Map.get(agent, :runtime_id) do runtime_id when is_binary(runtime_id) and runtime_id != "" -> true _other -> false end end defp runtime_identifier_or_local(agent_id) do case ClusterInfo.parse_runtime_id(agent_id) do {:ok, ref} -> ref.runtime_id :error -> agent_id end end end