The Durable Batch

The Durable Batch: six billable "LLM calls" run under an ExecutionJournal. Mid-batch, the process dies for real - exit!, no cleanup, the honest kill -9. Then a second process replays the journal and finishes the batch WITHOUT re-paying for completed work.

source on github

a real captured run

run 1: processing 6 invoice(s): invoice-1, invoice-2, invoice-3, invoice-4, invoice-5, invoice-6
  !! power cut during invoice-4 - process dying with exit!(97)
  child exited with status 97 (journal survived on disk)

journal replay: 3 invoice(s) already paid: invoice-1, invoice-2, invoice-3
                3 remain (including the one mid-flight when we died)

run 2: processing 3 invoice(s): invoice-4, invoice-5, invoice-6

RECEIPT
  run 1 paid:  4 calls (3 journaled + 1 lost to the crash)
  run 2 paid:  3 calls
  total spend: $1.75 for 6 invoices (naive rerun-everything: $2.50)
  journal: /tmp/agentic_durable_batch.journal.jsonl

source

# frozen_string_literal: true

# The Durable Batch: six billable "LLM calls" run under an
# ExecutionJournal. Mid-batch, the process dies for real - exit!, no
# cleanup, the honest kill -9. Then a second process replays the
# journal and finishes the batch WITHOUT re-paying for completed work.
#
#   bundle exec ruby examples/durable_batch.rb
#
# Runs offline; the "API" is sleep plus an invoice counter. The number
# to watch is the last one: total paid should equal the batch size,
# crash or no crash.

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

# exit! discards buffered IO - the child's narration would die with it
$stdout.sync = true

JOURNAL = File.join(Dir.tmpdir, class="s">"agentic_durable_batch.journal.jsonl")
File.delete(JOURNAL) if File.exist?(JOURNAL)

INVOICES = %w[invoice-1 invoice-2 invoice-3 invoice-4 invoice-5 invoice-6].freeze
COST_PER_CALL = 0.25 # dollars of imaginary tokens

# Bills the imaginary API, then optionally dies like a deploy
BillableAgent = Struct.new(class="y">:crash_on, class="y">:billed) do
  def execute(prompt)
    invoice = prompt[/invoice-\d+/]
    sleep(0.05) # the API call we pay for
    billed << invoice

    if invoice == crash_on
      puts class="s">"  !! power cut during #{invoice} - process dying with exit!(97)"
      Process.exit!(97) # no ensure blocks, no at_exit - a real crash
    end

    {class="s">"invoice" => invoice, class="s">"status" => class="s">"paid"}
  end
end

Desk = Struct.new(class="y">:agent) do
  def get_agent_for_task(_task)
    agent
  end
end

def run_batch(label, skip: [], crash_on: nil)
  journal = Agentic:class="y">:ExecutionJournal.new(path: JOURNAL)
  billed = []
  orchestrator = Agentic:class="y">:PlanOrchestrator.new(
    concurrency_limit: 1, # deterministic order, one call in flight
    lifecycle_hooks: journal.lifecycle_hooks
  )

  todo = INVOICES - skip
  todo.each do |invoice|
    orchestrator.add_task(Agentic:class="y">:Task.new(
      description: invoice,
      agent_spec: {class="s">"name" => class="s">"Biller", class="s">"instructions" => class="s">"Process #{invoice}"},
      input: {}
    ))
  end

  puts class="s">"#{label}: processing #{todo.size} invoice(s): #{todo.join(", class="s">")}"
  orchestrator.execute_plan(Desk.new(BillableAgent.new(crash_on, billed)))
  billed
end

# Maps journal task ids back to invoice names via task_started events
def completed_invoices(state)
  names = state.events.each_with_object({}) do |event, map|
    map[event[class="y">:task_id]] = event[class="y">:description] if event[class="y">:event] == class="s">"task_started"
  end
  state.completed_task_ids.map { |id| names[id] }
end

# --- Run 1: a child process that will not survive invoice-4 ---------------
child = fork do
  run_batch(class="s">"run 1", crash_on: class="s">"invoice-4")
end
_, status = Process.wait2(child)
puts class="s">"  child exited with status #{status.exitstatus} (journal survived on disk)"
puts

# --- The journal knows exactly what was paid for --------------------------
state = Agentic:class="y">:ExecutionJournal.replay(path: JOURNAL)
paid = completed_invoices(state)
puts class="s">"journal replay: #{paid.size} invoice(s) already paid: #{paid.join(", class="s">")}"
puts class="s">"                #{INVOICES.size - paid.size} remain (including the one mid-flight when we died)"
puts

# --- Run 2: same batch, skip what the journal proves is done --------------
billed_in_run2 = run_batch(class="s">"run 2", skip: paid)
puts

total_calls = paid.size + 1 + billed_in_run2.size # +1: paid for, then died during
naive_calls = paid.size + 1 + INVOICES.size # rerunning the whole batch
puts class="s">"RECEIPT"
puts class="s">"  run 1 paid:  #{paid.size + 1} calls (#{paid.size} journaled + 1 lost to the crash)"
puts class="s">"  run 2 paid:  #{billed_in_run2.size} calls"
puts format(class="s">"  total spend: $%.2f for %d invoices (naive rerun-everything: $%.2f)",
  total_calls * COST_PER_CALL, INVOICES.size, naive_calls * COST_PER_CALL)
puts class="s">"  journal: #{JOURNAL}"