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 selects one participant result or returns a value computed from intermediate answers.

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 result runs the child and returns its RunResult; output returns RunResult#output. Repeated reads reuse the completed result. Return an invocation to select its conversation, state, and structured result as the workflow answer, or return a String or JSON-compatible value computed in Ruby. Every step contributes usage.

All participating Agents publish live :agent_stream progress to the owning Run, regardless of how results are accessed. Selecting a result does not replay its progress. Applications choose which sources to show the caller; see Run for how to identify participants and choose which data to share.

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 nil, an unsupported value, an invocation owned by another workflow, or a failed invocation 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 244
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 161
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 172
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|
    @workflow_events = events
    ensure_open!
    check_execution!
    final_value = perform
    check_execution!
    if final_value.is_a?(Invocation)
      unless owned_invocation?(final_value)
        raise ProtocolError, "#{self.class} returned an invocation owned by another workflow"
      end
      result = final_value.result
      emit_completed_result(result.with(usage: workflow_usage, steps: workflow_steps), events)
    else
      emit_direct_result(final_value, events)
    end
  rescue => error
    events << StreamEvent.build(:invocation_error, error:, usage: workflow_usage, metadata: {})
    raise
  ensure
    @workflow_events = nil
  end
end

Streams the workflow once as StreamEvent objects.

perform may return an invocation owned by this workflow, a String, or a JSON-compatible value. The final :invocation_stop contains the selected or computed result; it does not replay text events. A selected invocation runs only if it has not started, and checkpoints its conversation and state. Usage includes every workflow step. Cancellation and deadlines are checked before child execution and final publication. The returned Enumerator is lazy, but calling stream reserves the single-use instance before enumeration starts.

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 283
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.

Read result or output to execute the child and inspect its answer. Return the invocation from perform to select that result without repeating work or replaying progress. as names the participant 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 318
def parallel(*invocations, max_concurrency: 8)
  raise ArgumentError, "parallel requires at least one invocation" if invocations.empty?
  unless invocations.all? { |invocation| owned_invocation?(invocation) && !invocation.consumed? }
    raise ArgumentError, "parallel accepts unconsumed invocations owned by this workflow"
  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|
      result = invocation.each do |event|
        if event.type.to_s.start_with?("assembly_")
          enqueue_assembly_event(queue, [:event, event], token)
        end
      end
      {output: result.output, steps: result.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 271
def perform
  raise AbstractMethodError, "#{self.class} must implement #perform"
end

Implements the composition and returns its final invocation or a directly computed String or JSON-compatible value. Subclasses must override this hook.