Class: Ace::Assign::Organisms::AssignmentExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/ace/assign/organisms/assignment_executor.rb

Overview

Orchestrates workflow operations on the work queue.

Implements the state machine for queue operations: start → advance → complete (with fail/add/retry branches)

Constant Summary collapse

DEFAULT_DYNAMIC_STEP_INSTRUCTIONS = 
"Complete this step and finish with: ace-assign finish --message report.md".freeze
PROJECT_ROOT_SIGNAL = 
"project_root".freeze
CATALOG_SIGNAL = 
"catalog".freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(cache_base: nil, skill_source_resolver: nil, step_catalog: nil) ⇒ AssignmentExecutor

Returns a new instance of AssignmentExecutor.



40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 40

def initialize(cache_base: nil, skill_source_resolver: nil, step_catalog: nil)
  @assignment_manager = Molecules::AssignmentManager.new(cache_base: cache_base)
  @queue_scanner = Molecules::QueueScanner.new
  @step_writer = Molecules::StepWriter.new
  @skill_source_resolver = skill_source_resolver || Molecules::SkillAssignSourceResolver.new
  @step_catalog = nil
  @step_catalog_from_fixture = step_catalog
  @step_catalog_from_fixture_set = !step_catalog.nil?
  @step_catalog_loaded = false
  @step_renumberer = Molecules::StepRenumberer.new(
    step_writer: @step_writer,
    queue_scanner: @queue_scanner
  )
end

Instance Attribute Details

#assignment_manager ⇒ Object (readonly)

Returns the value of attribute assignment_manager.



18
19
20
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 18

def assignment_manager
  @assignment_manager
end

#queue_scanner ⇒ Object (readonly)

Returns the value of attribute queue_scanner.



18
19
20
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 18

def queue_scanner
  @queue_scanner
end

#skill_source_resolver ⇒ Object (readonly)

Returns the value of attribute skill_source_resolver.



18
19
20
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 18

def skill_source_resolver
  @skill_source_resolver
end

#step_renumberer ⇒ Object (readonly)

Returns the value of attribute step_renumberer.



18
19
20
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 18

def step_renumberer
  @step_renumberer
end

#step_writer ⇒ Object (readonly)

Returns the value of attribute step_writer.



18
19
20
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 18

def step_writer
  @step_writer
end

Class Method Details

.cache_store ⇒ Object 



25
26
27
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 25

def cache_store
  @cache_store ||= { step_catalog_cache: {} }
end

.clear_caches! ⇒ Object 



21
22
23
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 21

def clear_caches!
  @cache_store = { step_catalog_cache: {} }
end

Instance Method Details

#add(name, instructions, after: nil, as_child: false, added_by: nil, extra: {}, fork_root: nil) ⇒ Hash

Add a new step dynamically

Parameters:

  • name (String) —

    Step name

  • instructions (String) —

    Step instructions

  • after (String, nil) (defaults to: nil) —

    Insert after this step number (optional)

  • as_child (Boolean) (defaults to: false) —

    Insert as child of ‘after’ step (default: false, sibling)

Returns:

  • (Hash) —

    Result with new step

Raises:



273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 273

def add(name, instructions, after: nil, as_child: false, added_by: nil, extra: {}, fork_root: nil)
  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  step_name = name.to_s.strip
  raise Error, "Step name cannot be empty." if step_name.empty?

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  after, as_child = normalize_scoped_insertion_anchor(state, after: after, as_child: as_child, fork_root: fork_root)
  existing_numbers = queue_scanner.step_numbers(assignment.steps_dir)

  # Validate --after step exists
  if after && !existing_numbers.include?(after)
    raise StepErrors::NotFound, "Step #{after} not found. Available steps: #{existing_numbers.join(", ")}"
  end

  new_number, renumbered = calculate_insertion_point(
    after: after,
    as_child: as_child,
    state: state,
    existing_numbers: existing_numbers
  )

  # Renumber existing steps if needed (uses molecule with rollback support)
  if renumbered.any?
    step_renumberer.renumber(assignment.steps_dir, renumbered)
    # Refresh existing numbers after renumbering
    queue_scanner.step_numbers(assignment.steps_dir)
  end

  # Build added_by metadata for audit trail
  added_by ||= if after && as_child
    "child_of:#{after}"
  elsif after
    "injected_after:#{after}"
  else
    "dynamic"
  end

  extra_frontmatter = normalize_batch_extra_fields(extra)

  # Create new step file with correct status
  step_writer.create(
    steps_dir: assignment.steps_dir,
    number: new_number,
    name: step_name,
    instructions: instructions,
    status: :pending,
    added_by: added_by,
    parent: as_child ? after : nil,
    extra: extra_frontmatter
  )

  rebalance_after_child_injection(assignment: assignment, state: state, parent_number: after) if as_child && after

  # Update assignment timestamp
  assignment_manager.update(assignment)

  # Return updated state
  new_state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  new_step = new_state.steps.find { |s| s.number == new_number }

  {
    assignment: assignment,
    state: new_state,
    added: new_step,
    renumbered: renumbered
  }
end

#add_batch(steps:, after: nil, as_child: false, source_file: nil, fork_root: nil) ⇒ Hash

Note:

Structural validation is performed for the full batch before any writes. Runtime I/O failures can still interrupt insertion after partial writes.

Add multiple steps dynamically from a pre-parsed steps array.

Parameters:

  • steps (Array<Hash>) —

    Step definitions loaded from YAML

  • after (String, nil) (defaults to: nil) —

    Insert after this step number

  • as_child (Boolean) (defaults to: false) —

    Insert as children of after

  • source_file (String, nil) (defaults to: nil) —

    Source YAML path (for added_by audit metadata)

Returns:

  • (Hash) —

    Result with added steps and final state

Raises:



352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 352

def add_batch(steps:, after: nil, as_child: false, source_file: nil, fork_root: nil)
  unless steps.is_a?(Array) && steps.any?
    source_label = source_file.to_s.strip.empty? ? "batch input" : source_file
    raise Error, "No steps defined in #{source_label}"
  end

  if as_child && (after.nil? || after.to_s.strip.empty?)
    raise Error, "Child insertion requires an after step reference."
  end

  prevalidate_batch_trees!(steps)

  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  after, as_child = normalize_scoped_insertion_anchor(state, after: after, as_child: as_child, fork_root: fork_root)

  added_steps = []
  renumbered = []
  sibling_cursor = after

  steps.each_with_index do |step_config, index|
    inserted = insert_batch_step_tree(
      step_config,
      after: as_child ? after : sibling_cursor,
      as_child: as_child,
      added_by: nil,
      location: "steps[#{index}]"
    )
    added_steps.concat(inserted[:added])
    renumbered.concat(inserted[:renumbered])
    sibling_cursor = inserted[:root_number] unless as_child
  end

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  {
    assignment: assignment,
    state: state,
    added: added_steps,
    renumbered: renumbered.uniq
  }
end

#advance(report_path, fork_root: nil) ⇒ Hash

Complete current step with report content from a file.

Parameters:

  • report_path (String) —

    Path to report file

  • fork_root (String, nil) (defaults to: nil) —

    Optional subtree root to constrain advancement

Returns:

  • (Hash) —

    Result with updated state

Raises:



233
234
235
236
237
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 233

def advance(report_path, fork_root: nil)
  raise ConfigErrors::NotFound, "Report file not found: #{report_path}" unless File.exist?(report_path)

  finish_step(report_content: File.read(report_path), fork_root: fork_root)
end

#fail(message, fork_root: nil) ⇒ Hash

Mark current step as failed

Parameters:

  • message (String) —

    Error message

Returns:

  • (Hash) —

    Result with updated state

Raises:



243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 243

def fail(message, fork_root: nil)
  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  current = find_target_step_for_fail(state, fork_root)
  raise Error, "No step currently active. Try 'ace-assign add' to add a new step or 'ace-assign retry' to retry a failed step." unless current

  # Mark step as failed
  step_writer.mark_failed(current.file_path, error_message: message)

  # Update assignment timestamp
  assignment_manager.update(assignment)

  # Return updated state (no automatic advancement after failure)
  new_state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  {
    assignment: assignment,
    state: new_state,
    failed: current
  }
end

#finish_step(report_content:, step_number: nil, fork_root: nil) ⇒ Hash

Finish an active step and update queue state.

Parameters:

  • report_content (String) —

    Completion report content

  • step_number (String, nil) (defaults to: nil) —

    Optional active step number to finish

  • fork_root (String, nil) (defaults to: nil) —

    Optional subtree root to constrain advancement

Returns:

  • (Hash) —

    Result with completed step and updated state

Raises:



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
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 190

def finish_step(report_content:, step_number: nil, fork_root: nil)
  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  current = find_target_step_for_finish(state, step_number, fork_root)
  raise Error, "No step currently active. Try 'ace-assign start' or 'ace-assign retry'." unless current

  # Enforce hierarchy: cannot mark parent as done with incomplete children
  if state.has_incomplete_children?(current.number)
    incomplete = state.children_of(current.number).reject { |c| c.status == :done }
    incomplete_nums = incomplete.map(&:number).join(", ")
    raise Error, "Cannot complete step #{current.number}: has incomplete children (#{incomplete_nums}). Complete children first or use 'ace-assign fail' to mark as failed."
  end

  # Mark current step as done
  step_writer.mark_done(current.file_path, report_content: report_content, reports_dir: assignment.reports_dir)

  # Rescan to get updated state after marking done
  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)

  # Auto-complete parent steps if all their children are done
  auto_complete_parents(state, assignment)

  # Re-scan to get fresh state after auto-completions
  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)

  assignment_manager.update(assignment)

  new_state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  {
    assignment: assignment,
    state: new_state,
    completed: current,
    current: new_state.current
  }
end

#retry_step(step_ref, fork_root: nil) ⇒ Hash

Retry a failed step (creates new step linked to original)

Parameters:

  • step_ref (String) —

    Step number or reference to retry

Returns:

  • (Hash) —

    Result with new retry step

Raises:



400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 400

def retry_step(step_ref, fork_root: nil)
  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)

  # Find the step to retry
  original = state.find_by_number(step_ref.to_s)
  raise StepErrors::NotFound, "Step #{step_ref} not found in queue" unless original
  validate_retry_scope!(state, original, fork_root)

  # Get existing numbers
  existing_numbers = queue_scanner.step_numbers(assignment.steps_dir)

  # Insert after all current steps (at end of queue before pending)
  # Find last done or failed step
  scoped_state = scoped_retry_state(state, assignment: assignment, fork_root: fork_root)
  base_number = if scoped_state.current
    scoped_state.current.number
  elsif scoped_state.last_done
    scoped_state.last_done.number
  else
    original.number
  end

  new_number = Atoms::NumberGenerator.next_after(base_number, existing_numbers)

  # Create retry step with link to original
  step_writer.create(
    steps_dir: assignment.steps_dir,
    number: new_number,
    name: original.name,
    instructions: original.instructions,
    status: :pending,
    added_by: "retry_of:#{original.number}"
  )

  # Update assignment timestamp
  assignment_manager.update(assignment)

  # Return updated state
  new_state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  retry_step = new_state.steps.find { |s| s.number == new_number }

  {
    assignment: assignment,
    state: new_state,
    retry: retry_step,
    original: original
  }
end

#start(config_path, parent_id: nil) ⇒ Hash

Start a new workflow assignment from config file

Parameters:

  • config_path (String) —

    Path to job.yaml config

  • parent_id (String, nil) (defaults to: nil) —

    Parent assignment ID for hierarchy linking

Returns:

  • (Hash) —

    Result with assignment and first step

Raises:



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 60

def start(config_path, parent_id: nil)
  raise ConfigErrors::NotFound, "Config file not found: #{config_path}" unless File.exist?(config_path)

  config = YAML.safe_load_file(config_path, permitted_classes: [Time, Date])

  assignment_config = config["assignment"] || {}
  steps_config = config["steps"] || []

  raise Error, "No steps defined in config" if steps_config.empty?

  # Enrich steps using declared workflow/skill assign metadata.
  steps_config = enrich_declared_sub_steps(steps_config)

  # Expand sub-step declarations into batch parent + child steps
  steps_config = expand_sub_steps(steps_config)
  steps_config = materialize_skill_backed_steps(steps_config)

  # Create assignment
  assignment = assignment_manager.create(
    name: assignment_config["name"] || File.basename(config_path, ".yaml"),
    description: assignment_config["description"],
    source_config: config_path,
    parent: parent_id
  )

  # Create initial step files
  # Steps may have pre-assigned numbers (from expansion) or need auto-numbering
  steps_config.each_with_index do |step, index|
    # Use pre-assigned number if present, otherwise generate from index
    number = step["number"] || Atoms::NumberGenerator.from_index(index)
    extra = step.reject { |k, _| %w[name instructions number].include?(k) }
    step_writer.create(
      steps_dir: assignment.steps_dir,
      number: number,
      name: step["name"],
      instructions: normalize_instructions(step["instructions"]),
      status: :pending,
      extra: extra
    )
  end

  # Archive source config into task's steps directory and update assignment metadata
  archived_path = archive_source_config(config_path, assignment.id)
  assignment = Models::Assignment.new(
    id: assignment.id,
    name: assignment.name,
    description: assignment.description,
    created_at: assignment.created_at,
    updated_at: assignment.updated_at,
    source_config: archived_path,
    cache_dir: assignment.cache_dir,
    parent: assignment.parent
  )
  assignment_manager.update(assignment)

  # Return result
  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  {
    assignment: assignment,
    state: state,
    current: state.current
  }
end

#start_step(step_number: nil, fork_root: nil) ⇒ Hash

Start a pending step.

Rules:

  • Starts an explicit pending target when provided

  • Otherwise starts the next workable pending step

Parameters:

  • step_number (String, nil) (defaults to: nil) —

    Optional target step number

  • fork_root (String, nil) (defaults to: nil) —

    Optional subtree root scope

Returns:

  • (Hash) —

    Result with started step and updated state

Raises:



148
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
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 148

def start_step(step_number: nil, fork_root: nil)
  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)

  fork_root = fork_root&.strip
  target_step = if step_number && !step_number.to_s.strip.empty?
    find_target_step_for_start(state, step_number, fork_root)
  elsif fork_root && !fork_root.empty?
    raise StepErrors::NotFound, "Subtree root #{fork_root} not found in assignment." unless state.find_by_number(fork_root)
    state.next_workable_in_subtree(fork_root)
  else
    state.next_workable
  end

  unless target_step
    if fork_root && !fork_root.empty?
      raise StepErrors::InvalidState, "No pending workable step found in subtree #{fork_root}."
    end
    raise StepErrors::InvalidState, "No pending workable step found."
  end

  validate_start_activation!(state, target_step, fork_root: fork_root)
  step_writer.mark_active(target_step.file_path)
  assignment_manager.update(assignment)

  new_state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  {
    assignment: assignment,
    state: new_state,
    started: new_state.find_by_number(target_step.number),
    current: new_state.current
  }
end

#status ⇒ Hash

Get current assignment and queue state

Returns:

  • (Hash) —

    Result with assignment and state

Raises:



127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/ace/assign/organisms/assignment_executor.rb', line 127

def status
  assignment = assignment_manager.find_active
  raise AssignmentErrors::NoActive, "No active assignment. Use 'ace-assign create --yaml <job.yaml>' to begin." unless assignment

  state = queue_scanner.scan(assignment.steps_dir, assignment: assignment)
  {
    assignment: assignment,
    state: state,
    current: state.current
  }
end