// 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) + "}"
}