feat(tasks): #408 increment 1b — suspend/resume, scheduler, join, deadlock

CHANGELOG.md

@@ -5,6 +5,20 @@ All notable changes to EigenScript are documented here.
 ## [Unreleased]
 
 ### Added
+- **Deterministic cooperative tasks — increment 1 (#408).** `task_spawn`
+  / `task_yield` / `task_join` / `task_alive`: a single-OS-thread
+  cooperative scheduler whose interleaving is deterministic **by
+  construction** — no trace records, byte-identical across runs. Tasks
+  are reified VM contexts (a copying-stack save-buffer per suspended
+  task, not a 1.28 MB VM each); the JIT and non-atomic refcount fast
+  paths stay live because the scheduler never flips the multithreaded
+  flag. Args and results cross the boundary deep-copied (share-nothing,
+  like channels). `task_join` returns the worker's result or re-raises
+  its uncaught error as the same `{kind, message, line}`. All-tasks-
+  blocked is a new `deadlock` error kind (loud, not a hang); suspending
+  inside an `arena_mark`…`arena_reset` scope or a nested evaluation is a
+  `value` error. Not yet: mailboxes/`task_send` (increment 2), virtual
+  time, the pluggable seeded-strategy hook for the DST consumer.
 - **E003 is scope-precise (#404, increment two)** — the undefined-name
   pass models the runtime's real scope rules instead of a whole-file
   binding set: function-locals (fresh-name `is` and `local`) are

docs/BUILTINS.md

@@ -570,6 +570,8 @@ Requires full build. Transformer model inference and training.
 | `channel_closed` | `channel_closed of channel` | Returns 1 if closed, 0 otherwise. |
 | `task_spawn` | `task_spawn of fn` or `task_spawn of [fn, arg1, ...]` | Create a cooperative task (#408) running `fn` on the single OS thread — deterministic by construction, unlike `spawn`'s OS thread. Args are deep-COPIED (share-nothing, like channel sends), not shared by reference. Returns a numeric task id. (Increment 1a: the task is recorded and reported by `task_alive`; the copying-stack scheduler that runs and interleaves tasks — `task_yield`/`task_join` — lands in a later increment.) |
 | `task_alive` | `task_alive of id` | Returns 1 while the task is runnable or suspended, 0 once it has finished (or for an unknown id). |
+| `task_yield` | `task_yield of null` | Cooperatively hand control to the next ready task; this task resumes round-robin. A no-op when no task has been spawned. Forbidden inside an `arena_mark`…`arena_reset` scope or a nested evaluation (raises `value`). |
+| `task_join` | `task_join of id` | Block until task `id` finishes, then return its deep-copied result — or re-raise its uncaught error (as the same `{kind, message, line}` it died with). Joining an already-finished task returns immediately; an unknown id (or self) returns null. All tasks blocked with none runnable is a `deadlock` error, not a hang. |
 
 **Thread safety:** Values sent through a channel (or returned through
 `thread_join`) are deep-COPIED via `val_clone_for_send` — messages are

docs/DIAGNOSTICS.md

@@ -67,6 +67,7 @@ below are the contract.)
 | `sandbox` | sandbox policy denial or budget | `blocked in sandbox`, `sandbox memory budget exceeded` |
 | `interrupt` | host-requested abort (`eigs_abort`) | `aborted` |
 | `assert` | `assert` builtin failure | `ASSERT FAIL: ...` |
+| `deadlock` | #408 all cooperative tasks blocked, none runnable | `all tasks are blocked — deadlock` |
 | `internal` | VM invariant broke — report it | `unknown opcode`, `SET_LOCAL slot out of range` |
 
 `throw` stamps kind `user` for host/embed introspection

examples/task_pipeline.eigs

@@ -0,0 +1,23 @@
+# Deterministic cooperative tasks (#408, increment 1). Three workers each
+# compute a stage, cooperatively yielding so their progress interleaves —
+# but the interleaving is a pure function of program order, so this program
+# prints byte-identically on every run (no threads, no clock, no tape).
+# `task_join` collects each worker's result; main sums them.
+
+define stage(name, base) as:
+    print of f"{name}: begin"
+    task_yield of null                 # cooperatively let a sibling run
+    local doubled is base * 2
+    print of f"{name}: computed {doubled}"
+    return doubled
+
+a is task_spawn of [stage, "alpha", 10]
+b is task_spawn of [stage, "beta", 20]
+c is task_spawn of [stage, "gamma", 30]
+
+print of "main: workers spawned, collecting"
+ra is task_join of a
+rb is task_join of b
+rc is task_join of c
+
+print of f"results: {ra} + {rb} + {rc} = {ra + rb + rc}"

src/builtins.c

@@ -4449,10 +4449,25 @@ void task_free(Task *t) {
             if (t->args[i]) val_decref(t->args[i]);
         free(t->args);
     }
+    if (t->run_env) env_decref(t->run_env);
     if (t->result) val_decref(t->result);
     if (t->error_value) val_decref(t->error_value);
-    free(t->saved_stack);           /* NULL in 1a; populated in 1b */
-    free(t->saved_frames);
+    /* A task torn down while still SUSPENDED (kill-outstanding at main's end,
+     * or an unjoined task at exit) still owns the counted refs sitting in its
+     * saved slice — decref them so the leak tally stays 0. */
+    if (t->saved_stack) {
+        for (int i = 0; i < t->saved_stack_len; i++)
+            slot_decref(t->saved_stack[i]);
+        free(t->saved_stack);
+    }
+    if (t->saved_frames) {
+        for (int i = 0; i < t->saved_frame_count; i++) {
+            CallFrame *f = &t->saved_frames[i];
+            if (f->owns_env && f->env) env_decref(f->env);
+            if (f->chunk) chunk_decref(f->chunk);
+        }
+        free(t->saved_frames);
+    }
     free(t);
 }
 
@@ -4496,9 +4511,63 @@ Value* builtin_task_spawn(Value *arg) {
         return make_null();
     }
     t->id = id;
+    task_sched_on_spawn(id);   /* enqueue + arm the scheduler */
     return make_num((double)id);
 }
 
+/* task_yield of null — cooperatively hand control to the next ready task.
+ * Returns null (the value of the yield expression); the scheduler saves and
+ * re-enqueues this task at the tail. Forbidden inside an active arena scope
+ * (arena_mark…arena_reset): a suspended task's arena values would be reclaimed
+ * by another task's arena_reset — the task layer and the training arena are
+ * separate tools (Lua-style "can't yield across a C boundary"). */
+Value* builtin_task_yield(Value *arg) {
+    (void)arg;
+    if (!g_task_sched) return make_null();   /* no scheduler: yield is a no-op */
+    if (g_arena.active) {
+        rt_error(EK_VALUE, 0, "cannot task_yield inside an arena scope "
+                 "(arena_mark…arena_reset)");
+        return make_null();
+    }
+    task_request_yield();
+    return make_null();   /* placeholder: execution resumes right after this */
+}
+
+/* task_join of id — block until task `id` finishes; return its (deep-copied)
+ * result, or re-raise its uncaught error. Joining an already-finished task
+ * returns immediately. Joining an unknown id, self, or with no scheduler
+ * returns null. Same arena/nesting restriction as task_yield. */
+Value* builtin_task_join(Value *arg) {
+    if (!arg || arg->type != VAL_NUM || !g_task_sched) return make_null();
+    int target = (int)arg->data.num;
+    Task *t = (Task*)handle_lookup(target, HANDLE_TASK);
+    if (!t) return make_null();
+    if (t->state == TASK_DONE || t->state == TASK_DEAD) {
+        /* Already finished: deliver its result / error now, no suspend. */
+        if (t->has_error) {
+            if (t->error_value) {
+                val_incref(t->error_value);
+                eigs_clear_error_value();
+                g_error_value = t->error_value;
+                g_error_kind = (int)EK_USER;
+            }
+            snprintf(g_error_msg, sizeof(g_error_msg), "joined task %d failed", target);
+            g_has_error = 1;
+            return make_null();
+        }
+        Value *r = t->result ? t->result : make_null();
+        val_incref(r);
+        return r;
+    }
+    if (g_arena.active) {
+        rt_error(EK_VALUE, 0, "cannot task_join inside an arena scope "
+                 "(arena_mark…arena_reset)");
+        return make_null();
+    }
+    if (!task_request_join(target)) return make_null();
+    return make_null();   /* placeholder: the scheduler fills it with the result on resume */
+}
+
 /* task_alive of id → 1 while the task is READY/RUNNING/SUSPENDED, else 0
  * (DONE, DEAD, or an unknown id). */
 Value* builtin_task_alive(Value *arg) {
@@ -5511,6 +5580,8 @@ void register_builtins(Env *env) {
     env_set_local_owned(env, "spawn", make_builtin(builtin_spawn));
     env_set_local_owned(env, "task_spawn", make_builtin(builtin_task_spawn));
     env_set_local_owned(env, "task_alive", make_builtin(builtin_task_alive));
+    env_set_local_owned(env, "task_yield", make_builtin(builtin_task_yield));
+    env_set_local_owned(env, "task_join", make_builtin(builtin_task_join));
     env_set_local_owned(env, "thread_join", make_builtin(builtin_thread_join));
     env_set_local_owned(env, "channel", make_builtin(builtin_channel));
     env_set_local_owned(env, "send", make_builtin(builtin_send));

src/eigenscript.c

@@ -87,6 +87,7 @@ const char* err_kind_name(ErrKind k) {
         case EK_SANDBOX:        return "sandbox";
         case EK_INTERRUPT:      return "interrupt";
         case EK_ASSERT:         return "assert";
+        case EK_DEADLOCK:       return "deadlock";
         case EK_USER:           return "user";
     }
     return "internal";

src/eigenscript.h

@@ -624,6 +624,11 @@ struct EigsThread {
     int                  vts_depth;
     int                  json_depth;
     int                  native_call_depth;
+    /* #408 cooperative task scheduler — opaque (TaskScheduler defined in
+     * vm.c; Task lives in vm.h, not visible here). Allocated lazily on the
+     * first task_spawn, freed at thread detach. Per-OS-thread because tasks
+     * are single-threaded by construction. */
+    void                *task_sched;
     /* Registry list — set by eigs_thread_attach. */
     EigsThread *next;
 };
@@ -695,6 +700,7 @@ extern __thread EigsThread *eigs_current;
 #define g_vts_depth           (eigs_current->vts_depth)
 #define g_json_depth          (eigs_current->json_depth)
 #define g_native_call_depth   (eigs_current->native_call_depth)
+#define g_task_sched          (eigs_current->task_sched)
 #define g_entry_threshold     (eigs_current->state->jit_entry_threshold)
 #define g_iter_threshold      (eigs_current->state->jit_iter_threshold)
 #define g_osr_threshold       (eigs_current->state->jit_osr_threshold)
@@ -984,6 +990,7 @@ typedef enum {
     EK_SANDBOX,           /* sandbox policy denial or budget exhaustion */
     EK_INTERRUPT,         /* host-requested abort (eigs_abort) */
     EK_ASSERT,            /* assert builtin failure */
+    EK_DEADLOCK,          /* #408 all cooperative tasks blocked, none runnable */
     EK_USER,              /* `throw` — catch binds the thrown value, not a dict */
 } ErrKind;
 const char* err_kind_name(ErrKind k);

src/state.c

@@ -139,6 +139,7 @@ void eigs_thread_detach(void) {
      * so any bridge-macro reads inside the destructors stay valid). */
     jit_thread_destroy(th);
     vm_thread_destroy(th);
+    task_sched_thread_free();   /* #408: release the cooperative task scheduler */
 
     /* Phase 8: release freelist + intern memory before the EigsThread
      * struct itself goes. Must run while eigs_current still points at th