vm.rs - CodeWhale

vm.rs

1	//! The sandboxed QuickJS VM that executes Workflow scripts.
2	//!
3	//! Threading model (design §2.2): `rquickjs` contexts and every `'js` value
4	//! are `!Send`, so each run gets a dedicated OS thread with its own
5	//! current-thread tokio reactor. Host functions do no heavy work inline —
6	//! only `Send` data (JSON strings, [`TaskRequest`]s, oneshot replies) crosses
7	//! to the driver; conversion back into JS values happens on the VM thread
8	//! after the await resolves.
9	//!
10	//! Sandbox: the context registers only standard ECMAScript intrinsics plus
11	//! the Workflow globals (`task`, `parallel`, `pipeline`, `log`, `phase`,
12	//! `budget`, `args`). There is no module loader, no fs/net/process access,
13	//! and `Date`/`Math.random` are overridden to throw so recorded runs stay
14	//! deterministic for replay.
15
16	use std::cell::Cell;
17	use std::env;
18	use std::rc::Rc;
19	use std::sync::atomic::{AtomicBool, Ordering};
20	use std::sync::{Arc, OnceLock};
21
22	use rquickjs::function::{Async, Func};
23	use rquickjs::{AsyncContext, AsyncRuntime, CatchResultExt, CaughtError, Ctx, Promise, Value};
24	use serde::Deserialize;
25	use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot, watch};
26
27	use crate::driver::{ProgressEvent, TaskCompletion, TaskRequest, WorkflowDriver};
28	use crate::error::WorkflowJsError;
29	use crate::schema::{compile_schema, decode_reply};
30	use crate::{PARALLEL_MAX_ITEMS, WORKFLOW_LIFETIME_CAP, normalize_profile};
31
32	const DEFAULT_VM_MEMORY_LIMIT_BYTES: usize = 32 * 1024 * 1024;
33	const MIN_VM_MEMORY_LIMIT_BYTES: usize = 4 * 1024 * 1024;
34	const MAX_VM_MEMORY_LIMIT_BYTES: usize = 512 * 1024 * 1024;
35	const DEFAULT_VM_STACK_BYTES: usize = 1024 * 1024;
36	const MIN_VM_STACK_BYTES: usize = 128 * 1024;
37	const MAX_VM_STACK_BYTES: usize = 8 * 1024 * 1024;
38	const DEFAULT_VM_THREAD_STACK_BYTES: usize = 2 * 1024 * 1024;
39	const MIN_VM_THREAD_STACK_BYTES: usize = 512 * 1024;
40	const MAX_VM_THREAD_STACK_BYTES: usize = 16 * 1024 * 1024;
41	const DEFAULT_MAX_CONCURRENT_VMS: usize = 4;
42	const MAX_CONCURRENT_VMS: usize = 256;
43
44	const VM_MEMORY_LIMIT_MB_ENV: &str = "CODEWHALE_WORKFLOW_JS_MEMORY_LIMIT_MB";
45	const VM_STACK_KB_ENV: &str = "CODEWHALE_WORKFLOW_JS_STACK_KB";
46	const VM_THREAD_STACK_KB_ENV: &str = "CODEWHALE_WORKFLOW_JS_THREAD_STACK_KB";
47	const VM_MAX_CONCURRENT_ENV: &str = "CODEWHALE_WORKFLOW_JS_MAX_CONCURRENT";
48
49	/// Resource limits applied to the QuickJS runtime before any script runs.
50	///
51	/// There is deliberately no wall-clock timeout here: cancellation (dropping
52	/// the run future, or the driver's cancel cascade) is the deadline mechanism.
53	#[derive(Debug, Clone, Copy)]
54	pub struct VmLimits {
55	/// QuickJS heap ceiling in bytes (default 32 MiB).
56	pub memory_limit_bytes: usize,
57	/// Maximum interpreter stack in bytes (default 1 MiB).
58	pub max_stack_bytes: usize,
59	}
60
61	impl Default for VmLimits {
62	fn default() -> Self {
63	Self::from_env()
64	}
65	}
66
67	impl VmLimits {
68	pub fn from_env() -> Self {
69	Self {
70	memory_limit_bytes: env_usize_bytes(
71	VM_MEMORY_LIMIT_MB_ENV,
72	1024 * 1024,
73	MIN_VM_MEMORY_LIMIT_BYTES,
74	MAX_VM_MEMORY_LIMIT_BYTES,
75	DEFAULT_VM_MEMORY_LIMIT_BYTES,
76	),
77	max_stack_bytes: env_usize_bytes(
78	VM_STACK_KB_ENV,
79	1024,
80	MIN_VM_STACK_BYTES,
81	MAX_VM_STACK_BYTES,
82	DEFAULT_VM_STACK_BYTES,
83	),
84	}
85	}
86	}
87
88	fn env_usize_bytes(name: &str, unit: usize, min: usize, max: usize, default: usize) -> usize {
89	env::var(name)
90	.ok()
91	.and_then(\|raw\| raw.parse::<usize>().ok())
92	.and_then(\|value\| value.checked_mul(unit))
93	.map(\|bytes\| bytes.clamp(min, max))
94	.unwrap_or(default)
95	}
96
97	fn max_concurrent_vms() -> usize {
98	env::var(VM_MAX_CONCURRENT_ENV)
99	.ok()
100	.and_then(\|raw\| raw.parse::<usize>().ok())
101	.map(\|value\| value.clamp(1, MAX_CONCURRENT_VMS))
102	.unwrap_or(DEFAULT_MAX_CONCURRENT_VMS)
103	}
104
105	fn vm_thread_stack_bytes() -> usize {
106	env_usize_bytes(
107	VM_THREAD_STACK_KB_ENV,
108	1024,
109	MIN_VM_THREAD_STACK_BYTES,
110	MAX_VM_THREAD_STACK_BYTES,
111	DEFAULT_VM_THREAD_STACK_BYTES,
112	)
113	}
114
115	fn vm_admission() -> &'static Arc<Semaphore> {
116	static ADMISSION: OnceLock<Arc<Semaphore>> = OnceLock::new();
117	ADMISSION.get_or_init(\|\| Arc::new(Semaphore::new(max_concurrent_vms())))
118	}
119
120	/// Executes Workflow scripts, one isolated QuickJS runtime per run.
121	///
122	/// Every [`WorkflowVm::run_script`] call spins up a fresh interpreter on a
123	/// dedicated thread, so runs share nothing (globals, heap, interned atoms)
124	/// and a wedged script can never stall a sibling run.
125	#[derive(Debug, Clone, Default)]
126	pub struct WorkflowVm {
127	limits: VmLimits,
128	}
129
130	impl WorkflowVm {
131	/// A VM with the default [`VmLimits`].
132	pub fn new() -> Self {
133	Self::default()
134	}
135
136	/// A VM with explicit resource limits.
137	pub fn with_limits(limits: VmLimits) -> Self {
138	Self { limits }
139	}
140
141	/// Run one Workflow script to completion.
142	///
143	/// * `source` is the script body; it is wrapped in an async function, so
144	/// top-level `await` and `return` both work. The returned value is the
145	/// script's `return` value, JSON-encoded (`undefined` becomes `null`).
146	/// * `args` is exposed verbatim to the script as the `args` global.
147	/// * `driver` executes `task()` spawns and receives progress events. A
148	/// driver instance is scoped to exactly one run: `cancel_all` is always
149	/// invoked at run teardown (success, script error, or cancellation), so
150	/// stray children never outlive the script that spawned them.
151	///
152	/// Cancellation cascade (design §9): dropping the returned future cancels
153	/// the run — the interrupt handler aborts executing JS, pending `task()`
154	/// awaits resolve to errors, and `driver.cancel_all()` is invoked
155	/// immediately from the dropping thread.
156	pub async fn run_script(
157	&self,
158	source: &str,
159	args: serde_json::Value,
160	driver: Arc<dyn WorkflowDriver>,
161	) -> Result<serde_json::Value, WorkflowJsError> {
162	self.run_script_with_cancel(source, args, driver, WorkflowRunCancel::new())
163	.await
164	}
165
166	/// Like [`Self::run_script`], but accepts an external cancel handle so the
167	/// host can interrupt the VM without dropping the run future.
168	pub async fn run_script_with_cancel(
169	&self,
170	source: &str,
171	args: serde_json::Value,
172	driver: Arc<dyn WorkflowDriver>,
173	cancel: WorkflowRunCancel,
174	) -> Result<serde_json::Value, WorkflowJsError> {
175	let args_json = serde_json::to_string(&args)
176	.map_err(\|err\| WorkflowJsError::InvalidArgs(err.to_string()))?;
177	let cancel = cancel.0;
178	let (result_tx, result_rx) = oneshot::channel();
179	let mut guard = RunGuard {
180	cancel: cancel.clone(),
181	driver: driver.clone(),
182	armed: true,
183	};
184
185	let permit = vm_admission()
186	.clone()
187	.acquire_owned()
188	.await
189	.map_err(\|_\| WorkflowJsError::VmInit("VM admission gate closed".to_string()))?;
190	let limits = self.limits;
191	let source = source.to_string();
192	let thread_driver = driver.clone();
193	let thread_cancel = cancel.clone();
194	let spawned = std::thread::Builder::new()
195	.name("workflow-js-vm".to_string())
196	.stack_size(vm_thread_stack_bytes())
197	.spawn(move \|\| {
198	let _permit: OwnedSemaphorePermit = permit;
199	let outcome = vm_thread_main(
200	source,
201	args_json,
202	thread_driver.clone(),
203	thread_cancel,
204	limits,
205	);
206	// Run teardown: this driver is scoped to one run, so any task
207	// still in flight is unreachable now — cancel the cascade.
208	thread_driver.cancel_all();
209	let _ = result_tx.send(outcome);
210	});
211	if let Err(err) = spawned {
212	guard.armed = false;
213	return Err(WorkflowJsError::VmInit(format!(
214	"failed to spawn VM thread: {err}"
215	)));
216	}
217
218	match result_rx.await {
219	Ok(outcome) => {
220	// The VM thread has already torn down and cancelled children.
221	guard.armed = false;
222	outcome
223	}
224	// VM thread panicked before reporting; leave the guard armed so
225	// its drop (right now, at return) cancels outstanding tasks.
226	Err(_) => Err(WorkflowJsError::VmTerminated(
227	"VM thread exited without reporting a result".to_string(),
228	)),
229	}
230	}
231	}
232
233	/// Cooperative cancel signal shared by the run future (guard side) and the VM
234	/// thread. The atomic flag feeds the QuickJS interrupt handler (sync, called
235	/// mid-bytecode); the watch channel wakes host futures parked on driver
236	/// completions.
237	#[derive(Clone)]
238	pub struct WorkflowRunCancel(CancelHandle);
239
240	impl WorkflowRunCancel {
241	#[must_use]
242	pub fn new() -> Self {
243	Self(CancelHandle::new())
244	}
245
246	pub fn cancel(&self) {
247	self.0.cancel();
248	}
249	}
250
251	impl Default for WorkflowRunCancel {
252	fn default() -> Self {
253	Self::new()
254	}
255	}
256
257	#[derive(Clone)]
258	struct CancelHandle {
259	flag: Arc<AtomicBool>,
260	tx: Arc<watch::Sender<bool>>,
261	}
262
263	impl CancelHandle {
264	fn new() -> Self {
265	let (tx, _rx) = watch::channel(false);
266	Self {
267	flag: Arc::new(AtomicBool::new(false)),
268	tx: Arc::new(tx),
269	}
270	}
271
272	fn cancel(&self) {
273	self.flag.store(true, Ordering::SeqCst);
274	self.tx.send_replace(true);
275	}
276
277	fn is_cancelled(&self) -> bool {
278	self.flag.load(Ordering::SeqCst)
279	}
280
281	async fn cancelled(&self) {
282	let mut rx = self.tx.subscribe();
283	let _ = rx.wait_for(\|cancelled\| *cancelled).await;
284	}
285
286	fn flag_arc(&self) -> Arc<AtomicBool> {
287	self.flag.clone()
288	}
289	}
290
291	/// Fires the cancel cascade if the caller drops the run future before the VM
292	/// reports a result.
293	struct RunGuard {
294	cancel: CancelHandle,
295	driver: Arc<dyn WorkflowDriver>,
296	armed: bool,
297	}
298
299	impl Drop for RunGuard {
300	fn drop(&mut self) {
301	if self.armed {
302	self.cancel.cancel();
303	self.driver.cancel_all();
304	}
305	}
306	}
307
308	fn vm_thread_main(
309	source: String,
310	args_json: String,
311	driver: Arc<dyn WorkflowDriver>,
312	cancel: CancelHandle,
313	limits: VmLimits,
314	) -> Result<serde_json::Value, WorkflowJsError> {
315	let reactor = tokio::runtime::Builder::new_current_thread()
316	.enable_all()
317	.build()
318	.map_err(\|err\| WorkflowJsError::VmInit(format!("failed to build VM reactor: {err}")))?;
319	reactor.block_on(run_in_vm(source, args_json, driver, cancel, limits))
320	}
321
322	async fn run_in_vm(
323	source: String,
324	args_json: String,
325	driver: Arc<dyn WorkflowDriver>,
326	cancel: CancelHandle,
327	limits: VmLimits,
328	) -> Result<serde_json::Value, WorkflowJsError> {
329	let runtime = AsyncRuntime::new().map_err(\|err\| WorkflowJsError::VmInit(err.to_string()))?;
330	runtime.set_memory_limit(limits.memory_limit_bytes).await;
331	runtime.set_max_stack_size(limits.max_stack_bytes).await;
332	let interrupt_flag = cancel.flag_arc();
333	runtime
334	.set_interrupt_handler(Some(Box::new(move \|\| {
335	interrupt_flag.load(Ordering::Acquire)
336	})))
337	.await;
338	let context = AsyncContext::full(&runtime)
339	.await
340	.map_err(\|err\| WorkflowJsError::VmInit(err.to_string()))?;
341
342	let result = context
343	.async_with(async \|ctx\| run_in_ctx(ctx, source, args_json, driver, cancel).await)
344	.await;
345	drop(context);
346	runtime.run_gc().await;
347	result
348	}
349
350	async fn run_in_ctx(
351	ctx: Ctx<'_>,
352	source: String,
353	args_json: String,
354	driver: Arc<dyn WorkflowDriver>,
355	cancel: CancelHandle,
356	) -> Result<serde_json::Value, WorkflowJsError> {
357	install_host(&ctx, driver, cancel.clone(), &args_json)?;
358	ctx.eval::<(), _>(prelude())
359	.catch(&ctx)
360	.map_err(\|err\| WorkflowJsError::VmInit(format!("prelude failed: {err}")))?;
361
362	let desugared = desugar_export_default(&source);
363	let wrapped = format!("(async () => {{\n{desugared}\n}})()");
364	let promise = ctx
365	.eval::<Promise, _>(wrapped)
366	.catch(&ctx)
367	.map_err(\|err\| script_error(&cancel, err))?;
368	let value = promise
369	.into_future::<Value>()
370	.await
371	.catch(&ctx)
372	.map_err(\|err\| script_error(&cancel, err))?;
373	js_value_to_json(&ctx, value)
374	}
375
376	/// Rewrite the documented module-style authoring shape
377	/// (`export default async function (args) { ... }`) into the script form the
378	/// VM actually evals. Sources are wrapped in an async IIFE, where the
379	/// module-only `export` keyword is a syntax error, so without this every
380	/// imperative `export default` workflow (including the #4131 dogfood
381	/// fixtures) failed to parse. The default export is captured, invoked with
382	/// the `args` global when it is a function, and its result becomes the run
383	/// result; a non-function default export is returned as-is.
384	fn desugar_export_default(source: &str) -> String {
385	const EXPORT_DEFAULT: &str = "export default";
386	let Some(offset) = line_leading_export_default(source) else {
387	return source.to_string();
388	};
389	let mut out = source.to_string();
390	out.replace_range(
391	offset..offset + EXPORT_DEFAULT.len(),
392	"globalThis.__workflow_default =",
393	);
394	out.push('\n');
395	out.push_str(
396	";{\n const __wf_default = globalThis.__workflow_default;\n delete globalThis.__workflow_default;\n if (typeof __wf_default === \"function\") {\n return await __wf_default(args);\n }\n if (__wf_default !== undefined) {\n return __wf_default;\n }\n}\n",
397	);
398	out
399	}
400
401	/// Return the byte offset of a line-leading `export default` token that is
402	/// actual JavaScript syntax, not text inside a string, template literal, or
403	/// comment. This intentionally recognizes only the documented authoring shape
404	/// instead of attempting to implement a general JavaScript module parser.
405	fn line_leading_export_default(source: &str) -> Option<usize> {
406	const EXPORT_DEFAULT: &[u8] = b"export default";
407	let bytes = source.as_bytes();
408	let mut idx = 0usize;
409	let mut quote = None;
410	let mut escaped = false;
411	let mut line_comment = false;
412	let mut block_comment = false;
413	let mut line_has_only_whitespace = true;
414
415	while idx < bytes.len() {
416	let byte = bytes[idx];
417
418	if line_comment {
419	if byte == b'\n' {
420	line_comment = false;
421	line_has_only_whitespace = true;
422	}
423	idx += 1;
424	continue;
425	}
426
427	if block_comment {
428	if byte == b'*' && bytes.get(idx + 1) == Some(&b'/') {
429	block_comment = false;
430	line_has_only_whitespace = false;
431	idx += 2;
432	continue;
433	}
434	if byte == b'\n' {
435	line_has_only_whitespace = true;
436	} else if !byte.is_ascii_whitespace() {
437	line_has_only_whitespace = false;
438	}
439	idx += 1;
440	continue;
441	}
442
443	if let Some(active_quote) = quote {
444	if byte == b'\n' {
445	line_has_only_whitespace = true;
446	escaped = false;
447	} else {
448	if !byte.is_ascii_whitespace() {
449	line_has_only_whitespace = false;
450	}
451	if escaped {
452	escaped = false;
453	} else if byte == b'\\' {
454	escaped = true;
455	} else if byte == active_quote {
456	quote = None;
457	}
458	}
459	idx += 1;
460	continue;
461	}
462
463	if byte == b'\n' {
464	line_has_only_whitespace = true;
465	idx += 1;
466	continue;
467	}
468	if line_has_only_whitespace && byte.is_ascii_whitespace() {
469	idx += 1;
470	continue;
471	}
472	if line_has_only_whitespace && bytes[idx..].starts_with(EXPORT_DEFAULT) {
473	return Some(idx);
474	}
475
476	line_has_only_whitespace = false;
477	if byte == b'/' && bytes.get(idx + 1) == Some(&b'/') {
478	line_comment = true;
479	idx += 2;
480	} else if byte == b'/' && bytes.get(idx + 1) == Some(&b'*') {
481	block_comment = true;
482	idx += 2;
483	} else {
484	if matches!(byte, b'\'' \| b'"' \| b'`') {
485	quote = Some(byte);
486	}
487	idx += 1;
488	}
489	}
490
491	None
492	}
493
494	fn script_error(cancel: &CancelHandle, err: CaughtError<'_>) -> WorkflowJsError {
495	if cancel.is_cancelled() {
496	WorkflowJsError::Cancelled
497	} else {
498	WorkflowJsError::Script(err.to_string())
499	}
500	}
501
502	fn js_value_to_json<'js>(
503	ctx: &Ctx<'js>,
504	value: Value<'js>,
505	) -> Result<serde_json::Value, WorkflowJsError> {
506	if value.is_undefined() {
507	return Ok(serde_json::Value::Null);
508	}
509	let text = ctx
510	.json_stringify(value)
511	.map_err(\|err\| WorkflowJsError::ResultEncoding(err.to_string()))?;
512	match text {
513	None => Ok(serde_json::Value::Null),
514	Some(text) => {
515	let text = text
516	.to_string()
517	.map_err(\|err\| WorkflowJsError::ResultEncoding(err.to_string()))?;
518	serde_json::from_str(&text)
519	.map_err(\|err\| WorkflowJsError::ResultEncoding(err.to_string()))
520	}
521	}
522	}
523
524	fn install_host(
525	ctx: &Ctx<'_>,
526	driver: Arc<dyn WorkflowDriver>,
527	cancel: CancelHandle,
528	args_json: &str,
529	) -> Result<(), WorkflowJsError> {
530	let globals = ctx.globals();
531
532	let args_value: Value = ctx
533	.json_parse(args_json)
534	.map_err(\|err\| WorkflowJsError::InvalidArgs(err.to_string()))?;
535	globals.set("args", args_value).map_err(init_err)?;
536
537	// Per-run lifetime counter (design §4.3): counts spawn attempts, and the
538	// check + increment happen with no await in between so a parallel burst
539	// cannot slip past the cap on the single-threaded VM.
540	let spawned = Rc::new(Cell::new(0u64));
541
542	let task_driver = driver.clone();
543	let task_cancel = cancel.clone();
544	globals
545	.set(
546	"__workflow_task",
547	Func::from(Async(move \|opts_json: String\| {
548	let driver = task_driver.clone();
549	let cancel = task_cancel.clone();
550	let spawned = spawned.clone();
551	async move { task_host(opts_json, driver, cancel, spawned).await }
552	})),
553	)
554	.map_err(init_err)?;
555
556	let log_driver = driver.clone();
557	globals
558	.set(
559	"__workflow_log",
560	Func::from(move \|message: String\| {
561	log_driver.progress(ProgressEvent::Log { message });
562	}),
563	)
564	.map_err(init_err)?;
565
566	let phase_driver = driver.clone();
567	globals
568	.set(
569	"__workflow_phase",
570	Func::from(move \|title: String\| {
571	phase_driver.progress(ProgressEvent::Phase { title });
572	}),
573	)
574	.map_err(init_err)?;
575
576	// Budget reads are live driver snapshots (design §5.2). NaN encodes
577	// "no ceiling" for `total`; the prelude maps it to `null`.
578	let total_driver = driver.clone();
579	globals
580	.set(
581	"__workflow_budget_total",
582	Func::from(move \|\| -> f64 {
583	match total_driver.budget().total {
584	Some(total) => total as f64,
585	None => f64::NAN,
586	}
587	}),
588	)
589	.map_err(init_err)?;
590
591	let spent_driver = driver.clone();
592	globals
593	.set(
594	"__workflow_budget_spent",
595	Func::from(move \|\| -> f64 { spent_driver.budget().spent as f64 }),
596	)
597	.map_err(init_err)?;
598
599	globals
600	.set(
601	"__workflow_budget_remaining",
602	Func::from(move \|\| -> f64 {
603	match driver.budget().remaining() {
604	Some(remaining) => remaining as f64,
605	None => f64::INFINITY,
606	}
607	}),
608	)
609	.map_err(init_err)?;
610
611	Ok(())
612	}
613
614	fn init_err(err: rquickjs::Error) -> WorkflowJsError {
615	WorkflowJsError::VmInit(err.to_string())
616	}
617
618	/// The `task()` host call. Everything that can go wrong is reported through
619	/// the JSON envelope (`{"error": ...}`) so the prelude re-throws it as a real
620	/// JS `Error` with a script-side stack.
621	async fn task_host(
622	opts_json: String,
623	driver: Arc<dyn WorkflowDriver>,
624	cancel: CancelHandle,
625	spawned: Rc<Cell<u64>>,
626	) -> String {
627	let outcome = task_host_inner(opts_json, driver, cancel, spawned).await;
628	let envelope = match outcome {
629	Ok(value) => serde_json::json!({ "value": value }),
630	Err(message) => serde_json::json!({ "error": message }),
631	};
632	envelope.to_string()
633	}
634
635	/// Best-effort `label`/`phase` from raw `task()` options, for rejection
636	/// receipts when the options never survived parsing.
637	fn task_identity_hint(opts_json: &str) -> (Option<String>, Option<String>) {
638	let value: serde_json::Value =
639	serde_json::from_str(opts_json).unwrap_or(serde_json::Value::Null);
640	let pluck = \|key: &str\| {
641	value
642	.get(key)
643	.and_then(serde_json::Value::as_str)
644	.map(str::trim)
645	.filter(\|text\| !text.is_empty())
646	.map(str::to_string)
647	};
648	(pluck("label"), pluck("phase"))
649	}
650
651	/// Record a pre-spawn `task()` rejection on the host ledger, then hand the
652	/// message back for the JS throw. Rejections that never reach `spawn_task`
653	/// would otherwise be invisible to the run record (#5035's surviving gap).
654	fn reject_task(driver: &Arc<dyn WorkflowDriver>, opts_json: &str, message: String) -> String {
655	let (label, phase) = task_identity_hint(opts_json);
656	driver.progress(ProgressEvent::TaskRejected {
657	label,
658	phase,
659	message: message.clone(),
660	});
661	message
662	}
663
664	async fn task_host_inner(
665	opts_json: String,
666	driver: Arc<dyn WorkflowDriver>,
667	cancel: CancelHandle,
668	spawned: Rc<Cell<u64>>,
669	) -> Result<serde_json::Value, String> {
670	let request = parse_task_options(&opts_json)
671	.map_err(\|message\| reject_task(&driver, &opts_json, message))?;
672	// Compile the schema before spawning so a malformed one fails fast
673	// instead of burning a subagent.
674	let validator = request
675	.response_schema
676	.as_ref()
677	.map(compile_schema)
678	.transpose()
679	.map_err(\|message\| reject_task(&driver, &opts_json, message))?;
680
681	// Lifetime backstop (design §4.3) — checked and bumped before any await.
682	if spawned.get() >= WORKFLOW_LIFETIME_CAP {
683	return Err(reject_task(
684	&driver,
685	&opts_json,
686	format!(
687	"task(): Workflow lifetime agent cap ({WORKFLOW_LIFETIME_CAP}) reached for this run"
688	),
689	));
690	}
691	// Fast-fail budget gate. The authoritative reservation lives in the
692	// driver (design §5.3); this only stops obviously-doomed spawns early.
693	let snapshot = driver.budget();
694	if snapshot.exhausted() {
695	return Err(reject_task(
696	&driver,
697	&opts_json,
698	format!(
699	"task(): budget exhausted ({} of {} tokens spent)",
700	snapshot.spent,
701	snapshot.total.unwrap_or(0)
702	),
703	));
704	}
705	if cancel.is_cancelled() {
706	return Err("task(): run cancelled".to_string());
707	}
708	spawned.set(spawned.get() + 1);
709
710	let spawned_task = driver
711	.spawn_task(request)
712	.await
713	.map_err(\|err\| err.to_string())?;
714	let task_id = spawned_task.task_id;
715	let completion_rx = spawned_task.completion;
716	let completion = tokio::select! {
717	_ = cancel.cancelled() => return Err("task(): run cancelled".to_string()),
718	completion = completion_rx => completion
719	.map_err(\|_\| "task(): driver dropped the completion channel".to_string())?,
720	};
721
722	match completion {
723	TaskCompletion::Completed { text } => match &validator {
724	None => Ok(serde_json::Value::String(text)),
725	Some(validator) => match decode_reply(&text, validator) {
726	Ok(value) => Ok(value),
727	Err(message) => {
728	driver.progress(ProgressEvent::TaskSchemaValidationFailed {
729	task_id,
730	message: message.clone(),
731	});
732	Err(message)
733	}
734	},
735	},
736	TaskCompletion::Failed { message } => Err(format!("task(): subagent failed: {message}")),
737	TaskCompletion::Cancelled => Err("task(): subagent cancelled".to_string()),
738	TaskCompletion::BudgetExhausted { message } => {
739	Err(format!("task(): budget exhausted: {message}"))
740	}
741	}
742	}
743
744	/// JS-facing option names for `task()` (design §3.3). Unknown fields are
745	/// rejected so a typo (`responseschema`) fails loudly instead of being
746	/// silently dropped. Every multi-word field also accepts its snake_case
747	/// spelling, and the `agent` tool's `workspace_policy` name is accepted as an
748	/// alias for worktree isolation — the two spawn surfaces are written by the
749	/// same authors (often models), so a schema that runs on one must not be an
750	/// unknown-field error on the other.
751	#[derive(Debug, Deserialize)]
752	#[serde(rename_all = "camelCase", deny_unknown_fields)]
753	struct TaskOptions {
754	#[serde(alias = "title")]
755	description: Option<String>,
756	prompt: Option<String>,
757	#[serde(alias = "type", alias = "subagent_type")]
758	subagent_type: Option<String>,
759	/// Fleet role name (#4177). Preferred step identity field.
760	role: Option<String>,
761	profile: Option<String>,
762	model: Option<String>,
763	#[serde(alias = "model_strength")]
764	model_strength: Option<String>,
765	thinking: Option<String>,
766	cwd: Option<String>,
767	#[serde(default)]
768	worktree: bool,
769	/// `agent`-tool alias for worktree isolation: "shared" \| "worktree".
770	#[serde(default, alias = "workspace_policy")]
771	workspace_policy: Option<String>,
772	#[serde(alias = "write_authority")]
773	write_authority: Option<String>,
774	#[serde(default, alias = "write_roots")]
775	write_roots: Vec<String>,
776	#[serde(default, alias = "exact_files")]
777	exact_files: Vec<String>,
778	#[serde(default, alias = "coordination_contracts")]
779	coordination_contracts: Vec<String>,
780	#[serde(default)]
781	dependencies: Vec<String>,
782	#[serde(default)]
783	acceptance: Vec<String>,
784	#[serde(alias = "allowed_tools")]
785	allowed_tools: Option<Vec<String>>,
786	#[serde(alias = "max_depth")]
787	max_depth: Option<u32>,
788	#[serde(alias = "token_budget")]
789	token_budget: Option<u64>,
790	#[serde(alias = "max_steps")]
791	max_steps: Option<u32>,
792	#[serde(alias = "wall_time_secs")]
793	wall_time_secs: Option<u64>,
794	#[serde(alias = "response_schema")]
795	response_schema: Option<serde_json::Value>,
796	label: Option<String>,
797	phase: Option<String>,
798	}
799
800	fn parse_task_options(opts_json: &str) -> Result<TaskRequest, String> {
801	let mut options: TaskOptions =
802	serde_json::from_str(opts_json).map_err(\|err\| format!("task(): invalid options: {err}"))?;
803	if let Some(policy) = options.workspace_policy.take() {
804	match policy.trim().to_ascii_lowercase().as_str() {
805	"worktree" => options.worktree = true,
806	"shared" => {
807	if options.worktree {
808	return Err(
809	"task(): workspacePolicy 'shared' conflicts with worktree: true"
810	.to_string(),
811	);
812	}
813	}
814	other => {
815	return Err(format!(
816	"task(): workspacePolicy must be shared or worktree; got {other:?}"
817	));
818	}
819	}
820	}
821	let description = options
822	.prompt
823	.or(options.description)
824	.filter(\|description\| !description.trim().is_empty())
825	.ok_or_else(\|\| "task(): 'description' (or 'prompt') is required".to_string())?;
826	let role = options
827	.role
828	.as_deref()
829	.map(normalize_profile)
830	.transpose()
831	.map_err(\|err\| format!("task(): role: {err}"))?;
832	let profile = options
833	.profile
834	.as_deref()
835	.map(normalize_profile)
836	.transpose()
837	.map_err(\|err\| format!("task(): {err}"))?;
838	options.write_roots = normalize_task_paths("writeRoots", options.write_roots, 32)?;
839	options.exact_files = normalize_task_paths("exactFiles", options.exact_files, 32)?;
840	let cwd = options
841	.cwd
842	.take()
843	.map(\|value\| normalize_task_paths("cwd", vec![value], 1))
844	.transpose()?
845	.and_then(\|mut paths\| paths.pop());
846	options.coordination_contracts =
847	normalize_task_string_list("coordinationContracts", options.coordination_contracts, 16)?;
848	options.dependencies = normalize_task_string_list("dependencies", options.dependencies, 8)?;
849	options.acceptance = normalize_task_string_list("acceptance", options.acceptance, 8)?;
850	let write_authority = options
851	.write_authority
852	.as_deref()
853	.map(\|value\| value.trim().to_ascii_lowercase())
854	.map(\|value\| match value.as_str() {
855	"read_only" \| "workspace_write" \| "worktree_write" => Ok(value),
856	_ => Err(format!(
857	"task(): writeAuthority must be read_only, workspace_write, or worktree_write; got {value:?}"
858	)),
859	})
860	.transpose()?;
861	if write_authority.as_deref() == Some("worktree_write") && !options.worktree {
862	return Err("task(): writeAuthority worktree_write requires worktree: true".to_string());
863	}
864	let role_kind = role.as_deref().and_then(task_role_kind);
865	let type_kind = options.subagent_type.as_deref().and_then(task_role_kind);
866	if let (Some(role_kind), Some(type_kind)) = (role_kind, type_kind)
867	&& role_kind != type_kind
868	{
869	return Err("task(): role and subagentType declare contradictory authorities".to_string());
870	}
871	let declared_kind = role_kind.or(type_kind);
872	if matches!(declared_kind, Some(TaskRoleKind::ReadOnly))
873	&& write_authority
874	.as_deref()
875	.is_some_and(\|authority\| authority != "read_only")
876	{
877	return Err("task(): read-only roles cannot declare write-capable authority".to_string());
878	}
879	if write_authority
880	.as_deref()
881	.is_some_and(\|authority\| authority != "read_only")
882	&& options.write_roots.is_empty()
883	&& options.exact_files.is_empty()
884	&& options.coordination_contracts.is_empty()
885	{
886	return Err(
887	"task(): write-capable authority requires writeRoots, exactFiles, or coordinationContracts"
888	.to_string(),
889	);
890	}
891	let explicit_write_identity = declared_kind == Some(TaskRoleKind::Implementer)
892	\|\| (declared_kind == Some(TaskRoleKind::General)
893	&& (role.is_some() \|\| options.subagent_type.is_some()))
894	\|\| (profile.is_some() && declared_kind.is_none());
895	if explicit_write_identity
896	&& write_authority.as_deref() != Some("read_only")
897	&& options.write_roots.is_empty()
898	&& options.exact_files.is_empty()
899	&& options.coordination_contracts.is_empty()
900	{
901	return Err(
902	"task(): explicit write-capable identities require writeRoots, exactFiles, or coordinationContracts"
903	.to_string(),
904	);
905	}
906	Ok(TaskRequest {
907	description,
908	subagent_type: options.subagent_type,
909	role,
910	profile,
911	model: options.model,
912	model_strength: options.model_strength,
913	thinking: options.thinking,
914	cwd,
915	worktree: options.worktree,
916	write_authority,
917	write_roots: options.write_roots,
918	exact_files: options.exact_files,
919	coordination_contracts: options.coordination_contracts,
920	dependencies: options.dependencies,
921	acceptance: options.acceptance,
922	allowed_tools: options.allowed_tools,
923	// Host-imposed only: a script cannot set (or clear) a deny list.
924	disallowed_tools: Vec::new(),
925	max_depth: options.max_depth,
926	token_budget: options.token_budget,
927	max_steps: options.max_steps,
928	wall_time_secs: options.wall_time_secs,
929	response_schema: options.response_schema,
930	label: options.label,
931	phase: options.phase,
932	})
933	}
934
935	fn normalize_task_string_list(
936	field: &str,
937	values: Vec<String>,
938	limit: usize,
939	) -> Result<Vec<String>, String> {
940	if values.len() > limit {
941	return Err(format!("task(): {field} accepts at most {limit} entries"));
942	}
943	let mut normalized = Vec::new();
944	for value in values {
945	let value = value.trim();
946	if value.is_empty() \|\| value.chars().count() > 512 {
947	return Err(format!(
948	"task(): {field} entries must be 1..=512 characters"
949	));
950	}
951	if !normalized.iter().any(\|existing\| existing == value) {
952	normalized.push(value.to_string());
953	}
954	}
955	Ok(normalized)
956	}
957
958	fn normalize_task_paths(
959	field: &str,
960	values: Vec<String>,
961	limit: usize,
962	) -> Result<Vec<String>, String> {
963	if values.len() > limit {
964	return Err(format!("task(): {field} accepts at most {limit} entries"));
965	}
966	let mut normalized = Vec::new();
967	for raw in values {
968	let raw = raw.trim().replace('\\', "/");
969	let windows_drive = raw.as_bytes().get(1) == Some(&b':')
970	&& raw.as_bytes().first().is_some_and(u8::is_ascii_alphabetic);
971	if raw.is_empty()
972	\|\| raw.chars().count() > 512
973	\|\| raw.starts_with('/')
974	\|\| raw.starts_with("//")
975	\|\| windows_drive
976	\|\| raw.chars().any(\|ch\| matches!(ch, '\0' \| '\r' \| '\n'))
977	{
978	return Err(format!(
979	"task(): {field} entries must be bounded repo-relative paths"
980	));
981	}
982	let mut segments = Vec::new();
983	for segment in raw.split('/') {
984	match segment {
985	"" \| "." => {}
986	".." => {
987	return Err(format!(
988	"task(): {field} paths cannot contain parent traversal"
989	));
990	}
991	value => segments.push(value),
992	}
993	}
994	let path = if segments.is_empty() {
995	".".to_string()
996	} else {
997	segments.join("/")
998	};
999	if !normalized.contains(&path) {
1000	normalized.push(path);
1001	}
1002	}
1003	Ok(normalized)
1004	}
1005
1006	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1007	enum TaskRoleKind {
1008	ReadOnly,
1009	General,
1010	Implementer,
1011	}
1012
1013	fn task_role_kind(value: &str) -> Option<TaskRoleKind> {
1014	match value.trim().to_ascii_lowercase().as_str() {
1015	"explore" \| "explorer" \| "scout" \| "plan" \| "planner" \| "review" \| "reviewer"
1016	\| "verify" \| "verifier" => Some(TaskRoleKind::ReadOnly),
1017	"general" \| "worker" => Some(TaskRoleKind::General),
1018	"implement" \| "implementer" \| "builder" => Some(TaskRoleKind::Implementer),
1019	_ => None,
1020	}
1021	}
1022
1023	/// The JS prelude injected before every script: determinism bans, the
1024	/// `task`/`parallel`/`pipeline`/`log`/`phase` stdlib (design §7), and the
1025	/// `budget` global.
1026	fn prelude() -> String {
1027	PRELUDE_TEMPLATE.replace("__MAX_ITEMS__", &PARALLEL_MAX_ITEMS.to_string())
1028	}
1029
1030	const PRELUDE_TEMPLATE: &str = r#""use strict";
1031	(() => {
1032	const banned = (name) => () => {
1033	throw new Error(name + " is unavailable in Workflow scripts: runs must be deterministic for record/replay");
1034	};
1035	const BannedDate = function Date() {
1036	throw new Error("new Date()/Date() is unavailable in Workflow scripts: runs must be deterministic for record/replay");
1037	};
1038	BannedDate.now = banned("Date.now()");
1039	BannedDate.parse = banned("Date.parse()");
1040	BannedDate.UTC = banned("Date.UTC()");
1041	globalThis.Date = BannedDate;
1042	Math.random = banned("Math.random()");
1043
1044	// Capture temporary host bindings into this closure, then strip them from
1045	// globalThis so scripts only see the documented Workflow surface (#4129).
1046	const hostTask = __workflow_task;
1047	const hostLog = __workflow_log;
1048	const hostPhase = __workflow_phase;
1049	const hostBudgetTotal = __workflow_budget_total;
1050	const hostBudgetSpent = __workflow_budget_spent;
1051	const hostBudgetRemaining = __workflow_budget_remaining;
1052
1053	const MAX_ITEMS = __MAX_ITEMS__;
1054	const taskErrorText = (err) => String(err && err.message !== undefined ? err.message : err);
1055	const isFatalTaskError = (err) => {
1056	const text = taskErrorText(err);
1057	return text.includes("responseSchema") \|\| text.includes("run cancelled");
1058	};
1059
1060	globalThis.task = async (opts) => {
1061	if (opts === null \|\| typeof opts !== "object") {
1062	throw new TypeError("task(): expected an options object");
1063	}
1064	const envelope = JSON.parse(await hostTask(JSON.stringify(opts)));
1065	if (envelope.error !== undefined) {
1066	throw new Error(envelope.error);
1067	}
1068	return envelope.value;
1069	};
1070
1071	globalThis.parallel = (thunks) => {
1072	if (!Array.isArray(thunks)) {
1073	throw new TypeError("parallel(): expected an array of thunks");
1074	}
1075	if (thunks.length > MAX_ITEMS) {
1076	throw new Error("parallel(): max " + MAX_ITEMS + " items per call");
1077	}
1078	return Promise.all(thunks.map((thunk) => {
1079	try {
1080	return Promise.resolve(typeof thunk === "function" ? thunk() : thunk).catch((err) => {
1081	if (isFatalTaskError(err)) throw err;
1082	hostLog("parallel(): dropped a failed slot as null: " + String((err && err.message) \|\| err));
1083	return null;
1084	});
1085	} catch (err) {
1086	if (isFatalTaskError(err)) return Promise.reject(err);
1087	hostLog("parallel(): dropped a failed slot as null: " + String((err && err.message) \|\| err));
1088	return null;
1089	}
1090	}));
1091	};
1092
1093	globalThis.pipeline = (items, ...stages) => {
1094	if (!Array.isArray(items)) {
1095	throw new TypeError("pipeline(): expected an array of items");
1096	}
1097	if (items.length > MAX_ITEMS) {
1098	throw new Error("pipeline(): max " + MAX_ITEMS + " items per call");
1099	}
1100	return Promise.all(items.map(async (item, index) => {
1101	let value = item;
1102	for (const stage of stages) {
1103	try {
1104	value = await stage(value, item, index);
1105	} catch (err) {
1106	if (isFatalTaskError(err)) throw err;
1107	hostLog("pipeline(): dropped item " + index + " as null: " + String((err && err.message) \|\| err));
1108	return null;
1109	}
1110	}
1111	return value;
1112	}));
1113	};
1114
1115	globalThis.log = (message) => {
1116	hostLog(typeof message === "string" ? message : (JSON.stringify(message) ?? String(message)));
1117	};
1118	globalThis.phase = (title) => {
1119	hostPhase(String(title));
1120	};
1121
1122	const total = hostBudgetTotal();
1123	globalThis.budget = Object.freeze({
1124	total: Number.isNaN(total) ? null : total,
1125	spent: () => hostBudgetSpent(),
1126	remaining: () => hostBudgetRemaining(),
1127	});
1128
1129	for (const name of [
1130	"__workflow_task",
1131	"__workflow_log",
1132	"__workflow_phase",
1133	"__workflow_budget_total",
1134	"__workflow_budget_spent",
1135	"__workflow_budget_remaining",
1136	]) {
1137	try {
1138	delete globalThis[name];
1139	} catch (_) {
1140	// Non-configurable bindings stay; the inventory test will fail closed.
1141	}
1142	}
1143	})();
1144	"#;
1145

1145 lines RUST