Class: Nexo::Workflow

Inherits:
Object
  • Object
show all
Defined in:
lib/nexo/workflow.rb

Overview

A finite-job lifecycle primitive. Subclass Workflow, implement #call(payload), and run it with MyWorkflow.run(payload) to get back a persisted run record carrying a stable runId, status, payload, result, error, and an ordered, inspectable event log.

class SummarizeDocument < Nexo::Workflow
  def call(payload)
    emit(:started, doc_id: payload[:doc_id])
    summary = payload[:text].to_s.slice(0, 280)
    emit(:summarized, length: summary.length)
    { summary: summary }
  end
end

run = SummarizeDocument.run(doc_id: 123, text: "Long text…")
run.status  # => "done"
run.result  # => { "summary" => "Long text…" }  (string keys after round-trip)

#call receives a symbol-keyed payload, but run.payload and run.result read back string-keyed (top-level) in both backends — the Hash keys are stringified before storage, matching what the ActiveRecord json column would round-trip to, so a single workflow drives either store consistently.

The lifecycle records failures but never swallows them: a raising #call leaves the run "failed" with error set and re-raises. (This is the opposite of a Nexo tool failure, which returns { error: … } and never raises into the agent loop.) An orphaned "running" run (crashed worker) is abandoned — sweep it with ::reconcile_interrupted!.

A run can pause and continue on purpose, though (Spec 13): call #suspend! mid-#call to leave the run "suspended" (not failed), then ::resume / ::resume_later to re-enter #call from the top. #checkpoint guards the expensive/side-effectful steps so resume skips already-paid-for work — see the "Durable workflows" README section for the honest resume semantics.

Defined Under Namespace

Classes: Suspended

Constant Summary collapse

RESERVED_STATE_KEYS =

State keys Nexo reserves for lifecycle metadata, never a caller's data: "__suspend__" (suspend reason/resume_key — Spec 13), "__approval__" (pending approval call — Spec 16), and "__buffer_events__" (the persisted buffering choice — Spec 5). ::cleared_state strips these when a run reaches "done"; #checkpoint_all refuses a step named after any of them (Spec 21).

%w[__suspend__ __approval__ __buffer_events__].freeze
ARTIFACTS_DIR =

Where #artifact drops its in-sandbox copy, relative to the sandbox root.

"artifacts"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(run, buffer_events: false) ⇒ Workflow

Binds the instance to its persisted run. buffer_events: accumulates emitted events in memory and flushes them once at the end of the run rather than persisting each immediately. Prefer the ::run / ::resume entry points over constructing directly.



436
437
438
439
440
441
442
443
444
445
 
# File 'lib/nexo/workflow.rb', line 436

def initialize(run, buffer_events: false)
  @run = run
  @buffer_events = buffer_events
  @event_buffer = []
  # Serializes the read-current → merge → assign → save_state! sequence across
  # the concurrent fibers #checkpoint_all runs each step on (Spec 21). A plain
  # Mutex is fiber-safe under the async reactor (Group 0 verified it serializes
  # without stalling); #checkpoint (singular, no concurrency) does not use it.
  @checkpoint_mutex = Mutex.new
end

Class Method Details

.agent(klass = nil) ⇒ Object

The Agent subclass this workflow drives (Spec 8). Follows the same read-vs-write ivar convention as ::sandbox/::cwd: with no argument it reads (default nil — a workflow need not drive an agent); with one it sets. Consumed by #run_agent, which binds the agent to the run's shared sandbox. There is no per-call override — this macro is the only source.



98
99
100
 
# File 'lib/nexo/workflow.rb', line 98

def agent(klass = nil)
  klass.nil? ? @agent : (@agent = klass)
end

.artifact_body(art) ⇒ Object

The bytes of a recorded artifact, decoding the Base64 form when present. The counterpart to the encoding #artifact applies; use it rather than reading art["content"] directly, which is Base64 text for binary.



693
694
695
696
 
# File 'lib/nexo/workflow.rb', line 693

def self.artifact_body(art)
  body = art["content"].to_s
  (art["encoding"] == "base64") ? body.unpack1("m") : body
end

.cwd(value = nil) ⇒ Object

The working directory used when this workflow's sandbox is :local. The Dir.pwd default is evaluated at read time (in the macro body), so the working directory is captured when the sandbox is actually resolved, not at class-definition time. Never read for a :virtual workflow.



89
90
91
 
# File 'lib/nexo/workflow.rb', line 89

def cwd(value = nil)
  value.nil? ? (@cwd || Dir.pwd) : (@cwd = value)
end

.execute(run, payload:, buffer_events: Nexo.config.buffer_workflow_events, resume_input: {}) ⇒ Object

Executes an already-created run: "running" → #call → "done"/"failed", flushing buffered events in the ensure (on both success and failure) and firing a status notification on each transition. Shared by ::run (sync) and WorkflowJob#perform (async). Re-raises on failure.

payload: is symbol-keyed: ::run passes the caller's original (nested Ruby values intact); the job passes the JSON-normalized run.payload symbolized.

resume_input: (Spec 13, default {}) is the symbol-keyed input handed to ::resume/::resume_later; it is exposed to #call via #resume_input and is {} on the first (non-resume) pass. A #call that raises Suspended (via #suspend!) leaves the run "suspended" (a non-failure outcome) and returns it — completed #checkpoints persist and are skipped on resume.



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
342
343
344
345
346
347
348
349
 
# File 'lib/nexo/workflow.rb', line 296

def execute(run, payload:, buffer_events: Nexo.config.buffer_workflow_events, resume_input: {})
  # Remember a non-default buffering choice on the run so a later resume
  # (sync or job) honors it instead of silently reverting to the config
  # default. Only written when buffering is on, so the unbuffered Spec 2 hot
  # path takes no extra state write (see #resume, which reads it back).
  persist_buffer_choice(run, buffer_events)
  run.update!(status: "running")
  notify_status(run)

  instance = new(run, buffer_events: buffer_events)
  instance.instance_variable_set(:@resume_input, resume_input)
  result = instance.call(payload)
  run.update!(status: "done", result: stringify_keys(result), **cleared_state(run))
  notify_status(run)
  run
rescue Nexo::Workflow::Suspended => s
  # A durable pause, NOT a failure: mark "suspended", record the suspend
  # metadata under the reserved "__suspend__" state key (alongside any
  # completed checkpoints), broadcast the transition, and RETURN the run —
  # never re-raise. The $!-aware ensure below still flushes buffered events
  # ($! is nil here, since the signal is caught, not propagating).
  run.update!(status: "suspended",
    state: (run.state || {}).merge("__suspend__" => {
      "reason" => s.reason, "resume_key" => s.resume_key, "at" => Time.now.utc.iso8601
    }))
  notify_status(run)
  run
rescue StandardError, ScriptError, SecurityError => e
  # The sandbox seam raises OUTSIDE StandardError: SecurityError on a path
  # escape (Local/Container#absolute) and NotImplementedError (a
  # ScriptError) from a shell-less sandbox. Catch those too, so an escaping
  # #stage/#artifact or a Virtual shell marks the run "failed" instead of
  # leaving it stranded in "running" in a live, healthy process.
  if run
    run.update!(status: "failed", error: e.message)
    notify_status(run)
  end
  raise
ensure
  # Flush buffered events on both success and failure. $! is the exception
  # already propagating out of #call (nil on the success path); a flush
  # failure must never mask that original error, so it is only surfaced
  # when the workflow itself succeeded.
  pending = $!
  begin
    instance&.flush_events!
  rescue
    raise if pending.nil?
  end
  # Release the run's sandbox (if one was built) on every terminal path —
  # done, suspended, or failed — so a container/remote sandbox never leaks.
  # Best-effort inside #release_sandbox!, so it can't mask `pending`.
  instance&.release_sandbox!
end

.logs(id) ⇒ Object

Looks up a run by its UUID string id through whichever store RunStore.default selects, yields each event when a block is given, and returns the ordered events array. Works identically in plain Ruby and under Rails.



355
356
357
358
359
 
# File 'lib/nexo/workflow.rb', line 355

def logs(id)
  run = Nexo::RunStore.default.find(id)
  run.events.each { |ev| yield ev if block_given? }
  run.events
end

.reconcile_interrupted! ⇒ Object

One-shot boot/deploy sweep (Spec 7 R6) that rewrites runs orphaned in "running" to "interrupted" so a crashed worker doesn't leave zombie runs. Touches only "running" rows — "done" and "failed" are never rewritten. Under Rails it is a single update_all; offline it iterates the Memory store. NEVER auto-invoked — call it from a boot hook or the shipped nexo:reconcile rake task.

This is not a liveness check: it cannot distinguish an orphaned run from one genuinely running in another process. Run it once at boot, before any worker starts new runs. Returns the number of runs rewritten and fires a nexo.workflow.status notification per swept run so a dashboard learns of it.



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/nexo/workflow.rb', line 113

def reconcile_interrupted!
  if defined?(::ActiveRecord::Base) && defined?(Nexo::WorkflowRun)
    ids = Nexo::WorkflowRun.where(status: "running").pluck(:id)
    Nexo::WorkflowRun.where(id: ids).update_all(status: "interrupted")
    ids.each { |id| instrument_status(id, "interrupted") }
    ids.size
  else
    running = Nexo::RunStore::Memory.runs.each_value.select { |run| run.status == "running" }
    running.each do |run|
      run.update!(status: "interrupted")
      notify_status(run)
    end
    running.size
  end
end

.resume(run_id, input = {}) ⇒ Object

Continues a "suspended" run synchronously (Spec 13): re-instantiates the workflow from run.workflow_class and re-runs #call from the top with the run's original payload, making input available via #resume_input.

This is re-entry, NOT replay — Ruby has no transparent continuation capture. Everything outside a #checkpoint re-runs; only checkpoint-guarded work is skipped (its stored result is returned without re-running the block). Guard every side effect and expensive step with a checkpoint; idempotency of the non-checkpointed code is the author's responsibility.

Loads the run through RunStore.default, so it works with either store — but durable cross-process resume needs the ActiveRecord store (a Memory run doesn't survive the process); Memory resume is valid in-process. Raises Nexo::Error for a run that is not currently "suspended".



227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/nexo/workflow.rb', line 227

def resume(run_id, input = {})
  store = Nexo::RunStore.default
  run = store.find(run_id)
  unless run.status == "suspended"
    raise Nexo::Error, "run #{run_id} is not suspended (#{run.status})"
  end
  # Atomically claim the run (compare-and-set "suspended" → "running") so a
  # concurrent resume/resume_later can't both re-enter #call and double-run
  # the non-checkpointed work. The loser sees the claim fail and stops here.
  unless store.claim_for_resume!(run)
    raise Nexo::Error, "run #{run_id} is already being resumed"
  end
  # Object.const_get (not String#constantize) so resume works in plain Ruby
  # without ActiveSupport's core_ext — mirrors Nexo::Session (Spec 10).
  klass = Object.const_get(run.workflow_class)
  # Honor the original run's buffering choice (persisted by ::execute) so a
  # run started with buffer_events: true under a fiber reactor doesn't revert
  # to per-emit DB writes on resume; falls back to the config default.
  buffered = (run.state || {}).fetch("__buffer_events__", Nexo.config.buffer_workflow_events)
  klass.execute(run, payload: symbolize(run.payload), resume_input: input, buffer_events: buffered)
end

.resume_later(run_id, input = {}, queue: Nexo.config.job_queue, wait: nil, wait_until: nil) ⇒ Object

Enqueues a durable, cross-process resume of a "suspended" run (Spec 13 Q4), mirroring ::run_later: the job carries the run id plus the resume input (the payload still lives on the run — only the input travels). input must be ActiveJob/json-serializable. Returns the run (still "suspended" until the job picks it up and re-enters ::resume's guard).

Requires ActiveJob (Rails): without it this raises Nexo::MissingDependencyError pointing at ::resume for synchronous execution. queue: (default Nexo.config.job_queue) routes the job exactly like ::run_later.

wait: / wait_until: (Spec 21) defer the resume via the installed ActiveJob's own .set(...) — wait: a duration (resume_later(id, input, wait: 1.hour) so a suspended run wakes itself on a timer), wait_until: an absolute time. Nexo adds no scheduler and no retry; a crashed scheduled-resume job remains the host's reconcile_interrupted! / retry_on story. Passing both raises ArgumentError (the installed ActiveJob would silently keep one), checked before the job is enqueued. With neither given the enqueue is byte-for-byte the pre-Spec-21 immediate one. The return value is unchanged: the run stays "suspended" until the job fires and re-enters ::resume's atomic claim.



269
270
271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/nexo/workflow.rb', line 269

def resume_later(run_id, input = {}, queue: Nexo.config.job_queue, wait: nil, wait_until: nil)
  unless defined?(::ActiveJob)
    raise Nexo::MissingDependencyError,
      "resume_later requires ActiveJob (Rails). Use `resume` for synchronous execution."
  end
  if wait && wait_until
    raise ArgumentError,
      "pass either `wait:` or `wait_until:`, not both (got both)"
  end
  run = Nexo::RunStore.default.find(run_id)
  enqueue_job(Nexo::WorkflowJob, queue: queue, wait: wait, wait_until: wait_until).perform_later(run.id, input)
  run
end

.run(payload = nil, buffer_events: Nexo.config.buffer_workflow_events, **kwargs) ⇒ Object

Runs the workflow end to end: creates a run record (status "pending"), marks it "running", invokes the subclass's #call with a symbol-keyed payload, and records the outcome. On success the run is "done" with the return value as result; on any raised error it is "failed" with the message as error and the exception is re-raised. Returns the run.

buffer_events: (Spec 5, default Nexo.config.buffer_workflow_events) controls persistence of the event log. When false each emit persists immediately (Spec 2 behavior). When true events are buffered in memory and flushed to the store exactly once — the flush runs in the ensure so events land on both success and failure. Buffering avoids a blocking per-event DB write under a fiber reactor.

The payload keeps its Spec 2 shape: it may be passed as an explicit Hash (run({doc_id: 1}, buffer_events: true)) or as bare keywords (run(doc_id: 1)) — when no positional Hash is given, the collected keywords (minus buffer_events:) become the payload. This keeps the documented Workflow.run(doc_id: …, text: …) form working now that buffer_events: is a real keyword.



148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/nexo/workflow.rb', line 148

def run(payload = nil, buffer_events: Nexo.config.buffer_workflow_events, **kwargs)
  if payload && !kwargs.empty?
    raise ArgumentError,
      "pass the payload either as a positional Hash or as keywords, not both (got both)"
  end
  payload ||= kwargs
  run = Nexo::RunStore.default.create(workflow_class: name, payload: stringify(payload))
  # Pass the caller's ORIGINAL symbolized payload (nested Ruby values intact)
  # — the sync path never crosses a JSON/DB boundary, so it must not read back
  # the store-normalized run.payload. Only the async job symbolizes run.payload.
  execute(run, payload: symbolize(payload), buffer_events: buffer_events)
end

.run_later(payload = nil, queue: Nexo.config.job_queue, wait: nil, wait_until: nil, **kwargs) ⇒ Object

Enqueues the workflow on the host's ActiveJob adapter and hands back the run immediately (status "queued") so a controller can return while the work happens in the background. The job carries only the run id — the payload lives on the run record.

Requires ActiveJob (Rails): with no ActiveJob loaded this raises Nexo::MissingDependencyError pointing at ::run for synchronous execution. queue: (default Nexo.config.job_queue) routes the job to a named queue; nil uses ActiveJob's default queue.

It is only meaningful with a shared run store — the AR store plus a real adapter, so a worker in another process finds the run in the database. Under the :inline/:test adapter the job runs in-process, so the Memory store is also reachable. Not resumable: a crashed or retried job re-runs #call from scratch (Nexo adds no retry_on); pair with ::reconcile_interrupted! to catch runs orphaned in "running".

wait: / wait_until: (Spec 21) defer the enqueue via the installed ActiveJob's own .set(...) scheduler — wait: takes a duration (wait: 1.hour), wait_until: an absolute time (wait_until: tomorrow_9am). Nexo adds no scheduler of its own; it just forwards these to .set. The run is still "queued" (no "scheduled" status is invented). Passing both in one call raises ArgumentError (the installed ActiveJob would silently keep one) — checked before any run is created. With neither given the enqueue is byte-for-byte the pre-Spec-21 immediate one (no :at on the job).

wait:/wait_until:/queue: share the bare-keyword/positional ambiguity that queue: already carried: a bare-keyword call like run_later(wait: 60) consumes wait as the scheduling option (the payload stays {}). A payload that legitimately needs a key literally named "wait" must be passed as an explicit positional Hash — run_later({wait: "value"}).



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/nexo/workflow.rb', line 192

def run_later(payload = nil, queue: Nexo.config.job_queue, wait: nil, wait_until: nil, **kwargs)
  unless defined?(::ActiveJob)
    raise Nexo::MissingDependencyError,
      "run_later requires ActiveJob (Rails). Use `run` for synchronous execution."
  end
  if wait && wait_until
    raise ArgumentError,
      "pass either `wait:` or `wait_until:`, not both (got both)"
  end
  if payload && !kwargs.empty?
    raise ArgumentError,
      "pass the payload either as a positional Hash or as keywords, not both (got both)"
  end
  payload ||= kwargs
  run = Nexo::RunStore.default.create(workflow_class: name, payload: stringify(payload))
  run.update!(status: "queued")
  notify_status(run) # let a dashboard learn the run was enqueued
  enqueue_job(Nexo::WorkflowJob, queue: queue, wait: wait, wait_until: wait_until).perform_later(run.id)
  run
end

.sandbox(value = nil, **opts) ⇒ Object

The sandbox this workflow's runs stage inputs into and write artifacts to (Spec 7). Follows the same read-vs-write ivar convention as Agent's macros: with no argument (and no opts) it reads (default :virtual — safe, in-memory); with a bare value it stores a symbol/instance; with keywords it stores an options Hash ({ type: value, **opts }) — e.g. sandbox :docker, image: "node:22-slim" (Spec 15). Resolution is lazy — a data-only workflow that never stages or emits artifacts builds nothing (see #sandbox).



79
80
81
82
83
 
# File 'lib/nexo/workflow.rb', line 79

def sandbox(value = nil, **opts)
  return @sandbox || :virtual if value.nil? && opts.empty?

  @sandbox = opts.empty? ? value : {type: value, **opts}
end

Instance Method Details

#artifact(name, content: nil, from: nil, path: nil, locals: {}) ⇒ Object

Records a named deliverable on the run (Spec 7 R3). The body comes from either content: (used verbatim) or from: (a trusted, developer-authored ERB template — a real disk file when File.exist?(from), else a staged sandbox path via sandbox.read — rendered with the given locals).

SECURITY: ERB executes arbitrary Ruby. A from: template must be a trusted file you control — NEVER model output or user-uploaded content. Templates are code, not data (see README).

path: is the third mode and the one for agent output: it copies a sandbox path VERBATIM, with no ERB and no rendering of any kind. from: cannot be used for this — it evaluates the file as ERB, which is exactly what the security note above forbids for model-written content, and would corrupt any file containing <% regardless. Use path: for anything an agent produced; from: only for templates you wrote.

The body is written to the sandbox at the workspace-relative artifacts/<name> (so scripts/agents can read it during the run) and recorded on the run as a string-keyed hash {"name" =>, "content" =>, "at" =>}, matching how #emit string-keys events so Memory and the AR json column round-trip identically. Bytes that are not valid UTF-8 are Base64-encoded and marked "encoding" => "base64", because an AR json column cannot hold arbitrary binary — see .artifact_body for the decode side. Artifacts persist immediately (never buffered). Raises Nexo::Error when no mode produces a body. Returns the artifact hash.

Raises:



668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
 
# File 'lib/nexo/workflow.rb', line 668

def artifact(name, content: nil, from: nil, path: nil, locals: {})
  body = content
  if from
    template = File.exist?(from) ? File.read(from) : sandbox.read(from)
    body = ERB.new(template, trim_mode: "-").result_with_hash(locals)
  end
  body = sandbox.read(path) if path
  raise Nexo::Error, "artifact #{name} needs content:, from: or path:" if body.nil?

  # Workspace-relative on purpose. This was an ABSOLUTE "/artifacts/#{name}",
  # which every real sandbox rejects — Local#absolute and Container#guard_path
  # both raise SecurityError ("path escapes sandbox") for a path outside the
  # single root. #artifact therefore only ever worked on :virtual, whose
  # in-memory paths are unguarded; on :local and :container it raised. Relative
  # resolves under the sandbox root on all four tiers.
  sandbox.write(File.join(ARTIFACTS_DIR, name.to_s), body)
  art = {"name" => name.to_s, "at" => Time.now.utc.iso8601}.merge(encoded_body(body))
  @run.push_artifact(art)
  @run.save_artifacts! if @run.respond_to?(:save_artifacts!)
  art
end

#call(payload) ⇒ Object

Subclasses implement the work here. The payload is symbol-keyed; the returned value becomes the run's result (read back string-keyed).

Raises:

  • (NotImplementedError) 


449
450
451
 
# File 'lib/nexo/workflow.rb', line 449

def call(payload)
  raise NotImplementedError, "#{self.class} must implement #call(payload)"
end

#checkpoint(name) ⇒ Object

Runs name's block once and stores its json-serializable result under name.to_s in the run's state — the primitive that makes resume cheap and side-effect-safe (Spec 13). On a later run/resume of the same run a present checkpoint returns the stored value without re-running the block:

fetched = checkpoint(:fetch) { expensive_api_call(payload[:id]) }
published = checkpoint(:publish) { publish!(fetched) }

Persists immediately (like artifacts, not buffered like events), so a completed checkpoint survives a subsequent #suspend!. Values must be json-serializable — they round-trip the store exactly like result/events.

Do NOT call #suspend! inside a checkpoint block (undefined — v1 unsupported), and do NOT name a checkpoint "__suspend__" (reserved for suspend metadata) or "__approval__" (reserved for the pending approval call — Spec 16). A crash inside a checkpoint re-runs that checkpoint on resume (at-least-once for the in-flight step) — guard side effects accordingly.



472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
 
# File 'lib/nexo/workflow.rb', line 472

def checkpoint(name)
  key = name.to_s
  store = @run.state || {}
  return store[key] if store.key?(key)

  # JSON round-trip the block's value before storing so it reads back the SAME
  # shape (string keys, no symbols) whether the run stays in the Memory store
  # or is reloaded from the AR json column on a cross-process resume — and so
  # the first pass and the resume pass see identical data. Values must be
  # json-serializable (documented above).
  value = json_normalize(yield)
  @run.state = store.merge(key => value)
  @run.save_state! if @run.respond_to?(:save_state!)
  value
end

#checkpoint_all(steps) ⇒ Object

Runs several independent checkpoints concurrently on the first pass and, crucially, persists each step as it completes — so a resume after a partial failure only re-runs the steps that never landed (Spec 21):

fetched = checkpoint_all(
  account: -> { (payload[:id]) },
  usage:   -> { fetch_usage(payload[:id]) }
)
fetched[:account] # => the account value (this pass or a prior one)

steps is a Hash of name => callable (each value a Proc/lambda); names may be symbols or strings and are stringified for storage exactly like #checkpoint (name.to_s). The returned Hash is keyed by the original (un-stringified) names with values read back from @run.state, so a caller gets the same shape whether a value came from this call or a prior pass.

The pending steps (those not already in @run.state) run through the existing Nexo.concurrent driver, all in flight at once — callers bound the batch by how many keys they pass; there is no separate rate knob. Each step persists on its own through the same read-current → merge → assign → save_state! sequence as #checkpoint, serialized across the concurrent fibers by an internal Mutex, and emits a :checkpoint event naming the step (Spec 21 R3). Concurrency (and the async gem) is touched only when something is pending — an all-persisted pass returns the prior-pass values directly without requiring async.

Known trade-off: this is per-step persistence, not an atomic batch. If step B raises after step A persisted, A stays in run.state, B is absent, the run goes "failed", and the exception propagates through the workflow's normal failure path (Nexo.concurrent's "first failure re-raises, the rest stop" — not rescued away). A subsequent ::execute of the SAME run re-submits only the still-missing names — A is skipped, B re-runs. Do NOT treat a batch as all-or-nothing.

Same restrictions as #checkpoint: values must be json-serializable (they round-trip the store), a step must NOT be named after a RESERVED_STATE_KEYS entry (raises Nexo::Error before any step runs), and do NOT call #suspend! inside a step (undefined — v1 unsupported, documented not enforced).



527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
 
# File 'lib/nexo/workflow.rb', line 527

def checkpoint_all(steps)
  if (reserved = steps.keys.find { |name| RESERVED_STATE_KEYS.include?(name.to_s) })
    raise Nexo::Error,
      "checkpoint name #{reserved.to_s.inspect} is reserved (#{RESERVED_STATE_KEYS.join(", ")})"
  end

  state = @run.state || {}
  pending = steps.reject { |name, _| state.key?(name.to_s) }

  unless pending.empty?
    Nexo.concurrent(max_in_flight: pending.size) do |c|
      pending.each { |name, callable| c.add { persist_checkpoint(name, callable.call) } }
    end
  end

  current = @run.state || {}
  steps.keys.to_h { |name| [name, current[name.to_s]] }
end

#collect_artifacts(agent) ⇒ Object

Copies every artifact an agent DECLARED into the run, immediately after that agent runs and before anything can tear its sandbox down.

This is the missing half of Skills.materialize. That gets a skill's files INTO a sandbox; nothing got results back OUT, and on an ephemeral tier there is nowhere else for them to live: Container#close is rm -f, and Workflow.execute releases the run's sandbox on EVERY terminal path — done, failed, and suspended. So a durable-approval pause (the whole point of Spec 16) destroyed everything the run had produced up to the approval, while the identical code on :local kept it, because there the sandbox is just a directory. Same workflow, durable on one tier and lossy on another, with no error either way.

Declared, never inferred: a sweep of the sandbox would collect staged skill scripts, templates and scratch files, and an agent naming its outputs is also the only honest way to say "this run produced nothing". A name may be a glob ("out/*.json"), and an agent may declare as many as it likes. A declared artifact that does not exist is skipped rather than fatal — a run can legitimately not produce one, and this must never be what fails a run. Returns the artifacts recorded.



718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
 
# File 'lib/nexo/workflow.rb', line 718

def collect_artifacts(agent)
  # Guarded like #run_agent's `agent.close`: an injected test double or a
  # non-Nexo::Agent duck simply declares nothing.
  return [] unless agent.class.respond_to?(:produces)

  names = agent.class.produces
  return [] if names.empty?

  names.flat_map { |name| artifact_paths(name) }.uniq.filter_map do |path|
    artifact(File.basename(path), path: path)
  rescue => e
    emit(:artifact_skipped, {"path" => path.to_s, "error" => e.message})
    nil
  end
end

#emit(type, data = {}) ⇒ Object

Appends an event to the run's ordered log and persists it incrementally. The event's own keys ("type"/"data"/"at") are strings so the record reads back the same shape after the ActiveRecord backend's JSON round-trip; the caller-supplied data hash is stored verbatim (symbol keys survive the in-memory store, stringify through the json column). Returns the event hash.



569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
 
# File 'lib/nexo/workflow.rb', line 569

def emit(type, data = {})
  ev = {"type" => type.to_s, "data" => data, "at" => Time.now.utc.iso8601}
  if @buffer_events
    # Defer the DB hit — accumulate in memory and persist once in
    # #flush_events! (called from Workflow.run's ensure).
    @event_buffer << ev
  else
    @run.push_event(ev)
    @run.save_events! if @run.respond_to?(:save_events!)
  end
  # Live broadcast (Spec 11 R2): fires regardless of buffering — persistence
  # stays separate/buffered above, but the notification is live. A no-op without
  # ActiveSupport, so this is exactly the Spec 2 emit in the plain-Ruby core.
  notify_event(ev)
  ev
end

#flush_events! ⇒ Object

Replays any buffered events through the run and persists them in a single save_events!, then clears the buffer. A no-op when buffering is off or the buffer is empty. Called from Workflow.run's ensure, so buffered events are saved on both success and failure. Idempotent (a second call, e.g. if a workflow calls it explicitly, finds an empty buffer and does nothing).



591
592
593
594
595
596
597
 
# File 'lib/nexo/workflow.rb', line 591

def flush_events!
  return unless @buffer_events && @event_buffer.any?

  @event_buffer.each { |ev| @run.push_event(ev) }
  @run.save_events! if @run.respond_to?(:save_events!)
  @event_buffer.clear
end

#release_sandbox! ⇒ Object

Releases the run's sandbox at the end of #execute — but ONLY if one was actually built. Reads @sandbox directly instead of calling #sandbox so a data-only workflow (which never resolved a sandbox) doesn't construct one just to close it. Idempotent and best-effort: a container/remote sandbox is force-removed/closed here so a #run_agent-driven container doesn't leak — #run_agent BORROWS this shared sandbox, so Agent#close leaves teardown to the owner (this workflow). A failing teardown must never raise out of #execute.



615
616
617
618
619
 
# File 'lib/nexo/workflow.rb', line 615

def release_sandbox!
  @sandbox&.close
rescue
  nil
end

#restore_artifacts(only: nil, into: ".") ⇒ Object

Materializes previously collected artifacts back into the run's sandbox, so a later stage can read what an earlier one produced. The hand-off half: on a persistent tier stage N+1 just reads the path stage N wrote, and this makes that true on an ephemeral tier too — including across a suspend/resume, where the sandbox that held them no longer exists. Names nothing by default (every recorded artifact); pass only: to restore a subset. Returns the paths written.



740
741
742
743
744
745
746
747
748
749
750
 
# File 'lib/nexo/workflow.rb', line 740

def restore_artifacts(only: nil, into: ".")
  wanted = only && Array(only).map(&:to_s)
  @run.artifacts.filter_map do |art|
    name = art["name"].to_s
    next if wanted && !wanted.include?(name)

    path = File.join(into, name)
    sandbox.write(path, self.class.artifact_body(art))
    path
  end
end

#resume_input ⇒ Object

The symbol-keyed input passed to ::resume/::resume_later (Spec 13). It is {} on the first (non-resume) pass, so a workflow gates on it to decide whether to #suspend! or proceed:

suspend!(reason: "needs approval") unless resume_input[:approved]



562
 
# File 'lib/nexo/workflow.rb', line 562

def resume_input = @resume_input || {}

#run_agent(prompt, max_turns: 25) ⇒ Object

Drives the workflow's declared ::agent (Spec 8), bound to this run's sandbox (Spec 7), forwarding every (type, payload) event the agent's loop yields into the run log as an agent_* event, and ensuring the agent is closed afterward (tearing down any memoized MCP servers from Spec 6). Returns the agent's response object (read response.content).

Composition only — no new loop, no orchestration engine: it wires the existing Agent#prompt + before_tool_call/after_tool_result seam (source: Loops::RubyLLM) through the existing #emit path, so the events honor the run's buffer_events setting and persist in both run stores with no extra wiring.

Shared-sandbox precedence: under run_agent the agent uses the workflow's sandbox; the agent's own sandbox class macro is ignored (it only applies when the agent runs standalone via .new.prompt). The agent keeps its own permissions/skills/mcp/mcp_allow — the workflow provides the where (sandbox), the agent owns the what (permissions) and how (skills). Driving an agent never widens its authority; its safe default (:read_only) is untouched. Raises ConfigurationError when no agent is declared. Durable approval (Spec 16): when the driven agent runs under an :approve permission gate and hits a sensitive capability with no decision yet, the gate raises Nexo::ApprovalRequired, which propagates out of the ruby_llm tool loop (Group 0: the loop does not rescue tool exceptions) and out of Agent#prompt. run_agent rescues it, records the pending call under the reserved "__approval__" state key, and #suspend!s the run — so the worker returns and a host renders "approval pending" from run.state. On ::resume/::resume_later with {approved: …} the decision is threaded into the agent (via decision:), so the same gate now allows (→ run completes) or denies (→ the tool returns {error:}, the model adapts, run still completes).



806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
 
# File 'lib/nexo/workflow.rb', line 806

def run_agent(prompt, max_turns: 25)
  klass = self.class.agent or raise Nexo::ConfigurationError,
    "#{self.class} has no `agent` declared; add `agent MyAgent`"
  agent = klass.new(sandbox: sandbox, **agent_decision_kwargs)
  agent.prompt(prompt, max_turns: max_turns) do |type, payload|
    emit(:"agent_#{type}", serializable(type, payload))
  end
rescue Nexo::ApprovalRequired => a
  # Persist the pending call (immediately, like a checkpoint) so a host can
  # render the approval prompt, then suspend — the existing Workflow.execute
  # `rescue Suspended` path marks the run "suspended" and returns it.
  @run.state = (@run.state || {}).merge(
    "__approval__" => {
      "capability" => a.capability.to_s,
      "tool" => a.detail.to_s,
      "args" => a.args
    }
  )
  @run.save_state! if @run.respond_to?(:save_state!)
  suspend!(reason: "approval: #{a.detail}", resume_key: a.detail.to_s)
ensure
  # Collect BEFORE teardown, and in the ensure so it also runs when the agent
  # suspended for approval or raised — those are the paths where the sandbox
  # dies with results still in it. Never allowed to mask the original outcome.
  begin
    collect_artifacts(agent) if agent
  rescue => e
    emit(:artifact_skipped, {"error" => e.message})
  end
  # Guarded so it's safe if Spec 6 (Agent#close) isn't present in the agent,
  # or if agent construction itself raised (agent is nil).
  agent.close if agent.respond_to?(:close)
end

#sandbox ⇒ Object

The run's sandbox (Spec 7), resolved lazily from the class-level ::sandbox macro and memoized. Built on first touch — by #stage, #artifact, or a workflow reading/writing files directly — so a data-only workflow that never calls any of them constructs nothing new and keeps the Spec 2 hot path byte-for-byte unchanged.



604
605
606
 
# File 'lib/nexo/workflow.rb', line 604

def sandbox
  @sandbox ||= resolve_sandbox(self.class.sandbox)
end

#stage(files) ⇒ Object

Stages provided files into the run's sandbox before work begins (Spec 7 R2). Accepts either a Hash { "path" => "content" } or an Array of { path:, content: } hashes; both normalize to [path, content] pairs. Each pair is written via sandbox.write. Emits a :staged event carrying the count (reusing the existing #emit path) and returns the number of files staged.



627
628
629
630
631
632
633
634
635
636
637
638
 
# File 'lib/nexo/workflow.rb', line 627

def stage(files)
  pairs = if files.is_a?(Hash)
    files.to_a
  else
    # Tolerate symbol- OR string-keyed hashes (e.g. an Array parsed from JSON),
    # matching how the Hash form already accepts string paths.
    files.map { |f| [f[:path] || f["path"], f[:content] || f["content"]] }
  end
  pairs.each { |path, content| sandbox.write(path, content) }
  emit(:staged, count: pairs.size)
  pairs.size
end

#suspend!(reason:, resume_key: nil) ⇒ Object

Pauses the run durably (Spec 13): raises Suspended, which ::execute catches to mark the run "suspended" (a non-failure outcome) and return it to the caller. Call this outside a checkpoint block. reason is surfaced to a host UI; the optional resume_key is persisted so a host can correlate which resume it is awaiting. Continue the run later with ::resume/::resume_later.

Raises:



551
552
553
 
# File 'lib/nexo/workflow.rb', line 551

def suspend!(reason:, resume_key: nil)
  raise Suspended.new(reason, resume_key)
end