class LittleGhost::Subagents::Manager

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

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

# 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

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

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

# File lib/little_ghost/subagents/manager.rb, line 707
def close
  synchronize_close { perform_close }
end

# 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

# 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

# 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

# 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

# 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

# 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