|
1 | 1 | defmodule Quoracle.Agent.ConsensusHandler.ChildrenInjector do |
2 | 2 | @moduledoc """ |
3 | | - Handles children injection into consensus messages. |
4 | | - Extracted from ConsensusHandler to maintain <500 line modules. |
| 3 | + Injects children context into consensus messages as a `<children>` XML block. |
5 | 4 |
|
6 | | - v2.0: Message enrichment — cross-references state.messages to show |
7 | | - latest_message_preview and latest_message_at per child in consensus context. |
| 5 | + Discovers children from two sources: `state.children` (tracked via casts) and |
| 6 | + Registry (fallback for race conditions where casts haven't arrived yet). Merges |
| 7 | + both sources, filters to live children, enriches each with the latest inbox |
| 8 | + message preview, and prepends the block to the last consensus message. |
8 | 9 | """ |
9 | 10 |
|
10 | 11 | alias Quoracle.Agent.ConsensusHandler.Helpers |
11 | 12 |
|
12 | 13 | @max_message_length 100 |
13 | 14 |
|
14 | | - @doc "Injects children context (up to 20) into last message. Returns messages unchanged only if messages empty." |
| 15 | + @doc "Injects children context into last message. Returns messages unchanged only if messages empty." |
15 | 16 | @spec inject_children_context(map(), list(map())) :: list(map()) |
16 | 17 | def inject_children_context(state, messages) do |
17 | 18 | if messages == [], do: messages, else: inject_children_block(state, messages) |
18 | 19 | end |
19 | 20 |
|
20 | 21 | defp inject_children_block(state, messages) do |
21 | | - children = Map.get(state, :children, []) |
22 | 22 | registry = Map.get(state, :registry) |
23 | | - live_children = if children == [], do: [], else: filter_live_children(children, registry) |
| 23 | + registry_children = discover_registry_children(state, registry) |
| 24 | + |
| 25 | + live_children = |
| 26 | + state |
| 27 | + |> Map.get(:children, []) |
| 28 | + |> filter_live_children(registry) |
| 29 | + |> merge_registry_children(registry_children) |
24 | 30 |
|
25 | 31 | if live_children == [] do |
26 | 32 | inject_empty_children(messages) |
27 | 33 | else |
28 | 34 | inbox = Map.get(state, :messages, []) |
29 | | - enriched = enrich_with_messages(Enum.take(live_children, 20), inbox) |
| 35 | + enriched = enrich_with_messages(live_children, inbox) |
30 | 36 | inject_into_last_message(messages, enriched) |
31 | 37 | end |
32 | 38 | end |
@@ -93,6 +99,61 @@ defmodule Quoracle.Agent.ConsensusHandler.ChildrenInjector do |
93 | 99 | end |
94 | 100 | end |
95 | 101 |
|
| 102 | + @spec discover_registry_children(map(), atom() | pid() | nil) :: list(map()) |
| 103 | + defp discover_registry_children(%{agent_id: nil}, _registry), do: [] |
| 104 | + defp discover_registry_children(_state, nil), do: [] |
| 105 | + |
| 106 | + defp discover_registry_children(state, registry) do |
| 107 | + case Map.get(state, :agent_id) do |
| 108 | + nil -> |
| 109 | + [] |
| 110 | + |
| 111 | + agent_id -> |
| 112 | + case safe_registry_lookup(registry, agent_id) do |
| 113 | + {:ok, parent_pid} -> |
| 114 | + registry |
| 115 | + |> safe_find_children_by_parent(parent_pid) |
| 116 | + |> Enum.map(&build_registry_child/1) |
| 117 | + |
| 118 | + :not_found -> |
| 119 | + [] |
| 120 | + end |
| 121 | + end |
| 122 | + end |
| 123 | + |
| 124 | + @spec safe_find_children_by_parent(atom() | pid(), pid()) :: list({String.t(), map()}) |
| 125 | + defp safe_find_children_by_parent(registry, parent_pid) do |
| 126 | + try do |
| 127 | + Registry.select(registry, [ |
| 128 | + {{{:agent, :"$1"}, :"$2", :"$3"}, [{:==, {:map_get, :parent_pid, :"$3"}, parent_pid}], |
| 129 | + [{{:"$1", :"$3"}}]} |
| 130 | + ]) |
| 131 | + rescue |
| 132 | + _ -> [] |
| 133 | + catch |
| 134 | + _, _ -> [] |
| 135 | + end |
| 136 | + end |
| 137 | + |
| 138 | + # Registry composite is a map with keys :pid, :agent_id, :task_id, :parent_pid, |
| 139 | + # :parent_id, :registered_at — but NOT :spawned_at or :budget_allocated, |
| 140 | + # so we use approximate values for the child entry. |
| 141 | + @spec build_registry_child({String.t(), map()}) :: map() |
| 142 | + defp build_registry_child({agent_id, _composite}) do |
| 143 | + %{ |
| 144 | + agent_id: agent_id, |
| 145 | + spawned_at: DateTime.utc_now(), |
| 146 | + budget_allocated: nil |
| 147 | + } |
| 148 | + end |
| 149 | + |
| 150 | + @spec merge_registry_children(list(map()), list(map())) :: list(map()) |
| 151 | + defp merge_registry_children(children, registry_children) do |
| 152 | + tracked_ids = MapSet.new(children, & &1.agent_id) |
| 153 | + |
| 154 | + children ++ Enum.reject(registry_children, &MapSet.member?(tracked_ids, &1.agent_id)) |
| 155 | + end |
| 156 | + |
96 | 157 | # Enrich children with latest message data from the agent's inbox. |
97 | 158 | # Cross-references each child's agent_id against inbox messages to find |
98 | 159 | # the most recent message (by timestamp) from that child. |
|
0 commit comments