The Live Dashboard

The Live Dashboard: lifecycle hooks publish events onto an Async::Queue; a consumer task IN THE SAME REACTOR renders the plan's state as it changes. This is the "streaming observability" the architecture documents promise, built from a queue and the hooks that already exist - about thirty structural lines.

source on github

awaiting its first recording

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

LIVE DASHBOARD (plan and renderer sharing one reactor, concurrency 2)

    0ms  ~ queued   resize:thumbnails
    0ms  > running  resize:thumbnails
    0ms  ~ queued   transcode:video
    0ms  > running  transcode:video
    0ms  ~ queued   extract:captions
  151ms  + done     resize:thumbnails    (ran 150ms)
  151ms  > running  extract:captions
  253ms  + done     extract:captions     (ran 101ms)
  253ms  ~ queued   compose:preview
  253ms  > running  compose:preview
  300ms  + done     transcode:video      (ran 300ms)
  374ms  + done     compose:preview      (ran 120ms)
  374ms  ~ queued   publish:episode
  374ms  > running  publish:episode
  424ms  + done     publish:episode      (ran 50ms)
  425ms  = plan completed in 424ms

every line above was printed WHILE the plan ran - the hooks are a
live event stream, not a post-mortem log

source

# frozen_string_literal: true

# The Live Dashboard: lifecycle hooks publish events onto an
# Async::Queue; a consumer task IN THE SAME REACTOR renders the plan's
# state as it changes. This is the "streaming observability" the
# architecture documents promise, built from a queue and the hooks
# that already exist - about thirty structural lines.
#
#   bundle exec ruby examples/live_dashboard.rb
#
# Runs offline; watch the states flip while the plan executes.

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

WORK = {
  class="s">"resize:thumbnails" => {sleep: 0.15, deps: []},
  class="s">"transcode:video" => {sleep: 0.30, deps: []},
  class="s">"extract:captions" => {sleep: 0.10, deps: []},
  class="s">"compose:preview" => {sleep: 0.12, deps: [class="s">"resize:thumbnails", class="s">"extract:captions"]},
  class="s">"publish:episode" => {sleep: 0.05, deps: [class="s">"compose:preview", class="s">"transcode:video"]}
}.freeze

events = Async:class="y">:Queue.new

hooks = {
  before_task_execution: ->(task_id:, task:) { events.enqueue([class="y">:queued, task.description]) },
  after_agent_build: ->(task_id:, task:, agent:, build_duration:) { events.enqueue([class="y">:running, task.description]) },
  after_task_success: ->(task_id:, task:, result:, duration:) {
    events.enqueue([class="y">:done, task.description, duration])
  },
  plan_completed: ->(plan_id:, status:, execution_time:, tasks:, results:) {
    events.enqueue([class="y">:plan_done, status, execution_time])
  }
}

orchestrator = Agentic:class="y">:PlanOrchestrator.new(concurrency_limit: 2, lifecycle_hooks: hooks)
tasks = {}
WORK.each do |name, spec|
  task = Agentic:class="y">:Task.new(
    description: name,
    agent_spec: {class="s">"name" => name, class="s">"instructions" => class="s">"process"},
    payload: spec[class="y">:sleep]
  )
  tasks[name] = task
  orchestrator.add_task(task, spec[class="y">:deps].map { |d| tasks.fetch(d) }, agent: ->(t) {
    sleep(t.payload)
    class="y">:ok
  })
end

started = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC)
stamp = -> { format(class="s">"%5dms", (Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - started) * 1000) }

puts class="s">"LIVE DASHBOARD (plan and renderer sharing one reactor, concurrency 2)"
puts

Sync do |host|
  # The renderer: a sibling task consuming the event stream live
  renderer = host.async do
    loop do
      event = events.dequeue
      case event.first
      when class="y">:queued then puts class="s">"#{stamp.call}  ~ queued   #{event[1]}"
      when class="y">:running then puts class="s">"#{stamp.call}  > running  #{event[1]}"
      when class="y">:done then puts format(class="s">"%s  + done     %-20s (ran %dms)", stamp.call, event[1], event[2] * 1000)
      when class="y">:plan_done
        puts format(class="s">"%s  = plan %s in %dms", stamp.call, event[1], event[2] * 1000)
        break
      end
    end
  end

  orchestrator.execute_plan
  renderer.wait
end

puts
puts class="s">"every line above was printed WHILE the plan ran - the hooks are a"
puts class="s">"live event stream, not a post-mortem log"