Class: LittleGhost::Workflow

Inherits:
Assembly show all
Defined in:
lib/little_ghost/workflow.rb

Overview

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 route a difficult request through research before the responder writes the caller-visible answer:

class ResponseWorkflow < LittleGhost::Workflow
private

def perform
  route = invoke(RouterAgent).output
  return invoke(CustomerSupportAgent) unless route["research"]

  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 # => "Transfer 481 is waiting for the receiving bank."

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.

Every child inherits input, history, settings, cancellation, deadline, template paths, and trace parentage unless invoke overrides its input. 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 an invocation twice raises ProtocolError. Each child assembly closes after its execution attempt; cleanup failures surface from that attempt. Closing the Workflow closes its lightweight invocation wrappers in reverse declaration order. Composition errors emit an invocation_error event and then re-raise.

Defined Under Namespace

Classes: Invocation

Constant Summary

Constants inherited from Assembly

Assembly::MAX_STEP_EVENTS, Assembly::MAX_STEP_EVENT_BYTES, Assembly::MAX_STEP_OUTPUT_BYTES

Instance Attribute Summary collapse

Attributes inherited from Assembly

#sandbox, #workspace

Instance Method Summary collapse

Methods inherited from Assembly

#as_tool, ask, #ask, assembly_id, assembly_kind, #build_run, #call, definition, description, #entrypoint_name, #interrupt, #interrupt_response, #start_execution, stream_ask, #stream_ask, to_builder, validate_step_policy!

Methods included from Support::ClassAttributes

#class_attribute, included

Constructor Details

#initialize(run: nil, runtime: nil) ⇒ Workflow

:nodoc:



132
133
134
135
136
137
138
 
# File 'lib/little_ghost/workflow.rb', line 132

def initialize(run: nil, runtime: nil) # :nodoc:
  super(run:, runtime:, standalone: run.nil?)
  @mutex = Mutex.new
  @closed = false
  @started = false
  @invocations = []
end

Instance Attribute Details

#run ⇒ Object (readonly)

Owning run and the runtime used to resolve workflow agents.



130
131
132
 
# File 'lib/little_ghost/workflow.rb', line 130

def run
  @run
end

#runtime ⇒ Object (readonly)

Owning run and the runtime used to resolve workflow agents.



130
131
132
 
# File 'lib/little_ghost/workflow.rb', line 130

def runtime
  @runtime
end

Instance Method Details

#close ⇒ Object

Closes all declared invocations in reverse order.

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



238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/little_ghost/workflow.rb', line 238

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

#prompt_locals ⇒ Object

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



142
 
# File 'lib/little_ghost/workflow.rb', line 142

def prompt_locals = {}

#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) ⇒ Object

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.

Raises:

  • (ArgumentError) 


149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/little_ghost/workflow.rb', line 149

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