Chat Format Converter — Ruby source

# chat-format-converter — Ruby polyglot showcase port.
#
# Converts one chat transcript between provider shapes through a shared internal
# message list:
#
# - openai    — {"messages": [{"role": "system|user|assistant|tool", ...}]}
# - anthropic — {"system": "...", "messages": [{"role": "user|assistant", ...}]}
# - gemini    — {"contents": [{"role": "user|model", "parts": [...]}]}
# - markdown  — a plain `**role**: text` transcript
#
# Language: Ruby (3.2, standard library only — json)
# Source:   CosmoDev polyglot showcase port of the Chat Format Converter tool,
#           ported from src/lib/chatFormatConverter.ts (the canonical
#           TypeScript that powers the live tool).
# License:  display source — part of CosmoDev's polyglot tool pages.
#
# Roles map faithfully (Gemini has no assistant — it is `model`; Anthropic
# system lives top-level). Fields a target format cannot represent (e.g. tool
# calls in Markdown) are flagged as warnings, never dropped silently. Ruby
# Hashes preserve insertion order, so key order survives a round-trip exactly
# like the TypeScript original. Parse failures raise ArgumentError.

require 'json'

CHAT_FORMATS = %w[openai anthropic gemini markdown].freeze

# One assistant-emitted tool call, provider-agnostic. `args` is a JSON string.
ToolCall = Struct.new(:id, :name, :args) do
  def initialize(id: '', name: '', args: '')
    super(id, name, args)
  end
end

# The shared internal shape every format parses into and serializes from.
InternalMessage = Struct.new(:role, :content, :name, :tool_calls, :tool_call_id,
                             keyword_init: true) do
  def initialize(role:, content: '', name: nil, tool_calls: nil, tool_call_id: nil)
    super
  end
end

# The conversion output plus human-readable notes about unmapped fields.
ConversionResult = Struct.new(:output, :warnings, keyword_init: true)

def parse_transcript_json(text, label)
  JSON.parse(text)
rescue JSON::ParserError => e
  raise ArgumentError, "#{label}: invalid JSON — #{e.message}"
end

def as_record(v)
  v if v.is_a?(Hash)
end

# The TS `?? {}` capture: null/missing becomes {}, any other value passes
# through and stringifies as-is.
def args_of(v)
  v.nil? ? {} : v
end

# Parse a JSON-string argument back to a Hash, `{}` when it is not an object.
def parse_args_or_empty(args)
  v = JSON.parse(args)
  v.is_a?(Hash) ? v : {}
rescue JSON::ParserError, TypeError
  {}
end

# ---------------------------------- parse -----------------------------------

# OpenAI messages[] (bare list, or wrapped in {"messages": [...]}).
def parse_openai(text, warnings)
  parsed = parse_transcript_json(text, 'OpenAI transcript')
  root = as_record(parsed)
  raw = root && root['messages'].is_a?(Array) ? root['messages'] : parsed
  unless raw.is_a?(Array)
    raise ArgumentError,
          'OpenAI transcript must be a messages[] array or an object with a "messages" array'
  end
  out = []
  raw.each_with_index do |entry, i|
    m = as_record(entry)
    raise ArgumentError, "messages[#{i}] is not an object" if m.nil?
    where = "messages[#{i}]"
    case m['role']
    when 'system', 'developer'
      out << InternalMessage.new(role: 'system', content: content_text(m['content'], where, warnings))
    when 'user', 'assistant', 'tool'
      msg = InternalMessage.new(role: m['role'], content: content_text(m['content'], where, warnings))
      msg.name = m['name'] if m['name'].is_a?(String)
      msg.tool_call_id = m['tool_call_id'] if m['tool_call_id'].is_a?(String)
      calls = parse_openai_tool_calls(m['tool_calls'])
      msg.tool_calls = calls unless calls.nil?
      out << msg
    else
      raise ArgumentError, "messages[#{i}] has unsupported role: #{JSON.generate(m['role'])}"
    end
  end
  out
end

# OpenAI content → plain text (string, or text parts joined; other parts flagged).
def content_text(content, where, warnings)
  case content
  when String then content
  when nil then ''
  when Array
    texts = []
    dropped = 0
    content.each do |part|
      p = as_record(part)
      if p && p['type'] == 'text' && p['text'].is_a?(String)
        texts << p['text']
      else
        dropped += 1
      end
    end
    warnings << "#{where}: dropped #{dropped} non-text content part(s)" if dropped.positive?
    texts.join
  else
    raise ArgumentError, "#{where}: content must be a string, an array of parts, or null"
  end
end

def parse_openai_tool_calls(raw)
  return nil if raw.nil?
  raise ArgumentError, 'tool_calls must be an array' unless raw.is_a?(Array)

  raw.map do |entry|
    tc = as_record(entry)
    fn = tc ? as_record(tc['function']) : nil
    ToolCall.new(
      id: tc && tc['id'].is_a?(String) ? tc['id'] : '',
      name: fn && fn['name'].is_a?(String) ? fn['name'] : '',
      args: fn && fn['arguments'].is_a?(String) ? fn['arguments'] : ''
    )
  end
end

# Anthropic messages[] (+ top-level `system`).
def parse_anthropic(text, warnings)
  root = as_record(parse_transcript_json(text, 'Anthropic transcript'))
  if root.nil? || !root['messages'].is_a?(Array)
    raise ArgumentError, 'Anthropic transcript must be an object with a "messages" array'
  end
  out = []
  out << InternalMessage.new(role: 'system', content: blocks_to_text(root['system'], 'system', warnings)) if root.key?('system')

  root['messages'].each_with_index do |entry, i|
    m = as_record(entry)
    raise ArgumentError, "messages[#{i}] is not an object" if m.nil?
    where = "messages[#{i}]"
    case m['role']
    when 'assistant'
      texts = []
      calls = []
      content_blocks(m['content'], where).each do |block|
        case block[0]
        when 'text' then texts << block[1]
        when 'tool_use' then calls << tool_call_from_block(block[1])
        when 'other' then warnings << "#{where}: dropped unsupported #{block_type_name(block[1])} block"
        end
      end
      msg = InternalMessage.new(role: 'assistant', content: texts.join("\n"))
      msg.tool_calls = calls unless calls.empty?
      out << msg
    when 'user'
      texts = []
      flush = lambda do
        unless texts.empty?
          out << InternalMessage.new(role: 'user', content: texts.join("\n"))
          texts.clear
        end
      end
      content_blocks(m['content'], where).each do |block|
        case block[0]
        when 'text'
          texts << block[1]
        when 'tool_use'
          warnings << "#{where}: tool_use block inside a user message moved to an assistant tool call"
          flush.call
          out << InternalMessage.new(role: 'assistant', content: '',
                                     tool_calls: [tool_call_from_block(block[1])])
        when 'tool_result'
          flush.call
          raw = block[1]
          out << InternalMessage.new(
            role: 'tool',
            content: blocks_to_text(raw['content'] || '', where, warnings),
            tool_call_id: raw['tool_use_id'].is_a?(String) ? raw['tool_use_id'] : ''
          )
        else
          warnings << "#{where}: dropped unsupported #{block_type_name(block[1])} block"
        end
      end
      flush.call
    else
      raise ArgumentError, "#{where} has unsupported role: #{JSON.generate(m['role'])}"
    end
  end
  out
end

# Normalize Anthropic content (string | block list) into typed pairs:
# ['text', text] | ['tool_use' | 'tool_result' | 'other', raw_hash].
def content_blocks(content, where)
  return [['text', content]] if content.is_a?(String)
  raise ArgumentError, "#{where}: content must be a string or an array of blocks" unless content.is_a?(Array)

  content.map do |block|
    b = as_record(block)
    if b && b['type'] == 'text' && b['text'].is_a?(String)
      ['text', b['text']]
    elsif b && %w[tool_use tool_result].include?(b['type'])
      [b['type'], b]
    else
      ['other', b || {}]
    end
  end
end

def block_type_name(raw)
  raw['type'].nil? ? 'content' : raw['type'].to_s
end

# Anthropic string-or-block-list content → plain text.
def blocks_to_text(content, where, warnings)
  return content if content.is_a?(String)

  texts = []
  content_blocks(content, where).each do |kind, raw|
    case kind
    when 'text' then texts << raw
    when 'other' then warnings << "#{where}: dropped unsupported #{block_type_name(raw)} block"
    else warnings << "#{where}: dropped #{kind} block from text-only content"
    end
  end
  texts.join("\n")
end

def tool_call_from_block(raw)
  ToolCall.new(
    id: raw['id'].is_a?(String) ? raw['id'] : '',
    name: raw['name'].is_a?(String) ? raw['name'] : '',
    args: JSON.generate(args_of(raw['input']))
  )
end

# Gemini contents[] (+ optional systemInstruction).
def parse_gemini(text, warnings)
  root = as_record(parse_transcript_json(text, 'Gemini transcript'))
  if root.nil? || !root['contents'].is_a?(Array)
    raise ArgumentError, 'Gemini transcript must be an object with a "contents" array'
  end
  out = []
  if root.key?('systemInstruction')
    out << InternalMessage.new(role: 'system',
                               content: gemini_text(root['systemInstruction'], 'systemInstruction', warnings))
  end

  root['contents'].each_with_index do |entry, i|
    m = as_record(entry)
    raise ArgumentError, "contents[#{i}] is not an object" if m.nil?
    where = "contents[#{i}]"
    unless %w[user model].include?(m['role'])
      raise ArgumentError,
            "#{where} has unsupported role: #{JSON.generate(m['role'])} (Gemini uses \"user\" or \"model\")"
    end
    role = m['role'] == 'model' ? 'assistant' : 'user'
    parts = m['parts']
    raise ArgumentError, "#{where}: parts must be an array" unless parts.is_a?(Array)

    if role == 'assistant'
      # A model turn keeps its text and function calls in ONE message,
      # mirroring an OpenAI assistant message with tool_calls.
      texts = []
      calls = []
      parts.each do |part|
        p = as_record(part)
        if p && p['text'].is_a?(String)
          texts << p['text']
        elsif p && (fc = as_record(p['functionCall']))
          calls << ToolCall.new(
            name: fc['name'].is_a?(String) ? fc['name'] : '',
            args: JSON.generate(args_of(fc['args']))
          )
        else
          warnings << "#{where}: dropped unsupported part (inlineData or similar)"
        end
      end
      msg = InternalMessage.new(role: 'assistant', content: texts.join("\n"))
      msg.tool_calls = calls unless calls.empty?
      out << msg
      next
    end

    texts = []
    flush_text = lambda do
      unless texts.empty?
        out << InternalMessage.new(role: 'user', content: texts.join("\n"))
        texts.clear
      end
    end
    parts.each do |part|
      p = as_record(part)
      if p && p['text'].is_a?(String)
        texts << p['text']
      elsif p && (fc = as_record(p['functionCall']))
        flush_text.call
        out << InternalMessage.new(role: 'assistant', content: '',
                                   tool_calls: [ToolCall.new(
                                     name: fc['name'].is_a?(String) ? fc['name'] : '',
                                     args: JSON.generate(args_of(fc['args']))
                                   )])
      elsif p && (fr = as_record(p['functionResponse']))
        flush_text.call
        name = fr['name'].is_a?(String) ? fr['name'] : ''
        out << InternalMessage.new(
          role: 'tool',
          content: JSON.generate(args_of(fr['response'])),
          name: name,
          tool_call_id: name
        )
      else
        warnings << "#{where}: dropped unsupported part (inlineData or similar)"
      end
    end
    flush_text.call
  end
  out
end

# Gemini systemInstruction (string or {parts}) → plain text.
def gemini_text(v, where, warnings)
  return v if v.is_a?(String)

  if (rec = as_record(v))
    return rec['text'] if rec['text'].is_a?(String)
    return rec['parts'].map { |part| p = as_record(part); p && p['text'].is_a?(String) ? p['text'] : '' }.join("\n") if rec['parts'].is_a?(Array)
  end
  warnings << "#{where}: unsupported systemInstruction shape, treated as empty"
  ''
end

MARKDOWN_HEADER = /^\*\*(system|user|assistant|model|tool)\*\*:\s*(.*)$/.freeze

# Markdown transcript: `**role**: text` header lines with continuation lines
# belonging to the same message. `model` maps to assistant.
def parse_markdown(text, _warnings)
  out = []
  current = nil
  text.split("\n", -1).each do |line|
    if (match = MARKDOWN_HEADER.match(line))
      out << finish_message(*current) if current
      current = [match[1] == 'model' ? 'assistant' : match[1], [match[2]]]
    elsif current
      current[1] << line
    elsif line.strip != ''
      raise ArgumentError, 'Markdown transcript must start with a `**role**:` header line'
    end
  end
  out << finish_message(*current) if current
  out
end

def finish_message(role, lines)
  # Trim leading/trailing blank continuation lines but keep inner blank lines.
  lines.shift while lines.length > 1 && lines.first.strip.empty?
  lines.pop while lines.length > 1 && lines.last.strip.empty?
  InternalMessage.new(role: role, content: lines.join("\n"))
end

# -------------------------------- serialize ---------------------------------

def serialize_openai(messages, _warnings)
  arr = messages.map do |m|
    if m.role == 'tool'
      o = { 'role' => 'tool', 'content' => m.content, 'tool_call_id' => m.tool_call_id || '' }
      o['name'] = m.name unless m.name.nil?
      o
    elsif m.role == 'assistant' && !m.tool_calls.nil?
      o = {
        'role' => 'assistant',
        'content' => m.content == '' ? nil : m.content,
        'tool_calls' => m.tool_calls.map do |tc|
          { 'id' => tc.id, 'type' => 'function',
            'function' => { 'name' => tc.name, 'arguments' => tc.args } }
        end
      }
      o['name'] = m.name unless m.name.nil?
      o
    else
      o = { 'role' => m.role, 'content' => m.content }
      o['name'] = m.name unless m.name.nil?
      o
    end
  end
  JSON.pretty_generate('messages' => arr) + "\n"
end

def serialize_anthropic(messages, warnings)
  system = messages.select { |m| m.role == 'system' }.map(&:content)
  out = []
  messages.each_with_index do |m, i|
    next if m.role == 'system'

    warnings << "message #{i}: \"name\" has no Anthropic equivalent and was dropped" if !m.name.nil? && m.role != 'tool'
    case m.role
    when 'user'
      out << { 'role' => 'user', 'content' => m.content }
    when 'assistant'
      if m.tool_calls
        blocks = []
        blocks << { 'type' => 'text', 'text' => m.content } if m.content != ''
        m.tool_calls.each do |tc|
          blocks << { 'type' => 'tool_use', 'id' => tc.id, 'name' => tc.name,
                      'input' => parse_args_or_empty(tc.args) }
        end
        out << { 'role' => 'assistant', 'content' => blocks }
      else
        out << { 'role' => 'assistant', 'content' => m.content }
      end
    else
      out << { 'role' => 'user',
               'content' => [{ 'type' => 'tool_result',
                               'tool_use_id' => m.tool_call_id || '',
                               'content' => m.content }] }
    end
  end
  root = {}
  root['messages'] = out unless out.empty?
  root['system'] = system.join("\n\n") unless system.empty?
  JSON.pretty_generate(root) + "\n"
end

def serialize_gemini(messages, warnings)
  system = messages.select { |m| m.role == 'system' }.map(&:content)
  contents = []

  push = lambda do |role, part|
    if !contents.empty? && contents.last['role'] == role
      contents.last['parts'] << part
    else
      contents << { 'role' => role, 'parts' => [part] }
    end
  end

  messages.each_with_index do |m, i|
    next if m.role == 'system'

    warnings << "message #{i}: \"name\" has no Gemini equivalent and was dropped" if !m.name.nil? && m.role != 'tool'
    case m.role
    when 'user'
      push.call('user', { 'text' => m.content })
    when 'assistant'
      push.call('model', { 'text' => m.content }) if m.content != ''
      (m.tool_calls || []).each do |tc|
        push.call('model', { 'functionCall' => { 'name' => tc.name,
                                                 'args' => parse_args_or_empty(tc.args) } })
      end
    else
      push.call('user', { 'functionResponse' => { 'name' => m.name || m.tool_call_id || '',
                                                  'response' => parse_args_or_empty(m.content) } })
    end
  end
  root = {}
  root['contents'] = contents unless contents.empty?
  unless system.empty?
    root['systemInstruction'] = { 'parts' => [{ 'text' => system.join("\n\n") }] }
  end
  JSON.pretty_generate(root) + "\n"
end

def serialize_markdown(messages, warnings)
  messages.each_with_index do |m, i|
    if m.tool_calls && !m.tool_calls.empty?
      warnings << "message #{i}: tool calls are not representable in Markdown and were dropped"
    end
    if m.role == 'tool' && !m.tool_call_id.nil?
      warnings << "message #{i}: tool result id is not representable in Markdown and was dropped"
    end
  end
  messages.map { |m| "**#{m.role}**: #{m.content}" }.join("\n\n") + "\n"
end

PARSERS = {
  'openai' => method(:parse_openai),
  'anthropic' => method(:parse_anthropic),
  'gemini' => method(:parse_gemini),
  'markdown' => method(:parse_markdown)
}.freeze

SERIALIZERS = {
  'openai' => method(:serialize_openai),
  'anthropic' => method(:serialize_anthropic),
  'gemini' => method(:serialize_gemini),
  'markdown' => method(:serialize_markdown)
}.freeze

# Convert a transcript between chat formats through the shared internal shape.
#
# Raises ArgumentError when the transcript is empty, the source format fails
# to parse, or from_format/to_format is not a known format.
def convert(transcript, from_format, to_format)
  raise ArgumentError, 'Transcript is empty — paste a transcript first' if transcript.to_s.strip.empty?

  parser = PARSERS[from_format]
  raise ArgumentError, "Unknown source format: #{from_format}" if parser.nil?
  serializer = SERIALIZERS[to_format]
  raise ArgumentError, "Unknown target format: #{to_format}" if serializer.nil?

  warnings = []
  messages = parser.call(transcript, warnings)
  output = serializer.call(messages, warnings)
  ConversionResult.new(output: output, warnings: warnings)
end