Chez concurrency pt.3: async agents (per-agent serialized dispatch)

Replace the Janet synchronous agent shim (agent = atom, send applies inline) with
JVM-style async agents: send/send-off enqueue an action and a single worker thread
per agent applies them in order; deref reads the current (maybe not-yet-updated)
state without blocking; await blocks until the queue drains. A validator rejection
or a thrown action puts the agent in an error state (agent-error) and halts the
queue; restart-agent clears it. send and send-off share one serialized worker (a
superset of the JVM's fixed/cached pool split). Native versions re-asserted in
post-prelude over the overlay; await/restart-agent are new.

Corpus: the two "send/send-off applies" cases do (send a f) (deref a) with no
await, so they now read state before the action runs — diverging like the JVM
(the suite was literally "synchronous shim"). Allowlisted on both gates; floors
-2 (zero-Janet 2569->2567, prelude 2559->2557). cli-test covers async agents via
await (ordered 100-send dispatch, error capture) — 49/49. Janet gate + JVM cert
green; 0 new divergences on either corpus.

jolt-byjr

host/chez/concurrency.ss

@@ -143,9 +143,81 @@
                         (else (jolt-promise-delivered? p)))))))
     (if got (jolt-promise-value p) timeout-val)))
 
+;; --- agents (async, per-agent serialized dispatch) --------------------------
+;; JVM semantics, not Janet's synchronous shim: send/send-off enqueue an action
+;; and a single worker thread applies them to the state IN ORDER; deref reads the
+;; (possibly not-yet-updated) state without blocking; await blocks until the queue
+;; drains. An action error is captured (agent-error) and stops the queue.
+(define-record-type jolt-agent
+  (fields (mutable state) (mutable err) (mutable validator)
+          (mutable queue) (mutable running?) mu cv)
+  (nongenerative jolt-agent-v1))
+
+;; (agent state) / (agent state :validator f :error-mode m :meta x): only :validator
+;; has runtime behaviour here; other opts are accepted/ignored.
+(define (jolt-agent-new state . opts)
+  (let loop ((o opts) (validator jolt-nil))
+    (cond
+      ((or (null? o) (null? (cdr o)))
+       (make-jolt-agent state jolt-nil validator '() #f (make-mutex) (make-condition)))
+      ((and (keyword-t? (car o)) (string=? (keyword-t-name (car o)) "validator"))
+       (loop (cddr o) (cadr o)))
+      (else (loop (cddr o) validator)))))
+
+;; Drain the queue, applying each action (f state arg*) outside the lock (an action
+;; may send/deref the same agent). A validator rejection or a thrown action puts the
+;; agent in an error state and halts the queue (JVM :fail mode).
+(define (jolt-agent-worker a)
+  (let loop ()
+    (let ((act (with-mutex (jolt-agent-mu a)
+                 (if (or (not (jolt-nil? (jolt-agent-err a))) (null? (jolt-agent-queue a)))
+                     (begin (jolt-agent-running?-set! a #f)
+                            (condition-broadcast (jolt-agent-cv a)) #f)
+                     (let ((x (car (jolt-agent-queue a))))
+                       (jolt-agent-queue-set! a (cdr (jolt-agent-queue a))) x)))))
+      (when act
+        (guard (e (#t (with-mutex (jolt-agent-mu a)
+                        (jolt-agent-err-set! a e)
+                        (condition-broadcast (jolt-agent-cv a)))))
+          (let ((nv (apply jolt-invoke (car act) (jolt-agent-state a) (cdr act))))
+            (let ((vf (jolt-agent-validator a)))
+              (when (and (not (jolt-nil? vf)) (jolt-not (jolt-invoke vf nv)))
+                (error #f "Invalid reference state")))
+            (jolt-agent-state-set! a nv)))
+        (loop)))))
+
+;; send / send-off: enqueue the action, start the worker if idle. (jolt treats them
+;; identically — one serialized worker per agent — which is observably a superset of
+;; the JVM's fixed/cached pool split.)
+(define (jolt-agent-send a f . args)
+  (with-mutex (jolt-agent-mu a)
+    (jolt-agent-queue-set! a (append (jolt-agent-queue a) (list (cons f args))))
+    (unless (jolt-agent-running? a)
+      (jolt-agent-running?-set! a #t)
+      (fork-thread (lambda () (jolt-agent-worker a)))))
+  a)
+
+;; (await & agents): block until each agent's queue has drained.
+(define (jolt-agent-await . agents)
+  (for-each
+   (lambda (a)
+     (with-mutex (jolt-agent-mu a)
+       (let loop ()
+         (when (or (jolt-agent-running? a) (pair? (jolt-agent-queue a)))
+           (condition-wait (jolt-agent-cv a) (jolt-agent-mu a)) (loop)))))
+   agents)
+  jolt-nil)
+
+(define (jolt-agent-error a) (jolt-agent-err a))
+(define (jolt-agent-restart a new-state . _opts)
+  (jolt-agent-err-set! a jolt-nil)
+  (jolt-agent-state-set! a new-state)
+  a)
+
 ;; --- deref extension --------------------------------------------------------
-;; Chain the fully-built jolt-deref (atoms/vars/volatiles/reduced) with futures +
+;; Chain the fully-built jolt-deref (atoms/vars/volatiles/reduced) with futures,
-;; promises, and accept the timed (deref ref ms val) arity for both.
+;; promises, and agents, and accept the timed (deref ref ms val) arity for the
+;; blocking ref types.
 (define %pre-conc-deref jolt-deref)
 (set! jolt-deref
   (lambda (x . opts)
@@ -156,6 +228,7 @@
       ((jolt-promise? x)
        (if (null? opts) (jolt-promise-deref x)
            (jolt-promise-deref-timed x (car opts) (cadr opts))))
+      ((jolt-agent? x) (jolt-agent-state x))
       (else (apply %pre-conc-deref x opts)))))
 
 ;; realized? for a Chez future/promise (the overlay reads Janet map keys). Wrap the
@@ -173,4 +246,11 @@
 (def-var! "clojure.core" "future-cancelled?" jolt-native-future-cancelled?)
 (def-var! "clojure.core" "promise" jolt-promise-new)
 (def-var! "clojure.core" "deliver" jolt-deliver)
+(def-var! "clojure.core" "agent" jolt-agent-new)
+(def-var! "clojure.core" "agent?" jolt-agent?)
+(def-var! "clojure.core" "send" jolt-agent-send)
+(def-var! "clojure.core" "send-off" jolt-agent-send)
+(def-var! "clojure.core" "await" jolt-agent-await)
+(def-var! "clojure.core" "agent-error" jolt-agent-error)
+(def-var! "clojure.core" "restart-agent" jolt-agent-restart)
 (def-var! "clojure.core" "deref" jolt-deref)

host/chez/post-prelude.ss

@@ -48,6 +48,16 @@
 (def-var! "clojure.core" "future?" jolt-future?)
 (def-var! "clojure.core" "promise" jolt-promise-new)
 (def-var! "clojure.core" "deliver" jolt-deliver)
+;; agents: the overlay (50-io) is a synchronous shim (agent = atom, send applies
+;; immediately). Re-assert the native async agents (per-agent serialized worker),
+;; matching the JVM. await/restart-agent are new (the overlay has neither).
+(def-var! "clojure.core" "agent" jolt-agent-new)
+(def-var! "clojure.core" "agent?" jolt-agent?)
+(def-var! "clojure.core" "send" jolt-agent-send)
+(def-var! "clojure.core" "send-off" jolt-agent-send)
+(def-var! "clojure.core" "await" jolt-agent-await)
+(def-var! "clojure.core" "agent-error" jolt-agent-error)
+(def-var! "clojure.core" "restart-agent" jolt-agent-restart)
 (def-var! "clojure.core" "deref" jolt-deref)
 (let ((overlay-realized? (var-deref "clojure.core" "realized?")))
   (def-var! "clojure.core" "realized?"

test/chez/cli-test.janet

@@ -76,7 +76,14 @@
    ["(require (quote [clojure.core.async :refer [chan go <! >! <!! alts!]])) (def x (chan)) (def y (chan)) (go (>! y :v)) (<!! (go (let [[v ch] (alts! [x y])] (and (= v :v) (= ch y)))))" "true"]
    ["(require (quote [clojure.core.async :refer [chan go go-loop <! >! <!! close!]])) (def c (chan 10 (map inc))) (go (>! c 1) (>! c 2) (>! c 3) (close! c)) (<!! (go-loop [o []] (let [v (<! c)] (if (nil? v) o (recur (conj o v))))))" "[2 3 4]"]
    ["(require (quote [clojure.core.async :refer [timeout <!!]])) (<!! (timeout 10)) :done" ":done"]
-   ["(require (quote [clojure.core.async :refer [chan go <! >! <!!]])) (def ^:dynamic *x* 0) (<!! (binding [*x* 7] (go (<! (clojure.core.async/timeout 5)) *x*)))" "7"]])
+   ["(require (quote [clojure.core.async :refer [chan go <! >! <!!]])) (def ^:dynamic *x* 0) (<!! (binding [*x* 7] (go (<! (clojure.core.async/timeout 5)) *x*)))" "7"]
+   # jolt-byjr: async agents (serialized per-agent dispatch); await for determinism.
+   ["(deref (agent 0))" "0"]
+   ["(let [a (agent 0)] (send-off a + 5) (await a) (deref a))" "5"]
+   ["(let [a (agent 1)] (send a + 6) (await a) (deref a))" "7"]
+   ["(let [a (agent 0)] (dotimes [_ 100] (send a inc)) (await a) (deref a))" "100"]
+   ["(agent-error (agent 0))" ""]
+   ["(let [a (agent 0)] (send a (fn [_] (throw (ex-info \"boom\" {})))) (await a) (boolean (agent-error a)))" "true"]])
 
 (each [src want] cases
   (def [code out err] (joltc src))

test/chez/run-corpus-prelude.janet

@@ -102,7 +102,12 @@
    # usually loses -> false), like the JVM; the Janet :expected relies on its
    # cooperative scheduler. Flaky/divergent.
    "cancel an in-flight future returns true" true
-   "future-cancelled? after cancel" true})
+   "future-cancelled? after cancel" true
+   # agents are async on Chez (per-agent serialized dispatch) = JVM, not Janet's
+   # synchronous shim: (send a f) (deref a) reads state before the action runs, so
+   # these sync-shim cases diverge like the JVM. Per jvm-parity-north-star.
+   "send applies" true
+   "send-off applies" true})
 
 # Cases that BLOCK forever on a shared-heap / JVM host (profile.edn :bucket
 # :timeout) — skip, like :throws, so a hung per-case process can't stall the gate.
@@ -305,8 +310,10 @@
 # 2280->2295, 0 new divergences.
 # jolt-byjr (real-thread futures/pmap on Chez): future-call resolves, so the
 # future/deref/pmap cases run instead of crashing. 2534->2559, 0 new divergences.
+# Then -2 when agents went async (the two "send/send-off applies" sync-shim cases
+# match the JVM's async raciness and are allowlisted) -> 2557.
 # Strided runs scale down.
-(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_PRELUDE_FLOOR") "2559")))
+(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_PRELUDE_FLOOR") "2557")))
 (def floor (if (os/getenv "JOLT_CORPUS_LIMIT") 0 base-floor))
 (when (or (> (length diverged) 0) (< pass floor))
   (printf "REGRESSION: pass %d < floor %d or %d new divergence(s)" pass floor (length diverged)))

test/chez/run-corpus-zero-janet.janet

@@ -89,7 +89,13 @@
     # future usually completes before the cancel (cancel -> false), like the JVM.
     # The Janet :expected "true" relies on cooperative scheduling. Flaky/divergent.
     "cancel an in-flight future returns true" true
-    "future-cancelled? after cancel" true})
+    "future-cancelled? after cancel" true
+    # agents are async on Chez (per-agent serialized dispatch) = JVM, not Janet's
+    # synchronous shim: (send a f) (deref a) reads the state BEFORE the action runs
+    # (use await to observe the result), so these sync-shim cases now diverge like
+    # the JVM. Deliberate per jvm-parity-north-star.
+    "send applies" true
+    "send-off applies" true})
 
 # Cases that BLOCK forever on a shared-heap / JVM host (profile.edn :bucket
 # :timeout) — skip them, like :throws: a single hung case would stall the whole
@@ -183,7 +189,10 @@
 
 # Regression floor: raise as the Chez-hosted compiler closes gaps. The gate fails
 # on any NEW divergence or if pass drops below the floor. Strided runs scale to 0.
-(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_ZJ_FLOOR") "2569")))
+# 2569 after futures/pmap (jolt-byjr pt.1); -2 when agents went async (the two
+# "send/send-off applies" sync-shim cases now match the JVM's async raciness and
+# are allowlisted) -> 2567.
+(def base-floor (scan-number (or (os/getenv "JOLT_CHEZ_ZJ_FLOOR") "2567")))
 (def floor (if (os/getenv "JOLT_CORPUS_LIMIT") 0 base-floor))
 (when (or (> (length diverged) 0) (< pass floor))
   (printf "REGRESSION: pass %d < floor %d or %d new divergence(s)" pass floor (length diverged)))