roder_dynamic_workflows/
runner.rs

1use crate::host_api::{RawWorkflowExecution, WorkflowExecution};
2use crate::model::{
3    WorkflowRunInput, WorkflowRuntimeError, WorkflowRuntimeErrorKind, WorkflowRuntimeOptions,
4    WorkflowRuntimeResult,
5};
6use crate::script::{
7    eval_js, install_definition_prelude, new_context, parse_definition_json, preflight_script,
8    read_global_string,
9};
10
11#[derive(Debug, Clone, Default)]
12pub struct WorkflowScriptRuntime {
13    options: WorkflowRuntimeOptions,
14}
15
16impl WorkflowScriptRuntime {
17    pub fn new(options: WorkflowRuntimeOptions) -> Self {
18        Self { options }
19    }
20
21    pub fn options(&self) -> &WorkflowRuntimeOptions {
22        &self.options
23    }
24
25    pub fn run(
26        &self,
27        source: &str,
28        input: WorkflowRunInput,
29    ) -> WorkflowRuntimeResult<WorkflowExecution> {
30        preflight_script(source)?;
31
32        let mut context = new_context(&self.options);
33        install_definition_prelude(&mut context)?;
34        install_host_prelude(&mut context, &self.options, &input)?;
35        eval_js(
36            &mut context,
37            source,
38            WorkflowRuntimeErrorKind::ScriptExecution,
39        )?;
40
41        let metadata = read_global_string(&mut context, "globalThis.__roderWorkflowMetadataJson")?
42            .filter(|value| !value.is_empty())
43            .ok_or_else(|| {
44                WorkflowRuntimeError::new(
45                    WorkflowRuntimeErrorKind::MissingDefinition,
46                    "script must call workflow.define(metadata, handler)",
47                )
48            })?;
49        let definition = parse_definition_json(&metadata, &self.options)?;
50        let run_source = workflow_run_source();
51        eval_js(
52            &mut context,
53            &run_source,
54            WorkflowRuntimeErrorKind::ScriptExecution,
55        )?;
56
57        for _ in 0..self.options.max_promise_drains {
58            context.run_jobs().map_err(|err| {
59                crate::script::classify_js_error(
60                    err.to_string(),
61                    WorkflowRuntimeErrorKind::ScriptExecution,
62                )
63            })?;
64            if let Some(error) =
65                read_global_string(&mut context, "globalThis.__roderWorkflowError")?
66            {
67                return Err(crate::script::classify_js_error(
68                    error,
69                    WorkflowRuntimeErrorKind::ScriptExecution,
70                ));
71            }
72            if let Some(json) =
73                read_global_string(&mut context, "globalThis.__roderWorkflowResultJson")?
74            {
75                let raw = parse_execution_json(&json)?;
76                return Ok(WorkflowExecution {
77                    definition,
78                    report: raw.report,
79                    phases: raw.phases,
80                    agent_launches: raw.agent_launches,
81                    checkpoints: raw.checkpoints,
82                });
83            }
84        }
85
86        Err(WorkflowRuntimeError::new(
87            WorkflowRuntimeErrorKind::ScriptExecution,
88            "workflow handler did not settle within the promise drain limit",
89        ))
90    }
91}
92
93fn install_host_prelude(
94    context: &mut boa_engine::Context,
95    options: &WorkflowRuntimeOptions,
96    input: &WorkflowRunInput,
97) -> WorkflowRuntimeResult<()> {
98    let run_id = serde_json::to_string(&input.run_id).expect("run id serializes");
99    let arguments = serde_json::to_string(&input.arguments).map_err(|err| {
100        WorkflowRuntimeError::new(
101            WorkflowRuntimeErrorKind::InvalidMetadata,
102            format!("workflow arguments must be JSON serializable: {err}"),
103        )
104    })?;
105    let limits = serde_json::to_string(&options.limits).expect("limits serialize");
106    let checkpoints = serde_json::to_string(&input.checkpoints).map_err(|err| {
107        WorkflowRuntimeError::new(
108            WorkflowRuntimeErrorKind::InvalidMetadata,
109            format!("workflow checkpoints must be JSON serializable: {err}"),
110        )
111    })?;
112    let abort = if input.abort_before_start {
113        "true"
114    } else {
115        "false"
116    };
117    let max_report_bytes = options.max_report_bytes;
118
119    let prelude = format!(
120        r#"
121globalThis.__roderRunId = {run_id};
122globalThis.__roderArguments = {arguments};
123globalThis.__roderLimits = Object.freeze({limits});
124globalThis.__roderAbort = {abort};
125globalThis.__roderMaxReportBytes = {max_report_bytes};
126globalThis.__roderState = {{
127  launches: [],
128  checkpoints: {{}},
129  checkpointRecords: [],
130  phases: [],
131  currentPhase: null,
132  report: null,
133  agentCount: 0
134}};
135for (const checkpoint of {checkpoints}) {{
136  globalThis.__roderState.checkpoints[checkpoint.key] = checkpoint.value;
137}}
138
139globalThis.__roderCheckAbort = function() {{
140  if (globalThis.__roderAbort) {{
141    throw new Error("abort:workflow aborted before host call");
142  }}
143}};
144
145globalThis.__roderLaunchAgent = function(role, descriptor, input, index) {{
146  globalThis.__roderCheckAbort();
147  const spec = descriptor || {{}};
148  if (globalThis.__roderState.agentCount >= globalThis.__roderLimits.maxAgentsPerRun) {{
149    throw new Error("limit:maxAgentsPerRun");
150  }}
151  const launch = {{
152    index,
153    role,
154    lane: spec.lane || role,
155    phase: globalThis.__roderState.currentPhase,
156    description: spec.description || "",
157    prompt: spec.prompt || "",
158    model: spec.model || null,
159    timeoutSeconds: spec.timeoutSeconds || globalThis.__roderLimits.defaultAgentTimeoutSeconds,
160    input: input === undefined ? null : input,
161    output: spec.output || `result:${{role}}:${{index}}`
162  }};
163  globalThis.__roderState.agentCount += 1;
164  globalThis.__roderState.launches.push(launch);
165  return Object.freeze({{
166    agentId: `agent-${{globalThis.__roderState.agentCount}}`,
167    role: launch.role,
168    lane: launch.lane,
169    input: launch.input,
170    prompt: launch.prompt,
171    output: launch.output
172  }});
173}};
174
175globalThis.__roderCreateContext = function() {{
176  const agents = Object.freeze({{
177    run(role, descriptor) {{
178      return globalThis.__roderLaunchAgent(role, descriptor, null, 0);
179    }},
180    map(role, items, mapper) {{
181      globalThis.__roderCheckAbort();
182      if (!Array.isArray(items)) {{
183        throw new Error("ctx.agents.map requires an array of items");
184      }}
185      if (typeof mapper !== "function") {{
186        throw new Error("ctx.agents.map requires a mapper function");
187      }}
188      return items.map((item, index) => {{
189        const descriptor = mapper(item, index) || {{}};
190        return globalThis.__roderLaunchAgent(role, descriptor, item, index);
191      }});
192    }},
193    reduce(role, items, mapper, reducer, initial) {{
194      if (typeof reducer !== "function") {{
195        throw new Error("ctx.agents.reduce requires a reducer function");
196      }}
197      return agents.map(role, items, mapper).reduce(reducer, initial);
198    }}
199  }});
200
201  const checkpoint = Object.freeze({{
202    save(key, value) {{
203      globalThis.__roderCheckAbort();
204      const json = JSON.stringify(value);
205      if (json.length > globalThis.__roderLimits.defaultCheckpointBytes) {{
206        throw new Error("limit:checkpointBytes");
207      }}
208      const record = {{ key, value, byteCount: json.length }};
209      globalThis.__roderState.checkpoints[key] = value;
210      globalThis.__roderState.checkpointRecords.push(record);
211      return value;
212    }},
213    read(key) {{
214      return Object.prototype.hasOwnProperty.call(globalThis.__roderState.checkpoints, key)
215        ? globalThis.__roderState.checkpoints[key]
216        : null;
217    }}
218  }});
219
220  const report = Object.freeze({{
221    markdown(value) {{
222      globalThis.__roderCheckAbort();
223      let text;
224      if (Array.isArray(value)) {{
225        text = value.map((item) => typeof item === "string" ? item : (item.output || JSON.stringify(item))).join("\n");
226      }} else {{
227        text = String(value);
228      }}
229      if (text.length > globalThis.__roderMaxReportBytes) {{
230        throw new Error("limit:reportBytes");
231      }}
232      globalThis.__roderState.report = text;
233      return text;
234    }}
235  }});
236
237  return Object.freeze({{
238    run: Object.freeze({{ id: globalThis.__roderRunId, arguments: globalThis.__roderArguments }}),
239    phase: Object.freeze({{ start(name) {{ const phase = String(name); globalThis.__roderState.currentPhase = phase; globalThis.__roderState.phases.push(phase); return phase; }} }}),
240    agents,
241    results: Object.freeze({{
242      all() {{ return globalThis.__roderState.launches.slice(); }},
243      vote(items, selector) {{
244        if (!Array.isArray(items)) {{
245          throw new Error("ctx.results.vote requires an array of items");
246        }}
247        const choose = typeof selector === "function" ? selector : (item) => item;
248        const counts = {{}};
249        for (const item of items) {{
250          const key = String(choose(item));
251          counts[key] = (counts[key] || 0) + 1;
252        }}
253        let winner = null;
254        let winnerCount = -1;
255        for (const [key, count] of Object.entries(counts)) {{
256          if (count > winnerCount) {{
257            winner = key;
258            winnerCount = count;
259          }}
260        }}
261        return {{ winner, counts }};
262      }}
263    }}),
264    checkpoint,
265    report,
266    limits: globalThis.__roderLimits,
267    abortSignal: Object.freeze({{ get aborted() {{ return globalThis.__roderAbort; }} }})
268  }});
269}};
270"#
271    );
272
273    eval_js(context, &prelude, WorkflowRuntimeErrorKind::ScriptExecution)
274}
275
276fn workflow_run_source() -> String {
277    r#"
278(async () => {
279  try {
280    if (typeof globalThis.__roderWorkflowHandler !== "function") {
281      throw new Error("workflow.define must install a handler before execution");
282    }
283    const ctx = globalThis.__roderCreateContext();
284    const result = await globalThis.__roderWorkflowHandler(ctx);
285    const report = globalThis.__roderState.report !== null
286      ? globalThis.__roderState.report
287      : (typeof result === "string" ? result : JSON.stringify(result));
288    globalThis.__roderWorkflowResultJson = JSON.stringify({
289      report,
290      phases: globalThis.__roderState.phases,
291      agentLaunches: globalThis.__roderState.launches,
292      checkpoints: globalThis.__roderState.checkpointRecords
293    });
294  } catch (error) {
295    globalThis.__roderWorkflowError = String(error && error.message ? error.message : error);
296  }
297})();
298"#
299    .to_string()
300}
301
302fn parse_execution_json(json: &str) -> WorkflowRuntimeResult<RawWorkflowExecution> {
303    serde_json::from_str(json).map_err(|err| {
304        WorkflowRuntimeError::new(
305            WorkflowRuntimeErrorKind::ScriptExecution,
306            format!("workflow produced invalid execution JSON: {err}"),
307        )
308    })
309}
roder_dynamic_workflows/runner.rs

roder_dynamic_workflows/
runner.rs
