From 708722b7ffc089f8a77b00aba7c50d1848925f3e Mon Sep 17 00:00:00 2001 From: bigmerge Date: Sat, 15 Aug 2026 14:25:55 -0500 Subject: [PATCH] Add native interruptibility for dispatched agent workers Cancellation-token control channel checked at every step boundary lets a coordinator PAUSE/RESUME/REDIRECT/KILL a running worker mid-task instead of waiting for the whole (possibly wrong) plan to finish. Bounded purviews mean no half-committed state to unwind on interrupt. Includes a proof harness (proof.el, run.sh) comparing a broken non-interruptible worker against the new one under identical kill/redirect/pause timing. Distinct from the already-preserved swarm-ccr orchestrator (fan-out/ converge dispatch): this is single-worker interruptibility, a complementary mechanism, not a duplicate. --- lang/runtime/swarm.el | 269 ++++++++++++++++++++++++++++++++++++++ lang/tests/swarm/proof.el | 180 +++++++++++++++++++++++++ lang/tests/swarm/run.sh | 56 ++++++++ 3 files changed, 505 insertions(+) create mode 100644 lang/runtime/swarm.el create mode 100644 lang/tests/swarm/proof.el create mode 100755 lang/tests/swarm/run.sh diff --git a/lang/runtime/swarm.el b/lang/runtime/swarm.el new file mode 100644 index 0000000..668a8fb --- /dev/null +++ b/lang/runtime/swarm.el @@ -0,0 +1,269 @@ +// swarm.el — Native interruptibility for Neuron-dispatched agents. +// +// CANON: Neuron memory 1ca4d3e9 (native-interruptibility). A dispatched worker +// must be re-steerable OR killable the INSTANT a correction / new signal arrives, +// MID-TASK — like a human who stops the moment they're told "that's wrong", not +// one who finishes the wrong workflow first. Claude's sub-agents cannot do this: +// reset off a wrong path, they finish the current workflow before absorbing the +// correction, wasting work on a KNOWN-WRONG thing. Neuron's must never. +// +// MECHANISM (Will's steer, 2026-08-15 — "not a hard problem, don't over-engineer"): +// CancellationToken + always-on input. +// (1) INPUT ALWAYS ON — every worker holds a CONTROL CHANNEL (the token). The +// coordinator can push a signal at ANY time; the channel is NEVER gated by +// whether the worker is busy. Like a person who keeps hearing while talking. +// (2) COOPERATIVE CANCELLATION — the worker CHECKS the token at every step +// boundary (a non-blocking poll, __channel_try_recv). On a signal it STOPS, +// ABSORBS the correction (re-plans) or TERMINATES cleanly — with ZERO wasted +// continuation of known-wrong work. +// +// The SAME mechanism serves the conversational speech loop: a mic barge-in simply +// invokes the token on the running render (see peripheral loop 66155ba9). +// +// SAFETY (Rule-4 single-writer / bounded purview): a worker holds its task as a +// PURVIEW (a meaning-plan). It writes ONLY to its own out-channel and its own +// local purview — it holds NO global write-lock. So it can be interrupted or +// killed mid-step with NO half-committed global state; the coordinator just +// signals stop. Bounded purviews are what make live interruption clean. +// +// Built on the runtime concurrency primitives: +// __channel_new / __channel_send / __channel_recv / __channel_try_recv (channels) +// __thread_create / __thread_join (workers) +// +// Contrast: the pre-existing channel.el `_channel_worker` loop spawns+joins the +// WHOLE task with no signal check — the exact broken pattern. This module checks +// the token BETWEEN every bounded step, so the interrupt latency is one step, not +// one whole workflow. + +// ── low-level worker/thread wrappers (call seed prims directly; no collision) ── + +fn _swarm_spawn(fn_name: String, arg: String) -> Int { + return __thread_create(fn_name, arg) +} + +fn _swarm_join(tid: Int) -> String { + return __thread_join(tid) +} + +// ── Cancellation token — the always-on control channel ───────────────────────── + +// token_new — create a cancellation/pause/redirect token for one worker. +// Unbounded so the coordinator NEVER blocks when it signals (always-on input). +fn token_new() -> Int { + return __channel_new(0) +} + +// ── Coordinator ops — INVOKE the token (interrupt / re-steer / kill) ─────────── + +// token_signal — send a raw signal JSON. Low-level; prefer the named helpers. +fn token_signal(tok: Int, sig_json: String) { + __channel_send(tok, sig_json) +} + +// token_interrupt — ask the worker to stop at the next step boundary and hold. +fn token_interrupt(tok: Int) { + __channel_send(tok, "{\"sig\":\"PAUSE\"}") +} + +// token_pause — alias for interrupt: stop stepping, hold state, wait for RESUME. +fn token_pause(tok: Int) { + __channel_send(tok, "{\"sig\":\"PAUSE\"}") +} + +// token_resume — resume a paused worker on its held plan. +fn token_resume(tok: Int) { + __channel_send(tok, "{\"sig\":\"RESUME\"}") +} + +// token_kill — terminate the worker cleanly at the next step boundary. +fn token_kill(tok: Int) { + __channel_send(tok, "{\"sig\":\"KILL\"}") +} + +// token_redirect — re-steer the worker onto a NEW plan MID-TASK. Progress already +// made (completed steps, emitted outputs) is PRESERVED; only the remaining plan is +// replaced. new_steps is a JSON array string, e.g. "[\"step-a\",\"step-b\"]". +fn token_redirect(tok: Int, new_goal: String, new_steps: String) { + let msg: String = "{\"sig\":\"REDIRECT\",\"goal\":\"" + + json_escape_string(new_goal) + "\",\"steps\":" + new_steps + "}" + __channel_send(tok, msg) +} + +// ── Worker ops — CHECK the token (cooperative cancellation point) ────────────── + +// token_check — non-blocking poll of the token. Returns the pending signal JSON, +// or "" when there is no signal. Called at every step boundary. This is the +// always-on check: it never blocks the worker, so work proceeds at full speed +// until — and only until — a signal actually arrives. +fn token_check(tok: Int) -> String { + return __channel_try_recv(tok) +} + +// token_wait — BLOCK until the next signal. Used only while PAUSED (the worker is +// idle, so blocking is correct — it is not spinning). +fn token_wait(tok: Int) -> String { + return __channel_recv(tok) +} + +// ── Purview — the worker's meaning-plan (held task, resumable) ───────────────── + +// purview_new — build a bounded meaning-plan the worker carries. +// id — purview id +// goal — what the worker is trying to achieve (steerable) +// steps — JSON array string of step descriptors (the remaining plan) +// A redirect UPDATES this plan; it is not lost. +fn purview_new(id: String, goal: String, steps: String) -> String { + return "{\"id\":\"" + json_escape_string(id) + + "\",\"goal\":\"" + json_escape_string(goal) + + "\",\"steps\":" + steps + + ",\"idx\":0,\"completed\":0,\"status\":\"ready\"}" +} + +// ── The interruptible worker loop ────────────────────────────────────────────── + +// swarm_worker_run — the generic natively-interruptible worker. +// +// arg is a JSON object carrying: +// "tok" — the cancellation token (control channel handle) +// "out" — the worker's own output channel handle (its ONLY write surface) +// "step_fn" — name of an El fn (String)->String that performs ONE bounded step +// "purview" — the meaning-plan (from purview_new) +// +// The loop: at EVERY iteration it first polls the token (always-on check). Only +// then does it execute ONE bounded step. So a KILL/REDIRECT/PAUSE is absorbed +// within a single step — never after finishing the whole (possibly wrong) plan. +// +// step_fn is invoked as a child thread joined immediately: exactly ONE step of +// work is in flight, so the interrupt latency is bounded by a single step. +// +// Returns the final purview JSON (status: complete | killed | redirected). +fn swarm_worker_run(arg: String) -> String { + let tok: Int = str_to_int(json_get(arg, "tok")) + let out_ch: Int = str_to_int(json_get(arg, "out")) + let step_fn: String = json_get(arg, "step_fn") + let purview: String = json_get_raw(arg, "purview") + + let pid: String = json_get(purview, "id") + let goal: String = json_get(purview, "goal") + let steps: String = json_get_raw(purview, "steps") + let n: Int = json_array_len(steps) + let idx: Int = 0 + let completed: Int = 0 + let status: String = "running" + let stop: Int = 0 + + while stop == 0 { + // ── ALWAYS-ON INTERRUPT CHECK (the cooperative cancellation point) ── + let sig: String = token_check(tok) + if str_eq(sig, "") { + // no signal — proceed with one bounded step (or finish) + if idx >= n { + let status = "complete" + let stop = 1 + } else { + let step: String = json_array_get(steps, idx) + let step_arg: String = "{\"goal\":\"" + json_escape_string(goal) + + "\",\"idx\":" + int_to_str(idx) + + ",\"step\":" + step + "}" + let tid: Int = _swarm_spawn(step_fn, step_arg) + let result: String = _swarm_join(tid) + __channel_send(out_ch, result) + let idx = idx + 1 + let completed = completed + 1 + } + } else { + // a signal arrived — ABSORB it immediately, before any more work + let kind: String = json_get(sig, "sig") + if str_eq(kind, "KILL") { + // terminate cleanly — commit nothing further. Bounded purview = + // no half-committed global state to unwind. + __channel_send(out_ch, "[KILL absorbed @step " + int_to_str(idx) + + " — stopped, no wasted continuation]") + let status = "killed" + let stop = 1 + } else { + if str_eq(kind, "REDIRECT") { + // ABSORB the correction: re-plan onto the new goal/steps. + // Completed steps + emitted outputs are PRESERVED; only the + // remaining plan is replaced. + let kept: Int = completed + let goal = json_get(sig, "goal") + let steps = json_get_raw(sig, "steps") + let n = json_array_len(steps) + let idx = 0 + let status = "redirected" + __channel_send(out_ch, "[REDIRECT absorbed @step " + + int_to_str(kept) + " — kept " + + int_to_str(kept) + + " done, re-planned to goal=" + goal + "]") + } else { + if str_eq(kind, "PAUSE") { + // hold state; idle-wait for the next signal + __channel_send(out_ch, "[PAUSE absorbed @step " + + int_to_str(idx) + " — holding plan]") + let status = "paused" + let resumed: Int = 0 + while resumed == 0 { + let s2: String = token_wait(tok) + let k2: String = json_get(s2, "sig") + if str_eq(k2, "RESUME") { + __channel_send(out_ch, "[RESUME @step " + + int_to_str(idx) + "]") + let status = "running" + let resumed = 1 + } else { + if str_eq(k2, "KILL") { + __channel_send(out_ch, "[KILL absorbed while paused]") + let status = "killed" + let stop = 1 + let resumed = 1 + } else { + if str_eq(k2, "REDIRECT") { + let goal = json_get(s2, "goal") + let steps = json_get_raw(s2, "steps") + let n = json_array_len(steps) + let idx = 0 + __channel_send(out_ch, "[REDIRECT absorbed while paused — goal=" + goal + "]") + let status = "running" + let resumed = 1 + } + } + } + } + } + } + } + } + } + + // final purview snapshot + let final: String = "{\"id\":\"" + pid + + "\",\"goal\":\"" + json_escape_string(goal) + + "\",\"idx\":" + int_to_str(idx) + + ",\"completed\":" + int_to_str(completed) + + ",\"planned\":" + int_to_str(n) + + ",\"status\":\"" + status + "\"}" + __channel_send(out_ch, "[DONE status=" + status + + " completed=" + int_to_str(completed) + "]") + return final +} + +// ── Coordinator convenience — dispatch an interruptible worker ───────────────── + +// swarm_dispatch — spawn a natively-interruptible worker on a purview. +// step_fn — name of the per-step executor (String)->String +// purview — the meaning-plan (from purview_new) +// Returns a handle JSON: {"tok":T,"out":O,"tid":D} — the coordinator holds `tok` +// to interrupt/redirect/kill live, and drains `out` for results. +fn swarm_dispatch(step_fn: String, purview: String) -> String { + let tok: Int = token_new() + let out_ch: Int = __channel_new(0) + let arg: String = "{\"tok\":" + int_to_str(tok) + + ",\"out\":" + int_to_str(out_ch) + + ",\"step_fn\":\"" + step_fn + + "\",\"purview\":" + purview + "}" + let tid: Int = _swarm_spawn("swarm_worker_run", arg) + return "{\"tok\":" + int_to_str(tok) + + ",\"out\":" + int_to_str(out_ch) + + ",\"tid\":" + int_to_str(tid) + "}" +} diff --git a/lang/tests/swarm/proof.el b/lang/tests/swarm/proof.el new file mode 100644 index 0000000..0429757 --- /dev/null +++ b/lang/tests/swarm/proof.el @@ -0,0 +1,180 @@ +// proof.el — Proof of native interruptibility for Neuron-dispatched agents. +// +// Concatenated after runtime/swarm.el (see run.sh). Demonstrates, with raw +// before/after counts, that a dispatched worker interrupted MID-TASK stops +// instantly (not after finishing the wrong workflow), absorbs a redirect +// (re-plans, keeping progress), or terminates cleanly. +// +// Three scenarios on the SAME 12-step plan, SAME per-step cost, SAME signal +// timing (sent ~50ms in ≈ after step 3): +// A. BASELINE — the broken Claude-style worker: no mid-loop signal check. +// A kill sent at step ~3 is ignored until the whole plan finishes → all 12 +// steps run = wasted work on a known-wrong task. +// B. INTERRUPTIBLE KILL — swarm_worker_run: the kill is absorbed within one +// step → stops at ~3, status=killed, ~9 steps of waste AVOIDED. +// C. INTERRUPTIBLE REDIRECT — mirrors the live incident (build-python-synth → +// native-fetch-render): the correction is absorbed mid-task; the 3 done +// steps are kept; the worker re-plans onto the new goal and finishes it. + +// ── per-step work — one bounded unit (~15ms) ────────────────────────────────── +fn demo_step(arg: String) -> String { + let goal: String = json_get(arg, "goal") + let idx: String = json_get(arg, "idx") + let step: String = json_get(arg, "step") + sleep_ms(15) + return "did[" + goal + "] step#" + idx + " (" + step + ")" +} + +// ── BASELINE: the broken, non-interruptible worker (Claude-style) ───────────── +// Same shape as swarm_worker_run BUT it never polls the token during the loop. +// It "finishes the workflow" and only notices the signal at the very end — the +// exact pattern Will called out. Reports how many steps it wasted post-signal. +fn broken_worker_run(arg: String) -> String { + let tok: Int = str_to_int(json_get(arg, "tok")) + let out_ch: Int = str_to_int(json_get(arg, "out")) + let step_fn: String = json_get(arg, "step_fn") + let purview: String = json_get_raw(arg, "purview") + let goal: String = json_get(purview, "goal") + let steps: String = json_get_raw(purview, "steps") + let n: Int = json_array_len(steps) + let idx: Int = 0 + // NO token check inside the loop — this is the bug. + while idx < n { + let step: String = json_array_get(steps, idx) + let step_arg: String = "{\"goal\":\"" + goal + "\",\"idx\":" + + int_to_str(idx) + ",\"step\":" + step + "}" + let tid: Int = __thread_create(step_fn, step_arg) + let result: String = __thread_join(tid) + __channel_send(out_ch, result) + let idx = idx + 1 + } + // Only NOW does it look at the control signal — too late. + let late: String = token_check(tok) + if str_eq(late, "") { + __channel_send(out_ch, "[no signal]") + } else { + __channel_send(out_ch, "[TOO LATE: absorbed " + json_get(late, "sig") + + " only after running ALL " + int_to_str(n) + + " steps — wasted work]") + } + return "{\"status\":\"ran-to-completion\",\"completed\":" + int_to_str(idx) + "}" +} + +// ── helpers ─────────────────────────────────────────────────────────────────── + +// drain_count — drain out_ch, print each line, return count of real step outputs +// (lines beginning with "did"). +fn drain_and_report(out_ch: Int) -> Int { + let done: Int = 0 + let did: Int = 0 + while done == 0 { + let m: String = __channel_try_recv(out_ch) + if str_eq(m, "") { + let done = 1 + } else { + println(" | " + m) + if str_starts_with(m, "did") { + let did = did + 1 + } + } + } + return did +} + +fn twelve_steps() -> String { + return "[\"s0\",\"s1\",\"s2\",\"s3\",\"s4\",\"s5\",\"s6\",\"s7\",\"s8\",\"s9\",\"s10\",\"s11\"]" +} + +fn main() -> Void { + println("=====================================================================") + println(" NATIVE INTERRUPTIBILITY — PROOF (canon 1ca4d3e9)") + println(" plan: 12 bounded steps @ ~15ms; signal sent ~50ms in (≈ after step 3)") + println("=====================================================================") + + // ── A. BASELINE — broken, non-interruptible ── + println("") + println("[A] BASELINE broken worker (no mid-loop signal check) — Claude-style") + let tokA: Int = token_new() + let outA: Int = __channel_new(0) + let pvA: String = purview_new("A", "build-wrong-thing", twelve_steps()) + let argA: String = "{\"tok\":" + int_to_str(tokA) + ",\"out\":" + int_to_str(outA) + + ",\"step_fn\":\"demo_step\",\"purview\":" + pvA + "}" + let t0A: Int = time_now() + let tidA: Int = __thread_create("broken_worker_run", argA) + sleep_ms(50) + println(" -> coordinator sends KILL at +" + int_to_str(time_now() - t0A) + "ms (≈step 3)") + token_kill(tokA) + let finA: String = __thread_join(tidA) + let elapA: Int = time_now() - t0A + let didA: Int = drain_and_report(outA) + println(" RESULT: executed " + int_to_str(didA) + "/12 steps, wall=" + + int_to_str(elapA) + "ms, final=" + finA) + println(" >> ignored the kill, RAN ALL 12 — ~9 steps of KNOWN-WRONG waste") + + // ── B. INTERRUPTIBLE KILL ── + println("") + println("[B] INTERRUPTIBLE swarm_worker_run + token_kill") + let pvB: String = purview_new("B", "build-wrong-thing", twelve_steps()) + let hB: String = swarm_dispatch("demo_step", pvB) + let tokB: Int = str_to_int(json_get(hB, "tok")) + let outB: Int = str_to_int(json_get(hB, "out")) + let tidB: Int = str_to_int(json_get(hB, "tid")) + let t0B: Int = time_now() + sleep_ms(50) + println(" -> coordinator sends KILL at +" + int_to_str(time_now() - t0B) + "ms (≈step 3)") + token_kill(tokB) + let finB: String = __thread_join(tidB) + let elapB: Int = time_now() - t0B + let didB: Int = drain_and_report(outB) + println(" RESULT: executed " + int_to_str(didB) + "/12 steps, wall=" + + int_to_str(elapB) + "ms, final=" + finB) + println(" >> STOPPED within one step of the signal — no wasted continuation") + + // ── C. INTERRUPTIBLE REDIRECT (the live incident) ── + println("") + println("[C] INTERRUPTIBLE redirect mid-task: build-python-synth -> native-fetch-render") + let pvC: String = purview_new("C", "build-python-synth-renderer", twelve_steps()) + let hC: String = swarm_dispatch("demo_step", pvC) + let tokC: Int = str_to_int(json_get(hC, "tok")) + let outC: Int = str_to_int(json_get(hC, "out")) + let tidC: Int = str_to_int(json_get(hC, "tid")) + let t0C: Int = time_now() + sleep_ms(50) + println(" -> coordinator REDIRECTS at +" + int_to_str(time_now() - t0C) + "ms (≈step 3)") + token_redirect(tokC, "native-fetch-render", "[\"fetch\",\"realize\",\"cohere\"]") + let finC: String = __thread_join(tidC) + let elapC: Int = time_now() - t0C + let didC: Int = drain_and_report(outC) + println(" RESULT: executed " + int_to_str(didC) + " steps total, wall=" + + int_to_str(elapC) + "ms, final=" + finC) + println(" >> absorbed the correction mid-task: kept early progress,") + println(" re-planned onto native-fetch-render, finished the RIGHT plan") + + // ── D. INTERRUPTIBLE PAUSE -> RESUME (hold state, then continue) ── + println("") + println("[D] INTERRUPTIBLE pause mid-task, hold state, then resume to finish") + let pvD: String = purview_new("D", "long-render", twelve_steps()) + let hD: String = swarm_dispatch("demo_step", pvD) + let tokD: Int = str_to_int(json_get(hD, "tok")) + let outD: Int = str_to_int(json_get(hD, "out")) + let tidD: Int = str_to_int(json_get(hD, "tid")) + let t0D: Int = time_now() + sleep_ms(50) + println(" -> coordinator PAUSES at +" + int_to_str(time_now() - t0D) + "ms (≈step 3)") + token_pause(tokD) + sleep_ms(60) + println(" -> worker held idle for ~60ms; coordinator RESUMES at +" + + int_to_str(time_now() - t0D) + "ms") + token_resume(tokD) + let finD: String = __thread_join(tidD) + let didD: Int = drain_and_report(outD) + println(" RESULT: executed " + int_to_str(didD) + "/12 steps, final=" + finD) + println(" >> paused on the spot, held its plan, resumed and finished it") + + println("") + println("=====================================================================") + println(" A ran all 12 wrong steps. B stopped at ~3. C re-planned at ~3.") + println(" Same plan, same timing, same signal — only the interruptible worker") + println(" stops the instant it's told, like a human. QED.") + println("=====================================================================") +} diff --git a/lang/tests/swarm/run.sh b/lang/tests/swarm/run.sh new file mode 100755 index 0000000..a86a626 --- /dev/null +++ b/lang/tests/swarm/run.sh @@ -0,0 +1,56 @@ +#!/usr/bin/env bash +# run.sh — build and run the native-interruptibility proof. +# +# Concatenates runtime/swarm.el + proof.el into one translation unit (the El +# multi-file strategy — elc does not resolve cross-directory imports), compiles +# via the canonical elc, links against el_runtime.c, and runs. +# +# A tiny forward-declaration prelude is prepended to the generated C for the +# __channel_* seed primitives (they are defined in el_runtime.c but not declared +# in el_runtime.h). This touches nothing shared — it is local to this build. + +set -uo pipefail +cd "$(dirname "$0")" + +EL_HOME="${EL_HOME:-$(cd ../.. && pwd)}" +ELC="${EL_HOME}/dist/platform/elc" +RT="${EL_HOME}/el-compiler/runtime" +SWARM="${EL_HOME}/runtime/swarm.el" + +OSSL="$(brew --prefix openssl@3 2>/dev/null || brew --prefix openssl 2>/dev/null || echo /usr/local)" +LDF=(); [ -d "${OSSL}/lib" ] && LDF=(-L"${OSSL}/lib") + +BUILD="$(mktemp -d -t swarmproof.XXXXXX)" +trap 'rm -rf "${BUILD}"' EXIT + +if [ ! -x "${ELC}" ]; then echo "elc not found at ${ELC}" >&2; exit 1; fi + +# 1. Concatenate library + proof into one .el +cat "${SWARM}" proof.el > "${BUILD}/combined.el" + +# 2. elc emit -> C +if ! "${ELC}" "${BUILD}/combined.el" > "${BUILD}/body.c" 2>"${BUILD}/elc.err"; then + echo "elc FAILED:"; sed 's/^/ /' "${BUILD}/elc.err"; exit 1 +fi + +# 3. Prepend forward-decl prelude for the __channel_* seed primitives +cat > "${BUILD}/prog.c" <<'PRELUDE' +#include +typedef int64_t el_val_t; +el_val_t __channel_new(el_val_t); +el_val_t __channel_send(el_val_t, el_val_t); +el_val_t __channel_recv(el_val_t); +el_val_t __channel_try_recv(el_val_t); +el_val_t __channel_close(el_val_t); +PRELUDE +cat "${BUILD}/body.c" >> "${BUILD}/prog.c" + +# 4. cc link +if ! cc -O2 -Wno-implicit-function-declaration -I "${RT}" "${LDF[@]}" \ + "${BUILD}/prog.c" "${RT}/el_runtime.c" \ + -lcurl -lssl -lcrypto -lpthread -lm -o "${BUILD}/proof" 2>"${BUILD}/cc.err"; then + echo "cc FAILED:"; tail -20 "${BUILD}/cc.err"; exit 1 +fi + +# 5. run +"${BUILD}/proof"