Live Import Mapper

Live Import Mapper: bulk_import's sibling with the stub removed. The mechanical 80% of an import (batch, upsert, journal) never needed a mind - but the semantic 20% does: marketing's CSV says "E-Mail Addr" and "Tier", one row has name and email swapped, another writes "Pro " with a trailing space. Here that step is a REAL LLM task: an agent spec resolved by DefaultAgentProvider, a structured-output schema registered in TaskOutputSchemas, one live call that returns canonical …

source on github

awaiting its first recording

this example drives real LLM calls through the full stack. recording it once (bin/record live_import_mapper, the only step that needs a key) commits a cassette; every build after that replays the same real interaction - keyless and deterministic.

LIVE IMPORT MAPPER - not yet recorded

  bulk_import without the stub: a real LLM normalizes the messy CSV,
  deterministic tasks batch/upsert/journal the result.
  record it once (the only step that needs a key):
    OPENAI_ACCESS_TOKEN=... bin/record live_import_mapper
  after that every run replays the recording, offline, byte-for-byte.

source

# frozen_string_literal: true

# Live Import Mapper: bulk_import's sibling with the stub removed. The
# mechanical 80% of an import (batch, upsert, journal) never needed a
# mind - but the semantic 20% does: marketing's CSV says "E-Mail Addr"
# and "Tier", one row has name and email swapped, another writes "Pro "
# with a trailing space. Here that step is a REAL LLM task: an agent
# spec resolved by DefaultAgentProvider, a structured-output schema
# registered in TaskOutputSchemas, one live call that returns canonical
# rows plus its own repair log - then the boring disciplines take over.
# Right tool per task is the whole point of a plan.
#
#   bundle exec ruby examples/live_import_mapper.rb
#
# Replays offline from its cassette; exits 1 unless every row lands
# canonically (valid emails, plans in {free,pro}, unique ids) and the
# repairs were declared. Before the first recording it explains itself
# and exits 0. Record once: OPENAI_ACCESS_TOKEN=... bin/record live_import_mapper

require class="s">"bundler/setup"
require class="s">"agentic"
require class="s">"vcr"
require class="s">"tmpdir"

Agentic.logger.level = class="y">:fatal

NAME = File.basename(__FILE__, class="s">".rb")
CASSETTES = File.expand_path(class="s">"cassettes", __dir__)
RECORDING = ENV[class="s">"RECORD"] == class="s">"1"

unless RECORDING || File.exist?(File.join(CASSETTES, class="s">"#{NAME}.yml"))
  puts class="s">"LIVE IMPORT MAPPER - not yet recorded"
  puts
  puts class="s">"  bulk_import without the stub: a real LLM normalizes the messy CSV,"
  puts class="s">"  deterministic tasks batch/upsert/journal the result."
  puts class="s">"  record it once (the only step that needs a key):"
  puts class="s">"    OPENAI_ACCESS_TOKEN=... bin/record #{NAME}"
  puts class="s">"  after that every run replays the recording, offline, byte-for-byte."
  exit 0
end

VCR.configure do |c|
  c.cassette_library_dir = CASSETTES
  c.hook_into class="y">:webmock
  c.filter_sensitive_data(class="s">"<LLM_TOKEN>") { Agentic.configuration.access_token }
  c.before_record { |i| i.request.headers.delete(class="s">"Authorization") }
  # match on path, not full uri: a cassette recorded against a local model
  # replays fine in CI, where the client points at the default endpoint
  c.default_cassette_options = {match_requests_on: [class="y">:method, class="y">:path]}
end

# replay needs no credentials - every byte of HTTP comes from the cassette
Agentic.configure { |c| c.access_token ||= class="s">"vcr-replay" } unless RECORDING

# The CSV marketing sent at 5pm: alien headers, two damaged rows.
HEADERS = [class="s">"Customer Ref", class="s">"Full Name", class="s">"E-Mail Addr", class="s">"Tier"]
ROWS = [
  [class="s">"u1", class="s">"Ada Lovelace", class="s">"ada@example.com", class="s">"pro"],
  [class="s">"u2", class="s">"Grace Hopper", class="s">"grace@example.com", class="s">"free"],
  [class="s">"u3", class="s">"hopper2@example.com", class="s">"Mary Hopper", class="s">"free"],   # name/email swapped
  [class="s">"u4", class="s">"Jean Jennings", class="s">"jean@example.com", class="s">"Pro "],    # case + stray space
  [class="s">"u5", class="s">"Kay Antonelli", class="s">"kay@example.com", class="s">"premium"]   # not a plan we sell
]

Agentic:class="y">:TaskOutputSchemas.register(class="y">:canonical_rows,
  Agentic:class="y">:StructuredOutputs:class="y">:Schema.new(class="s">"canonical_rows") do |s|
    s.array class="y">:rows, items: {
      type: class="s">"object",
      properties: {id: {type: class="s">"string"}, email: {type: class="s">"string"}, plan: {type: class="s">"string"}},
      required: %w[id email plan]
    }
    s.array class="y">:repairs, items: {type: class="s">"string"}
  end)

DB = {}
JOURNAL_PATH = File.join(Dir.tmpdir, class="s">"agentic_live_import_journal.jsonl")
File.delete(JOURNAL_PATH) if File.exist?(JOURNAL_PATH)
journal = Agentic:class="y">:ExecutionJournal.new(path: JOURNAL_PATH)

normalize = Agentic:class="y">:Task.new(
  description: class="s">"Normalize this CSV export into canonical import rows",
  agent_spec: {class="s">"name" => class="s">"data steward",
               class="s">"instructions" => class="s">"Map the given headers onto the canonical schema {id, email, plan}. " \
                                 class="s">"Repair obviously damaged rows (swapped fields, whitespace, casing). " \
                                 class="s">"plan must be exactly 'free' or 'pro'; map unknown paid tiers to 'pro'. " \
                                 class="s">"List every repair you made, one short sentence each."},
  input: {headers: HEADERS, rows: ROWS},
  output_schema_name: class="y">:canonical_rows
)

import = Agentic:class="y">:Task.new(description: class="s">"batch upsert + journal",
  agent_spec: {class="s">"name" => class="s">"importer", class="s">"instructions" => class="s">"load canonical rows"})

orchestrator = Agentic:class="y">:PlanOrchestrator.new(concurrency_limit: 1, retry_policy: {max_retries: 0, retryable_errors: []})
orchestrator.add_task(normalize, []) # no agent injected: DefaultAgentProvider builds a real one
orchestrator.add_task(import, [normalize], agent: ->(_t) {
  rows = normalize.output[class="s">"rows"]
  rows.each { |r| DB[r[class="s">"id"]] = r } # upsert: keyed write, idempotent
  journal.record(class="y">:batch_done, description: class="s">"0", rows: rows.size)
  class="y">:imported
})

VCR.use_cassette(NAME, record: RECORDING ? class="y">:all : class="y">:none) do
  puts class="s">"LIVE IMPORT MAPPER (a real mind at the semantic step, discipline everywhere else)"
  puts
  status = orchestrator.execute_plan(Agentic:class="y">:DefaultAgentProvider.new).status
  if normalize.output.nil?
    puts class="s">"  the normalize task failed: #{normalize.failure&.message}"
    exit 1
  end
  repairs = normalize.output[class="s">"repairs"]

  puts class="s">"  the agent read #{HEADERS.inspect}"
  puts class="s">"  and returned #{DB.size} canonical rows; its own repair log:"
  repairs.each { |r| puts class="s">"    - #{r.gsub(/\s+/, " class="s">").strip[0, 76]}" }
  puts
  puts class="s">"  then the boring disciplines: upserted in 1 round-trip, journaled;"
  puts class="s">"  plan status: #{status}"
  puts

  failures = []
  failures << class="s">"plan status: #{status}" unless status == class="y">:completed
  failures << class="s">"lost rows (#{DB.size}/#{ROWS.size})" unless DB.size == ROWS.size
  failures << class="s">"an email doesn't look like one" unless DB.values.all? { |r| r[class="s">"email"].include?(class="s">"@") }
  failures << class="s">"plans not canonical: #{DB.values.map { |r| r["planclass="s">"] }.uniq}" unless DB.values.all? { |r| %w[free pro].include?(r[class="s">"plan"]) }
  failures << class="s">"repairs went undeclared" if repairs.size < 2

  puts class="s">"  the LLM did the one thing lambdas can't - understood 'E-Mail Addr',"
  puts class="s">"  unswapped row u3, read 'premium' as a paid tier - and the referee"
  puts class="s">"  still holds it to falsifiable claims. minds for meaning, machines"
  puts class="s">"  for mechanics, one plan for both."
  exit(failures.empty? ? 0 : 1)
end