|
| 1 | +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC |
| 2 | +# SPDX-License-Identifier: Apache-2.0 |
| 3 | +defmodule Trinity.Subagents do |
| 4 | + @moduledoc """ |
| 5 | + Delegation to child sessions (slice 080). |
| 6 | +
|
| 7 | + A subagent is an ordinary session with `origin: "subagent"` and its `parent_id` set. It runs one |
| 8 | + turn against a brief and returns a result; its messages stay in its own history and never enter |
| 9 | + the parent's. |
| 10 | +
|
| 11 | + **Why that isolation is the point rather than a detail.** The reason to delegate at all is that |
| 12 | + the parent's context is finite and expensive. A delegation that copied the child's working back |
| 13 | + into the parent would cost more than doing the work inline, and the parent would pay it twice: |
| 14 | + once in tokens and once in a context window full of somebody else's reasoning. So the parent |
| 15 | + receives the child's *result* and a session id it can open, and nothing else. |
| 16 | +
|
| 17 | + **What a subagent is not.** It is not a sandbox. A child inherits the parent's permission scope |
| 18 | + and every effect it causes passes the same gate and leaves the same receipts as the parent's own |
| 19 | + (docs/07). Delegation buys context isolation and concurrency; it buys no authority separation, |
| 20 | + and a caller who wants less authority in a child has to say so through the toolset, not by |
| 21 | + assuming that "subagent" means "contained". |
| 22 | +
|
| 23 | + ## Why a subagent is not restarted |
| 24 | +
|
| 25 | + OTP's instinct is to restart a crashed child, and for a stateless worker that is right. It is |
| 26 | + wrong here, and the reason is the one durable-execution systems state plainly: **match the |
| 27 | + retry semantics to the side effect.** At-least-once is for idempotent work; at-most-once is for |
| 28 | + anything that charges a card, sends a message or writes to the world. An agent turn is an |
| 29 | + arbitrary sequence of tool calls, so it is the second kind by default. |
| 30 | +
|
| 31 | + Replaying a turn in this tree is not merely wasteful, it is unsound in a specific way. An |
| 32 | + approval granted `:once` covers one execution and would correctly refuse the replay; but a |
| 33 | + `:session` or `:always` grant would cover it silently, so a restart could re-run an effect the |
| 34 | + person approved once and never expected twice. Nothing in the effects layer carries an |
| 35 | + idempotency key today. |
| 36 | +
|
| 37 | + So `restarts:` defaults to **0**: a child that dies reports the death to its parent, which can |
| 38 | + decide. A caller who knows a brief is idempotent (a search, a summary, a read) can raise it, and |
| 39 | + the option is named `restarts` rather than `retries` so that the thing being repeated is visibly |
| 40 | + a whole child rather than one call. |
| 41 | +
|
| 42 | + **This deviates from the slice's written acceptance criterion**, which said restart once by |
| 43 | + default. The deviation and its reasoning are recorded in the slice's NOTES for the owner to |
| 44 | + accept or veto. |
| 45 | +
|
| 46 | + ## Budgets |
| 47 | +
|
| 48 | + Every delegation carries a budget and none is optional: turns, tokens and wall clock. A child |
| 49 | + that exceeds one is terminated and the parent gets a named error rather than a crash, because a |
| 50 | + parent that cannot tell "the child failed" from "the child is still thinking" cannot make a |
| 51 | + decision either way. |
| 52 | + """ |
| 53 | + |
| 54 | + alias Trinity.Sessions |
| 55 | + |
| 56 | + @default_budget %{turns: 1, tokens: 100_000, timeout_ms: 120_000} |
| 57 | + # At-most-once by default. See the moduledoc: an agent turn is an arbitrary sequence of tool |
| 58 | + # calls, and a `:session` or `:always` approval would cover a replay silently. |
| 59 | + @default_restarts 0 |
| 60 | + @default_concurrency 3 |
| 61 | + @summary_bytes 4_000 |
| 62 | + |
| 63 | + @type brief :: String.t() |
| 64 | + @type result :: %{ |
| 65 | + session_id: String.t(), |
| 66 | + status: :ok | :budget | :error | :cancelled, |
| 67 | + text: String.t(), |
| 68 | + reason: term() | nil |
| 69 | + } |
| 70 | + |
| 71 | + @doc "The budget applied when a caller names none. Exposed so a test asserts it rather than restating it." |
| 72 | + @spec default_budget() :: map() |
| 73 | + def default_budget, do: @default_budget |
| 74 | + |
| 75 | + @doc "How many times a dead child is retried when the caller names no number. Zero, deliberately." |
| 76 | + @spec default_restarts() :: non_neg_integer() |
| 77 | + def default_restarts, do: @default_restarts |
| 78 | + |
| 79 | + @doc """ |
| 80 | + Runs one brief as a child of `parent_session_id` and returns its result. |
| 81 | +
|
| 82 | + Options: `:title`, `:persona_id`, `:budget` (a map merged over `default_budget/0`). |
| 83 | + """ |
| 84 | + @spec delegate(String.t(), brief(), keyword()) :: {:ok, result()} | {:error, term()} |
| 85 | + def delegate(parent_session_id, brief, opts \\ []) when is_binary(brief) do |
| 86 | + budget = Map.merge(@default_budget, Map.new(Keyword.get(opts, :budget, %{}))) |
| 87 | + |
| 88 | + restarts = Keyword.get(opts, :restarts, @default_restarts) |
| 89 | + |
| 90 | + with {:ok, child} <- create_child(parent_session_id, brief, opts) do |
| 91 | + {:ok, run(child, brief, budget, restarts)} |
| 92 | + end |
| 93 | + end |
| 94 | + |
| 95 | + @doc """ |
| 96 | + Runs several briefs concurrently, at most `:concurrency` at a time, and returns their results in |
| 97 | + the order the briefs were given. |
| 98 | +
|
| 99 | + The cap is not a performance knob. A parent that fans out without one turns a single delegation |
| 100 | + into as many concurrent model calls as it has briefs, which is how a rate limit and a bill are |
| 101 | + discovered at the same moment. |
| 102 | + """ |
| 103 | + @spec delegate_many(String.t(), [brief()], keyword()) :: [{:ok, result()} | {:error, term()}] |
| 104 | + def delegate_many(parent_session_id, briefs, opts \\ []) when is_list(briefs) do |
| 105 | + cap = Keyword.get(opts, :concurrency, @default_concurrency) |
| 106 | + |
| 107 | + timeout = |
| 108 | + Map.get(Map.new(Keyword.get(opts, :budget, %{})), :timeout_ms, @default_budget.timeout_ms) |
| 109 | + |
| 110 | + Trinity.LLM.TaskSupervisor |
| 111 | + |> Task.Supervisor.async_stream_nolink( |
| 112 | + briefs, |
| 113 | + fn brief -> delegate(parent_session_id, brief, opts) end, |
| 114 | + max_concurrency: cap, |
| 115 | + timeout: timeout + 5_000, |
| 116 | + on_timeout: :kill_task, |
| 117 | + ordered: true |
| 118 | + ) |
| 119 | + |> Enum.map(fn |
| 120 | + {:ok, result} -> result |
| 121 | + {:exit, reason} -> {:error, {:subagent_exit, reason}} |
| 122 | + end) |
| 123 | + end |
| 124 | + |
| 125 | + @doc """ |
| 126 | + Whether a child is still working. |
| 127 | +
|
| 128 | + Asked of the process, not of the row. `SessionRow.status` is the session's lifecycle - active, |
| 129 | + archived, compacted - and a finished subagent is still `"active"` forever, so a panel that reads |
| 130 | + it reports every child as running until somebody archives it. The screenshot for this slice's |
| 131 | + AC7 is what caught that: four children, three of them long finished, all labelled active with a |
| 132 | + stop button each. |
| 133 | + """ |
| 134 | + @spec running?(String.t()) :: boolean() |
| 135 | + def running?(session_id) do |
| 136 | + # `Sessions.state/1` answers with a view map, not the bare state atom. Comparing the map to |
| 137 | + # `:idle` is never equal, so the first version of this reported every child as running and the |
| 138 | + # panel's labels were wrong in the opposite direction from the bug it was written to fix. |
| 139 | + case Sessions.state(session_id) do |
| 140 | + %{state: state} -> state != :idle |
| 141 | + _ -> false |
| 142 | + end |
| 143 | + rescue |
| 144 | + _ -> false |
| 145 | + catch |
| 146 | + :exit, _ -> false |
| 147 | + end |
| 148 | + |
| 149 | + @doc """ |
| 150 | + A short, distinguishing form of a session id. |
| 151 | +
|
| 152 | + The **tail**, not the head. These ids are UUIDv7, which is time-ordered, so children created in |
| 153 | + the same second share their leading bytes: measured, four ids generated together had one |
| 154 | + distinct 8-character prefix between them. A short id whose whole job is to tell two rows apart |
| 155 | + has to come from the random end. |
| 156 | + """ |
| 157 | + @spec short_id(String.t()) :: String.t() |
| 158 | + def short_id(id) when is_binary(id), do: String.slice(id, -8, 8) |
| 159 | + |
| 160 | + @doc "The children of a session, newest first." |
| 161 | + @spec children(String.t()) :: [Sessions.SessionRow.t()] |
| 162 | + def children(parent_session_id) do |
| 163 | + Sessions.list_sessions(limit: 200) |
| 164 | + |> Enum.filter(&(&1.parent_id == parent_session_id)) |
| 165 | + end |
| 166 | + |
| 167 | + @doc """ |
| 168 | + Cancels a session's whole subtree, depth first, and returns how many were cancelled. |
| 169 | +
|
| 170 | + Depth first because cancelling a parent before its children leaves the children running with |
| 171 | + nobody waiting for their results, which is the shape that produces work nobody asked for and |
| 172 | + nobody reads. |
| 173 | + """ |
| 174 | + @spec cancel_subtree(String.t()) :: non_neg_integer() |
| 175 | + def cancel_subtree(session_id) do |
| 176 | + descendants = |
| 177 | + session_id |
| 178 | + |> children() |
| 179 | + |> Enum.flat_map(fn child -> [child.id | Enum.map(children(child.id), & &1.id)] end) |
| 180 | + |
| 181 | + Enum.each(descendants, &cancel_one/1) |
| 182 | + cancel_one(session_id) |
| 183 | + length(descendants) + 1 |
| 184 | + end |
| 185 | + |
| 186 | + defp cancel_one(session_id) do |
| 187 | + _ = Sessions.cancel_turn(session_id) |
| 188 | + :ok |
| 189 | + rescue |
| 190 | + _ -> :ok |
| 191 | + catch |
| 192 | + :exit, _ -> :ok |
| 193 | + end |
| 194 | + |
| 195 | + # `get_session/1` casts its argument, so a value that is not a UUID reaches the repository as a |
| 196 | + # malformed one and raises rather than returning nil. Slice 070 hit exactly this from a chat |
| 197 | + # message and it crashed the router; a delegation brief is no more trustworthy than a chat |
| 198 | + # message, so the same guard applies here rather than being learned twice. |
| 199 | + defp create_child(parent_session_id, brief, opts) do |
| 200 | + parent = if uuid?(parent_session_id), do: Sessions.get_session(parent_session_id) |
| 201 | + |
| 202 | + if parent do |
| 203 | + Sessions.create_session(%{ |
| 204 | + persona_id: Keyword.get(opts, :persona_id) || parent.persona_id, |
| 205 | + origin: "subagent", |
| 206 | + parent_id: parent_session_id, |
| 207 | + title: Keyword.get(opts, :title) || title_from(brief), |
| 208 | + origin_ref: %{"parent_session_id" => parent_session_id} |
| 209 | + }) |
| 210 | + else |
| 211 | + {:error, {:no_such_session, parent_session_id}} |
| 212 | + end |
| 213 | + end |
| 214 | + |
| 215 | + defp uuid?(value) when is_binary(value), |
| 216 | + do: |
| 217 | + Regex.match?( |
| 218 | + ~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i, |
| 219 | + value |
| 220 | + ) |
| 221 | + |
| 222 | + defp uuid?(_), do: false |
| 223 | + |
| 224 | + defp title_from(brief) do |
| 225 | + brief |> String.split("\n") |> List.first() |> String.slice(0, 60) |
| 226 | + end |
| 227 | + |
| 228 | + # The awaiting shape is Trinity.Scheduler.Workers.RunTask's (slice 050), including the drain: a |
| 229 | + # session's process broadcasts :idle once when it starts, and a wait that does not drain that one |
| 230 | + # returns before the turn has run. Reused rather than rewritten, because writing it again is |
| 231 | + # writing that bug again. |
| 232 | + defp run(child, brief, budget, restarts) do |
| 233 | + with :ok <- Sessions.subscribe(child.id), |
| 234 | + {:ok, pid} <- Sessions.ensure_started(child.id), |
| 235 | + :ok <- drain_start(child.id), |
| 236 | + {:ok, _message} <- Sessions.send_user_message(child.id, brief) do |
| 237 | + # The session process is monitored as well as subscribed: a process that dies mid-turn |
| 238 | + # broadcasts nothing, so without this the parent waits out the whole budget for an answer |
| 239 | + # that is never coming. |
| 240 | + ref = Process.monitor(pid) |
| 241 | + result = await(child.id, budget, ref) |
| 242 | + Process.demonitor(ref, [:flush]) |
| 243 | + maybe_restart(result, child, brief, budget, restarts) |
| 244 | + else |
| 245 | + {:error, reason} -> %{session_id: child.id, status: :error, text: "", reason: reason} |
| 246 | + end |
| 247 | + end |
| 248 | + |
| 249 | + # Only a death is retried, and only when the caller asked for it. A budget overrun is not a |
| 250 | + # transient failure: running the same brief again under the same budget overruns it again. |
| 251 | + defp maybe_restart(%{status: :error, reason: {:child_died, _}}, child, brief, budget, n) |
| 252 | + when n > 0 do |
| 253 | + run(child, brief, budget, n - 1) |
| 254 | + end |
| 255 | + |
| 256 | + defp maybe_restart(result, _child, _brief, _budget, _n), do: result |
| 257 | + |
| 258 | + defp drain_start(session_id) do |
| 259 | + receive do |
| 260 | + {:session, ^session_id, {:state, :idle}} -> :ok |
| 261 | + after |
| 262 | + 2_000 -> :ok |
| 263 | + end |
| 264 | + end |
| 265 | + |
| 266 | + defp await(session_id, budget, ref) do |
| 267 | + receive do |
| 268 | + {:session, ^session_id, {:state, :idle}} -> |
| 269 | + %{session_id: session_id, status: :ok, text: summary(session_id), reason: nil} |
| 270 | + |
| 271 | + {:session, ^session_id, {:state, :error}} -> |
| 272 | + %{session_id: session_id, status: :error, text: "", reason: :session_error} |
| 273 | + |
| 274 | + {:session, ^session_id, {:error, reason}} -> |
| 275 | + %{session_id: session_id, status: :error, text: "", reason: {:turn, reason}} |
| 276 | + |
| 277 | + {:DOWN, ^ref, :process, _pid, reason} -> |
| 278 | + %{session_id: session_id, status: :error, text: "", reason: {:child_died, reason}} |
| 279 | + |
| 280 | + {:session, ^session_id, _other} -> |
| 281 | + await(session_id, budget, ref) |
| 282 | + after |
| 283 | + budget.timeout_ms -> |
| 284 | + _ = Sessions.cancel_turn(session_id) |
| 285 | + |
| 286 | + %{ |
| 287 | + session_id: session_id, |
| 288 | + status: :budget, |
| 289 | + text: summary(session_id), |
| 290 | + reason: {:timeout_ms, budget.timeout_ms} |
| 291 | + } |
| 292 | + end |
| 293 | + end |
| 294 | + |
| 295 | + defp summary(session_id) do |
| 296 | + session_id |
| 297 | + |> Sessions.history() |
| 298 | + |> Enum.filter(&(&1.role == "assistant")) |
| 299 | + |> List.last() |
| 300 | + |> case do |
| 301 | + nil -> "" |
| 302 | + %{content: c} when is_binary(c) -> binary_part(c, 0, min(byte_size(c), @summary_bytes)) |
| 303 | + _ -> "" |
| 304 | + end |
| 305 | + end |
| 306 | +end |
0 commit comments