Class: Mistri::SubAgent

Inherits:
Object
  • Object
show all
Defined in:
lib/mistri/sub_agent.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])

Children never receive a spawn tool: delegation is one level deep by construction. Parallel fan-out costs nothing: several spawn calls in one turn run concurrently on the executor pool.

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).

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."

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.



50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/mistri/sub_agent.rb', line 50

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

Instance Attribute Details

#description ⇒ Object (readonly)

Returns the value of attribute description.



45
46
47
 
# File 'lib/mistri/sub_agent.rb', line 45

def description
  @description
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



45
46
47
 
# File 'lib/mistri/sub_agent.rb', line 45

def name
  @name
end

Class Method Details

.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.



191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/mistri/sub_agent.rb', line 191

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
                         ))
  end
end

.forbid_gated!(tools) ⇒ Object

Raises:



217
218
219
220
221
222
223
224
 
# File 'lib/mistri/sub_agent.rb', line 217

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

.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.



100
101
102
 
# File 'lib/mistri/sub_agent.rb', line 100

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

.run_child(label:, provider:, system:, tools:, task:, context:, schema: nil, **agent_options) ⇒ Object 



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/mistri/sub_agent.rb', line 113

def run_child(label:, provider:, system:, tools:, task:, context:, schema: nil,
              **agent_options)
  store = context.session ? context.session.store : Stores::Memory.new
  child = Session.new(store: store)
  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 = context.signal ? context.signal.derive : [AbortSignal.new, nil]
  result = begin
    execute_child(child: child, label: label, provider: provider, system: system,
                  tools: tools, task: task, schema: schema, signal: signal,
                  emit: context.emit, **agent_options)
  ensure
    context.signal&.remove_callback(cascade) if cascade
  end
  outcome = answer(result, label, child)
  child.append(Child::TERMINAL, terminal(result))
  outcome
end

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

The host job's way back in: reconstruct provider, system, and tools from the spec through host registries, then hand them here. Reopens the child session the spawn created, runs it exactly like an inline child (started entry, lease, stop watching, terminals), streams origin-tagged events to the emit the job supplies, and reports the outcome back to the parent (see report_back). A background child runs on its own signal: the parent's turn is long over, so only stop_agent and the stop flag end it early.

The child's lease is the exactly-once fence, so it is taken before anything else: refused means another process is running this child right now (a queue redelivered a live job), so leave its owner alone. Holding it, a terminal decides: present means the child was cancelled while queued or the queue retried a finished job, so there is nothing to run; absent means run, and that includes the child a crashed process left mid-run, which is exactly what queue retries are for. Either kind of no-op returns nil.



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/mistri/sub_agent.rb', line 151

def run_dispatched(spec, provider:, system:, tools:, store:, emit: nil, schema: nil,
                   **agent_options)
  child = Session.new(store: store, id: spec.fetch("session_id"))
  signal = AbortSignal.new
  lease = Locks.hold(Child.lease_key(child.id),
                     stop_key: Child.stop_key(child.id), signal: signal)
  return nil if Mistri.locks && lease.nil?

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

  result = execute_child(child: child, label: spec.fetch("name"), provider: provider,
                         system: system, tools: tools, task: spec.fetch("task"),
                         schema: schema, signal: signal, emit: emit, lease: lease,
                         **agent_options)
  deny_pending(result, child)
  child.append(Child::TERMINAL, terminal(result))
  result
rescue StandardError => 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.
  if child && !Child.new(name: spec.fetch("name"), session_id: child.id,
                         store: store).finished?
    child.append(Child::TERMINAL,
                 "status" => "failed", "error" => "#{e.class}: #{e.message}")
  end
  lease&.release
  raise
ensure
  report_back(spec, store, emit) if child
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.



107
108
109
110
111
 
# File 'lib/mistri/sub_agent.rb', line 107

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.



94
95
96
 
# File 'lib/mistri/sub_agent.rb', line 94

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.



205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/mistri/sub_agent.rb', line 205

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 



83
84
85
86
87
88
 
# File 'lib/mistri/sub_agent.rb', line 83

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, context: context, schema: @schema,
                     **@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.



68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/mistri/sub_agent.rb', line 68

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