Ralph25

crates/runtime/include/scuzz_rt.h

@@ -1244,6 +1244,24 @@ void sz_scenario_run_setup(void);
 void sz_testrt_fault_hold(void);
 void sz_testrt_fault_release(void);
 
+/* Fuzz hooks for the evaluator probe (`scuzz eval --probe`). The
+ * registration nodes register when they are built and return IO[Unit].
+ * Sequence them before `sz_fuzz_probe` in one for-comprehension.
+ * Closures are (fn, env) pairs with the SzCont shape. */
+SzIo *sz_fuzz_setup(SzIo *setup); /* setup IO; its value is Scenario.context */
+/* fn(List[String] tokens, env) gives IO[Unit]. nargs 0: empty list. nargs 1:
+ * the rest of the line. More: whitespace tokens; a missing token is "". */
+SzIo *sz_fuzz_driver(SzString *name, int64_t nargs, void *fn, void *env);
+SzIo *sz_fuzz_verify(SzString *name, void *fn, void *env);     /* Timeline to Verdict */
+SzIo *sz_fuzz_verify_rel(SzString *name, void *fn, void *env); /* (Timeline, Timeline) pair to Verdict */
+/* Coverage hit with a key the evaluator interns. */
+void sz_fuzz_hit(SzString *key);
+/* One probe: copy SCUZZ_EV_* to SCUZZ_*, install TestRuntime under
+ * SCUZZ_TESTRT=1, refresh cached env reads, run setup, run the drive script
+ * or `program`, end the session, flush the dumps. Fails with the runtime
+ * message when `program` fails. */
+SzIo *sz_fuzz_probe(SzIo *program);
+
 /* Entrypoint helper used by @main codegen */
 int sz_runtime_main_args(SzIo *program, int argc, char **argv);
 

crates/runtime/src/runtime.c

@@ -2551,7 +2551,8 @@ typedef enum JoinKind {
   JOIN_NONE = 0,
   JOIN_RACE = 1,
   JOIN_BOTH = 2,
-  JOIN_TIMEOUT = 3
+  JOIN_TIMEOUT = 3,
+  JOIN_CANCEL_WAIT = 4 /* winner settled; wait for the loser's finalizers */
 } JoinKind;
 
 typedef struct Fiber {
@@ -2570,6 +2571,8 @@ typedef struct Fiber {
   void *child_val[2];
   SzError *child_err[2];
   int children_settled;
+  int win_slot;       /* JOIN_CANCEL_WAIT: child_val slot to resume with */
+  SzError *wait_err;  /* JOIN_CANCEL_WAIT: fail with this instead */
   int result_ok;
   void *result_value;
   SzError *result_error;
@@ -3134,12 +3137,15 @@ static SzError *deferred_copy_error(SzDeferred *d) {
                   : sz_error_new(1, "deferred failed");
 }
 
-/* Unique parent: take the slot. Shared parent (loop template): retain. */
+/* Unique parent: take the slot. Shared parent (loop template): retain.
+ * A raw (non-RC) env is single-use state that its continuation frees. Take
+ * it even from a shared parent so a later release of the parent does not
+ * touch freed memory. */
 static void *io_slot_child(SzIo *parent, void **slot) {
   void *e = *slot;
   if (!e)
     return NULL;
-  if (parent && sz_is_rc(parent) && sz_rc_hdr(parent)->rc > 1) {
+  if (parent && sz_is_rc(parent) && sz_rc_hdr(parent)->rc > 1 && sz_is_rc(e)) {
     sz_retain(e);
     return e;
   }
@@ -3236,6 +3242,11 @@ static void fiber_cancel(Sched *s, Fiber *f) {
     poller_remove(s, f);
   if (f->state == FIB_FWAIT && f->fwait)
     fiber_join_waiter_remove(f->fwait, f);
+  if (f->wait_err) {
+    sz_error_free(f->wait_err);
+    f->wait_err = NULL;
+  }
+  f->join_kind = JOIN_NONE;
   if (f->children[0])
     fiber_cancel(s, f->children[0]);
   if (f->children[1])
@@ -3299,14 +3310,52 @@ static void fiber_wake_joiners(Sched *s, Fiber *target, int ok, void *val,
   }
 }
 
+static void parent_after_cancel_wait(Sched *s, Fiber *p) {
+  SzError *err = p->wait_err;
+  p->join_kind = JOIN_NONE;
+  p->wait_err = NULL;
+  if (err) {
+    fiber_fail(s, p, err);
+    return;
+  }
+  p->state = FIB_READY;
+  fiber_set_pure_retained(p, p->child_val[p->win_slot]);
+  ready_enqueue(s, p);
+}
+
 static void fiber_settle_cancelled(Sched *s, Fiber *f) {
+  Fiber *p = f->parent;
   f->state = FIB_CANCELLED;
   f->result_ok = 0;
   if (!f->result_error)
     f->result_error = fiber_interrupt_err();
   if (f->forked)
     forked_live_remove(s, f);
   fiber_wake_joiners(s, f, 0, NULL, f->result_error);
+  if (p && p->join_kind == JOIN_CANCEL_WAIT && p->state == FIB_JOIN &&
+      p->children[f->child_slot] == f)
+    parent_after_cancel_wait(s, p);
+}
+
+/* Cancel the loser. The parent resumes after the loser's finalizers run, so
+ * `race`, `timeout`, and `both` return only when no child is still active.
+ * Returns 1 when the parent waits. */
+static int cancel_sibling_then(Sched *s, Fiber *p, int slot, SzError *err) {
+  Fiber *sib = p->children[1 - slot];
+  if (sib)
+    fiber_cancel(s, sib);
+  if (sib && sib->state == FIB_FINALIZING) {
+    p->join_kind = JOIN_CANCEL_WAIT;
+    p->win_slot = slot;
+    p->wait_err = err;
+    return 1;
+  }
+  p->join_kind = JOIN_NONE;
+  if (err) {
+    fiber_fail(s, p, err);
+    return 1;
+  }
+  return 0;
 }
 
 static void join_child_done(Sched *s, Fiber *child, int ok, void *val,
@@ -3324,12 +3373,10 @@ static void join_child_done(Sched *s, Fiber *child, int ok, void *val,
 
   if (p->join_kind == JOIN_RACE) {
     if (ok) {
-      Fiber *sib = p->children[1 - slot];
-      if (sib)
-        fiber_cancel(s, sib);
+      if (cancel_sibling_then(s, p, slot, NULL))
+        return;
       p->state = FIB_READY;
       fiber_set_pure_retained(p, val);
-      p->join_kind = JOIN_NONE;
       ready_enqueue(s, p);
       return;
     }
@@ -3347,33 +3394,25 @@ static void join_child_done(Sched *s, Fiber *child, int ok, void *val,
   }
 
   if (p->join_kind == JOIN_TIMEOUT) {
-    Fiber *sib = p->children[1 - slot];
-    if (sib)
-      fiber_cancel(s, sib);
-    p->join_kind = JOIN_NONE;
-    if (slot == 1) {
-      if (ok) {
-        p->state = FIB_READY;
-        fiber_set_pure_retained(p, val);
-        ready_enqueue(s, p);
-      } else {
-        fiber_fail(s, p, err ? error_copy_or_interrupt(err)
-                             : sz_error_new(1, "timeout inner failed"));
-      }
-    } else {
-      fiber_fail(s, p, sz_error_new(1, "timeout"));
-    }
+    SzError *out = NULL;
+    if (slot == 0)
+      out = sz_error_new(1, "timeout");
+    else if (!ok)
+      out = err ? error_copy_or_interrupt(err)
+                : sz_error_new(1, "timeout inner failed");
+    if (cancel_sibling_then(s, p, slot, out))
+      return;
+    p->state = FIB_READY;
+    fiber_set_pure_retained(p, val);
+    ready_enqueue(s, p);
     return;
   }
 
   if (p->join_kind == JOIN_BOTH) {
     if (!ok) {
-      Fiber *sib = p->children[1 - slot];
-      if (sib)
-        fiber_cancel(s, sib);
-      p->join_kind = JOIN_NONE;
-      fiber_fail(s, p, err ? error_copy_or_interrupt(err)
-                           : sz_error_new(7, "both failed"));
+      cancel_sibling_then(s, p, slot,
+                          err ? error_copy_or_interrupt(err)
+                              : sz_error_new(7, "both failed"));
       return;
     }
     if (p->children_settled >= 2) {

crates/runtime/src/stream.c

@@ -760,7 +760,7 @@ static SzIo *filter_into(SzStream *s, SzList *acc, int64_t remain,
       st->pred = pred;
       st->penv = penv;
       st->stopped = NULL;
-      return fm_drop((SzIo *)s->left, after_filter_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_filter_eval, st);
     }
     case SZ_ST_CONCAT: {
       StTWConcat *st = (StTWConcat *)sz_alloc(sizeof(StTWConcat));
@@ -876,7 +876,7 @@ static SzIo *dropwhile_into(SzStream *s, SzList *acc, int64_t remain,
       st->pred = pred;
       st->penv = penv;
       st->stopped = NULL;
-      return fm_drop((SzIo *)s->left, after_dw_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_dw_eval, st);
     }
     case SZ_ST_CONCAT: {
       StTWConcat *st = (StTWConcat *)sz_alloc(sizeof(StTWConcat));
@@ -1006,7 +1006,7 @@ static SzIo *compile_into(SzStream *s, SzList *acc, int64_t remain) {
       st->tail = (SzStream *)s->right;
       st->acc = acc;
       st->remain = remain_dec(remain);
-      return fm_drop((SzIo *)s->left, after_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_eval, st);
     }
     case SZ_ST_CONCAT: {
       StConcat *st = (StConcat *)sz_alloc(sizeof(StConcat));
@@ -1632,7 +1632,7 @@ static SzIo *stream_step(SzStream *s) {
       StStep *st = (StStep *)sz_alloc(sizeof(StStep));
       sz_retain(s->right);
       st->pin = (SzStream *)s->right;
-      return fm_drop((SzIo *)s->left, after_step_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_step_eval, st);
     }
     case SZ_ST_EVALMAP: {
       StStep *st = (StStep *)sz_alloc(sizeof(StStep));
@@ -2150,7 +2150,7 @@ static SzIo *mapconcat_into(SzStream *s, SzList *acc, int64_t remain,
       st->acc_len = 0;
       st->next = (SzStream *)s->right;
       st->acc = acc;
-      return fm_drop((SzIo *)s->left, after_mc_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_mc_eval, st);
     }
     case SZ_ST_CONCAT: {
       StMc *st = (StMc *)sz_alloc(sizeof(StMc));
@@ -2266,7 +2266,7 @@ static SzIo *changes_into(SzStream *s, SzList *acc, int64_t remain, void *prev,
       st->acc_len = 0;
       st->next = (SzStream *)s->right;
       st->acc = acc;
-      return fm_drop((SzIo *)s->left, after_ch_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_ch_eval, st);
     }
     case SZ_ST_CONCAT: {
       StCh *st = (StCh *)sz_alloc(sizeof(StCh));
@@ -2402,7 +2402,7 @@ static SzIo *flatmap_into(SzStream *s, SzList *acc, int64_t remain,
       st->next = (SzStream *)s->right;
       st->cur = NULL;
       st->acc = acc;
-      return fm_drop((SzIo *)s->left, after_fp_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_fp_eval, st);
     }
     case SZ_ST_CONCAT: {
       StFp *st = (StFp *)sz_alloc(sizeof(StFp));
@@ -2524,7 +2524,7 @@ static SzIo *takewhile_into(SzStream *s, SzList *acc, int64_t remain,
       st->pred = pred;
       st->penv = penv;
       st->stopped = stopped;
-      return fm_drop((SzIo *)s->left, after_tw_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_tw_eval, st);
     }
     case SZ_ST_CONCAT: {
       StTWConcat *st = (StTWConcat *)sz_alloc(sizeof(StTWConcat));
@@ -2705,7 +2705,7 @@ static SzIo *find_into(SzStream *s, SzList *acc, int64_t remain,
       st->pred = pred;
       st->penv = penv;
       st->stopped = found;
-      return fm_drop((SzIo *)s->left, after_find_eval, st);
+      return sz_io_flatmap((SzIo *)s->left, after_find_eval, st);
     }
     case SZ_ST_CONCAT: {
       StTWConcat *st = (StTWConcat *)sz_alloc(sizeof(StTWConcat));
@@ -2738,21 +2738,29 @@ static void *st_release_io(void *env) {
   return NULL;
 }
 
+/* Build the pull graph at run time. The graph carries single-use state in
+ * raw envs, so a shared or repeated compile node must build a fresh graph
+ * on every run. The outer node keeps only the stream (RC). */
+static SzIo *compile_build(void *value, void *env) {
+  SzStream *s = (SzStream *)env;
+  SzIo *body = fm_drop(compile_into(s, sz_list_nil(), -1), reverse_acc, NULL);
+  SzIo *fin = sz_io_delay(st_release_io, s);
+  SzIo *ens = sz_io_ensure(body, fin);
+  (void)value;
+  sz_release(body);
+  sz_release(fin);
+  return ens;
+}
+
 SzIo *sz_stream_compile_to_list(SzStream *s) {
-  SzIo *body;
+  SzIo *io;
   if (!s)
     s = sz_stream_nil();
   else
     sz_retain(s);
-  body = fm_drop(compile_into(s, sz_list_nil(), -1), reverse_acc, NULL);
-  {
-    SzIo *fin = sz_io_delay(st_release_io, s);
-    SzIo *ens = sz_io_ensure(body, fin);
-    sz_release(body);
-    sz_release(fin);
-    sz_release(s);
-    return ens;
-  }
+  io = fm_drop(sz_io_pure(NULL), compile_build, s);
+  sz_release(s);
+  return io;
 }
 
 static SzIo *drain_discard(void *list, void *env) {
@@ -2807,6 +2815,7 @@ static SzIo *fold_from_list(void *list, void *env) {
   void *out = sz_list_fold_left(xs, z, (SzListMapFn)fn, st->outer);
   sz_release(xs);
   sz_release(z);
+  sz_release(st->outer);
   sz_free(st);
   return pure_drop(out);
 }
@@ -2816,6 +2825,7 @@ SzIo *sz_stream_fold(SzStream *s, void *z, SzStreamMapFn f, void *env) {
   if (!f)
     sz_panic("sz_stream_fold(null fn)");
   st = (StLift *)sz_alloc(sizeof(StLift));
+  sz_retain(env);
   st->outer = env;
   st->remain = 0;
   st->tag = 0;