Files
el/lang/swarm/swarm.el
T
bigmerge e5c80359a8 swarm: Rule 4 — engram-write is @manager-ONLY, enforced by capability
New hard invariant (Will): only the orchestrator mutates global engram state;
workers are read-only against the full engram + write only their own local
geometry. This is an AUTHORITY gate (capability), not a health gate — a worker
is STRUCTURALLY UNABLE to mutate global engram state regardless of engram health.

- containment.el: scope tokens now carry a caps set. Orchestrator token holds
  engram:write + dharma:emit (@manager-only, the VBD rule that only the manager
  mutates global state); worker token holds ONLY engram:read. Rule 4:
  containment_check_engram_write / _dharma_emit reject any caller lacking the
  capability — same scope-token mechanism as the live Rule-2 denial.
- swarm.el: swarm_engram_write is the ONLY engram write path, gated by Rule 4;
  a worker token is denied before any HTTP is issued (no mutation). The curated
  merge (commit=1) is the sole writer: the orchestrator commits approved
  geometry via its write-capable token. Workers' full-engram READ stays intact.
- reshape_surface.el: compose op_write (json_escape_string) for the commit path.
- harness: Rule-4 suite proven — worker engram-write DENIED by capability, no
  node created, violation journalled; orchestrator passes the gate as sole
  writer. 24/24 green on the :8901 clone with real cognition.

Authority gate holds independent of daemon write-health (proven with daemon
both alive and, earlier, crashed). Prod :8742 untouched.
2026-08-14 21:46:03 -05:00

484 lines
21 KiB
EmacsLisp

// swarm.el the swarm orchestrator: bounded parallel agent execution.
//
// Implements Swarm Architecture's single pattern fan out, execute independently,
// converge on El's NATIVE concurrency (thread.el spawn/join). No external
// orchestrator: a swarm is a coordinator (this file, the main thread) that mints
// a correlation identity, compiles a bounded CCR context per worker, dispatches
// workers as native pthreads, tracks every unit of work, and converges the
// results before returning control to the parent step.
//
// The five properties of every swarm (Swarm §2.1) are all present:
// parent step -> swarm_run is called from one process step
// task blueprint -> `blueprint` name + knowledge refs, run by every worker
// input set -> `inputs_json`, one item per worker
// convergence -> `strategy` in config (collect|merge|vote|reduce)
// correlation ID -> minted here, threaded through tracking + every worker
//
// Containment (Swarm §3) is enforced: the caller must hold a coordinator/absent
// token to open a swarm (Rule 2), each worker is stamped a closed worker token
// (Rules 1+3), and workers share no mutable state (the coordinator is the only
// journal writer).
// worker entry the top-level (String)->String fn native threads run
//
// Every El fn compiles to a global C symbol; spawn() resolves this by name via
// dlsym and runs it in a pthread. The envelope carries everything the worker is
// permitted to see its compiled context and nothing else (§9.3).
//
// Returns a result JSON: {worker_id, status:"completed"|"failed", output|error}.
fn swarm_worker_entry(envelope_json: String) -> String {
let worker_id: String = json_get_string(envelope_json, "worker_id")
let ctx: String = json_get_raw(envelope_json, "ctx")
// The worker holds a CLOSED worker token (Rules 1+3): it shares no state
// with siblings and may not open/join a swarm. That boundary is enforced at
// the point of attempt swarm_run rejects any swarm opened under a worker
// token (Rule 2). A worker simply executing its blueprint is not opening a
// swarm, so it proceeds. Its only outward edge is this returned result
// (the vertical worker->coordinator path).
let out: String = swarm_run_blueprint(ctx)
// A worker reports failed iff its blueprint signalled failure. This is the
// vertical status edge the coordinator reads during convergence (§4.3, §7).
let bstatus: String = json_get_string(out, "blueprint_status")
let status: String = "completed"
if str_eq(bstatus, "failed") {
let status = "failed"
}
let kv: [String] = el_list_empty()
let kv = el_list_append(kv, "worker_id")
let kv = el_list_append(kv, worker_id)
let kv = el_list_append(kv, "status")
let kv = el_list_append(kv, status)
let res: String = json_build_object(kv)
return json_set(res, "output", out)
}
// swarm_run_blueprint execute the task blueprint over a compiled context.
// The default blueprint is the CCR execution cycle: think -> intend -> act over
// the worker's bounded context. Specialise by dispatching on
// json_get_string(ctx,"blueprint"). Idempotent: reads ctx, writes only its
// returned output (§7.3).
fn swarm_run_blueprint(ctx: String) -> String {
let blueprint: String = json_get_string(ctx, "blueprint")
let input_item: String = json_get_string(ctx, "input")
let knowledge: String = json_get_string(ctx, "knowledge")
// classify deterministic verdict for the `vote` convergence strategy:
// verdict is "long" if the input has >4 chars, else "short".
if str_eq(blueprint, "classify") {
let verdict: String = "short"
if str_len(input_item) > 4 {
let verdict = "long"
}
let kv: [String] = el_list_empty()
let kv = el_list_append(kv, "verdict")
let kv = el_list_append(kv, verdict)
let kv = el_list_append(kv, "blueprint_status")
let kv = el_list_append(kv, "ok")
return json_build_object(kv)
}
// faildemo a worker that fails on inputs beginning with "x" (exercises the
// failure threshold + partial convergence path). Idempotent, side-effect-free.
if str_eq(blueprint, "faildemo") {
let st: String = "ok"
if str_starts_with(input_item, "x") {
let st = "failed"
}
return json_set_str("{}", "blueprint_status", st)
}
// cognize REAL-COGNITION blueprint. Routes think through the seam (bound to
// op_think in decorated mode) over the worker's NODE-ID anchor, then derives a
// vote verdict from the gradient's confidence. In stub mode there is no
// gradient, so the verdict falls back to a deterministic slice hash the
// same blueprint runs green on either side of the seam.
if str_eq(blueprint, "cognize") {
let thought: String = seam_think(ctx, "reason over " + input_item)
// Derive the vote verdict from the REAL gradient's support count
// (json_get_int, since n_support is numeric). Different anchors have
// different support -> genuine, cognition-driven vote diversity. In stub
// mode there is no gradient (n_support -> 0) -> "uncertain".
let nsup: Int = json_get_int(thought, "n_support")
let verdict: String = "uncertain"
if nsup >= 10 {
let verdict = "confident"
}
let ck: [String] = el_list_empty()
let ck = el_list_append(ck, "verdict")
let ck = el_list_append(ck, verdict)
let ck = el_list_append(ck, "blueprint_status")
let ck = el_list_append(ck, "ok")
let cout0: String = json_build_object(ck)
let cout1: String = json_set_str(cout0, "n_support", int_to_str(nsup))
let cout2: String = json_set_str(cout1, "seam_mode", json_get_string(thought, "seam_mode"))
return json_set_str(cout2, "afferent", json_get_string(thought, "afferent"))
}
// default (analyze_item): the CCR execution cycle think -> intend -> act,
// with `think` routed through the CONFIGURABLE PRIMITIVE SEAM. Telemetry
// (seam_mode + afferent tick) rides the worker's returned output.
let instruction: String = "process input: " + input_item
let thought: String = seam_think(ctx, instruction)
let intent: String = primitive_intend(thought)
let effect: String = primitive_act(intent, input_item)
let e1: String = json_set_str(effect, "blueprint_status", "ok")
let e2: String = json_set_str(e1, "seam_mode", json_get_string(thought, "seam_mode"))
let e3: String = json_set_str(e2, "afferent", json_get_string(thought, "afferent"))
return e3
}
// native-thread fan-out, bounded by concurrency, order-preserving
//
// parallel_map (thread.el) spawns ALL threads at once. The swarm honours the
// blueprint's `concurrency` cap (§5.1: a resource constraint, not a parallelism
// constraint all items are processed, at most N at a time) by dispatching in
// waves of N native threads, joining each wave before the next. Results are
// returned in input order.
fn swarm_fanout(worker_fn: String, envelopes: [String], concurrency: Int) -> [String] {
let n: Int = el_list_len(envelopes)
let cap: Int = concurrency
if cap < 1 {
let cap = 1
}
let results: [String] = el_list_empty()
let base = 0
while base < n {
// spawn a wave of up to `cap` workers
let tids: [String] = el_list_empty()
let k = 0
while k < cap {
let idx: Int = base + k
if idx < n {
let env_item: String = el_list_get(envelopes, idx)
let tid: Int = spawn(worker_fn, env_item)
let tids = el_list_append(tids, int_to_str(tid))
}
let k = k + 1
}
// join the wave in order
let j = 0
let jn: Int = el_list_len(tids)
while j < jn {
let tid: Int = str_to_int(el_list_get(tids, j))
let r: String = join(tid)
let results = el_list_append(results, r)
let j = j + 1
}
let base = base + cap
}
return results
}
// convergence strategies (Swarm §4.2)
// swarm_converge_collect ordered list, no transformation.
fn swarm_converge_collect(results: [String]) -> String {
let n: Int = el_list_len(results)
let arr: String = "[]"
let i = 0
while i < n {
let arr = json_array_push(arr, el_list_get(results, i))
let i = i + 1
}
return arr
}
// swarm_converge_merge combine worker outputs into a single joined string.
fn swarm_converge_merge(results: [String]) -> String {
let n: Int = el_list_len(results)
let merged: String = ""
let i = 0
while i < n {
let out: String = json_get_raw(el_list_get(results, i), "output")
if i > 0 {
let merged = merged + " | "
}
let merged = merged + out
let i = i + 1
}
return json_set_str("{}", "merged", merged)
}
// swarm_converge_vote tally a field across worker outputs, pick the majority.
// Each worker output is expected to carry a "verdict" string field.
fn swarm_converge_vote(results: [String]) -> String {
let n: Int = el_list_len(results)
// Collect verdicts (no mutable tally: json_set can't update an existing key
// and there is no el_list_set). Then count each verdict by rescanning.
let verdicts: [String] = el_list_empty()
let i = 0
while i < n {
let out: String = json_get_raw(el_list_get(results, i), "output")
let v: String = json_get_string(out, "verdict")
if str_eq(v, "") {
let i = i + 1
} else {
let verdicts = el_list_append(verdicts, v)
let i = i + 1
}
}
// pick the verdict with the highest count (first-past-the-post)
let vn: Int = el_list_len(verdicts)
let best: String = ""
let bestc = 0
let a = 0
while a < vn {
let cand: String = el_list_get(verdicts, a)
// count occurrences of cand
let c = 0
let b = 0
while b < vn {
if str_eq(el_list_get(verdicts, b), cand) {
let c = c + 1
}
let b = b + 1
}
if c > bestc {
let bestc = c
let best = cand
}
let a = a + 1
}
let kv: [String] = el_list_empty()
let kv = el_list_append(kv, "winner")
let kv = el_list_append(kv, best)
let kv = el_list_append(kv, "votes")
let kv = el_list_append(kv, int_to_str(bestc))
return json_build_object(kv)
}
// swarm_converge_reduce fold outputs into an accumulator (count + concat).
fn swarm_converge_reduce(results: [String]) -> String {
let n: Int = el_list_len(results)
let acc: String = ""
let i = 0
while i < n {
let out: String = json_get_raw(el_list_get(results, i), "output")
let acc = acc + out
let i = i + 1
}
let kv: [String] = el_list_empty()
let kv = el_list_append(kv, "count")
let kv = el_list_append(kv, int_to_str(n))
let kv = el_list_append(kv, "accumulated")
let kv = el_list_append(kv, acc)
return json_build_object(kv)
}
// ratio_to_permille parse a decimal ratio string ("1.0", "0.8") into an
// integer per-mille (1000, 800) so failure thresholds use exact integer math.
// (El float division is unreliable in this runtime int_to_float(n)/int_to_float(n)
// does not equal 1.0 so the swarm deliberately avoids floats.)
fn ratio_to_permille(s: String) -> Int {
if str_eq(s, "") {
return 1000
}
let parts: [String] = str_split(s, ".")
let whole: Int = str_to_int(el_list_get(parts, 0))
let permille: Int = whole * 1000
if el_list_len(parts) > 1 {
let frac_raw: String = el_list_get(parts, 1)
let frac3: String = str_slice(str_pad_right(frac_raw, 3, "0"), 0, 3)
let permille = permille + str_to_int(frac3)
}
return permille
}
// swarm_converge dispatch on strategy name.
fn swarm_converge(strategy: String, results: [String]) -> String {
if str_eq(strategy, "merge") {
return swarm_converge_merge(results)
}
if str_eq(strategy, "vote") {
return swarm_converge_vote(results)
}
if str_eq(strategy, "reduce") {
return swarm_converge_reduce(results)
}
// default: collect
return swarm_converge_collect(results)
}
// the ONLY global-engram write path (Rule 4, @manager-only)
//
// Every engram mutation flows through here and is gated by the caller's token
// capability. Only the orchestrator's token carries engram:write, so a worker
// (engram:read only) calling this is DENIED by capability before any HTTP is
// issued structurally unable to mutate global engram state, regardless of
// engram health. This is the curated-merge write: the orchestrator committing
// the geometry it approved. Workers never reach a successful branch here.
fn swarm_engram_write(token: String, corr_id: String, content: String, typ: String, importance: Float) -> String {
let deny: String = containment_guard_engram_write(token, corr_id, "engram.write")
if str_eq(deny, "") {
// authorized (orchestrator) perform the write
let res: String = op_write(content, typ, importance)
let new_id: String = json_get_string(res, "id")
let cp: String = json_set_str("{}", "node_id", new_id)
worktrack_append("swarm.committed", corr_id, "orchestrator", cp)
return res
}
// denied by capability return the rejection, no engram mutation performed
return json_set_str("{}", "denied", deny)
}
// the coordinator: fan out -> track -> converge
//
// blueprint : task blueprint name run by every worker
// knowledge_refs : JSON array of retrieval queries for CCR compilation
// inputs_json : JSON array of input items (one per worker)
// config_json : { concurrency, strategy, min_success_ratio,
// failure_action, caller_token }
//
// Returns: { corr_id, status:"completed"|"aborted", merged, report }.
fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, config_json: String) -> String {
let corr_id: String = "swarm-" + uuid_v4()
let caller_token: String = json_get_raw(config_json, "caller_token")
let concurrency: Int = str_to_int(json_get_string(config_json, "concurrency"))
if concurrency < 1 {
let concurrency = 4
}
let strategy: String = json_get_string(config_json, "strategy")
// Containment Rule 2: only a coordinator/absent token may open a swarm
let deny: String = containment_guard_open(caller_token, corr_id)
if str_eq(deny, "") {
// allowed proceed
let n: Int = json_array_len(inputs_json)
// swarm.created
let cp: String = json_set_str("{}", "blueprint", blueprint)
let cp2: String = json_set(cp, "input_count", int_to_str(n))
worktrack_append("swarm.created", corr_id, corr_id, cp2)
// build per-worker envelopes: worker token + CCR-compiled bounded context
let envelopes: [String] = el_list_empty()
let i = 0
while i < n {
let worker_id: String = corr_id + "/worker-" + int_to_str(i)
let input_item: String = json_array_get_string(inputs_json, i)
let wtoken: String = containment_worker_token(corr_id, worker_id)
let ctx: String = ccr_compile(blueprint, knowledge_refs, input_item, corr_id, worker_id, wtoken)
// envelope: only this worker's compiled context + its closed token
let ekv: [String] = el_list_empty()
let ekv = el_list_append(ekv, "worker_id")
let ekv = el_list_append(ekv, worker_id)
let ekv = el_list_append(ekv, "corr_id")
let ekv = el_list_append(ekv, corr_id)
let env0: String = json_build_object(ekv)
let env1: String = json_set(env0, "scope_token", wtoken)
let env2: String = json_set(env1, "ctx", ctx)
let envelopes = el_list_append(envelopes, env2)
let sp: String = json_set_str("{}", "input", input_item)
worktrack_append("worker.started", corr_id, worker_id, sp)
let i = i + 1
}
// native-thread fan-out (bounded)
let results: [String] = swarm_fanout("swarm_worker_entry", envelopes, concurrency)
// record per-worker terminal status + aggregate AFFERENT telemetry.
// Afferent counters (primitive signals travelling toward the @manager)
// are summed from the vertical result path no shared bus register,
// so the aggregation is containment-safe.
let succ = 0
let afferent = 0
let seam_mode_seen: String = "stub"
let rn: Int = el_list_len(results)
let r = 0
while r < rn {
let res: String = el_list_get(results, r)
let wid: String = json_get_string(res, "worker_id")
let st: String = json_get_string(res, "status")
let out: String = json_get_raw(res, "output")
let aff: Int = str_to_int(json_get_string(out, "afferent"))
let afferent = afferent + aff
let sm: String = json_get_string(out, "seam_mode")
if str_eq(sm, "") {
let seam_mode_seen = seam_mode_seen
} else {
let seam_mode_seen = sm
}
if str_eq(st, "completed") {
let succ = succ + 1
worktrack_append("worker.completed", corr_id, wid, json_set_str("{}", "status", "completed"))
} else {
worktrack_append("worker.failed", corr_id, wid, json_set_str("{}", "error", json_get_string(res, "error")))
}
let r = r + 1
}
// swarm.converging
let vg: String = json_set("{}", "success_count", int_to_str(succ))
worktrack_append("swarm.converging", corr_id, corr_id, vg)
// swarm.telemetry afferent counters observed by the @manager.
let tkv: [String] = el_list_empty()
let tkv = el_list_append(tkv, "seam_mode")
let tkv = el_list_append(tkv, seam_mode_seen)
let telem0: String = json_build_object(tkv)
let telem1: String = json_set_str(telem0, "afferent_think", int_to_str(afferent))
let telemetry: String = json_set_str(telem1, "results_received", int_to_str(rn))
worktrack_append("swarm.telemetry", corr_id, corr_id, telemetry)
// failure threshold (Swarm §4.3), integer per-mille math
// require succ/n >= min_success_ratio <=> succ*1000 >= permille*n
let permille: Int = ratio_to_permille(json_get_string(config_json, "min_success_ratio"))
let status: String = "completed"
if succ * 1000 < permille * n {
let status = "aborted"
}
if str_eq(status, "aborted") {
let ap: String = json_set_str("{}", "reason", "success ratio below min_success_ratio")
worktrack_append("swarm.aborted", corr_id, corr_id, ap)
let rep: String = worktrack_swarm_report(corr_id)
let ok: [String] = el_list_empty()
let ok = el_list_append(ok, "corr_id")
let ok = el_list_append(ok, corr_id)
let ok = el_list_append(ok, "status")
let ok = el_list_append(ok, "aborted")
let out0: String = json_build_object(ok)
return json_set(out0, "report", rep)
}
// converge
let merged: String = swarm_converge(strategy, results)
let dp: String = json_set_str("{}", "strategy", strategy)
worktrack_append("swarm.completed", corr_id, corr_id, dp)
// curated merge = the ONLY engram write path (Rule 4)
// With "commit":"1", the ORCHESTRATOR (its token carries engram:write)
// commits the approved merged geometry back to the engram. This is the
// single writer. Workers returned geometry; only the orchestrator writes.
let commit_id: String = ""
if str_eq(json_get_string(config_json, "commit"), "1") {
let orch_token: String = containment_coordinator_token(corr_id)
let cres: String = swarm_engram_write(orch_token, corr_id, "swarm-merge " + corr_id + " :: " + merged, "memory", 0.5)
let commit_id = json_get_string(cres, "id")
}
let rep2: String = worktrack_swarm_report(corr_id)
let ok2: [String] = el_list_empty()
let ok2 = el_list_append(ok2, "corr_id")
let ok2 = el_list_append(ok2, corr_id)
let ok2 = el_list_append(ok2, "status")
let ok2 = el_list_append(ok2, "completed")
let out1: String = json_build_object(ok2)
let out2: String = json_set(out1, "report", rep2)
let out3: String = json_set(out2, "merged", merged)
let out4: String = json_set(out3, "telemetry", telemetry)
return json_set_str(out4, "committed_node", commit_id)
}
// denied: caller was a worker trying to open a swarm (Rule 2)
let dkv: [String] = el_list_empty()
let dkv = el_list_append(dkv, "corr_id")
let dkv = el_list_append(dkv, corr_id)
let dkv = el_list_append(dkv, "status")
let dkv = el_list_append(dkv, "denied")
let dkv = el_list_append(dkv, "error")
let dkv = el_list_append(dkv, deny)
return json_build_object(dkv)
}