Compiles again after command subsystem

terrified to start testing

orchid-host/Cargo.toml

@@ -27,6 +27,7 @@ orchid-extension = { version = "0.1.0", path = "../orchid-extension", optional =
 ordered-float = "5.1.0"
 pastey = "0.2.1"
 substack = "1.1.1"
+task-local = "0.1.0"
 tokio = { version = "1.49.0", features = ["process"], optional = true }
 tokio-util = { version = "0.7.18", features = ["compat"], optional = true }
 trait-set = "0.3.0"

orchid-host/src/atom.rs

@@ -94,7 +94,7 @@ impl AtomHand {
 	#[must_use]
 	pub fn ext(&self) -> &Extension { self.sys().ext() }
 	pub async fn req(&self, key: api::TStrv, req: Vec<u8>) -> Option<Vec<u8>> {
-		self.0.owner.client().request(api::Fwded(self.0.api_ref(), key, req)).await.unwrap()
+		self.0.owner.client().request(api::FinalFwded(self.0.api_ref(), key, req)).await.unwrap()
 	}
 	#[must_use]
 	pub fn api_ref(&self) -> api::Atom { self.0.api_ref() }

orchid-host/src/cmd_system.rs

@@ -0,0 +1,249 @@
+use std::borrow::Cow;
+use std::cell::RefCell;
+use std::collections::VecDeque;
+use std::fmt::Debug;
+use std::pin::pin;
+use std::rc::Rc;
+
+use async_event::Event;
+use async_fn_stream::stream;
+use futures::channel::mpsc;
+use futures::future::LocalBoxFuture;
+use futures::stream::FuturesUnordered;
+use futures::{SinkExt, StreamExt, select};
+use never::Never;
+use orchid_base::{OrcErrv, Receipt, ReqHandle, Sym};
+use orchid_extension::{self as ox, AtomicFeatures as _};
+
+use crate::ctx::Ctx;
+use crate::execute::{ExecCtx, ExecResult};
+use crate::expr::{Expr, ExprFromApiCtx, PathSetBuilder};
+use crate::tree::Root;
+
+struct CommandQueueState {
+	new: VecDeque<Expr>,
+	added: Rc<Event>,
+	wants_exit: bool,
+	ctx: Ctx,
+}
+#[derive(Clone)]
+struct CommandQueue(Rc<RefCell<CommandQueueState>>);
+impl CommandQueue {
+	fn new(ctx: Ctx, init: impl IntoIterator<Item = Expr>) -> Self {
+		Self(Rc::new(RefCell::new(CommandQueueState {
+			new: init.into_iter().collect(),
+			added: Rc::default(),
+			wants_exit: false,
+			ctx,
+		})))
+	}
+	pub fn push(&self, expr: Expr) {
+		let was_empty = {
+			let mut g = self.0.borrow_mut();
+			g.new.push_back(expr);
+			g.new.len() == 1
+		};
+		if was_empty {
+			let added = self.0.borrow_mut().added.clone();
+			added.notify_one();
+		}
+	}
+	pub async fn get_new(&self) -> Expr {
+		let added = {
+			let mut g = self.0.borrow_mut();
+			if let Some(waiting) = g.new.pop_front() {
+				return waiting;
+			}
+			g.added.clone()
+		};
+		added.wait_until(|| self.0.borrow_mut().new.pop_front()).await
+	}
+}
+impl Debug for CommandQueue {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("CommandQueue").finish_non_exhaustive()
+	}
+}
+
+pub enum CmdResult {
+	/// All command sequences settled
+	Settled,
+	/// Exit was requested explicitly by usercode
+	Exit,
+	/// Ran out of gas
+	Gas,
+	/// Received a value that wasn't a command
+	///
+	/// This is potentially user error, but implementors may choose to well-define
+	/// certain responses
+	NonCommand(Expr),
+	/// Received an Orchid error
+	///
+	/// This is definitely user error, but implementors may choose to continue
+	/// running other chains of execution after handling it
+	Err(OrcErrv),
+}
+
+pub struct CmdRunner {
+	root: Root,
+	queue: CommandQueue,
+	gas: Option<u64>,
+	interrupted: Option<ExecCtx>,
+	futures: FuturesUnordered<LocalBoxFuture<'static, Option<CmdResult>>>,
+}
+impl CmdRunner {
+	pub async fn new(root: Root, ctx: Ctx, init: impl IntoIterator<Item = Expr>) -> Self {
+		Self {
+			futures: FuturesUnordered::new(),
+			gas: None,
+			root,
+			interrupted: None,
+			queue: CommandQueue::new(ctx, init),
+		}
+	}
+	#[must_use]
+	pub fn get_gas(&self) -> u64 { self.gas.expect("queried gas but no gas was set") }
+	pub fn set_gas(&mut self, gas: u64) { self.gas = Some(gas) }
+	pub fn disable_gas(&mut self) { self.gas = None }
+	pub async fn execute(&mut self) -> CmdResult {
+		let waiting_on_queue = RefCell::new(false);
+		let (mut spawn, mut on_spawn) = mpsc::channel(1);
+		let mut normalize_stream = pin!(
+			stream(async |mut h| {
+				loop {
+					if self.queue.0.borrow().wants_exit {
+						h.emit(CmdResult::Exit).await;
+						break;
+					}
+					waiting_on_queue.replace(false);
+					let mut xctx = match self.interrupted.take() {
+						None => ExecCtx::new(self.root.clone(), self.queue.get_new().await).await,
+						Some(xctx) => xctx,
+					};
+					waiting_on_queue.replace(true);
+					xctx.set_gas(self.gas);
+					let res = xctx.execute().await;
+					match res {
+						ExecResult::Err(e, gas) => {
+							self.gas = gas;
+							h.emit(CmdResult::Err(e)).await;
+						},
+						ExecResult::Gas(exec) => {
+							self.interrupted = Some(exec);
+							h.emit(CmdResult::Gas).await;
+						},
+						ExecResult::Value(val, gas) => {
+							self.gas = gas;
+							let Some(atom) = val.as_atom().await else {
+								h.emit(CmdResult::NonCommand(val)).await;
+								continue;
+							};
+							let queue = self.queue.clone();
+							let ctx = queue.0.borrow_mut().ctx.clone();
+							spawn
+								.send(Box::pin(async move {
+									match atom.ipc(ox::std_reqs::RunCommand).await {
+										None => Some(CmdResult::NonCommand(val)),
+										Some(None) => None,
+										Some(Some(expr)) => {
+											let from_api_cx = ExprFromApiCtx { ctx, sys: atom.api_ref().owner };
+											queue.push(Expr::from_api(expr, PathSetBuilder::new(), from_api_cx).await);
+											None
+										},
+									}
+								}))
+								.await
+								.expect("Receiver is owned by the layer that polls this");
+						},
+					}
+				}
+			})
+			.fuse()
+		);
+		loop {
+			if self.queue.0.borrow().wants_exit {
+				break CmdResult::Exit;
+			}
+			let task = select!(
+				r_opt = self.futures.by_ref().next() => match r_opt {
+					Some(Some(r)) => break r,
+					None if *waiting_on_queue.borrow() => break CmdResult::Settled,
+					None | Some(None) => continue,
+				},
+				r = normalize_stream.by_ref().next() => break r.expect("infinite stream"),
+				task = on_spawn.by_ref().next() => task.expect("sender moved into infinite stream"),
+			);
+			self.futures.push(task)
+		}
+	}
+}
+
+#[derive(Default, Debug)]
+pub struct CmdSystemCard;
+impl ox::SystemCard for CmdSystemCard {
+	type Ctor = CmdSystemCtor;
+	type Req = Never;
+	fn atoms() -> impl IntoIterator<Item = Option<Box<dyn ox::AtomOps>>> {
+		[Some(CommandQueue::ops())]
+	}
+}
+
+#[derive(Debug)]
+pub struct CmdSystemCtor {
+	queue: CommandQueue,
+}
+impl ox::SystemCtor for CmdSystemCtor {
+	const NAME: &'static str = "orchid::cmd";
+	const VERSION: f64 = 0.1;
+	type Card = CmdSystemCard;
+	type Deps = ();
+	type Instance = CmdSystemInst;
+	fn inst(&self, _: <Self::Deps as orchid_extension::DepDef>::Sat) -> Self::Instance {
+		CmdSystemInst { queue: self.queue.clone() }
+	}
+}
+
+fn ox_get_queue() -> CommandQueue { ox::cted::<CmdSystemCtor>().inst.queue.clone() }
+
+#[derive(Debug)]
+pub struct CmdSystemInst {
+	queue: CommandQueue,
+}
+impl ox::System for CmdSystemInst {
+	type Ctor = CmdSystemCtor;
+	async fn env(&self) -> Vec<ox::tree::GenMember> {
+		ox::tree::prefix("orchid::cmd", [
+			ox::tree::cnst(false, "queue", ox::gen_expr::new_atom(ox_get_queue())),
+			ox::tree::fun(true, "spawn", async |side: ox::Expr, cont: ox::Expr| {
+				ox::cmd(async move || {
+					let queue = ox_get_queue();
+					let side_xtk = side.serialize().await;
+					let mut g = queue.0.borrow_mut();
+					let host_ex =
+						g.ctx.exprs.take_expr(side_xtk).expect("Host could not locate leaked expr by ID ");
+					g.new.push_back(host_ex);
+					Some(cont)
+				})
+			}),
+		])
+	}
+	async fn prelude(&self) -> Vec<Sym> { vec![] }
+	fn lexers(&self) -> Vec<ox::LexerObj> { vec![] }
+	fn parsers(&self) -> Vec<ox::ParserObj> { vec![] }
+	async fn request<'a>(
+		&self,
+		_hand: Box<dyn ReqHandle<'a> + 'a>,
+		req: ox::ReqForSystem<Self>,
+	) -> Receipt<'a> {
+		match req {}
+	}
+}
+
+impl ox::Atomic for CommandQueue {
+	type Data = ();
+	type Variant = ox::OwnedVariant;
+}
+impl ox::OwnedAtom for CommandQueue {
+	type Refs = Never;
+	async fn val(&self) -> Cow<'_, Self::Data> { Cow::Owned(()) }
+}

orchid-host/src/execute.rs

@@ -20,9 +20,9 @@ enum StackOp {
 }
 
 pub enum ExecResult {
-	Value(Expr),
+	Value(Expr, Option<u64>),
 	Gas(ExecCtx),
-	Err(OrcErrv),
+	Err(OrcErrv, Option<u64>),
 }
 
 pub struct ExecCtx {
@@ -46,17 +46,6 @@ impl ExecCtx {
 	#[must_use]
 	pub fn idle(&self) -> bool { self.did_pop }
 	#[must_use]
-	pub fn result(self) -> ExecResult {
-		if self.idle() {
-			match &*self.cur {
-				ExprKind::Bottom(errv) => ExecResult::Err(errv.clone()),
-				_ => ExecResult::Value(*self.cur.unbind()),
-			}
-		} else {
-			ExecResult::Gas(self)
-		}
-	}
-	#[must_use]
 	pub fn use_gas(&mut self, amount: u64) -> bool {
 		if let Some(gas) = &mut self.gas {
 			*gas -= amount;
@@ -79,7 +68,7 @@ impl ExecCtx {
 			Err(TryLockError) => panic!("Cycle encountered!"),
 		}
 	}
-	pub async fn execute(&mut self) {
+	pub async fn execute(mut self) -> ExecResult {
 		while self.use_gas(1) {
 			let mut kind_swap = ExprKind::Missing;
 			mem::swap(&mut kind_swap, &mut self.cur);
@@ -135,7 +124,7 @@ impl ExecCtx {
 				StackOp::Nop => (),
 				StackOp::Pop => match self.stack.pop() {
 					Some(top) => self.cur = top,
-					None => return,
+					None => return ExecResult::Value(*self.cur.unbind(), self.gas),
 				},
 				StackOp::Push(sub) => {
 					self.cur_pos = sub.pos();
@@ -150,10 +139,11 @@ impl ExecCtx {
 					}
 					*self.cur = ExprKind::Bottom(err.clone());
 					self.stack = vec![];
-					return;
+					return ExecResult::Err(err.clone(), self.gas);
 				},
 			}
 		}
+		ExecResult::Gas(self)
 	}
 }
 

orchid-host/src/extension.rs

@@ -162,8 +162,10 @@ impl Extension {
 										let sys =
 											ctx.system_inst(atom.owner).await.expect("owner of live atom dropped");
 										let client = sys.client();
-										let reply =
-											client.request(api::Fwded(fw.0.clone(), *key, body.clone())).await.unwrap();
+										let reply = client
+											.request(api::FinalFwded(fw.0.clone(), *key, body.clone()))
+											.await
+											.unwrap();
 										handle.reply(fw, &reply).await
 									},
 									api::ExtHostReq::SysFwd(ref fw @ api::SysFwd(id, ref body)) => {

orchid-host/src/inline.rs

@@ -1,28 +1,18 @@
-#[cfg(feature = "orchid-extension")]
+use std::io;
+use std::rc::Rc;
 use std::time::Duration;
 
-#[cfg(feature = "orchid-extension")]
-use orchid_base::on_drop;
-#[cfg(feature = "orchid-extension")]
+use futures::io::BufReader;
+use futures::{AsyncBufReadExt, StreamExt};
+use orchid_base::{log, on_drop};
 use orchid_extension as ox;
+use unsync_pipe::pipe;
 
-#[cfg(feature = "orchid-extension")]
 use crate::ctx::Ctx;
-#[cfg(feature = "orchid-extension")]
 use crate::extension::ExtPort;
-#[cfg(feature = "orchid-extension")]
 use crate::task_set::TaskSet;
 
-#[cfg(feature = "orchid-extension")]
-pub async fn ext_inline(builder: ox::entrypoint::ExtensionBuilder, ctx: Ctx) -> ExtPort {
-	use std::io;
-	use std::rc::Rc;
-
-	use futures::io::BufReader;
-	use futures::{AsyncBufReadExt, StreamExt};
-	use orchid_base::log;
-	use unsync_pipe::pipe;
-
+pub async fn ext_inline(builder: ox::ExtensionBuilder, ctx: Ctx) -> ExtPort {
 	let (in_stdin, out_stdin) = pipe(1024);
 	let (in_stdout, out_stdout) = pipe(1024);
 	let (in_stderr, out_stderr) = pipe(1024);
@@ -48,7 +38,7 @@ pub async fn ext_inline(builder: ox::entrypoint::ExtensionBuilder, ctx: Ctx) ->
 	std::mem::drop(ctx.clone().spawn(Duration::ZERO, async move {
 		let task_set2 = task_set1.clone();
 		builder
-			.run(ox::ext_port::ExtPort {
+			.run(ox::ExtPort {
 				input: Box::pin(out_stdin),
 				output: Box::pin(in_stdout),
 				log: Box::pin(in_stderr),

orchid-host/src/lib.rs

@@ -2,6 +2,8 @@ use orchid_api as api;
 
 pub mod atom;
 pub mod ctx;
+#[cfg(feature = "orchid-extension")]
+pub mod cmd_system;
 pub mod dealias;
 #[cfg(feature = "tokio")]
 pub mod dylib;
@@ -9,6 +11,7 @@ pub mod execute;
 pub mod expr;
 pub mod expr_store;
 pub mod extension;
+#[cfg(feature = "orchid-extension")]
 pub mod inline;
 pub mod lex;
 pub mod logger;