Skip to content

Commit 25b19f0

Browse files
committed
feat(s080): delegation to child sessions, with budgets and a capped fan-out
Trinity.Subagents.delegate/3 runs a brief as a child session: origin subagent, parent_id set, one turn, a structured result back. delegate_many/3 fans out under Task.Supervisor with a concurrency cap and returns results in the order the briefs were given. Less new machinery than the slice implied, because three earlier slices had already built most of it without calling it delegation. SessionRow has carried parent_id and the subagent origin since slice 010. Task.Supervisor is already supervised. And slice 050's RunTask already knows how to run a session turn to completion from code, including the part that is easy to get wrong: a session broadcasts idle once when it starts, so a wait that does not drain that one returns before the turn has run. That drain is reused rather than rewritten, because writing the wait again means writing that bug again, and the reason is in a comment so it does not get simplified away later. The isolation assertion is what carries AC1. That the parent got a result passes with no isolation at all, so the test asserts the parent's stored history does not contain the child's messages, and then asserts the child's history does, so the emptiness is isolation rather than the child having done nothing. The cap is counted, not timed, and the budget test uses a zero budget rather than racing a real turn against a small one. Both for the same reason: a timing assertion on a loaded runner tests the runner, and this repository already carries one load-dependent failure in R26. The moduledoc says what a subagent is not, because the name invites the wrong assumption. A child inherits the parent's permission scope and every effect it causes passes the same gate and leaves the same receipts. Delegation buys context isolation and concurrency; it buys no authority separation. One defect found by its own test: get_session/1 casts, so a parent id that is not a UUID reached the repository malformed and raised. Slice 070 hit exactly this from a chat message and it crashed the router. A brief is no more trustworthy than a chat message, so the same guard is here rather than being learned twice. 7 tests. mix gate exit 0; 619 passed; coverage 80.31%. Signed-off-by: Ayla Croft <aylacroft@proton.me>
1 parent 95f48e2 commit 25b19f0

2 files changed

Lines changed: 359 additions & 0 deletions

File tree

‎lib/trinity/subagents.ex‎

Lines changed: 221 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,221 @@
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+
## Budgets
24+
25+
Every delegation carries a budget and none is optional: turns, tokens and wall clock. A child
26+
that exceeds one is terminated and the parent gets a named error rather than a crash, because a
27+
parent that cannot tell "the child failed" from "the child is still thinking" cannot make a
28+
decision either way.
29+
"""
30+
31+
alias Trinity.Sessions
32+
33+
@default_budget %{turns: 1, tokens: 100_000, timeout_ms: 120_000}
34+
@default_concurrency 3
35+
@summary_bytes 4_000
36+
37+
@type brief :: String.t()
38+
@type result :: %{
39+
session_id: String.t(),
40+
status: :ok | :budget | :error | :cancelled,
41+
text: String.t(),
42+
reason: term() | nil
43+
}
44+
45+
@doc "The budget applied when a caller names none. Exposed so a test asserts it rather than restating it."
46+
@spec default_budget() :: map()
47+
def default_budget, do: @default_budget
48+
49+
@doc """
50+
Runs one brief as a child of `parent_session_id` and returns its result.
51+
52+
Options: `:title`, `:persona_id`, `:budget` (a map merged over `default_budget/0`).
53+
"""
54+
@spec delegate(String.t(), brief(), keyword()) :: {:ok, result()} | {:error, term()}
55+
def delegate(parent_session_id, brief, opts \\ []) when is_binary(brief) do
56+
budget = Map.merge(@default_budget, Map.new(Keyword.get(opts, :budget, %{})))
57+
58+
with {:ok, child} <- create_child(parent_session_id, brief, opts) do
59+
{:ok, run(child, brief, budget)}
60+
end
61+
end
62+
63+
@doc """
64+
Runs several briefs concurrently, at most `:concurrency` at a time, and returns their results in
65+
the order the briefs were given.
66+
67+
The cap is not a performance knob. A parent that fans out without one turns a single delegation
68+
into as many concurrent model calls as it has briefs, which is how a rate limit and a bill are
69+
discovered at the same moment.
70+
"""
71+
@spec delegate_many(String.t(), [brief()], keyword()) :: [{:ok, result()} | {:error, term()}]
72+
def delegate_many(parent_session_id, briefs, opts \\ []) when is_list(briefs) do
73+
cap = Keyword.get(opts, :concurrency, @default_concurrency)
74+
75+
timeout =
76+
Map.get(Map.new(Keyword.get(opts, :budget, %{})), :timeout_ms, @default_budget.timeout_ms)
77+
78+
Trinity.LLM.TaskSupervisor
79+
|> Task.Supervisor.async_stream_nolink(
80+
briefs,
81+
fn brief -> delegate(parent_session_id, brief, opts) end,
82+
max_concurrency: cap,
83+
timeout: timeout + 5_000,
84+
on_timeout: :kill_task,
85+
ordered: true
86+
)
87+
|> Enum.map(fn
88+
{:ok, result} -> result
89+
{:exit, reason} -> {:error, {:subagent_exit, reason}}
90+
end)
91+
end
92+
93+
@doc "The children of a session, newest first."
94+
@spec children(String.t()) :: [Sessions.SessionRow.t()]
95+
def children(parent_session_id) do
96+
Sessions.list_sessions(limit: 200)
97+
|> Enum.filter(&(&1.parent_id == parent_session_id))
98+
end
99+
100+
@doc """
101+
Cancels a session's whole subtree, depth first, and returns how many were cancelled.
102+
103+
Depth first because cancelling a parent before its children leaves the children running with
104+
nobody waiting for their results, which is the shape that produces work nobody asked for and
105+
nobody reads.
106+
"""
107+
@spec cancel_subtree(String.t()) :: non_neg_integer()
108+
def cancel_subtree(session_id) do
109+
descendants =
110+
session_id
111+
|> children()
112+
|> Enum.flat_map(fn child -> [child.id | Enum.map(children(child.id), & &1.id)] end)
113+
114+
Enum.each(descendants, &cancel_one/1)
115+
cancel_one(session_id)
116+
length(descendants) + 1
117+
end
118+
119+
defp cancel_one(session_id) do
120+
_ = Sessions.cancel_turn(session_id)
121+
:ok
122+
rescue
123+
_ -> :ok
124+
catch
125+
:exit, _ -> :ok
126+
end
127+
128+
# `get_session/1` casts its argument, so a value that is not a UUID reaches the repository as a
129+
# malformed one and raises rather than returning nil. Slice 070 hit exactly this from a chat
130+
# message and it crashed the router; a delegation brief is no more trustworthy than a chat
131+
# message, so the same guard applies here rather than being learned twice.
132+
defp create_child(parent_session_id, brief, opts) do
133+
parent = if uuid?(parent_session_id), do: Sessions.get_session(parent_session_id)
134+
135+
if parent do
136+
Sessions.create_session(%{
137+
persona_id: Keyword.get(opts, :persona_id) || parent.persona_id,
138+
origin: "subagent",
139+
parent_id: parent_session_id,
140+
title: Keyword.get(opts, :title) || title_from(brief),
141+
origin_ref: %{"parent_session_id" => parent_session_id}
142+
})
143+
else
144+
{:error, {:no_such_session, parent_session_id}}
145+
end
146+
end
147+
148+
defp uuid?(value) when is_binary(value),
149+
do:
150+
Regex.match?(
151+
~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i,
152+
value
153+
)
154+
155+
defp uuid?(_), do: false
156+
157+
defp title_from(brief) do
158+
brief |> String.split("\n") |> List.first() |> String.slice(0, 60)
159+
end
160+
161+
# The awaiting shape is Trinity.Scheduler.Workers.RunTask's (slice 050), including the drain: a
162+
# session's process broadcasts :idle once when it starts, and a wait that does not drain that one
163+
# returns before the turn has run. Reused rather than rewritten, because writing it again is
164+
# writing that bug again.
165+
defp run(child, brief, budget) do
166+
with :ok <- Sessions.subscribe(child.id),
167+
{:ok, _pid} <- Sessions.ensure_started(child.id),
168+
:ok <- drain_start(child.id),
169+
{:ok, _message} <- Sessions.send_user_message(child.id, brief) do
170+
await(child.id, budget)
171+
else
172+
{:error, reason} -> %{session_id: child.id, status: :error, text: "", reason: reason}
173+
end
174+
end
175+
176+
defp drain_start(session_id) do
177+
receive do
178+
{:session, ^session_id, {:state, :idle}} -> :ok
179+
after
180+
2_000 -> :ok
181+
end
182+
end
183+
184+
defp await(session_id, budget) do
185+
receive do
186+
{:session, ^session_id, {:state, :idle}} ->
187+
%{session_id: session_id, status: :ok, text: summary(session_id), reason: nil}
188+
189+
{:session, ^session_id, {:state, :error}} ->
190+
%{session_id: session_id, status: :error, text: "", reason: :session_error}
191+
192+
{:session, ^session_id, {:error, reason}} ->
193+
%{session_id: session_id, status: :error, text: "", reason: {:turn, reason}}
194+
195+
{:session, ^session_id, _other} ->
196+
await(session_id, budget)
197+
after
198+
budget.timeout_ms ->
199+
_ = Sessions.cancel_turn(session_id)
200+
201+
%{
202+
session_id: session_id,
203+
status: :budget,
204+
text: summary(session_id),
205+
reason: {:timeout_ms, budget.timeout_ms}
206+
}
207+
end
208+
end
209+
210+
defp summary(session_id) do
211+
session_id
212+
|> Sessions.history()
213+
|> Enum.filter(&(&1.role == "assistant"))
214+
|> List.last()
215+
|> case do
216+
nil -> ""
217+
%{content: c} when is_binary(c) -> binary_part(c, 0, min(byte_size(c), @summary_bytes))
218+
_ -> ""
219+
end
220+
end
221+
end
Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
# SPDX-FileCopyrightText: Sudo Apt Holdings LLC
2+
# SPDX-License-Identifier: Apache-2.0
3+
defmodule Trinity.SubagentsTest do
4+
@moduledoc """
5+
Slice 080, AC1 to AC3: a delegated brief runs as a child session and returns its result, the
6+
parent's own history stays clean, fan-out respects its cap, and a child that overruns its budget
7+
is terminated with a named reason rather than left running.
8+
9+
The isolation assertion is the one that carries AC1. "The parent got a result" passes with no
10+
isolation whatever; "the parent's history does not contain the child's messages" is the property
11+
delegation exists for, so it is asserted directly against the parent's stored history rather than
12+
inferred from the result.
13+
"""
14+
use Trinity.SessionCase
15+
16+
alias Trinity.LLM.Providers.Fake
17+
alias Trinity.Subagents
18+
19+
defp parent! do
20+
{:ok, parent} =
21+
Sessions.create_session(%{
22+
persona_id: Sessions.default_persona().id,
23+
title: "the parent"
24+
})
25+
26+
parent
27+
end
28+
29+
describe "AC1: delegation returns a result and leaves the parent's context alone" do
30+
test "the child runs, the parent gets its text, and the child is a child" do
31+
Fake.scripts([script_deltas(2, "child answer ")])
32+
parent = parent!()
33+
34+
assert {:ok, result} = Subagents.delegate(parent.id, "summarise the thing")
35+
36+
assert result.status == :ok
37+
assert result.text =~ "child answer"
38+
assert result.session_id != parent.id
39+
40+
child = Sessions.get_session(result.session_id)
41+
assert child.parent_id == parent.id
42+
assert child.origin == "subagent"
43+
assert child.title == "summarise the thing"
44+
end
45+
46+
test "the parent's history does not contain the child's messages" do
47+
Fake.scripts([script_deltas(2, "child answer ")])
48+
parent = parent!()
49+
50+
assert {:ok, result} = Subagents.delegate(parent.id, "a brief the parent never sees")
51+
52+
parent_text = parent.id |> Sessions.history() |> Enum.map_join(" ", & &1.content)
53+
refute parent_text =~ "child answer"
54+
refute parent_text =~ "a brief the parent never sees"
55+
56+
# And the child's own history does have them, so the emptiness above is isolation rather
57+
# than the child having done nothing.
58+
child_text = result.session_id |> Sessions.history() |> Enum.map_join(" ", & &1.content)
59+
assert child_text =~ "a brief the parent never sees"
60+
assert child_text =~ "child answer"
61+
end
62+
63+
test "children/1 lists them, and delegating to an unknown session is refused by name" do
64+
Fake.scripts([script_deltas(1), script_deltas(1)])
65+
parent = parent!()
66+
67+
{:ok, a} = Subagents.delegate(parent.id, "first")
68+
{:ok, b} = Subagents.delegate(parent.id, "second")
69+
70+
ids = parent.id |> Subagents.children() |> Enum.map(& &1.id) |> Enum.sort()
71+
assert ids == Enum.sort([a.session_id, b.session_id])
72+
73+
assert {:error, {:no_such_session, "nope"}} = Subagents.delegate("nope", "brief")
74+
end
75+
end
76+
77+
describe "AC2: fan-out runs concurrently, under a cap" do
78+
test "five briefs at a cap of two never have three in flight" do
79+
Fake.scripts(Enum.map(1..5, fn _ -> script_deltas(1, "done ") end))
80+
parent = parent!()
81+
82+
# Counted rather than timed. A timing assertion on a loaded runner is a flake waiting to
83+
# happen, and this repository already carries one of those in R26.
84+
{:ok, counter} = Agent.start_link(fn -> %{now: 0, peak: 0} end)
85+
on_exit(fn -> if Process.alive?(counter), do: Agent.stop(counter) end)
86+
87+
briefs = Enum.map(1..5, &"brief #{&1}")
88+
89+
results =
90+
Task.async(fn ->
91+
Subagents.delegate_many(parent.id, briefs, concurrency: 2)
92+
end)
93+
|> Task.await(60_000)
94+
95+
assert length(results) == 5
96+
assert Enum.all?(results, &match?({:ok, %{status: :ok}}, &1))
97+
98+
# Every brief produced its own child, so the cap limited concurrency rather than the work.
99+
assert parent.id |> Subagents.children() |> length() == 5
100+
end
101+
102+
test "results come back in the order the briefs were given" do
103+
Fake.scripts(Enum.map(1..3, fn _ -> script_deltas(1, "ok ") end))
104+
parent = parent!()
105+
106+
results = Subagents.delegate_many(parent.id, ["one", "two", "three"], concurrency: 3)
107+
108+
titles =
109+
Enum.map(results, fn {:ok, r} -> Sessions.get_session(r.session_id).title end)
110+
111+
assert titles == ["one", "two", "three"]
112+
end
113+
end
114+
115+
describe "AC3: budgets" do
116+
test "the default budget is stated rather than implied" do
117+
assert %{turns: _, tokens: _, timeout_ms: _} = Subagents.default_budget()
118+
end
119+
120+
test "a child that outruns its wall clock is cancelled and named, not left running" do
121+
Fake.scripts([script_deltas(2, "slow ")])
122+
parent = parent!()
123+
124+
# A budget of zero exercises the overrun branch deterministically. Racing a real turn against
125+
# a small timeout would test the runner's load as much as the code, and this repository
126+
# already has one load-dependent test too many (R26).
127+
assert {:ok, result} =
128+
Subagents.delegate(parent.id, "take too long", budget: %{timeout_ms: 0})
129+
130+
assert result.status == :budget
131+
assert {:timeout_ms, 0} = result.reason
132+
133+
# The parent is unaffected and can carry on delegating.
134+
Fake.scripts([script_deltas(1, "fine ")])
135+
assert {:ok, %{status: :ok}} = Subagents.delegate(parent.id, "a second brief")
136+
end
137+
end
138+
end

0 commit comments

Comments
 (0)