feat(tasks): #408 increment 3 — virtual time (task_sleep/task_now)

CHANGELOG.md

@@ -5,6 +5,20 @@ All notable changes to EigenScript are documented here.
 ## [Unreleased]
 
 ### Added
+- **Cooperative task virtual time — increment 3 (#408).** `task_sleep of
+  ticks` suspends a task on a **logical clock**; `task_now of null` reads
+  it. The clock starts at 0 and only jumps *forward to the earliest
+  sleeper* when nothing else is runnable (discrete-event simulation) —
+  so a program that sleeps runs in zero real time and stays deterministic
+  by construction: identical on two fresh processes, replays
+  byte-identically, and records **zero tape `N` records** (the clock is
+  not a nondeterministic source). Sleepers waiting on the clock are no
+  longer mistaken for a `deadlock`; ties at the same wake time break by
+  ascending task id. A negative sleep clamps to 0; `task_sleep of 0` is a
+  same-tick yield; with no task ever spawned it is a no-op like
+  `task_yield`. This is the election-timeout / heartbeat primitive the
+  deterministic Raft simulator (liferaft) needs. Not yet: the pluggable
+  seeded-strategy hook for the DST consumer.
 - **Cooperative task mailboxes — increment 2 (#408).** `task_send` /
   `task_recv` / `task_try_recv` / `task_kill`: unbounded FIFO mailboxes
   (Erlang-style; bounded/backpressure is a later add) with messages

docs/BUILTINS.md

@@ -576,6 +576,8 @@ Requires full build. Transformer model inference and training.
 | `task_recv` | `task_recv of null` | Return the next message from this task's mailbox, or block cooperatively until one arrives. Forbidden inside an `arena_mark`…`arena_reset` scope or a nested evaluation (raises `value`). |
 | `task_try_recv` | `task_try_recv of null` | Non-blocking receive: the next mailbox message, or `null` if empty. Never suspends. |
 | `task_kill` | `task_kill of id` | Tear down task `id`: drop its mailbox, mark it dead, wake any joiner with an `interrupt` error. Returns 1 if killed, 0 for a finished/unknown/self target. |
+| `task_sleep` | `task_sleep of ticks` | Suspend this task until the **virtual clock** advances by `ticks`. The clock is logical (discrete-event): it only jumps forward — to the earliest sleeper — when nothing else is runnable, so sleeping stays deterministic, not wall-clock. A negative sleep is treated as 0. A no-op when no task has been spawned. Forbidden inside an `arena_mark`…`arena_reset` scope. |
+| `task_now` | `task_now of null` | The current virtual-clock value (a number; 0 before any `task_sleep`). Deterministic — reads a logical counter, records no nondeterminism. |
 
 **Thread safety:** Values sent through a channel (or returned through
 `thread_join`) are deep-COPIED via `val_clone_for_send` — messages are

docs/SPEC.md

@@ -1189,6 +1189,39 @@ print of (task_join of worker_id)
 42
 ```
 
+### Virtual time
+
+`task_sleep of ticks` suspends a task until a **virtual clock** advances by
+`ticks`; `task_now of null` reads that clock. The clock is *logical*, not
+wall-clock: it starts at 0 and only ever jumps **forward to the earliest
+sleeper** when nothing else is runnable. So a program that sleeps runs in
+zero real time and — like the rest of the task layer — replays identically,
+with no dependence on how fast the machine is. Tasks therefore resume in
+virtual-time order regardless of the order they were spawned, and the clock
+lands on the last wake time.
+
+```eigenscript
+log is []
+define nap(tag, ticks) as:
+    task_sleep of ticks
+    t is task_now of null
+    log is append of [log, f"{tag}@{t}"]
+    return tag
+
+a is task_spawn of [nap, "a", 30]
+b is task_spawn of [nap, "b", 10]
+c is task_spawn of [nap, "c", 20]
+task_join of a
+task_join of b
+task_join of c
+print of log
+print of (task_now of null)
+```
+```output
+["b@10", "c@20", "a@30"]
+30
+```
+
 ## Buffers
 
 `buffer of count` allocates a flat array of `count` nums (all 0).

examples/task_virtual_time.eigs

@@ -0,0 +1,27 @@
+# Virtual time (#408). A watchdog and a worker race on a *logical* clock.
+# The worker sleeps 100 ticks of "work"; the watchdog sleeps a 40-tick
+# timeout. Because time is virtual — the scheduler jumps the clock forward to
+# the earliest sleeper only when nothing else can run — the whole race
+# resolves in ZERO real time and prints byte-identically on every machine and
+# every run (no threads, no wall clock, no tape nondeterminism). This is the
+# election-timeout / heartbeat shape a deterministic Raft simulator needs.
+
+deadline is 40
+
+define worker() as:
+    task_sleep of 100                      # a long job, measured in virtual ticks
+    return "done"
+
+define watchdog(target) as:
+    task_sleep of deadline
+    if task_alive of target:
+        task_kill of target                # timed out first — cancel the worker
+        return f"timeout at t={task_now of null}"
+    return f"worker finished by t={task_now of null}"
+
+w is task_spawn of worker
+d is task_spawn of [watchdog, w]
+
+verdict is task_join of d
+print of verdict
+print of f"clock rests at t={task_now of null}"

src/builtins.c

@@ -4633,6 +4633,36 @@ Value* builtin_task_alive(Value *arg) {
     return make_num(alive ? 1 : 0);
 }
 
+/* task_sleep of ticks — suspend the current task until the virtual clock
+ * advances by `ticks`. The clock is logical (discrete-event): it only jumps
+ * forward, to the earliest sleeper, when nothing else is runnable — so this is
+ * deterministic-by-construction like the rest of the task layer, NOT wall time.
+ * A negative sleep is treated as 0 (a same-tick yield to everything ready). No
+ * scheduler (no task ever spawned) → no time to pass, so it is a no-op, like
+ * task_yield. Same arena/nesting restriction as the other suspending builtins. */
+Value* builtin_task_sleep(Value *arg) {
+    if (!g_task_sched) return make_null();   /* no scheduler: nothing to wait for */
+    if (!arg || arg->type != VAL_NUM) {
+        rt_error(EK_TYPE, 0, "task_sleep requires a number of ticks");
+        return make_null();
+    }
+    if (g_arena.active) {
+        rt_error(EK_VALUE, 0, "cannot task_sleep inside an arena scope "
+                 "(arena_mark…arena_reset)");
+        return make_null();
+    }
+    task_request_sleep(arg->data.num);
+    return make_null();   /* placeholder: execution resumes when the clock wakes it */
+}
+
+/* task_now of null → the current virtual-clock value (a number, 0 before any
+ * task_sleep and 0 when no scheduler is active). Deterministic; reads a logical
+ * counter, so it records no tape nondet. */
+Value* builtin_task_now(Value *arg) {
+    (void)arg;
+    return make_num(task_virtual_now());
+}
+
 /* Deterministic teardown of OS-resource handles, run once the program has
  * finished executing (the full value world is still alive, so buffered-message
  * decrefs are safe). Channels and thread handles live in the process handle
@@ -5640,6 +5670,8 @@ void register_builtins(Env *env) {
     env_set_local_owned(env, "task_recv", make_builtin(builtin_task_recv));
     env_set_local_owned(env, "task_try_recv", make_builtin(builtin_task_try_recv));
     env_set_local_owned(env, "task_kill", make_builtin(builtin_task_kill));
+    env_set_local_owned(env, "task_sleep", make_builtin(builtin_task_sleep));
+    env_set_local_owned(env, "task_now", make_builtin(builtin_task_now));
     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/vm.c

@@ -5128,6 +5128,7 @@ typedef struct {
     int   live;                   /* spawned tasks not yet DONE/DEAD */
     int   active;                 /* armed on first spawn */
     int   dead_letters;           /* inc 2: sends to finished/unknown tasks */
+    double now;                   /* inc 3: virtual clock (logical, starts 0) */
     Task  main_task;              /* task 0 — save-buffer only, never "started" */
 } TaskScheduler;
 
@@ -5308,6 +5309,65 @@ void task_request_recv(void) {
     g_task_suspend_request = 1;
 }
 
+/* ---- Inc 3: virtual time ---------------------------------------------- */
+
+/* task_sleep: the current task becomes runnable again when the virtual clock
+ * reaches now + ticks. The trampoline advances the clock only when nothing is
+ * runnable (see sched_wake_sleepers), so time is a pure function of program
+ * order + sleep durations — no tape records, no wall clock. A negative sleep
+ * is clamped to 0 (a same-tick yield to everything currently ready). */
+void task_request_sleep(double ticks) {
+    TaskScheduler *s = sched_get();
+    Task *t = task_current_running();
+    if (!s || !t) return;
+    double dt = ticks > 0 ? ticks : 0;
+    t->wake_at = s->now + dt;
+    t->sleeping = 1;
+    g_task_suspend_request = 1;
+}
+
+double task_virtual_now(void) {
+    TaskScheduler *s = sched_get();
+    return s ? s->now : 0;
+}
+
+/* When the ready queue is empty, advance the virtual clock to the earliest
+ * sleeper's wake time and make every task due at (or before) that instant
+ * runnable. Returns 1 if any sleeper was woken (the trampoline then loops),
+ * 0 if there are no sleepers (genuine idle → done or deadlock). Ties at the
+ * same wake_at are broken by ascending task id (main = 0 first), so the
+ * interleaving stays deterministic. The clock only ever moves forward:
+ * wake_at = now + dt >= now, so the min is never behind the current now. */
+static int sched_wake_sleepers(TaskScheduler *s) {
+    double best = 0; int have_best = 0;
+    if (s->main_task.state == TASK_SUSPENDED && s->main_task.sleeping) {
+        best = s->main_task.wake_at; have_best = 1;
+    }
+    for (int i = 1; i < HANDLE_TABLE_SIZE; i++) {
+        Task *t = (Task *)handle_lookup(i, HANDLE_TASK);
+        if (t && t->state == TASK_SUSPENDED && t->sleeping &&
+            (!have_best || t->wake_at < best)) {
+            best = t->wake_at; have_best = 1;
+        }
+    }
+    if (!have_best) return 0;
+    s->now = best;
+    /* Wake main first, then tasks in ascending id order — a fixed tie-break. */
+    if (s->main_task.state == TASK_SUSPENDED && s->main_task.sleeping &&
+        s->main_task.wake_at <= s->now) {
+        s->main_task.sleeping = 0;
+        sched_ready_push(s, 0);
+    }
+    for (int i = 1; i < HANDLE_TABLE_SIZE; i++) {
+        Task *t = (Task *)handle_lookup(i, HANDLE_TASK);
+        if (t && t->state == TASK_SUSPENDED && t->sleeping && t->wake_at <= s->now) {
+            t->sleeping = 0;
+            sched_ready_push(s, t->id);
+        }
+    }
+    return 1;
+}
+
 /* Deterministic teardown of a task mid-run (task_kill): drop its mailbox and
  * saved slice, wake any joiner with an `interrupt` error, mark it DEAD. The
  * handle entry stays (task_alive → 0, joiners see DEAD); handle_table_drain
@@ -5478,9 +5538,13 @@ static Value *scheduler_trampoline(TaskScheduler *s) {
     for (;;) {
         int id = sched_ready_pop(s);
         if (id < 0) {
-            /* Nothing runnable. If main already finished, we're done. If tasks
-             * remain live (blocked on joins that can't resolve), that's a
-             * deadlock — raise it loudly rather than hang. */
+            /* Nothing runnable now. Sleepers waiting on the virtual clock are
+             * not a deadlock — advance time to the earliest wake and retry
+             * before deciding anything is stuck. */
+            if (sched_wake_sleepers(s)) continue;
+            /* Genuinely nothing runnable. If main already finished, we're done.
+             * If tasks remain live (blocked on joins that can't resolve), that's
+             * a deadlock — raise it loudly rather than hang. */
             if (s->main_task.state == TASK_DONE || s->main_task.state == TASK_DEAD) {
                 if (s->main_task.has_error) {
                     g_error_value = s->main_task.error_value;
@@ -5511,8 +5575,10 @@ static Value *scheduler_trampoline(TaskScheduler *s) {
         if (t->state == TASK_SUSPENDED) {
             /* It suspended again. task_yield → re-enqueue; task_join → stay
              * blocked (woken by sched_finish); task_recv on an empty mailbox →
-             * stay blocked (woken by task_deliver). */
-            if (t->join_target == 0 && !t->recv_blocked) sched_ready_push(s, id);
+             * stay blocked (woken by task_deliver); task_sleep → stay blocked
+             * (woken by sched_wake_sleepers when the clock reaches wake_at). */
+            if (t->join_target == 0 && !t->recv_blocked && !t->sleeping)
+                sched_ready_push(s, id);
         } else {
             sched_finish(s, t, r);
             /* kill-outstanding: main ending tears the rest down deterministically. */
@@ -5547,9 +5613,13 @@ Value *vm_execute(EigsChunk *chunk, Env *env) {
      * tasks are still live, drive them too (kill-outstanding at main's end). */
     if (s->main_task.state == TASK_SUSPENDED) {
         /* main's first suspension happened in the initial vm_run, outside the
-         * trampoline — enqueue it now (unless it blocked on a join, in which
-         * case sched_finish wakes it) so it resumes round-robin. */
-        if (s->main_task.join_target == 0) sched_ready_push(s, 0);
+         * trampoline — enqueue it now so it resumes round-robin, UNLESS it
+         * blocked on a join (sched_finish wakes it), a recv (task_deliver
+         * wakes it), or a sleep (sched_wake_sleepers wakes it). Mirrors the
+         * trampoline's re-enqueue guard. */
+        if (s->main_task.join_target == 0 && !s->main_task.recv_blocked &&
+            !s->main_task.sleeping)
+            sched_ready_push(s, 0);
         return scheduler_trampoline(s);
     }
     /* main ran to completion without ever suspending. Record its result and,

src/vm.h

@@ -417,6 +417,13 @@ typedef struct Task {
     Value    **mbox;
     int        mbox_head, mbox_count, mbox_cap;
     int        recv_blocked;
+    /* Inc 3: virtual time. `sleeping` is 1 while this task is suspended in
+     * task_sleep; `wake_at` is the virtual-clock value it becomes runnable at.
+     * The clock is logical (discrete-event) — it never tracks wall time and
+     * only jumps forward to the earliest sleeper when nothing else is runnable,
+     * so sleeping stays deterministic-by-construction (no tape records). */
+    int        sleeping;
+    double     wake_at;
     /* Completion. */
     Value     *result;             /* owned, deep-copied on normal end */
     int        has_error;          /* died of an uncaught error */
@@ -443,6 +450,9 @@ int   task_mbox_has(void);           /* current task has a queued message? */
 Value *task_mbox_pop(void);          /* pop current task's front message (owned ref) */
 void  task_request_recv(void);       /* current task blocks awaiting a message */
 int   task_do_kill(int tid);         /* deterministic teardown of `tid`; 0 = bad target */
+/* Inc 3 virtual time (builtins.c task_sleep/task_now). */
+void   task_request_sleep(double ticks); /* current task sleeps until virtual now + ticks */
+double task_virtual_now(void);           /* current virtual-clock value (0 with no scheduler) */
 
 /* ---- Public API ---- */
 

tests/run_all_tests.sh

@@ -2158,6 +2158,27 @@ else
     FAIL=$((FAIL + 1))
 fi
 
+# #408 increment 3 virtual time: task_sleep/task_now on a LOGICAL clock must be
+# deterministic (identical on two fresh processes), replay byte-identically,
+# and — since the clock is not a nondet source — record ZERO tape 'N' records.
+TOTAL=$((TOTAL + 1))
+VT_EX=../examples/task_virtual_time.eigs
+VT_TAPE=$(mktemp -t eigs_vt.XXXXXX)
+VTA=$(./eigenscript "$VT_EX" </dev/null 2>&1)
+VTB=$(./eigenscript "$VT_EX" </dev/null 2>&1)
+EIGS_TRACE="$VT_TAPE" ./eigenscript "$VT_EX" </dev/null >/dev/null 2>&1
+VTR=$(EIGS_REPLAY="$VT_TAPE" ./eigenscript "$VT_EX" </dev/null 2>&1)
+VT_NREC=$(grep -c '^N ' "$VT_TAPE" 2>/dev/null)
+rm -f "$VT_TAPE"
+if [ "$VTA" = "$VTB" ] && [ "$VTA" = "$VTR" ] && [ "$VT_NREC" -eq 0 ] && \
+   echo "$VTA" | grep -q "timeout at t=40"; then
+    echo "  PASS: virtual time is deterministic, replays, records zero nondet (#408 inc3)"
+    PASS=$((PASS + 1))
+else
+    echo "  FAIL: virtual time diverged/replayed wrong/leaked nondet (rec=$VT_NREC)"
+    FAIL=$((FAIL + 1))
+fi
+
 # [105] Builtin contract fixes (#312 negative indices, #316 predicate
 # type-rejection, #317 min/max N-ary reduction) + #314: a directory as the
 # script path must take the clean cannot-read-file exit, not xmalloc's

tests/test_tasks.eigs

@@ -214,4 +214,74 @@ catch e:
     assert_eq of [e.kind, "interrupt", "join of a killed task raises interrupt"]
 assert_eq of [kthrew, 1, "join of a killed task raises"]
 
+# ============================================================================
+# Increment 3: virtual time (task_sleep / task_now).
+# The virtual clock is logical: it starts at 0 and only jumps FORWARD to the
+# earliest sleeper when nothing else is runnable, so sleeps stay deterministic
+# (no wall clock). The scheduler is per-run, so `now` is monotonic across the
+# whole file — these checks reason from a captured baseline, not absolutes.
+# ============================================================================
+
+# --- sleepers wake in virtual-time order, not spawn order; each observes the
+#     clock advanced by exactly its own sleep; the clock rests at the last wake ---
+base is task_now of null
+woke is []
+define sleeper(tag, ticks) as:
+    task_sleep of ticks
+    woke is append of [woke, tag]
+    return task_now of null
+sa is task_spawn of [sleeper, "a", 30]
+sb is task_spawn of [sleeper, "b", 10]
+sc is task_spawn of [sleeper, "c", 20]
+ra is task_join of sa
+rb is task_join of sb
+rc is task_join of sc
+assert_eq of [woke, ["b", "c", "a"], "sleepers wake in virtual-time order, not spawn order"]
+assert_eq of [rb - base, 10, "b observes now advanced by exactly its sleep"]
+assert_eq of [rc - base, 20, "c observes now advanced by exactly its sleep"]
+assert_eq of [ra - base, 30, "a observes now advanced by exactly its sleep"]
+assert_eq of [(task_now of null) - base, 30, "the clock rests at the last wake and holds"]
+
+# --- task_sleep of 0 is a same-tick yield: both run, clock does not advance ---
+b0 is task_now of null
+tick0 is []
+define zzz(tag) as:
+    task_sleep of 0
+    tick0 is append of [tick0, tag]
+    return null
+z1 is task_spawn of [zzz, "x"]
+z2 is task_spawn of [zzz, "y"]
+task_join of z1
+task_join of z2
+assert_eq of [tick0, ["x", "y"], "sleep(0) tasks both run"]
+assert_eq of [task_now of null, b0, "sleep(0) does not advance the virtual clock"]
+
+# --- a negative sleep is clamped to 0 (no time travel) ---
+bn is task_now of null
+define nap_neg() as:
+    task_sleep of (0 - 5)
+    return task_now of null
+rn is task_join of (task_spawn of nap_neg)
+assert_eq of [rn, bn, "a negative sleep is clamped to 0"]
+
+# --- main itself can sleep to let an earlier-waking worker run first ---
+bm is task_now of null
+mlog is []
+define bg() as:
+    task_sleep of 5
+    mlog is append of [mlog, "worker"]
+    return null
+bgid is task_spawn of bg
+task_sleep of 15                             # main sleeps PAST the worker's wake
+mlog is append of [mlog, "main"]
+task_join of bgid
+assert_eq of [mlog, ["worker", "main"], "a sleeping main lets an earlier-waking worker run first"]
+assert_eq of [(task_now of null) - bm, 15, "main's own sleep advances the clock"]
+
+# --- a lone sleeper is woken by the clock, not reported as a deadlock ---
+define lone() as:
+    task_sleep of 100
+    return 7
+assert_eq of [task_join of (task_spawn of lone), 7, "a lone sleeper is woken by the clock, not a deadlock"]
+
 test_summary of null