The Command Bus

The Command Bus: every command is a composed capability with its OWN declared contract (new in this round - compositions used to be contract-less). The bus is just the registry: dispatching a command validates it at the boundary, routes it through its handler pipeline, and validates what comes back.

source on github

a real captured run

COMMAND BUS

  ACCEPTED PlaceOrder(sku: "widget", quantity: 3)
           -> StockReserved(7 left)
           -> OrderPlaced(#1)
  REJECTED PlaceOrder(sku: "widget", quantity: "many")
           -> CommandRejected: quantity must be Numeric
  ACCEPTED RestockShelf(sku: "widget", quantity: 5)
           -> ShelfRestocked(#2)
  REJECTED PlaceOrder(sku: "widget", quantity: 13)
           -> OrderRejected: insufficient stock for widget
  REJECTED ShipRocket(to: "the moon")
           -> UnknownCommand: ShipRocket

ledger: 2 entries | widget stock: 12

note the two different REJECTED shapes: the contract rejected
'many' before any handler ran; the business rule rejected 13 after
checking the shelf. types stop nonsense, domains stop mistakes.

source

# frozen_string_literal: true

# The Command Bus: every command is a composed capability with its OWN
# declared contract (new in this round - compositions used to be
# contract-less). The bus is just the registry: dispatching a command
# validates it at the boundary, routes it through its handler pipeline,
# and validates what comes back.
#
#   bundle exec ruby examples/command_bus.rb
#
# Runs offline. Watch PlaceOrder(quantity: "many") bounce off the
# boundary with the violation named.

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

registry = Agentic:class="y">:AgentCapabilityRegistry.instance

# --- primitive capabilities: small, reusable handler steps -----------------
def capability(name, inputs:, outputs:, &impl)
  spec = Agentic:class="y">:CapabilitySpecification.new(
    name: name, description: name, version: class="s">"1.0.0", inputs: inputs, outputs: outputs
  )
  Agentic.register_capability(
    spec, Agentic:class="y">:CapabilityProvider.new(capability: spec, implementation: impl)
  )
end

STOCK = Hash.new(10)
LEDGER = []

capability(class="s">"reserve_stock",
  inputs: {sku: {type: class="s">"string", required: true}, quantity: {type: class="s">"number", required: true}},
  outputs: {reserved: {type: class="s">"boolean", required: true}, remaining: {type: class="s">"number", required: true}}) do |input|
  available = STOCK[input[class="y">:sku]]
  reserved = available >= input[class="y">:quantity]
  STOCK[input[class="y">:sku]] -= input[class="y">:quantity] if reserved
  {reserved: reserved, remaining: STOCK[input[class="y">:sku]]}
end

capability(class="s">"record_entry",
  inputs: {entry: {type: class="s">"string", required: true}},
  outputs: {position: {type: class="s">"number", required: true}}) do |input|
  LEDGER << input[class="y">:entry]
  {position: LEDGER.size}
end

# --- commands: compositions with their own contracts ------------------------
registry.compose(
  class="s">"PlaceOrder", class="s">"Place an order for a SKU", class="s">"1.0.0",
  [{name: class="s">"reserve_stock", version: class="s">"1.0.0"}, {name: class="s">"record_entry", version: class="s">"1.0.0"}],
  lambda do |(reserve, record), command|
    reservation = reserve.execute(sku: command[class="y">:sku], quantity: command[class="y">:quantity])
    unless reservation[class="y">:reserved]
      next {accepted: false, events: [class="s">"OrderRejected: insufficient stock for #{command[class="y">:sku]}"]}
    end

    entry = record.execute(entry: class="s">"order #{command[class="y">:sku]} x#{command[class="y">:quantity]}")
    {accepted: true, events: [class="s">"StockReserved(#{reservation[class="y">:remaining]} left)", class="s">"OrderPlaced(##{entry[class="y">:position]})"]}
  end,
  inputs: {
    sku: {type: class="s">"string", required: true},
    quantity: {type: class="s">"number", required: true}
  },
  outputs: {
    accepted: {type: class="s">"boolean", required: true},
    events: {type: class="s">"array", required: true}
  }
)

registry.compose(
  class="s">"RestockShelf", class="s">"Add stock for a SKU", class="s">"1.0.0",
  [{name: class="s">"record_entry", version: class="s">"1.0.0"}],
  lambda do |(record), command|
    STOCK[command[class="y">:sku]] += command[class="y">:quantity]
    entry = record.execute(entry: class="s">"restock #{command[class="y">:sku]} +#{command[class="y">:quantity]}")
    {accepted: true, events: [class="s">"ShelfRestocked(##{entry[class="y">:position]})"]}
  end,
  inputs: {sku: {type: class="s">"string", required: true}, quantity: {type: class="s">"number", required: true}},
  outputs: {accepted: {type: class="s">"boolean", required: true}, events: {type: class="s">"array", required: true}}
)

# --- the bus: dispatch is validation + routing, nothing else ----------------
def dispatch(registry, command_name, payload)
  provider = registry.get_provider(command_name) or
    return {accepted: false, events: [class="s">"UnknownCommand: #{command_name}"]}
  provider.execute(payload)
rescue Agentic:class="y">:Errors:class="y">:ValidationError => e
  {accepted: false,
   events: e.violations.map { |key, msgs| class="s">"CommandRejected: #{key} #{Array(msgs).join(", class="s">")}" }}
end

COMMANDS = [
  [class="s">"PlaceOrder", {sku: class="s">"widget", quantity: 3}],
  [class="s">"PlaceOrder", {sku: class="s">"widget", quantity: class="s">"many"}], # violates the contract
  [class="s">"RestockShelf", {sku: class="s">"widget", quantity: 5}],
  [class="s">"PlaceOrder", {sku: class="s">"widget", quantity: 13}], # violates the business rule
  [class="s">"ShipRocket", {to: class="s">"the moon"}] # nobody handles this
].freeze

puts class="s">"COMMAND BUS"
puts
COMMANDS.each do |name, payload|
  result = dispatch(registry, name, payload)
  status = result[class="y">:accepted] ? class="s">"ACCEPTED" : class="s">"REJECTED"
  puts format(class="s">"  %-8s %s(%s)", status, name, payload.map { |k, v| class="s">"#{k}: #{v.inspect}" }.join(class="s">", "))
  result[class="y">:events].each { |event| puts class="s">"           -> #{event}" }
end

puts
puts class="s">"ledger: #{LEDGER.size} entries | widget stock: #{STOCK["widgetclass="s">"]}"
puts
puts class="s">"note the two different REJECTED shapes: the contract rejected"
puts class="s">"'many' before any handler ran; the business rule rejected 13 after"
puts class="s">"checking the shelf. types stop nonsense, domains stop mistakes."