class LittleGhost::Subagents::Manager

# File lib/little_ghost/subagents/manager.rb, line 199
def commit_session_id(conversation_id, slot)
  "lg_subagent_commit_#{conversation_id}_#{slot}"
end

# File lib/little_ghost/subagents/manager.rb, line 194
def conversation_session_id(conversation_id)
  "lg_subagent_conversation_#{conversation_id}"
end

# 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

# File lib/little_ghost/subagents/manager.rb, line 184
def parent_link(session)
  Digest::SHA256.hexdigest("#{session.actor_id}\0#{session.id}")
end

# File lib/little_ghost/subagents/manager.rb, line 189
def registry_session_id(session)
  "lg_subagent_registry_#{parent_link(session)}"
end

# 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

# 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

# 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

# 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

# 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

# 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

# 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