class Riffer::Rig::ACP

# File lib/riffer/rig/acp.rb, line 45
def self.build_runtime(cwd, host, servers, env:, home:)
  Riffer::Rig::Loader.runtime(
    cwd: cwd, host: host, env: env, home: home, extra_extensions: session_extensions(servers)
  )
end

# File lib/riffer/rig/acp.rb, line 35
def self.for(input:, output:, env:, home:, cwd:, build: nil, adopt: nil)
  new(input: input, output: output, env: env, home: home, cwd: cwd, build: build, adopt: adopt)
end

# File lib/riffer/rig/acp.rb, line 72
def initialize(input:, output:, env:, home:, cwd:, build: nil, adopt: nil)
  @input = input
  @output = output
  @env = env
  @home = home
  @cwd = cwd
  @build = build || lambda do |cwd, host, servers|
    own(self.class.build_runtime(cwd, host, servers, env: @env, home: @home))
  end
  @adopt = adopt
  @client = nil
  @sessions = {}
  @hosts = {}
  @owned = []
  @lock = Mutex.new
end

# File lib/riffer/rig/acp.rb, line 53
def self.session_extensions(servers)
  return [] if servers.empty?

  # A synthetic extension, so a reload rebuild re-declares the servers and an
  # extension's same-name declaration replaces them with the usual notify.
  extension = Riffer::Rig::Extension.new(SESSION_EXTENSION) do |rig|
    servers.each { |name, server| rig.mcp(name, command: server.command, args: server.args, env: server.env) }
  end
  [extension]
end

# File lib/riffer/rig/acp.rb, line 210
def cancel(notification)
  session(notification.session_id)&.cancel
end

# File lib/riffer/rig/acp.rb, line 185
def list_sessions(request)
  loader = Riffer::Rig::Loader.new(
    cwd: request.cwd || @cwd,
    host: Riffer::Rig::Hosts::Null.new,
    env: @env,
    home: @home
  )
  headers = request.cwd ? loader.list : loader.list(all: true)
  sessions = headers.map do |header|
    ::ACP::Types::SessionInfo.new(
      session_id: header.id,
      cwd: header.cwd,
      title: header.title,
      updated_at: header.updated&.utc&.iso8601
    )
  end
  ::ACP::Types::ListSessionsResponse.new(sessions: sessions)
rescue StandardError => e
  internal_error(e)
end

# File lib/riffer/rig/acp.rb, line 163
def load_session(request)
  host = Riffer::Rig::ACP::Host.new(client: client)
  loader = Riffer::Rig::Loader.new(cwd: request.cwd, host: host, env: @env, home: @home)
  runtime = adopted(request.session_id) || resumed(loader, request, host)
  return ::ACP::RequestError.resource_not_found unless runtime

  # A client need not answer a question while its own session/load is
  # pending, and acp-sdk never times a request out, so the host takes the
  # session id only once the build can no longer ask one.
  host.session_id = request.session_id
  @lock.synchronize { @sessions[request.session_id] = runtime }
  loader.messages(request.session_id)
        .filter_map { |message| Riffer::Rig::ACP::Mapper.replay_of(message) }
        .each { |update| client.update(request.session_id, update) }
  client.available_commands(request.session_id, available_commands(runtime))
  ::ACP::Types::LoadSessionResponse.new
rescue StandardError => e
  internal_error(e)
end

# File lib/riffer/rig/acp.rb, line 118
def new_session(request)
  host = Riffer::Rig::ACP::Host.new(client: client)
  servers = declarations(request.mcp_servers, host)
  runtime = announcing_job_exits(@build.call(request.cwd, host, servers), host)
  @lock.synchronize do
    @sessions[runtime.id] = runtime
    @hosts[runtime.id] = host
  end
  ::ACP::Types::NewSessionResponse.new(session_id: runtime.id)
rescue StandardError => e
  internal_error(e)
end

# File lib/riffer/rig/acp.rb, line 133
def prompt(request)
  runtime = session(request.session_id)
  return ::ACP::RequestError.resource_not_found unless runtime

  commands = available_commands(runtime).to_set(&:to_h)
  # A command that runs no turn sends no turn_end.
  stop_reason = :completed
  dispatch(runtime, Riffer::Rig::ACP::Mapper.prompt_text(request.prompt)) do |event|
    case event
    when Riffer::Rig::Events::TurnEnd then stop_reason = event.stop_reason
    end
    update = Riffer::Rig::ACP::Mapper.update_for(event)
    client.update(request.session_id, update) if update
  end
  # A reload in the dispatch, by /reload or the automatic check, can change
  # the commands; its rebuild also reorders them, which is no change.
  reloaded = available_commands(runtime)
  client.available_commands(request.session_id, reloaded) unless reloaded.to_set(&:to_h) == commands
  ::ACP::Types::PromptResponse.new(stop_reason: Riffer::Rig::ACP::Mapper.stop_reason_of(stop_reason))
rescue Riffer::Rig::Runtime::BusyError
  ::ACP::RequestError.new(
    code: ::ACP::RequestError::INVALID_REQUEST,
    message: 'a prompt is already running on this session; wait for it to finish or cancel it'
  )
rescue StandardError => e
  internal_error(e)
end

# File lib/riffer/rig/acp.rb, line 91
def run
  connection = ::ACP::AgentConnection.new(
    transport: ::ACP::Transport::Stdio.new(input: @input, output: @output),
    capabilities: ::ACP::Types::AgentCapabilities.new(
      load_session: true,
      session_capabilities: ::ACP::Types::SessionCapabilities.new(list: ::ACP::Types::SessionListCapabilities.new)
    ),
    agent_info: ::ACP::Types::Implementation.new(name: AGENT_NAME, version: Riffer::Rig::VERSION, title: AGENT_TITLE)
  ) do |client|
    @client = client
    self
  end
  connection.start.join
  yield connection if block_given?
  0
ensure
  # Background jobs run in process groups of their own, so only closing
  # their Runtime stops them outliving this process. A turn still running is
  # not waited for, so it is cancelled first to stop it starting another.
  @lock.synchronize { @owned.dup }.each do |runtime|
    runtime.cancel
    runtime.close
  end
end

# File lib/riffer/rig/acp.rb, line 219
def session_created(response)
  id = response.session_id
  # @type var found: [Riffer::Rig::ACP::Host?, Riffer::Rig::Runtime?]
  found = @lock.synchronize { [@hosts[id], @sessions[id]] }
  host, runtime = found
  return unless host && runtime

  host.session_id = id
  client.available_commands(id, available_commands(runtime))
end