class LittleGhost::Workflow

Coordinates Assembly participants with ordinary Ruby control flow.

A workflow is an Assembly whose perform method controls ordering, branching, parallel work, and local variables. Each participant may be an Agent or another coordinated Assembly. The workflow consumes intermediate answers and streams one final participant response.

A support workflow can guarantee that research happens before the responder writes the caller-visible answer:

class ResponseWorkflow < LittleGhost::Workflow
  private

  def perform
    evidence = invoke(ResearchAgent).output
    invoke CustomerSupportAgent, input: <<~PROMPT
      #{input.text}

      Research:
      #{evidence}
    PROMPT
  end
end

run = ResponseWorkflow.ask("Why is transfer 481 pending?")
run.response
# One possible response: Transfer 481 is waiting for the receiving bank.

Call a named Workflow with ask for its final Run, or the streaming entrypoint for live events.

invoke returns a lazy Workflow::Invocation. Reading output consumes an intermediate invocation and returns RunResult#output; perform must return its final invocation without consuming it so those events reach the caller. Intermediate usage is added to the final result.

A child receives the Workflow input unless invoke supplies another one. It also inherits history, settings, cancellation, deadline, template paths, and the parent tracing relationship. JSON-like context is copied for each child, preventing one intermediate Agent from mutating a sibling’s state. Non-JSON-like workflow context raises ArgumentError.

A Workflow instance streams once. Returning the wrong value, returning an already consumed invocation, or consuming one twice raises ProtocolError. A composition error fails the owning top-level Run. Each child Assembly closes after its attempt, and a cleanup failure raises from that attempt.

Attributes

run [R]

Run that owns this run-scoped Workflow.

runtime [R]

Runtime used to resolve child Assemblies.

Public Instance Methods

close ()

Source

# File lib/little_ghost/workflow.rb, line 241
def close
  invocations = @mutex.synchronize do
    return if @closed

    @closed = true
    @invocations.reverse
  end
  errors = []
  invocations.each do |invocation|
    invocation.close
  rescue => error
    errors << error
  end
  raise errors.first if errors.any?
end

Closes all declared invocations in reverse order.

The operation is idempotent, attempts every close, and raises the first cleanup failure.

prompt_locals ()

Source

# File lib/little_ghost/workflow.rb, line 145
def prompt_locals = {}

Additional prompt locals shared by agents invoked from the workflow. Subclasses may override this hook.

stream (input = nil, history: nil, context: nil, cancellation_token: Support::CancellationToken.new, deadline: nil, settings: nil, template_locals: nil, template_paths: nil, parent_operation_id: nil, checkpoint: nil)

Source

# File lib/little_ghost/workflow.rb, line 152
def stream(
  input = nil,
  history: nil,
  context: nil,
  cancellation_token: Support::CancellationToken.new,
  deadline: nil,
  settings: nil,
  template_locals: nil,
  template_paths: nil,
  parent_operation_id: nil,
  checkpoint: nil
)
  raise ArgumentError, "input is required" if input.nil?

  if standalone?
    return build_run(entrypoint_payload(input, {
      history:,
      context:,
      settings:,
      template_paths:,
      deadline_at: deadline,
      cancellation_token:
    }.compact)).each
  end

  @mutex.synchronize do
    raise Error, "workflow is already closed" if @closed
    raise Error, "workflow instances can only be streamed once" if @started

    @started = true
    @input = input.is_a?(Message) ? input : Message.new(role: :user, content: input)
    @history = normalize_history(history)
    @context = context || {}
    @cancellation_token = cancellation_token
    @deadline = deadline
    @settings = settings || {}
    @template_locals = template_locals || {}
    @template_paths = template_paths || []
    @parent_operation_id = parent_operation_id
    @checkpoint = checkpoint
    @intermediate_usage = Usage.new
    @workflow_steps = []
    @workflow_events = nil
  end

  Enumerator.new do |events|
    error_emitted = false
    observed_usage = nil
    @workflow_events = events
    ensure_open!
    final_invocation = perform
    unless final_invocation.is_a?(Invocation) && !final_invocation.consumed?
      raise ProtocolError, "#{self.class} must return its final invoke from perform"
    end

    final_invocation.each(checkpoint: @checkpoint) do |event|
      error_emitted = true if event.type == :invocation_error
      event = aggregate_usage(event)
      observed_usage = case event.type
      when :invocation_stop
        event.data.fetch(:result).usage
      when :invocation_error
        event.data[:usage] || observed_usage
      when :assembly_step_error
        event.data[:usage] || observed_usage
      else
        observed_usage
      end
      events << event
    end
  rescue => error
    unless error_emitted
      events << StreamEvent.build(
        :invocation_error,
        error:,
        usage: observed_usage || workflow_usage,
        metadata: {}
      )
    end
    raise
  ensure
    @workflow_events = nil
  end
end

Streams the workflow once as StreamEvent objects.

perform must return a final, unconsumed Workflow::Invocation. The returned Enumerator is lazy, but calling stream reserves the single-use workflow instance even when enumeration has not started yet.

Private Instance Methods

invoke (assembly, as: nil, input: self.input, history: self.history, context: self.context, timeout: nil, retries: 0, retry_on: nil, retry_delay: 0)

Source

# File lib/little_ghost/workflow.rb, line 278
def invoke(
  assembly,
  as: nil,
  input: self.input,
  history: self.history,
  context: self.context,
  timeout: nil,
  retries: 0,
  retry_on: nil,
  retry_delay: 0
)
  participant = as || assembly_identity(assembly)
  invocation = Invocation.new(
    reference: assembly,
    participant:,
    input:,
    history:,
    context: isolated_state(context),
    policies: {timeout:, retries:, retry_on:, retry_delay:},
    owner: self
  )
  @mutex.synchronize do
    raise Error, "workflow is already closed" if @closed

    @invocations << invocation
  end
  invocation
end

Creates a lazy invocation for assembly.

Intermediate calls may use output; the final call must be returned from perform without being consumed. as supplies the participant name used in steps and telemetry. Retries default to zero; a positive retries value requires explicit exception classes in retry_on.

parallel (*invocations, max_concurrency: 8)

Source

# File lib/little_ghost/workflow.rb, line 313
def parallel(*invocations, max_concurrency: 8)
  raise ArgumentError, "parallel requires at least one invocation" if invocations.empty?
  unless invocations.all? { |invocation| invocation.is_a?(Invocation) && !invocation.consumed? }
    raise ArgumentError, "parallel accepts unconsumed workflow invocations"
  end

  token = @cancellation_token.child
  queue = SizedQueue.new(1_000)
  worker = task_runner.spawn do
    results = Support::Executor.new(max_concurrency:, runner: task_runner).map(
      invocations,
      cancellation_token: token,
      on_result: ->(_index, execution) { record_workflow_steps(execution.fetch(:steps)) }
    ) do |invocation|
      invocation.each do |event|
        if event.type.to_s.start_with?("assembly_")
          enqueue_assembly_event(queue, [:event, event], token)
        end
      end
      {output: invocation.result&.output, steps: invocation.instance_variable_get(:@steps)}
    end
    enqueue_assembly_event(queue, [:done, results], token)
  rescue => error
    token.cancel
    enqueue_assembly_terminal(queue, [:error, error])
  end
  executions = loop do
    type, value = queue.pop
    emit_workflow_event(value) if type == :event
    raise value if type == :error
    break value if type == :done
  end
  executions.map { |execution| execution.fetch(:output) }
ensure
  token&.cancel
  worker&.wait
end

Consumes independent invocations concurrently and returns their outputs in declaration order.

max_concurrency bounds active child executions. A child failure cancels siblings cooperatively before the error is raised.

perform ()

Source

# File lib/little_ghost/workflow.rb, line 267
def perform
  raise AbstractMethodError, "#{self.class} must implement #perform"
end

Implements the composition and returns its final unconsumed invocation. Subclasses must override this hook.