Class: LittleGhost::Subagents::Manager

Inherits:

Object

Object
LittleGhost::Subagents::Manager

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, wait for, interrupt, and list research agents. The manager keeps each child identity stable across follow-up turns.

Follow-up messages are FIFO turns and never interrupt active work. #interrupt 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.

Defined Under Namespace

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

Constant Summary collapse

DEFAULT_MAX_CONCURRENT = :nodoc:

DEFAULT_MAX_IDENTITIES = :nodoc:

DEFAULT_MAX_TURNS = :nodoc:

DEFAULT_MAX_QUEUED_TURNS_PER_IDENTITY = :nodoc:

DEFAULT_MAX_MESSAGE_CHARS = :nodoc:

50_000

DEFAULT_MAX_RESPONSE_CHARS = :nodoc:

100_000

DEFAULT_WAIT_TIMEOUT = :nodoc:

20.0

DEFAULT_CLOSE_TIMEOUT = :nodoc:

5.0

DEFAULT_LIST_LIMIT = :nodoc:

MAX_LIST_LIMIT = :nodoc:

MAX_PROGRESS_CHARS = :nodoc:

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:

CURSOR_MAX_BYTES = :nodoc:

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

#definitions ⇒ Object readonly
Available definitions, indexed by kind.

Class Method Summary collapse

.commit_session_id(conversation_id, slot) ⇒ Object
Derives one of the rotating committed-state session IDs.
.conversation_session_id(conversation_id) ⇒ Object
Derives the framework-owned transcript session ID.
.parent_link(session) ⇒ Object
Produces a pseudonymous parent-session link for durable metadata.
.registry_session_id(session) ⇒ Object
Derives the framework-owned registry session ID.

Instance Method Summary collapse

#close ⇒ Object
Cancels queued work, cooperatively stops workers, and closes child agents.
#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 constructor
Configures a bounded manager.
#interrupt(subagent_id:, message:, cancellation_token: @cancellation_token, deadline: @deadline) ⇒ Object
Delivers message to one currently running turn and waits for the next model response.
#list(kind: nil, limit: DEFAULT_LIST_LIMIT, cursor: nil) ⇒ Object
Lists active and persisted identities newest-first without restoring inactive agents.
#send_message(subagent_id:, message:, mode:, parent_operation_id: nil, context: nil) ⇒ Object
Queues a FIFO follow-up for an active or durable identity.
#spawn(kind:, task_name:, task:, mode:, parent_operation_id: nil, context: nil) ⇒ Object
Creates a unique child identity and queues its first task.
#tools ⇒ Object
Builds spawn, follow-up, interrupt, wait, and list tools bound to this manager.
#wait(subagent_ids: nil) ⇒ Object
Long-polls selected identities, or all identities when omitted.

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.

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

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
  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
  @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
  @registry_mutex = Mutex.new
  @restore_mutex = Mutex.new
  @condition = ConditionVariable.new
  @identities = {}
  @reserved_agent_paths = {}
  @identity_slots = 0
  @turn_count = 0
  @closed = false
  restore_identities
end

Instance Attribute Details

#definitions ⇒ `Object` (readonly)

Available definitions, indexed by kind.



180
181
182

# File 'lib/little_ghost/subagents/manager.rb', line 180

def definitions
  @definitions
end

Class Method Details

.commit_session_id(conversation_id, slot) ⇒ `Object`

Derives one of the rotating committed-state session IDs.



199
200
201

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

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.



194
195
196

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

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.



184
185
186

# File 'lib/little_ghost/subagents/manager.rb', line 184

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.



189
190
191

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

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

Instance Method Details

#close ⇒ `Object`

Cancels queued work, cooperatively stops workers, and closes child agents. Raises CleanupError if workers do not stop within the bound.

# File 'lib/little_ghost/subagents/manager.rb', line 666

def close
  workers = @mutex.synchronize do
    return if @closed

    @closed = true
    @cancellation_token.cancel
    @identities.each_value do |identity|
      next if %w[idle failed cancelled persisting].include?(identity.status)

      turn = identity.current
      identity.status = "cancelled"
      turn&.completion&.resolve(cancelled_turn(identity, turn))
      identity.progress_message = nil
      identity.current_turn = nil
      identity.current = nil
      cancel_queued_turns(identity)
      emit("cancelled", identity, turn:)
    end
    @condition.broadcast
    @identities.values.filter_map(&:worker)
  end

  deadline = monotonic_time + @close_timeout
  cooperative_deadline = monotonic_time + (@close_timeout / 2.0)
  workers.each do |worker|
    remaining = cooperative_deadline - monotonic_time
    break unless remaining.positive?

    worker.join(remaining)
  end
  workers.select(&:alive?).each(&:kill)
  workers.each do |worker|
    remaining = deadline - monotonic_time
    break unless remaining.positive?

    worker.join(remaining)
  end
  first_error = nil
  survivors = workers.select(&:alive?)
  unless survivors.empty?
    first_error ||= CleanupError.new(
      "#{survivors.length} subagent worker(s) did not stop within #{@close_timeout} seconds"
    )
  end
  agents = @mutex.synchronize { @identities.values.map(&:agent).reverse.uniq(&:object_id) }
  agents.each do |agent|
    agent.close if agent.respond_to?(:close)
  rescue => error
    first_error ||= error
  end
  raise first_error if first_error
end

#interrupt(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.

# File 'lib/little_ghost/subagents/manager.rb', line 371

def interrupt(subagent_id:, message:, cancellation_token: @cancellation_token, deadline: @deadline)
  unless message.is_a?(String)
    raise ToolError, "Subagent messages must be strings."
  end
  if message.length > @max_message_chars
    raise ToolError, "Subagent messages cannot exceed #{@max_message_chars} characters."
  end

  exchange = InterruptExchange.new(message:, complete: false)
  identity, turn = @mutex.synchronize do
    ensure_open!
    value = fetch_identity!(subagent_id)
    unless value.agent.respond_to?(:interrupt_response)
      raise ToolError, "Subagent #{subagent_id.inspect} does not support interruptions."
    end
    unless value.status == "running"
      raise ToolError, "Subagent #{subagent_id.inspect} is not currently running."
    end
    if value.current.interrupts.length >= @max_queued_turns_per_identity
      raise ToolError, "Subagent #{subagent_id.inspect} has reached its interrupt limit."
    end
    interrupt_chars = value.current.interrupts.sum { |pending| pending.message.length }
    if interrupt_chars + message.length > @max_message_chars
      raise ToolError, "Subagent interrupt messages cannot exceed #{@max_message_chars} total characters."
    end

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

  interrupt_response = begin
    identity.agent.interrupt_response(
      message,
      cancellation_token:,
      deadline:,
      target_operation_id: turn.operation_id
    )
  rescue
    @mutex.synchronize do
      turn.interrupts.delete(exchange)
      @condition.broadcast
    end
    raise
  end
  response = interrupt_response.text
  truncated = response.length > @max_response_chars
  returned_response = truncated ? response[0, @max_response_chars] : response
  @mutex.synchronize do
    used_response_chars = turn.interrupts.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: "interruption_delivered",
    subagent_id: identity.subagent_id,
    kind: identity.definition.kind,
    subagent:,
    turn: turn.number,
    response: returned_response,
    response_disposition: interrupt_response.tool_calls? ? "text_with_tool_calls" : "text_only"
  }
  value[:response_truncated] = true if truncated
  value
rescue AgentInterruptError => 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.

# File 'lib/little_ghost/subagents/manager.rb', line 478

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, "limit must be between 1 and #{MAX_LIST_LIMIT}"
  end
  if kind && !definitions.key?(kind)
    raise ToolError, "Unknown subagent 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.

# File 'lib/little_ghost/subagents/manager.rb', line 344

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.

# File 'lib/little_ghost/subagents/manager.rb', line 269

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: "Subagent could not be created."
    }
  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

  turn, queued_snapshot = enqueue(
    identity,
    task,
    event: "spawned",
    count_turn: false,
    parent_operation_id:,
    context:
  )
  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, interrupt, wait, and list tools bound to this manager. Closing the first tool closes the shared manager.

# File 'lib/little_ghost/subagents/manager.rb', line 509

def tools
  manager = self
  kind_descriptions = definitions.values.map do |definition|
    "- #{definition.kind}: #{definition.description}"
  end.join("\n")
  tools = [
    Tool.define(
      name: "spawn_subagent",
      description: <<~DESCRIPTION.strip,
        Create a new subagent identity for an independent task. Mode controls delivery: sync waits for the
        response in this call, while async returns immediately and leaves the response for
        wait_for_subagents. Several sync spawns requested together can still run in parallel. Give the task a
        concise lowercase name. The returned identity is its canonical path beneath the current agent. Task
        names must be unique among that agent's children.
      DESCRIPTION
      input_schema: {
        type: "object",
        properties: {
          kind: {
            type: "string",
            enum: definitions.keys,
            description: "Kind of subagent to create.\n#{kind_descriptions}"
          },
          task_name: {
            type: "string",
            pattern: "^[a-z0-9_]+$",
            maxLength: AgentPath::MAX_NAME_LENGTH,
            description: "Friendly task name using lowercase letters, digits, and underscores."
          },
          task: {type: "string", description: "Independent task to delegate."},
          mode: {
            type: "string", enum: %w[sync async],
            description: "sync waits for the response; async returns while the subagent continues."
          }
        },
        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,
    Tool.define(
      name: "send_message_to_subagent",
      description: <<~DESCRIPTION.strip,
        Send a follow-up turn to an existing active or persisted subagent identity. Persisted conversations
        are restored transparently before the follow-up. Messages are processed in order after the
        current turn and never interrupt active work. Do not use this for status, steering, stopping, or
        finalization; use interrupt_subagent for an active subagent. Mode controls delivery: sync waits for the
        later turn's response, while async enqueues the turn and returns immediately.
      DESCRIPTION
      input_schema: {
        type: "object",
        properties: {
          subagent_id: {type: "string", description: "Existing subagent identity."},
          message: {type: "string", description: "Follow-up task or context."},
          mode: {
            type: "string", enum: %w[sync async],
            description: "sync waits for this turn; async enqueues it and returns immediately."
          }
        },
        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,
    Tool.define(
      name: "interrupt_subagent",
      description: <<~DESCRIPTION.strip,
        Interrupt an actively running subagent in its current turn. The message is added at the next model
        boundary. This call waits for that model response and reports its ordinary text, whether the same
        response also initiated tool work, and the subagent's current lifecycle state. Delivery is distinct
        from stopping: tool work from that response remains with the subagent and its current run may continue.
      DESCRIPTION
      input_schema: {
        type: "object",
        properties: {
          subagent_id: {type: "string", description: "Actively running subagent identity."},
          message: {type: "string", description: "Status question, steering context, or request to finish."}
        },
        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.interrupt(
        subagent_id: input.fetch("subagent_id"),
        message: input.fetch("message"),
        **options
      )
    end,
    Tool.define(
      name: "wait_for_subagents",
      description: <<~DESCRIPTION.strip,
        Wait briefly for selected subagents, or all subagents when omitted. A still_working response is expected
        when work takes longer than this check-in window. Call this tool again to keep waiting; timeout is not an
        error and does not cancel the subagents. A successful settled turn is returned as response. When newer
        work is queued, running, persisting, failed, or cancelled, the most recent successful result may instead
        appear as previous_response for context; it is not the result of that newer work. Inspect each subagent's
        status and keep waiting while selected work is active.
      DESCRIPTION
      input_schema: {
        type: "object",
        properties: {
          subagent_ids: {
            type: "array", items: {type: "string"},
            description: "Subagent identities to wait for; omit to wait for all."
          }
        },
        additionalProperties: false
      }
    ) { |input| manager.wait(subagent_ids: input["subagent_ids"]) },
    Tool.define(
      name: "list_subagents",
      description: <<~DESCRIPTION.strip,
        List active and persisted subagent conversations newest-first without restoring inactive agents.
        Use kind to filter. Omit cursor for the first page; to continue, pass the exact non-empty next_cursor
        from the preceding result.
      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
  ]
  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.

# File 'lib/little_ghost/subagents/manager.rb', line 447

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

Class: LittleGhost::Subagents::Manager

Overview

Durability and cleanup

Defined Under Namespace

Constant Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

Instance Attribute Details

#definitions ⇒ Object (readonly)

Class Method Details

.commit_session_id(conversation_id, slot) ⇒ Object

.conversation_session_id(conversation_id) ⇒ Object

.parent_link(session) ⇒ Object

.registry_session_id(session) ⇒ Object

Instance Method Details

#close ⇒ Object

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

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

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

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

#tools ⇒ Object

#wait(subagent_ids: nil) ⇒ Object