Class: Mistri::SubAgent

Inherits:
Object
  • Object
show all
Defined in:
lib/mistri/sub_agent.rb,
lib/mistri/sub_agent/runtime.rb,
lib/mistri/sub_agent/execution.rb

Overview

Delegation with a clean context: a child agent runs on its own session (the caller's store, linked in the caller's transcript), and only its final answer returns to the parent; exploration never fills the parent's window. Compaction rescues a full context after the fact; spawning avoids filling it in the first place. A child session is its own single-provider session, so delegating to a cheaper model is the sanctioned way to mix models.

Two shapes on one mechanism. A named specialist the host curates:

researcher = Mistri::SubAgent.new(
name: "researcher", description: "Answers factual questions.",
provider: Mistri.provider("claude-haiku-4-5-20251001"),
system: "Research. Report findings only.", tools: [fetch_page],
)
agent = Mistri::Agent.new(provider:, tools: [researcher.tool])

and the open spawn tool, where the model composes each worker: names it, writes its instructions, picks a tool subset, and may pick a model:

spawn = Mistri::SubAgent.spawner(provider:, tools: [fetch_page, search])

The open spawner never grants itself, preventing accidental recursion. Hosts may deliberately nest fixed specialists. Several spawn calls in one turn are scheduled concurrently; provider instances decide whether their network requests overlap.

Child events forward into the parent's stream tagged with origin ("researcher#ab12cd34"; nesting of named specialists joins with ">"). Approval-gated tools cannot ride inside a child: a child answers its parent synchronously and cannot fire-and-forget suspend, so statically gated tools are refused at construction and a child that suspends at runtime is denied and reported in band. Gate the delegation itself instead (needs_approval: on the definition or spawner).

Defined Under Namespace

Classes: Runtime

Constant Summary collapse

SPAWNER_DESCRIPTION = 
"Delegate a self-contained task to a focused child agent with a clean " \
"context. The child starts blank: give it complete instructions and " \
"every fact it needs. Use it to keep exploration out of your own " \
"context, and spawn several in one turn to fan out independent " \
"angles. Only the child's final answer comes back."
DISPATCH_SPEC_VERSION = 
1

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name:, description:, provider:, system: nil, tools: [], schema: nil, **agent_options) ⇒ SubAgent

schema: makes the specialist answer in validated JSON (task mode underneath), so fan-out children return a uniform shape the parent synthesizes instead of five styles of prose.



53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/mistri/sub_agent.rb', line 53

def initialize(name:, description:, provider:, system: nil, tools: [], schema: nil,
               **agent_options)
  SubAgent.forbid_gated!(tools)
  @gate = agent_options.delete(:needs_approval) || false
  ChildAgentOptions.validate!(agent_options)
  @name = name.to_s
  @description = description
  @provider = provider
  @system = system
  @tools = tools
  @schema = schema
  @agent_options = agent_options
end

Instance Attribute Details

#description ⇒ Object (readonly)

Returns the value of attribute description.



48
49
50
 
# File 'lib/mistri/sub_agent.rb', line 48

def description
  @description
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



48
49
50
 
# File 'lib/mistri/sub_agent.rb', line 48

def name
  @name
end

Class Method Details

.abort_before_child_start!(child, signal, lease) ⇒ Object 



52
53
54
55
56
57
 
# File 'lib/mistri/sub_agent/execution.rb', line 52

def abort_before_child_start!(child, signal, lease)
  return unless Mistri.locks
  return if lease && !Mistri.locks.flag?(Child.stop_key(child.id))

  signal.abort!("stopped by user")
end

.deny_pending(result, child) ⇒ Object

A child cannot wait for a human, whichever door it entered by: any calls parked for approval are denied AND settled with the denial as their tool result, so no approval request stays open on a finished child and its transcript replays without repair.



257
258
259
260
261
262
263
264
265
266
267
 
# File 'lib/mistri/sub_agent.rb', line 257

def deny_pending(result, child)
  return unless result.awaiting_approval?

  result.pending.each do |call|
    child.deny(call.id, note: "sub-agents cannot pause for human approval")
    child.append_message(Message.tool(
                           content: "Denied: sub-agents cannot pause for human approval.",
                           tool_call_id: call.id, tool_name: call.name, tool_error: true
                         ))
  end
end

.execute_child(child:, label:, provider:, system:, tools:, task:, schema:, signal:, emit:, lease: nil, started: false, agent_options: {}) ⇒ Object

The hardened execution path every child uses. Inline work acquires and cleans up its lease here; a dispatched owner supplies its lease and retains it through terminal persistence and parent reporting.



9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/mistri/sub_agent/execution.rb', line 9

def execute_child(child:, label:, provider:, system:, tools:, task:, schema:,
                  signal:, emit:, lease: nil, started: false, agent_options: {})
  ChildAgentOptions.validate!(agent_options)
  child.append(Child::STARTED, {}) unless started
  owns_lease = lease.nil?
  primary_error = nil
  begin
    lease ||= Locks.hold(Child.lease_key(child.id),
                         stop_key: Child.stop_key(child.id), signal: signal)
    abort_before_child_start!(child, signal, lease)
    if signal.aborted?
      Result.new(message: nil, status: :aborted)
    else
      run_child_agent(child:, label:, provider:, system:, tools:, task:, schema:,
                      signal:, emit:, agent_options:)
    end
  rescue StandardError => e
    primary_error = e
    reported = EventDelivery.original(e)
    child.append(Child::TERMINAL, "status" => "failed",
                                  "error" => "#{reported.class}: #{reported.message}")
    raise
  ensure
    cleanup_error = release_inline_lease(child, lease) if owns_lease
    raise cleanup_error if cleanup_error && primary_error.nil?
  end
end

.forbid_gated!(tools) ⇒ Object

Raises:



283
284
285
286
287
288
289
290
 
# File 'lib/mistri/sub_agent.rb', line 283

def forbid_gated!(tools)
  gated = tools.select { |tool| statically_gated?(tool) }
  return if gated.empty?

  raise ConfigurationError,
        "approval-gated tools cannot run inside a sub-agent " \
        "(#{gated.map(&:name).join(", ")}); gate the delegation instead"
end

.forward(event, origin, emit) ⇒ Object 



70
71
72
73
74
75
 
# File 'lib/mistri/sub_agent/execution.rb', line 70

def forward(event, origin, emit)
  return unless emit

  tagged = event.origin ? "#{origin}>#{event.origin}" : origin
  emit.call(event.with(origin: tagged))
end

.pack(provider:, console: {}, **spawner_options) ⇒ Object

The whole kit in one call: the spawn tool plus the management console, so a host hands its agent everything workers need.



104
105
106
 
# File 'lib/mistri/sub_agent.rb', line 104

def pack(provider:, console: {}, **spawner_options)
  [spawner(provider: provider, **spawner_options), *Console.tools(**console)]
end

.release_inline_lease(child, lease) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/mistri/sub_agent/execution.rb', line 37

def release_inline_lease(child, lease)
  errors = []
  begin
    lease&.release
  rescue StandardError => e
    errors << e
  end
  begin
    Mistri.locks&.clear_flag(Child.stop_key(child.id))
  rescue StandardError => e
    errors << e
  end
  errors.first
end

.run_child(label:, provider:, system:, tools:, task:, parent_context:, schema: nil, agent_options: {}) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/mistri/sub_agent.rb', line 117

def run_child(label:, provider:, system:, tools:, task:, parent_context:, schema: nil,
              agent_options: {})
  ChildAgentOptions.validate!(agent_options)
  store = parent_context.session ? parent_context.session.store : Stores::Memory.new
  child = Session.new(store: store)
  parent_context.session&.append("subagent", "name" => label, "session_id" => child.id)
  # An inline child runs on a signal derived from the parent's: the
  # parent's abort cascades down through the handle, while stopping
  # the child alone leaves the parent running.
  signal, cascade = if parent_context.signal
                      parent_context.signal.derive
                    else
                      [AbortSignal.new, nil]
                    end
  result = begin
    execute_child(child: child, label: label, provider: provider, system: system,
                  tools: tools, task: task, schema: schema, signal: signal,
                  emit: parent_context.emit, agent_options: agent_options)
  rescue StandardError => e
    unless child.entries.any? { |entry| entry["type"] == Child::TERMINAL }
      child.append(Child::TERMINAL,
                   "status" => "failed", "error" => "#{e.class}: #{e.message}")
    end
    raise
  ensure
    parent_context.signal&.remove_callback(cascade) if cascade
  end
  outcome = answer(result, label, child)
  child.append(Child::TERMINAL, terminal(result))
  outcome
end

.run_child_agent(child:, label:, provider:, system:, tools:, task:, schema:, signal:, emit:, agent_options:) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
 
# File 'lib/mistri/sub_agent/execution.rb', line 59

def run_child_agent(child:, label:, provider:, system:, tools:, task:, schema:,
                    signal:, emit:, agent_options:)
  agent = Agent.new(provider: provider, session: child, system: system,
                    tools: tools, **agent_options)
  origin = "#{label}##{child.id[0, 8]}"
  tagged = ->(event) { forward(event, origin, emit) }
  return agent.task(task, schema: schema, signal: signal, &tagged) if schema

  agent.run(task, signal: signal, &tagged)
end

.run_dispatched(spec, store:, emit: nil, runtime_factory: nil, provider: UNSET_RUNTIME_FIELD, system: UNSET_RUNTIME_FIELD, tools: UNSET_RUNTIME_FIELD, schema: UNSET_RUNTIME_FIELD, retry_factory_errors: true, **agent_options) ⇒ Object

The host job's way back in: a runtime factory turns the durable spec into live dependencies inside the worker. The compatible direct provider/system/tools form remains for hosts that already reconstruct those values in their job. Either form is checked against the spec before execution. The child then runs exactly like an inline child, streams origin-tagged events, and reports back to its parent.

The child lease is duplicate suppression, not an exactly-once claim: a refused delivery leaves the current holder alone. A terminal means a queued cancellation or finished retry and returns nil. The runner retains an acquired lease through terminal persistence and reporting, suppressing ordinary redelivery while that lease remains live. rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity -- ordered dispatch transitions are the safety contract



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/mistri/sub_agent.rb', line 162

def run_dispatched(spec, store:, emit: nil, runtime_factory: nil,
                   provider: UNSET_RUNTIME_FIELD, system: UNSET_RUNTIME_FIELD,
                   tools: UNSET_RUNTIME_FIELD, schema: UNSET_RUNTIME_FIELD,
                   retry_factory_errors: true, **agent_options)
  runtime = nil
  resolved = nil
  authorized = false
  terminalize = false
  factory_failed = false
  retryable_exit = false
  primary_error = nil
  delivery_owned = false
  spec = RuntimeContract.own_spec(spec)
  RuntimeContract.validate_identity!(spec)
  child = Session.new(store: store, id: spec.fetch("session_id"))
  spec = RuntimeContract.bind_spec(child.entries, spec)
  authorized = true
  direct = { provider: provider, system: system, tools: tools, schema: schema,
             agent_options: agent_options.empty? ? UNSET_RUNTIME_FIELD : agent_options }
  RuntimeContract.validate_resolution!(factory: runtime_factory, direct: direct)
  signal = AbortSignal.new
  terminalize = true
  delivery_owned = true
  lease = Locks.hold(Child.lease_key(child.id),
                     stop_key: Child.stop_key(child.id), signal: signal)
  if Mistri.locks && lease.nil?
    delivery_owned = false
    return nil
  end

  if Child.new(name: spec.fetch("name"), session_id: child.id, store: store).finished?
    return nil
  end

  child.append(Child::STARTED, {})
  RuntimeContract.validate_spec!(spec)
  signal.abort!("stopped by user") if Mistri.locks&.flag?(Child.stop_key(child.id))
  unless signal.aborted?
    runtime = RuntimeContract.resolve(spec, factory: runtime_factory, direct: direct) do
      factory_failed = true
    end
    resolved = RuntimeContract.validate_runtime!(runtime, spec)
    signal.abort!("stopped by user") if Mistri.locks&.flag?(Child.stop_key(child.id))
  end
  result = if signal.aborted?
             Result.new(message: nil, status: :aborted)
           else
             execute_child(child: child, label: spec.fetch("name"),
                           provider: resolved.provider, system: resolved.system,
                           tools: resolved.tools, task: spec.fetch("task"),
                           schema: resolved.schema, signal: signal, emit: emit,
                           lease: lease, started: true,
                           agent_options: resolved.agent_options)
           end
  deny_pending(result, child)
  child.append(Child::TERMINAL, terminal(result))
  result
rescue StandardError => e
  primary_error = e
  # Completion is a contract even when the runner dies in the
  # preamble: without this, a raise before the started entry (a lock
  # backend down, say) would leave the child reading :queued forever,
  # with nothing to report and nothing for a retry to heal.
  stopped = factory_failed && child_stop_requested?(child, signal)
  retryable = factory_failed && retry_factory_errors && !stopped
  retryable_exit = retryable
  if child && terminalize && !retryable &&
     !Child.new(name: spec.fetch("name"), session_id: child.id, store: store).finished?
    terminal = if stopped
                 { "status" => "stopped" }
               else
                 { "status" => "failed", "error" => "#{e.class}: #{e.message}" }
               end
    child.append(Child::TERMINAL, terminal)
  end
  raise
ensure
  cleanup_error = RuntimeContract.cleanup(runtime)
  begin
    report_back(spec, store, emit) if child && authorized && delivery_owned
  ensure
    begin
      Mistri.locks&.clear_flag(Child.stop_key(child.id)) if child && lease && !retryable_exit
    ensure
      lease&.release
    end
  end
  raise cleanup_error if cleanup_error && primary_error.nil?
end

.sanitize_label(text, fallback:) ⇒ Object

A worker's display name, made safe for origins: the label rides them as "label#id" and nesting joins with ">", so those separators squeeze to hyphens along with whitespace. Blank falls back.



111
112
113
114
115
 
# File 'lib/mistri/sub_agent.rb', line 111

def sanitize_label(text, fallback:)
  label = text.to_s.gsub(/[#>\s]+/, "-").squeeze("-")[0, 32]
  label = label.delete_prefix("-").delete_suffix("-")
  label.empty? ? fallback : label
end

.spawner(provider:) ⇒ Object

The open spawn tool: the model names each worker, grants it a tool subset from the host's pool, and may pick a type, a model, and a mode. All policy lives on Spawner; this is the front door.



98
99
100
 
# File 'lib/mistri/sub_agent.rb', line 98

def spawner(provider:, **)
  Spawner.new(provider: provider, **).tool
end

.terminal(result) ⇒ Object

Every child ends by writing its own terminal entry: completion is a contract, and status stays readable from the store forever.



271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/mistri/sub_agent.rb', line 271

def terminal(result)
  case result.status
  when :completed then { "status" => "done", "report" => result.text.to_s }
  when :aborted then { "status" => "stopped" }
  when :awaiting_approval
    { "status" => "failed",
      "error" => "needed human approval, which sub-agents cannot wait for" }
  else
    { "status" => "failed", "error" => (result.error_message || result.status).to_s }
  end
end

Instance Method Details

#run_child(task, context, name: nil) ⇒ Object 



87
88
89
90
91
92
 
# File 'lib/mistri/sub_agent.rb', line 87

def run_child(task, context, name: nil)
  SubAgent.run_child(label: SubAgent.sanitize_label(name, fallback: @name),
                     provider: @provider, system: @system,
                     tools: @tools, task: task, parent_context: context, schema: @schema,
                     agent_options: @agent_options)
end

#tool ⇒ Object

The delegate tool: each call runs a fresh child and answers with its final text, plus session_id on the ui channel so a host can link the child's transcript. The model may name each run, so two parallel researchers read as "Corgi" and "Beagle" in lanes and lists instead of "researcher" twice.



72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/mistri/sub_agent.rb', line 72

def tool
  sub = self
  blurb = "#{@description} Runs as a focused sub-agent with a clean " \
          "context: give it complete instructions, it starts blank."
  Tool.define(@name, blurb, needs_approval: @gate,
                            schema: lambda {
                              string :task, "Complete instructions for the sub-agent",
                                     required: true
                              string :name, "A short name for this run, shown wherever " \
                                            "its events appear (default: the tool's name)"
                            }) do |args, context|
    sub.run_child(args.fetch("task"), context, name: args["name"])
  end
end