Class: LittleGhost::Subagents::Manager

Inherits:
Object
  • Object
show all
Defined in:
lib/little_ghost/subagents/manager.rb

Overview

Manager coordinates delegated conversations without making an application build its own worker pool or message protocol. It runs bounded concurrent tasks, queues follow-ups, reports progress, and can restore durable children.

Applications normally enable it through the agent DSL:

class CustomerSupportAgent < LittleGhost::Agent
subagent ResearchAgent,
  kind: "research",
  description: "Investigates policies and account history"
end

LittleGhost then gives CustomerSupportAgent tools to spawn, message, check on, interject, and list research agents. The manager keeps each child identity stable across follow-up turns. A progress check returns after 30 seconds by default when the selected subagents are still working; it does not pause or restart them.

Follow-up messages are FIFO turns and never interject active work. #interject is the separate synchronous path for delivery at the next model boundary; delivery does not stop the child, and tool calls from that model response continue in the child run.

Durability and cleanup

With a parent session, durable definitions retain only committed compact transcripts and limited state snapshots. Failed or cancelled turns never become committed conversation history. Call #close to cancel and join workers owned by a directly constructed manager. If cooperative fiber cleanup exceeds the deadline, the manager remains closed to new work and a later #close retries cleanup.

Defined Under Namespace

Classes: Capacity, CleanupError, Completion, Identity, InterjectionExchange, Turn

Constant Summary collapse

DEFAULT_MAX_CONCURRENT =

:nodoc:

8
DEFAULT_MAX_IDENTITIES =

:nodoc:

20
DEFAULT_MAX_TURNS =

:nodoc:

100
DEFAULT_MAX_QUEUED_TURNS_PER_IDENTITY =

:nodoc:

8
DEFAULT_MAX_MESSAGE_CHARS =

:nodoc:

50_000
DEFAULT_MAX_RESPONSE_CHARS =

:nodoc:

100_000
DEFAULT_WAIT_TIMEOUT =

:nodoc:

30.0
DEFAULT_CLOSE_TIMEOUT =

:nodoc:

5.0
DEFAULT_LIST_LIMIT =

:nodoc:

20
MAX_LIST_LIMIT =

:nodoc:

100
MAX_PROGRESS_CHARS =

:nodoc:

160
MAX_PROGRESS_SOURCE_CHARS =

:nodoc:

4_096
PROGRESS_SEPARATOR =

:nodoc:

/[\p{Z}\p{Cc}\p{Cf}]/
CANCELLATION_POLL_INTERVAL =

:nodoc:

0.05
REGISTRY_VERSION =

:nodoc:

2
CURSOR_MAX_BYTES =

:nodoc:

512
UUID_PATTERN =

:nodoc:

/\A[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\z/

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(definitions, runtime: nil, max_concurrent: DEFAULT_MAX_CONCURRENT, max_identities: DEFAULT_MAX_IDENTITIES, max_turns: DEFAULT_MAX_TURNS, max_queued_turns_per_identity: DEFAULT_MAX_QUEUED_TURNS_PER_IDENTITY, max_message_chars: DEFAULT_MAX_MESSAGE_CHARS, max_response_chars: DEFAULT_MAX_RESPONSE_CHARS, wait_timeout: DEFAULT_WAIT_TIMEOUT, close_timeout: DEFAULT_CLOSE_TIMEOUT, cancellation_token: Support::CancellationToken.new, deadline: nil, observer: nil, parent_session: nil, parent_agent_path: AgentPath::ROOT) ⇒ Manager

Configures a bounded manager. Durable restoration is enabled only when parent_session is supplied.



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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
 
# File 'lib/little_ghost/subagents/manager.rb', line 211

def initialize(
  definitions,
  runtime: nil,
  max_concurrent: DEFAULT_MAX_CONCURRENT,
  max_identities: DEFAULT_MAX_IDENTITIES,
  max_turns: DEFAULT_MAX_TURNS,
  max_queued_turns_per_identity: DEFAULT_MAX_QUEUED_TURNS_PER_IDENTITY,
  max_message_chars: DEFAULT_MAX_MESSAGE_CHARS,
  max_response_chars: DEFAULT_MAX_RESPONSE_CHARS,
  wait_timeout: DEFAULT_WAIT_TIMEOUT,
  close_timeout: DEFAULT_CLOSE_TIMEOUT,
  cancellation_token: Support::CancellationToken.new,
  deadline: nil,
  observer: nil,
  parent_session: nil,
  parent_agent_path: AgentPath::ROOT
)
  @runtime = runtime
  @framework_prompts = FrameworkPrompts.for_runtime(runtime)
  @prompt_renderer = ->(key, **locals) { @framework_prompts.render(key, locals:) }
  validate_limit(:max_concurrent, max_concurrent)
  validate_limit(:max_identities, max_identities)
  validate_limit(:max_turns, max_turns)
  validate_limit(:max_queued_turns_per_identity, max_queued_turns_per_identity)
  validate_limit(:max_message_chars, max_message_chars)
  validate_limit(:max_response_chars, max_response_chars)
  validate_timeout(:wait_timeout, wait_timeout)
  validate_timeout(:close_timeout, close_timeout)

  @definitions = definitions.each_with_object({}) do |definition, index|
    raise ArgumentError, "Duplicate subagent kind: #{definition.kind}" if index.key?(definition.kind)

    index[definition.kind] = definition
  end.freeze
  @max_identities = max_identities
  @max_turns = max_turns
  @max_queued_turns_per_identity = max_queued_turns_per_identity
  @max_message_chars = max_message_chars
  @max_response_chars = max_response_chars
  @wait_timeout = wait_timeout
  @close_timeout = close_timeout
  @cancellation_token = cancellation_token.child
  @deadline = deadline
  @observer = observer
  @observer_events = []
  @observer_flushing = false
  @task_runner = runtime ? runtime.task_runner : Support::TaskRunner.new
  @parent_session = parent_session
  @parent_agent_path = AgentPath.validate!(parent_agent_path)
  @parent_link = parent_session && self.class.parent_link(parent_session)
  @registry_session = parent_session && registry_session
  @capacity = Capacity.new(max_concurrent)
  @mutex = Mutex.new
  @close_monitor = Monitor.new
  @close_active = false
  @registry_mutex = Mutex.new
  @restore_mutex = Mutex.new
  @condition = ConditionVariable.new
  @identities = {}
  @reserved_agent_paths = {}
  @identity_slots = 0
  @turn_count = 0
  @closed = false
  @close_complete = false
  @closed_agents = {}
  restore_identities
end

Instance Attribute Details

#definitions ⇒ Object (readonly)

Available definitions, indexed by kind.



185
186
187
 
# File 'lib/little_ghost/subagents/manager.rb', line 185

def definitions
  @definitions
end

Class Method Details

.commit_session_id(conversation_id, slot) ⇒ Object

Derives one of the rotating committed-state session IDs.



204
205
206
 
# File 'lib/little_ghost/subagents/manager.rb', line 204

def commit_session_id(conversation_id, slot)
  "lg_subagent_commit_#{conversation_id}_#{slot}"
end

.conversation_session_id(conversation_id) ⇒ Object

Derives the framework-owned transcript session ID.



199
200
201
 
# File 'lib/little_ghost/subagents/manager.rb', line 199

def conversation_session_id(conversation_id)
  "lg_subagent_conversation_#{conversation_id}"
end

.parent_link(session) ⇒ Object

Produces a pseudonymous parent-session link for durable metadata.



189
190
191
 
# File 'lib/little_ghost/subagents/manager.rb', line 189

def parent_link(session)
  Digest::SHA256.hexdigest("#{session.actor_id}\0#{session.id}")
end

.registry_session_id(session) ⇒ Object

Derives the framework-owned registry session ID.



194
195
196
 
# File 'lib/little_ghost/subagents/manager.rb', line 194

def registry_session_id(session)
  "lg_subagent_registry_#{parent_link(session)}"
end

Instance Method Details

#bind_prompt_renderer(renderer) ⇒ Object

:nodoc:



279
280
281
 
# File 'lib/little_ghost/subagents/manager.rb', line 279

def bind_prompt_renderer(renderer) # :nodoc:
  @prompt_renderer = renderer
end

#close ⇒ Object

Cancels queued work, cooperatively stops workers, and closes child agents. Raises CleanupError if workers do not stop within the bound. A later call retries unfinished cleanup without accepting new work.



707
708
709
 
# File 'lib/little_ghost/subagents/manager.rb', line 707

def close
  synchronize_close { perform_close }
end

#interject(subagent_id:, message:, cancellation_token: @cancellation_token, deadline: @deadline) ⇒ Object

Delivers message to one currently running turn and waits for the next model response. The returned response_disposition says whether that response also initiated tool calls; it does not imply the subagent has stopped.



403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
 
# File 'lib/little_ghost/subagents/manager.rb', line 403

def interject(subagent_id:, message:, cancellation_token: @cancellation_token, deadline: @deadline)
  unless message.is_a?(String)
    raise ToolError, prompt("subagents/feedback/message_type")
  end
  if message.length > @max_message_chars
    raise ToolError, prompt("subagents/feedback/message_limit", limit: @max_message_chars)
  end

  exchange = InterjectionExchange.new(message:, complete: false)
  identity, turn = @mutex.synchronize do
    ensure_open!
    value = fetch_identity!(subagent_id)
    unless value.agent.respond_to?(:interject)
      raise ToolError, prompt("subagents/feedback/interjection_unsupported", id: subagent_id)
    end
    unless value.status == "running"
      raise ToolError, prompt("subagents/feedback/not_running", id: subagent_id)
    end
    if value.current.interjections.length >= @max_queued_turns_per_identity
      raise ToolError, prompt("subagents/feedback/interjection_limit", id: subagent_id)
    end
    interjection_chars = value.current.interjections.sum { |pending| pending.message.length }
    if interjection_chars + message.length > @max_message_chars
      raise ToolError, prompt("subagents/feedback/interjection_chars_limit", limit: @max_message_chars)
    end

    value.current.interjections << exchange
    [value, value.current]
  end

  result = begin
    identity.agent.interject(
      message,
      cancellation_token:,
      deadline:,
      target_operation_id: turn.operation_id
    )
  rescue
    @mutex.synchronize do
      turn.interjections.delete(exchange)
      @condition.broadcast
    end
    raise
  end
  response = result.text
  truncated = response.length > @max_response_chars
  returned_response = truncated ? response[0, @max_response_chars] : response
  @mutex.synchronize do
    used_response_chars = turn.interjections.sum do |pending|
      pending.equal?(exchange) ? 0 : pending.response.to_s.length
    end
    remaining_response_chars = [@max_response_chars - used_response_chars, 0].max
    exchange.response = returned_response[0, remaining_response_chars]
    exchange.complete = true
    @condition.broadcast
  end
  subagent = @mutex.synchronize do
    snapshot(identity, include_response: true, include_progress: true)
  end
  value = {
    status: "interjection_delivered",
    subagent_id: identity.subagent_id,
    kind: identity.definition.kind,
    subagent:,
    turn: turn.number,
    response: returned_response,
    response_disposition: result.tool_calls? ? "text_with_tool_calls" : "text_only"
  }
  value[:response_truncated] = true if truncated
  value
rescue AgentInterjectionError => error
  raise ToolError, error.message
end

#list(kind: nil, limit: DEFAULT_LIST_LIMIT, cursor: nil) ⇒ Object

Lists active and persisted identities newest-first without restoring inactive agents. Cursors are opaque and must be passed back unchanged.



510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
 
# File 'lib/little_ghost/subagents/manager.rb', line 510

def list(kind: nil, limit: DEFAULT_LIST_LIMIT, cursor: nil)
  cursor = nil if cursor == ""
  unless limit.is_a?(Integer) && limit.between?(1, MAX_LIST_LIMIT)
    raise ToolError, prompt("subagents/feedback/list_limit", maximum: MAX_LIST_LIMIT)
  end
  if kind && !definitions.key?(kind)
    raise ToolError, prompt("subagents/feedback/unknown_kind", kind:)
  end

  @mutex.synchronize do
    identities = @identities.values
    identities = identities.select { |identity| identity.definition.kind == kind } if kind
    identities = identities.sort_by { |identity| [identity.updated_at.to_s, identity.subagent_id] }.reverse
    if cursor
      boundary = decode_cursor(cursor)
      identities = identities.drop_while do |identity|
        ([identity.updated_at.to_s, identity.subagent_id] <=> boundary) >= 0
      end
    end
    page = identities.first(limit)
    value = {
      status: "ok",
      subagents: page.map { |identity| snapshot(identity, include_progress: true) }
    }
    value[:next_cursor] = encode_cursor(page.last) if identities.length > page.length
    value
  end
end

#send_message(subagent_id:, message:, mode:, parent_operation_id: nil, context: nil) ⇒ Object

Queues a FIFO follow-up for an active or durable identity.



376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
 
# File 'lib/little_ghost/subagents/manager.rb', line 376

def send_message(subagent_id:, message:, mode:, parent_operation_id: nil, context: nil)
  validate_mode(mode)
  identity = @mutex.synchronize do
    ensure_open!
    fetch_identity!(subagent_id)
  end
  restore_agent!(identity)
  queued = enqueue(
    identity,
    message,
    event: "message_queued",
    enforce_limits: true,
    parent_operation_id:,
    context:
  )
  return queued if queued.is_a?(Hash)

  turn, queued_snapshot = queued
  return {status: "working", subagent: queued_snapshot} if mode == "async"

  turn.completion.value(cancellation_token: @cancellation_token, deadline: @deadline)
end

#spawn(kind:, task_name:, task:, mode:, parent_operation_id: nil, context: nil) ⇒ Object

Creates a unique child identity and queues its first task.

mode is "sync" or "async". Synchronous mode waits for the turn; asynchronous mode returns a working snapshot for later #wait calls.



287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
 
# File 'lib/little_ghost/subagents/manager.rb', line 287

def spawn(kind:, task_name:, task:, mode:, parent_operation_id: nil, context: nil)
  validate_mode(mode)
  definition, subagent_id = reserve_identity(kind, task, task_name:)
  return subagent_id unless definition

  conversation_id = SecureRandom.uuid
  begin
    agent = build_agent(definition, subagent_id, conversation_id)
    raise TypeError, "factory result must respond to call" unless agent.respond_to?(:call)
  rescue LittleGhost::CleanupError
    release_identity_reservation(subagent_id)
    raise
  rescue => error
    release_identity_reservation(subagent_id)
    warn_failure("factory", subagent_id, error)
    emit_factory_failure(definition, subagent_id, error, parent_operation_id:)
    return {
      status: "failed",
      subagent_id: subagent_id,
      kind: definition.kind,
      error: prompt("subagents/errors/create_failed")
    }
  end

  identity = Identity.new(
    subagent_id: subagent_id,
    conversation_id: conversation_id,
    definition: definition,
    agent: agent,
    session: definition.persist && @parent_session && child_session(conversation_id),
    durable: definition.persist && !!@parent_session,
    resumed: false,
    updated_at: Time.now.utc.iso8601(6),
    committed_count: 0,
    commit_slot: 1,
    history: [].freeze,
    state: {},
    queue: [],
    status: "idle",
    next_turn: 1,
    latest_response_truncated: false,
    progress_sequence: 0
  )
  observe_delegated_activity(identity)

  closed = @mutex.synchronize do
    if @closed
      @reserved_agent_paths.delete(subagent_id)
      @identity_slots -= 1
      @turn_count -= 1
      next true
    end
    @reserved_agent_paths.delete(subagent_id)
    @identities[subagent_id] = identity
    false
  end
  if closed
    agent.close if agent.respond_to?(:close)
    raise Error, "Subagent manager is closed"
  end

  begin
    turn, queued_snapshot = enqueue(
      identity,
      task,
      event: "spawned",
      count_turn: false,
      parent_operation_id:,
      context:
    )
  rescue
    synchronize_close do
      should_close = @mutex.synchronize do
        if @identities.delete(subagent_id)
          @identity_slots -= 1
          @turn_count -= 1
        end
        !@closed_agents.key?(agent.object_id)
      end
      agent.close if should_close && agent.respond_to?(:close)
    end
    raise
  end
  return {status: "working", subagent: queued_snapshot} if mode == "async"

  turn.completion.value(cancellation_token: @cancellation_token, deadline: @deadline)
end

#tools ⇒ Object

Builds spawn, follow-up, interject, wait, and list tools bound to this manager. Closing the first tool closes the shared manager.



541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
 
# File 'lib/little_ghost/subagents/manager.rb', line 541

def tools
  manager = self
  kind_descriptions = definitions.values.map do |definition|
    "- #{definition.kind}: #{definition.description}"
  end.join("\n")
  tools = [
    ControlTool.define(
      name: "spawn_subagent",
      description: "",
      input_schema: {
        type: "object",
        properties: {
          kind: {
            type: "string",
            enum: definitions.keys,
            description: ""
          },
          task_name: {
            type: "string",
            pattern: "^[a-z0-9_]+$",
            maxLength: AgentPath::MAX_NAME_LENGTH,
            description: ""
          },
          task: {type: "string", description: ""},
          mode: {
            type: "string", enum: %w[sync async],
            description: ""
          }
        },
        required: %w[kind task_name task mode],
        additionalProperties: false
      }
    ) do |input, context: nil|
      manager.spawn(
        kind: input.fetch("kind"),
        task_name: input.fetch("task_name"),
        task: input.fetch("task"),
        mode: input.fetch("mode"),
        context:,
        parent_operation_id: context&.agent_operation_id
      )
    end.tap do |tool|
      tool.framework_prompts(
        description: FrameworkPrompts.reference("subagents/tools/spawn/description"), manager:,
        schema: {
          %w[properties kind description] => FrameworkPrompts.reference("subagents/tools/spawn/inputs/kind/description", kinds: kind_descriptions),
          %w[properties task_name description] => FrameworkPrompts.reference("subagents/tools/spawn/inputs/task_name/description"),
          %w[properties task description] => FrameworkPrompts.reference("subagents/tools/spawn/inputs/task/description"),
          %w[properties mode description] => FrameworkPrompts.reference("subagents/tools/spawn/inputs/spawn_mode/description")
        }
      )
    end,
    ControlTool.define(
      name: "send_message_to_subagent",
      description: "",
      input_schema: {
        type: "object",
        properties: {
          subagent_id: {type: "string", description: ""},
          message: {type: "string", description: ""},
          mode: {
            type: "string", enum: %w[sync async],
            description: ""
          }
        },
        required: %w[subagent_id message mode],
        additionalProperties: false
      }
    ) do |input, context: nil|
      manager.send_message(
        subagent_id: input.fetch("subagent_id"),
        message: input.fetch("message"),
        mode: input.fetch("mode"),
        context:,
        parent_operation_id: context&.agent_operation_id
      )
    end.tap do |tool|
      tool.framework_prompts(
        description: FrameworkPrompts.reference("subagents/tools/send/description"), manager:,
        schema: {
          %w[properties subagent_id description] => FrameworkPrompts.reference("subagents/tools/send/inputs/id/description"),
          %w[properties message description] => FrameworkPrompts.reference("subagents/tools/send/inputs/message/description"),
          %w[properties mode description] => FrameworkPrompts.reference("subagents/tools/send/inputs/send_mode/description")
        }
      )
    end,
    ControlTool.define(
      name: "interject_subagent",
      description: "",
      input_schema: {
        type: "object",
        properties: {
          subagent_id: {type: "string", description: ""},
          message: {type: "string", description: ""}
        },
        required: %w[subagent_id message],
        additionalProperties: false
      }
    ) do |input, context: nil|
      options = {}
      options[:cancellation_token] = context.cancellation_token if context
      options[:deadline] = context.deadline if context&.deadline
      manager.interject(
        subagent_id: input.fetch("subagent_id"),
        message: input.fetch("message"),
        **options
      )
    end.tap do |tool|
      tool.framework_prompts(
        description: FrameworkPrompts.reference("subagents/tools/interject/description"), manager:,
        schema: {
          %w[properties subagent_id description] => FrameworkPrompts.reference("subagents/tools/interject/inputs/active_id/description"),
          %w[properties message description] => FrameworkPrompts.reference("subagents/tools/interject/inputs/message/description")
        }
      )
    end,
    ControlTool.define(
      name: "wait_for_subagents",
      description: "",
      input_schema: {
        type: "object",
        properties: {
          subagent_ids: {
            type: "array", items: {type: "string"},
            description: ""
          }
        },
        additionalProperties: false
      }
    ) { |input| manager.wait(subagent_ids: input["subagent_ids"]) }.tap do |tool|
      tool.framework_prompts(
        description: FrameworkPrompts.reference("subagents/tools/wait/description"), manager:,
        schema: {
          %w[properties subagent_ids description] => FrameworkPrompts.reference("subagents/tools/wait/inputs/ids/description")
        }
      )
    end,
    ControlTool.define(
      name: "list_subagents",
      description: "",
      input_schema: {
        type: "object",
        properties: {
          kind: {type: "string", enum: definitions.keys},
          limit: {type: "integer", minimum: 1, maximum: MAX_LIST_LIMIT},
          cursor: {type: "string"}
        },
        additionalProperties: false
      }
    ) do |input|
      manager.list(
        kind: input["kind"],
        limit: input.fetch("limit", DEFAULT_LIST_LIMIT),
        cursor: input["cursor"]
      )
    end.tap do |tool|
      tool.framework_prompts(description: FrameworkPrompts.reference("subagents/tools/list/description"), manager:)
    end
  ]
  tools.first.define_method(:close) { manager.close }
  tools
end

#wait(subagent_ids: nil) ⇒ Object

Long-polls selected identities, or all identities when omitted. still_working is an ordinary timeout result and does not cancel work.



479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
 
# File 'lib/little_ghost/subagents/manager.rb', line 479

def wait(subagent_ids: nil)
  identities = @mutex.synchronize do
    ensure_open!
    selected_identities(subagent_ids)
  end
  return {status: "finished", subagents: []} if identities.empty?

  deadline = monotonic_time + @wait_timeout
  @mutex.synchronize do
    until identities.all? { |identity| finished?(identity) }
      @cancellation_token.raise_if_cancelled!
      if @deadline && Time.now >= @deadline
        raise DeadlineExceededError, "The run deadline was reached"
      end

      remaining = deadline - monotonic_time
      remaining = [remaining, @deadline - Time.now].min if @deadline
      break unless remaining.positive?

      @condition.wait(@mutex, [remaining, CANCELLATION_POLL_INTERVAL].min)
    end
    status = (identities.all? { |identity| finished?(identity) }) ? "finished" : "still_working"
    {
      status: status,
      subagents: identities.map { |identity| snapshot(identity, include_response: true, include_progress: true) }
    }
  end
end