Spike: console /await, DECISIONS.md, SPIKE.md

cli/src/console.rs

@@ -28,12 +28,16 @@ const HELP: &str = "\
 /deploy <actor> <dir>  build a Rust folder with cargo, then spawn or upgrade
 /list                  list actors
 /sleep <seconds>       wait (for scripts)
+/await <actor> <secs>  wait for the next message from an actor (for scripts)
 /quit                  stop all actors and exit";
 
 pub struct Console {
     broker: Arc<Broker>,
     waiters: Mutex<HashMap<String, oneshot::Sender<Envelope>>>,
     seq: AtomicU64,
+    /// Number of messages received per sender, for `/await`.
+    seen: Mutex<HashMap<String, u64>>,
+    arrived: tokio::sync::Notify,
 }
 
 fn show(env: &Envelope) {
@@ -65,12 +69,20 @@ fn show(env: &Envelope) {
 
 impl Console {
     pub fn new(broker: Arc<Broker>) -> Arc<Self> {
-        Arc::new(Self { broker, waiters: Mutex::new(HashMap::new()), seq: AtomicU64::new(1) })
+        Arc::new(Self {
+            broker,
+            waiters: Mutex::new(HashMap::new()),
+            seq: AtomicU64::new(1),
+            seen: Mutex::new(HashMap::new()),
+            arrived: tokio::sync::Notify::new(),
+        })
     }
 
     pub async fn run_mailbox(self: Arc<Self>, mut mailbox: mpsc::Receiver<Envelope>) {
         while let Some(env) = mailbox.recv().await {
             show(&env);
+            *self.seen.lock().unwrap().entry(env.from.clone()).or_default() += 1;
+            self.arrived.notify_waiters();
             if let Some(r) = &env.reply_to
                 && let Some(tx) = self.waiters.lock().unwrap().remove(r)
             {
@@ -161,6 +173,24 @@ impl Console {
                     let secs: f64 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(1.0);
                     tokio::time::sleep(Duration::from_secs_f64(secs)).await;
                 }
+                Some("await") => {
+                    let from = parts.next().unwrap_or("agent").to_string();
+                    let secs: f64 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(60.0);
+                    let count = |c: &Self| c.seen.lock().unwrap().get(&from).copied().unwrap_or(0);
+                    let before = count(self);
+                    let deadline = tokio::time::Instant::now() + Duration::from_secs_f64(secs);
+                    loop {
+                        let notified = self.arrived.notified();
+                        tokio::pin!(notified);
+                        notified.as_mut().enable();
+                        if count(self) != before {
+                            break;
+                        }
+                        if tokio::time::timeout_at(deadline, notified).await.is_err() {
+                            bail!("no message from {from} within {secs} s");
+                        }
+                    }
+                }
                 Some("deploy") => {
                     let (Some(actor), Some(dir)) = (parts.next(), parts.next()) else {
                         bail!("usage: /deploy <actor> <dir>");