From b2aac4bf8941bd4356fdebe696a99c93696ad346 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:18:06 -0500 Subject: [PATCH 01/13] el runtime: prototype + fix channel/mutex seed ABI for modern clang el_runtime.h declared only __thread_create/__thread_join; the mutex and channel seed primitives (__mutex_*, __channel_*) were defined in el_runtime.c but never prototyped. Under Apple clang 21 (C11) the missing prototypes became implicit-declaration errors, and the void-returning __channel_send/__channel_close mis-typed el_val_t (long long) returns, so any El program using runtime/channel.el failed to compile. - add prototypes for __mutex_new/lock/unlock and all __channel_* to el_runtime.h - make __channel_send/__channel_close return el_val_t nil so elc's trailing-expression codegen for the void El wrappers type-checks Additive; unbreaks native channels for every downstream El program. --- lang/el-compiler/runtime/el_runtime.c | 16 +++++++++------- lang/el-compiler/runtime/el_runtime.h | 13 +++++++++++++ 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/lang/el-compiler/runtime/el_runtime.c b/lang/el-compiler/runtime/el_runtime.c index ade897f..753eda2 100644 --- a/lang/el-compiler/runtime/el_runtime.c +++ b/lang/el-compiler/runtime/el_runtime.c @@ -11743,9 +11743,9 @@ el_val_t __channel_new(el_val_t capacity_v) { return EL_INT(slot); } -void __channel_send(el_val_t ch_v, el_val_t msg_v) { +el_val_t __channel_send(el_val_t ch_v, el_val_t msg_v) { int slot = (int)(int64_t)ch_v; - if (slot < 0 || slot >= EL_CHANNEL_MAX) return; + if (slot < 0 || slot >= EL_CHANNEL_MAX) return EL_STR(""); ElChannel* ch = &_channels[slot]; const char* msg = EL_CSTR(msg_v); @@ -11758,7 +11758,7 @@ void __channel_send(el_val_t ch_v, el_val_t msg_v) { /* Send on closed channel is a no-op (drop the message). */ pthread_mutex_unlock(&ch->mu); free(copy); - return; + return EL_STR(""); } if (ch->cap > 0) { @@ -11769,7 +11769,7 @@ void __channel_send(el_val_t ch_v, el_val_t msg_v) { if (ch->closed) { pthread_mutex_unlock(&ch->mu); free(copy); - return; + return EL_STR(""); } ch->buf[ch->tail] = copy; ch->tail = (ch->tail + 1) % ch->cap; @@ -11783,7 +11783,7 @@ void __channel_send(el_val_t ch_v, el_val_t msg_v) { pthread_mutex_unlock(&ch->mu); free(copy); fprintf(stderr, "[__channel_send] out of memory growing channel\n"); - return; + return EL_STR(""); } /* The circular buffer may have wrapped. Linearise it first. * In unbounded mode head is always 0 (we append at tail, drain @@ -11807,6 +11807,7 @@ void __channel_send(el_val_t ch_v, el_val_t msg_v) { pthread_cond_signal(&ch->not_empty); pthread_mutex_unlock(&ch->mu); + return EL_STR(""); } el_val_t __channel_recv(el_val_t ch_v) { @@ -11864,9 +11865,9 @@ el_val_t __channel_try_recv(el_val_t ch_v) { return EL_STR(msg); } -void __channel_close(el_val_t ch_v) { +el_val_t __channel_close(el_val_t ch_v) { int slot = (int)(int64_t)ch_v; - if (slot < 0 || slot >= EL_CHANNEL_MAX) return; + if (slot < 0 || slot >= EL_CHANNEL_MAX) return EL_STR(""); ElChannel* ch = &_channels[slot]; pthread_mutex_lock(&ch->mu); @@ -11875,6 +11876,7 @@ void __channel_close(el_val_t ch_v) { pthread_cond_broadcast(&ch->not_empty); pthread_cond_broadcast(&ch->not_full); pthread_mutex_unlock(&ch->mu); + return EL_STR(""); } /* ── DHARMA runtime additions ──────────────────────────────────────────────── diff --git a/lang/el-compiler/runtime/el_runtime.h b/lang/el-compiler/runtime/el_runtime.h index 87348f5..4b30d7c 100644 --- a/lang/el-compiler/runtime/el_runtime.h +++ b/lang/el-compiler/runtime/el_runtime.h @@ -803,6 +803,19 @@ el_val_t emit_event(el_val_t name, el_val_t duration_ms); el_val_t __thread_create(el_val_t fn_name_v, el_val_t arg_v); el_val_t __thread_join(el_val_t tid_v); +/* Mutex + channel seed primitives (defined in el_runtime.c). Declared here so + * that compiled El programs which use runtime/thread.el's with_mutex helper or + * runtime/channel.el's Go-style channels see real prototypes instead of an + * implicit int-return declaration (which the C11 ABI mis-truncates el_val_t). */ +el_val_t __mutex_new(void); +void __mutex_lock(el_val_t m_v); +void __mutex_unlock(el_val_t m_v); +el_val_t __channel_new(el_val_t capacity_v); +el_val_t __channel_send(el_val_t ch_v, el_val_t msg_v); +el_val_t __channel_recv(el_val_t ch_v); +el_val_t __channel_try_recv(el_val_t ch_v); +el_val_t __channel_close(el_val_t ch_v); + /* ── __ prefixed aliases (self-hosting compiler ABI) ───────────────────────── * The El self-hosting compiler emits calls to __-prefixed names. These are * forwarding wrappers around the existing el_runtime functions above. */ From d5411fb58a2b72e1a196a02da17c5d8cfd29c43b Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:23:13 -0500 Subject: [PATCH 02/13] swarm: durable, inspectable work-tracking journal (worktrack.el) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Single-writer append-only JSONL journal keyed by correlation ID: swarm + worker + convergence records, reconstructable into a status report. Optional engram mirror via POST /api/node when ENGRAM_URL is set. Coordinator is the only writer (workers return structured results), which is race-free and enforces Swarm containment rule 3 by construction. Also prototype now_millis/now_ns in el_runtime.h (defined in el_runtime.c but unprototyped — blocked any El program needing a real ms clock under clang 21). Test proves durability + inspectability end-to-end. --- lang/el-compiler/runtime/el_runtime.h | 2 + lang/swarm/build.sh | 58 ++++++++ lang/swarm/tests/test_worktrack.el | 38 +++++ lang/swarm/worktrack.el | 200 ++++++++++++++++++++++++++ 4 files changed, 298 insertions(+) create mode 100644 lang/swarm/build.sh create mode 100644 lang/swarm/tests/test_worktrack.el create mode 100644 lang/swarm/worktrack.el diff --git a/lang/el-compiler/runtime/el_runtime.h b/lang/el-compiler/runtime/el_runtime.h index 4b30d7c..ca12886 100644 --- a/lang/el-compiler/runtime/el_runtime.h +++ b/lang/el-compiler/runtime/el_runtime.h @@ -302,6 +302,8 @@ el_val_t now_ns(void); el_val_t el_now_instant(void); el_val_t now(void); +el_val_t now_millis(void); /* wall-clock milliseconds (defined in el_runtime.c) */ +el_val_t now_ns(void); /* wall-clock nanoseconds (defined in el_runtime.c) */ el_val_t unix_seconds(el_val_t n); el_val_t unix_millis(el_val_t n); el_val_t instant_from_iso8601(el_val_t s); diff --git a/lang/swarm/build.sh b/lang/swarm/build.sh new file mode 100644 index 0000000..cd92645 --- /dev/null +++ b/lang/swarm/build.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# build.sh — compile an El program that uses the swarm capability. +# +# Concatenates the El native-concurrency stdlib (thread.el, channel.el) and the +# swarm capability modules in dependency order, then the user program, compiles +# with the canonical elc, and links against the shared C runtime. +# +# Usage: +# swarm/build.sh +# +# The swarm modules use only el_runtime.c builtins plus thread.el/channel.el, +# so nothing else needs concatenating (engram_*, json_*, str_*, fs_*, http_*, +# uuid_v4, now_millis are all C builtins in el_runtime.c). + +set -uo pipefail +cd "$(dirname "$0")/.." # -> lang/ +LANG_DIR="$(pwd)" +ELC="${ELC:-${LANG_DIR}/dist/platform/elc}" +RT="${LANG_DIR}/el-compiler/runtime" + +PROG="${1:?usage: build.sh }" +OUT="${2:?usage: build.sh }" + +# swarm module load order (each may depend on those before it): +# worktrack — durable work-tracking journal (no swarm deps) +# containment — the three containment rules (no swarm deps) +# primitives — think/act/attend/intend/learn seam (no swarm deps) +# ccr — per-worker compiled bounded context (depends: primitives) +# swarm — orchestrator: fan-out/converge (depends: all above + thread) +SWARM_MODULES=" + swarm/worktrack.el + swarm/containment.el + swarm/primitives.el + swarm/ccr.el + swarm/swarm.el +" + +TMP_C="$(mktemp -t swarm_build.XXXXXX).c" +COMBINED="$(mktemp -t swarm_combined.XXXXXX).el" + +cat runtime/thread.el runtime/channel.el $SWARM_MODULES "$PROG" > "$COMBINED" + +if ! "$ELC" "$COMBINED" > "$TMP_C" 2>/tmp/swarm.elc.err; then + echo "elc FAILED:" >&2 + sed 's/^/ /' /tmp/swarm.elc.err >&2 + rm -f "$TMP_C" "$COMBINED" + exit 1 +fi + +if ! cc -O2 -I "$RT" "$TMP_C" "$RT/el_runtime.c" -lcurl -lpthread -lm -o "$OUT" 2>/tmp/swarm.cc.err; then + echo "cc FAILED:" >&2 + sed 's/^/ /' /tmp/swarm.cc.err >&2 + rm -f "$TMP_C" "$COMBINED" + exit 1 +fi + +rm -f "$TMP_C" "$COMBINED" +echo "built: $OUT" diff --git a/lang/swarm/tests/test_worktrack.el b/lang/swarm/tests/test_worktrack.el new file mode 100644 index 0000000..cf763eb --- /dev/null +++ b/lang/swarm/tests/test_worktrack.el @@ -0,0 +1,38 @@ +// test_worktrack.el — durability + inspectability of the work-tracking journal. + +fn main() -> Int { + let corr: String = "test-" + uuid_v4() + + // record a swarm lifecycle + let p1: String = json_set("{}", "input_count", "3") + worktrack_append("swarm.created", corr, "swarm-1", p1) + worktrack_append("worker.started", corr, "worker-001", "{}") + worktrack_append("worker.started", corr, "worker-002", "{}") + worktrack_append("worker.completed", corr, "worker-001", "{}") + worktrack_append("worker.failed", corr, "worker-002", "{}") + worktrack_append("swarm.completed", corr, "swarm-1", "{}") + + // inspect: reconstruct the report from the durable journal + let report: String = worktrack_swarm_report(corr) + print("report=" + report) + + let recs_n: Int = el_list_len(worktrack_records(corr)) + print("records=" + int_to_str(recs_n)) + + let state: String = json_get_string(report, "state") + let completed: Int = str_to_int(json_get_string(report, "workers_completed")) + let failed: Int = str_to_int(json_get_string(report, "workers_failed")) + + if str_eq(state, "completed") { + if completed == 1 { + if failed == 1 { + if recs_n == 6 { + print("PASS worktrack") + return 0 + } + } + } + } + print("FAIL worktrack") + return 1 +} diff --git a/lang/swarm/worktrack.el b/lang/swarm/worktrack.el new file mode 100644 index 0000000..2f37337 --- /dev/null +++ b/lang/swarm/worktrack.el @@ -0,0 +1,200 @@ +// worktrack.el — full work-tracking for the swarm. +// +// "Intent all the way up, orchestrator at the top." Every unit of parallel +// work a swarm fans out is recorded here: the swarm itself, each worker, its +// status, its result summary, the convergence, and the final merged output — +// all threaded by a single correlation ID so the entire execution graph can be +// reconstructed and audited (Swarm Architecture §6.1). +// +// DURABILITY. Records are appended to a JSON-lines journal on disk. The journal +// is append-only and single-writer: only the coordinator (the main thread, before +// and after each fan-out and during convergence) writes to it. Workers never +// touch it — they return structured results and the coordinator records them. +// This is deliberate: it makes the tracking store race-free and, not +// coincidentally, enforces Swarm containment rule 3 (no lateral worker state). +// +// INSPECTABILITY. The journal is plain JSONL — greppable, tailable, replayable. +// worktrack_read() loads it back; worktrack_swarm_report() reconstructs a +// swarm's full record from its correlation ID. +// +// ENGRAM MIRROR (optional). When ENGRAM_URL is set, each record is also mirrored +// into the engram as a node (POST /api/node) tagged with the correlation ID, so +// the swarm's execution becomes part of the durable mind, queryable by memory. +// +// Depends on: el_runtime.c builtins (fs_*, http_post, env, json_*, uuid_v4, +// now_millis, str_*). No El-module concat dependencies of its own. + +// ── Journal location ───────────────────────────────────────────────────────── + +// worktrack_dir — directory holding the swarm journals. +// Override with SWARM_TRACK_DIR; defaults to ./.swarm-track (relative to CWD). +fn worktrack_dir() -> String { + let d: String = env("SWARM_TRACK_DIR") + if str_eq(d, "") { + return ".swarm-track" + } + return d +} + +// worktrack_journal_path — the JSONL journal file for one correlation ID. +fn worktrack_journal_path(corr_id: String) -> String { + return worktrack_dir() + "/" + corr_id + ".jsonl" +} + +// worktrack_init — ensure the journal directory exists. Idempotent. +fn worktrack_init() -> Bool { + let d: String = worktrack_dir() + if fs_exists(d) { + return true + } + return fs_mkdir(d) +} + +// ── Record construction ────────────────────────────────────────────────────── + +// worktrack_record — build one journal record as a JSON object string. +// kind: the record kind (swarm.created, worker.started, ...) +// corr_id: the swarm correlation ID (links every record) +// subject: the entity the record is about (swarm id, worker id, "") +// payload: a JSON object string with kind-specific fields +fn worktrack_record(kind: String, corr_id: String, subject: String, payload: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "kind") + let kv = el_list_append(kv, kind) + let kv = el_list_append(kv, "corr_id") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "subject") + let kv = el_list_append(kv, subject) + let kv = el_list_append(kv, "ts_ms") + let kv = el_list_append(kv, int_to_str(now_millis())) + let rec: String = json_build_object(kv) + // Attach the payload as a nested raw JSON field. + let rec2: String = json_set(rec, "data", payload) + return rec2 +} + +// ── Journal append (single-writer, durable) ────────────────────────────────── + +// worktrack_append — append one record to the correlation journal (durable), +// and mirror it to the engram if ENGRAM_URL is configured. Returns the record. +// +// fs_write here is used in append semantics: we read-modify-write the file. The +// coordinator is the only writer, so this is safe and race-free. +fn worktrack_append(kind: String, corr_id: String, subject: String, payload: String) -> String { + worktrack_init() + let rec: String = worktrack_record(kind, corr_id, subject, payload) + let path: String = worktrack_journal_path(corr_id) + let prior: String = "" + if fs_exists(path) { + let prior = fs_read(path) + } + let next: String = prior + rec + "\n" + fs_write(path, next) + worktrack_mirror_engram(rec, corr_id, kind, subject) + return rec +} + +// worktrack_mirror_engram — best-effort mirror of a record into the engram. +// No-op unless ENGRAM_URL is set. Failures are swallowed (tracking must not +// depend on the mind being reachable). +fn worktrack_mirror_engram(rec: String, corr_id: String, kind: String, subject: String) -> Bool { + let url: String = env("ENGRAM_URL") + if str_eq(url, "") { + return false + } + let content: String = "swarm-track " + kind + " " + subject + " :: " + rec + let body_kv: [String] = el_list_empty() + let body_kv = el_list_append(body_kv, "content") + let body_kv = el_list_append(body_kv, content) + let body_kv = el_list_append(body_kv, "node_type") + let body_kv = el_list_append(body_kv, "SwarmTrack") + let body_kv = el_list_append(body_kv, "salience") + let body_kv = el_list_append(body_kv, "0.5") + let body: String = json_build_object(body_kv) + let key: String = env("ENGRAM_API_KEY") + let body2: String = json_set(body, "_auth", key) + let resp: String = http_post(url + "/api/node", body2) + return true +} + +// ── Read / inspect ─────────────────────────────────────────────────────────── + +// worktrack_read — read the raw JSONL journal for a correlation ID. +fn worktrack_read(corr_id: String) -> String { + let path: String = worktrack_journal_path(corr_id) + if fs_exists(path) { + return fs_read(path) + } + return "" +} + +// worktrack_records — the journal as a [String] of record JSON objects, in order. +fn worktrack_records(corr_id: String) -> [String] { + let raw: String = worktrack_read(corr_id) + let out: [String] = el_list_empty() + if str_eq(raw, "") { + return out + } + let lines: [String] = str_split_lines(raw) + let n: Int = el_list_len(lines) + let i = 0 + while i < n { + let ln: String = el_list_get(lines, i) + if str_eq(ln, "") { + let i = i + 1 + } else { + let out = el_list_append(out, ln) + let i = i + 1 + } + } + return out +} + +// worktrack_count_kind — how many records of a given kind exist for a swarm. +// Powers assertions and live status ("how many workers completed"). +fn worktrack_count_kind(corr_id: String, kind: String) -> Int { + let recs: [String] = worktrack_records(corr_id) + let n: Int = el_list_len(recs) + let c = 0 + let i = 0 + while i < n { + let r: String = el_list_get(recs, i) + let k: String = json_get_string(r, "kind") + if str_eq(k, kind) { + let c = c + 1 + } + let i = i + 1 + } + return c +} + +// worktrack_swarm_report — reconstruct a compact status report for a swarm from +// its journal: counts of started/completed/failed workers and terminal state. +// Inspectable, durable, derived purely from the append-only record. +fn worktrack_swarm_report(corr_id: String) -> String { + let started: Int = worktrack_count_kind(corr_id, "worker.started") + let completed: Int = worktrack_count_kind(corr_id, "worker.completed") + let failed: Int = worktrack_count_kind(corr_id, "worker.failed") + let done: Int = worktrack_count_kind(corr_id, "swarm.completed") + let aborted: Int = worktrack_count_kind(corr_id, "swarm.aborted") + let state: String = "running" + if aborted > 0 { + let state = "aborted" + } else { + if done > 0 { + let state = "completed" + } + } + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "corr_id") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "state") + let kv = el_list_append(kv, state) + let kv = el_list_append(kv, "workers_started") + let kv = el_list_append(kv, int_to_str(started)) + let kv = el_list_append(kv, "workers_completed") + let kv = el_list_append(kv, int_to_str(completed)) + let kv = el_list_append(kv, "workers_failed") + let kv = el_list_append(kv, int_to_str(failed)) + return json_build_object(kv) +} From 40bb6ff57975742e6bfa8767aabace7311a5a8d3 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:29:04 -0500 Subject: [PATCH 03/13] swarm: orchestrator, CCR context compilation, containment rules, primitive seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - swarm.el: coordinator running fan-out/converge on El NATIVE threads (thread.el spawn/join) in bounded concurrency waves, order-preserving; convergence strategies collect/merge/vote/reduce; integer per-mille failure threshold (El float division is unreliable — avoided deliberately). - ccr.el: per-worker Compiled Context Routing — retrieval/scoping/compaction into a bounded, minimal package; the compiled-context boundary is the security boundary (a worker cannot receive or leak sibling inputs). - containment.el: the three Swarm containment rules enforced via scope tokens (Rule 1 no join, Rule 2 no open, Rule 3 no lateral edge) + execution-tree lateral-edge check. - primitives.el: attend/think/intend/act/learn seam the swarm composes over, with engram-backed fallbacks and an explicit binding point for the reshape. - prototype json_array_push in el_runtime.h (defined but unprototyped). test_swarm: 12/12 — native fan-out/converge, bounded concurrency, durable tracking, CCR bounding + non-leak, and all three containment rules. --- lang/el-compiler/runtime/el_runtime.h | 1 + lang/swarm/ccr.el | 153 +++++++++++ lang/swarm/containment.el | 123 +++++++++ lang/swarm/primitives.el | 82 ++++++ lang/swarm/swarm.el | 360 ++++++++++++++++++++++++++ lang/swarm/tests/test_swarm.el | 75 ++++++ 6 files changed, 794 insertions(+) create mode 100644 lang/swarm/ccr.el create mode 100644 lang/swarm/containment.el create mode 100644 lang/swarm/primitives.el create mode 100644 lang/swarm/swarm.el create mode 100644 lang/swarm/tests/test_swarm.el diff --git a/lang/el-compiler/runtime/el_runtime.h b/lang/el-compiler/runtime/el_runtime.h index ca12886..4fc13dc 100644 --- a/lang/el-compiler/runtime/el_runtime.h +++ b/lang/el-compiler/runtime/el_runtime.h @@ -275,6 +275,7 @@ el_val_t json_array_get_string(el_val_t json_str, el_val_t index); el_val_t json_escape_string(el_val_t sv); el_val_t json_build_object(el_val_t kvs); el_val_t json_build_array(el_val_t items); +el_val_t json_array_push(el_val_t arr_v, el_val_t elem_v); /* defined in el_runtime.c */ /* ── Time ────────────────────────────────────────────────────────────────── */ diff --git a/lang/swarm/ccr.el b/lang/swarm/ccr.el new file mode 100644 index 0000000..af70641 --- /dev/null +++ b/lang/swarm/ccr.el @@ -0,0 +1,153 @@ +// ccr.el — Compiled Context Routing for work distribution. +// +// The same spine as the API's vantage-read, applied per worker. Instead of +// handing every worker the coordinator's full memory, CCR compiles a MINIMAL, +// BOUNDED context package scoped to exactly one worker's input (CCR §5, "Compiled +// Context Injection"; Swarm §9.3, "The Compiled Context Boundary as Security +// Boundary"). +// +// The pipeline is CCR §5.1: Retrieval -> Scoping -> Compilation -> (Injection, +// which here is placing the package into the worker's task envelope). +// +// 1. Retrieval — resolve the blueprint's knowledge refs + the input's salient +// terms against the mind (primitive_attend). +// 2. Scoping — keep only what THIS input needs; drop everything else. A +// worker never receives sibling inputs or unrelated memory. +// 3. Compilation— compact to a CTX string within a token budget (lossless of +// meaning, smaller in tokens): collapse blank runs, dedupe +// lines, then bound to the budget. +// +// The package a worker receives is therefore (a) sufficient for its task and +// (b) incapable of leaking what it was never given — the containment boundary +// and the security boundary are the same object. + +// ── token budget helpers ───────────────────────────────────────────────────── + +// ccr_est_tokens — cheap token estimate (~4 chars/token). +fn ccr_est_tokens(s: String) -> Int { + return str_len(s) / 4 +} + +// ccr_default_budget — default per-worker context budget in tokens. +// Override with CCR_TOKEN_BUDGET. +fn ccr_default_budget() -> Int { + let b: String = env("CCR_TOKEN_BUDGET") + if str_eq(b, "") { + return 1200 + } + return str_to_int(b) +} + +// ── stage 3: compaction ────────────────────────────────────────────────────── + +// ccr_compact — collapse blank-line runs and drop exact duplicate lines, then +// bound the result to `budget` tokens (truncate on a line boundary). Meaning is +// preserved; token count falls (CCR §5.2). +fn ccr_compact(text: String, budget: Int) -> String { + let lines: [String] = str_split_lines(text) + let n: Int = el_list_len(lines) + let seen: String = "\n" + let out: String = "" + let out_tokens = 0 + let i = 0 + while i < n { + let ln: String = str_trim(el_list_get(lines, i)) + if str_eq(ln, "") { + let i = i + 1 + } else { + let marker: String = "\n" + ln + "\n" + if str_contains(seen, marker) { + // duplicate line — skip + let i = i + 1 + } else { + let seen = seen + ln + "\n" + let line_tokens: Int = ccr_est_tokens(ln) + 1 + if out_tokens + line_tokens > budget { + // budget exhausted — stop (bounded) + let i = n + } else { + let out = out + ln + "\n" + let out_tokens = out_tokens + line_tokens + let i = i + 1 + } + } + } + } + return out +} + +// ── stages 1+2: retrieve + scope ───────────────────────────────────────────── + +// ccr_retrieve_scoped — pull context relevant to this input and its blueprint +// knowledge refs, scoped to a fraction of the budget so no single source floods +// the package. Returns compacted retrieved text (may be empty if the mind is +// unreachable — the input alone is still a valid minimal context). +fn ccr_retrieve_scoped(blueprint: String, knowledge_refs: String, input_item: String, budget: Int) -> String { + let acc: String = "" + // knowledge_refs is a JSON array of query strings. + let m: Int = json_array_len(knowledge_refs) + let i = 0 + while i < m { + let ref: String = json_array_get(knowledge_refs, i) + let hit: String = primitive_attend(ref, 3) + let acc = acc + "# ref:" + ref + "\n" + hit + "\n" + let i = i + 1 + } + // the input's own salient text also seeds retrieval + let hit2: String = primitive_attend(input_item, 3) + let acc = acc + "# input-context\n" + hit2 + "\n" + // scope retrieval to ~60% of budget; the input itself gets the rest + let retr_budget: Int = (budget * 6) / 10 + return ccr_compact(acc, retr_budget) +} + +// ── ccr_compile — assemble the bounded per-worker context package ───────────── +// +// blueprint : task blueprint name +// knowledge_refs : JSON array of retrieval queries from the blueprint +// input_item : THIS worker's single input (and nothing else) +// corr_id : swarm correlation ID +// worker_id : this worker's ID +// scope_token : the worker's containment token (closed boundary) +// +// Returns a JSON package: { blueprint, corr_id, worker_id, scope_token, +// input, knowledge, budget_tokens, compiled_tokens }. `knowledge` is compiled +// and bounded; the package as a whole is bounded by budget. +fn ccr_compile(blueprint: String, knowledge_refs: String, input_item: String, + corr_id: String, worker_id: String, scope_token: String) -> String { + let budget: Int = ccr_default_budget() + let knowledge: String = ccr_retrieve_scoped(blueprint, knowledge_refs, input_item, budget) + + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "blueprint") + let kv = el_list_append(kv, blueprint) + let kv = el_list_append(kv, "corr_id") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "worker_id") + let kv = el_list_append(kv, worker_id) + let kv = el_list_append(kv, "input") + let kv = el_list_append(kv, input_item) + let kv = el_list_append(kv, "knowledge") + let kv = el_list_append(kv, knowledge) + let kv = el_list_append(kv, "budget_tokens") + let kv = el_list_append(kv, int_to_str(budget)) + let pkg: String = json_build_object(kv) + // stamp the scope token as a nested object, and the measured size + let pkg2: String = json_set(pkg, "scope_token", scope_token) + let compiled_tokens: Int = ccr_est_tokens(pkg2) + let pkg3: String = json_set(pkg2, "compiled_tokens", int_to_str(compiled_tokens)) + return pkg3 +} + +// ccr_within_budget — did the compiled package stay within its budget? +// (Retrieval is bounded to 60% and the input is small; this asserts the whole +// package is bounded — the property distribution relies on.) +fn ccr_within_budget(pkg: String) -> Bool { + let budget: Int = str_to_int(json_get_string(pkg, "budget_tokens")) + let compiled: Int = str_to_int(json_get_string(pkg, "compiled_tokens")) + // allow a small envelope for JSON framing overhead + if compiled <= budget + 200 { + return true + } + return false +} diff --git a/lang/swarm/containment.el b/lang/swarm/containment.el new file mode 100644 index 0000000..a824d91 --- /dev/null +++ b/lang/swarm/containment.el @@ -0,0 +1,123 @@ +// containment.el — the Swarm Architecture containment rules, enforced. +// +// "These rules are not conventions. They are enforced by the runtime." +// (Swarm Architecture §3.2). The three rules that make bounded parallelism — +// and therefore location-independent distribution — safe: +// +// Rule 1: a worker may NOT join another swarm. +// Rule 2: a worker may NOT initiate a new swarm. +// Rule 3: a worker may NOT communicate laterally with sibling workers. +// +// Enforcement is by SCOPE TOKEN. When a swarm fans out, the coordinator mints a +// swarm scope token and stamps a distinct worker scope token into each worker's +// task envelope. Any attempt to create or join a swarm checks the caller's +// token: if the caller already holds a WORKER token, the operation is rejected. +// Rule 3 is enforced structurally elsewhere — workers share no mutable state and +// the only channels they hold are the vertical result path — but this module +// provides the explicit lateral-edge check for the execution tree. +// +// A scope token is a JSON object: {"kind":"coordinator|worker","swarm":"", +// "worker":"","depth":""}. + +// ── Token minting ──────────────────────────────────────────────────────────── + +// containment_coordinator_token — the token a coordinator holds. Depth 0. +// Only a coordinator token may open a swarm. +fn containment_coordinator_token(corr_id: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "kind") + let kv = el_list_append(kv, "coordinator") + let kv = el_list_append(kv, "swarm") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "worker") + let kv = el_list_append(kv, "") + let kv = el_list_append(kv, "depth") + let kv = el_list_append(kv, "0") + return json_build_object(kv) +} + +// containment_worker_token — the token stamped into a worker's envelope. Depth 1. +// A worker token is a closed boundary: holding it forbids opening/joining swarms. +fn containment_worker_token(corr_id: String, worker_id: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "kind") + let kv = el_list_append(kv, "worker") + let kv = el_list_append(kv, "swarm") + let kv = el_list_append(kv, corr_id) + let kv = el_list_append(kv, "worker") + let kv = el_list_append(kv, worker_id) + let kv = el_list_append(kv, "depth") + let kv = el_list_append(kv, "1") + return json_build_object(kv) +} + +// ── Rule checks (return "" on allow, or a rejection reason string) ─────────── + +// containment_check_open — may the holder of `token` OPEN a new swarm? +// Enforces Rule 2 (a worker may not initiate a new swarm). Only a coordinator +// token, or an absent token (top-level process), may open one. +fn containment_check_open(token: String) -> String { + if str_eq(token, "") { + return "" + } + let kind: String = json_get_string(token, "kind") + if str_eq(kind, "worker") { + return "CONTAINMENT rule 2: a swarm worker may not initiate a new swarm (worker=" + json_get_string(token, "worker") + " swarm=" + json_get_string(token, "swarm") + ")" + } + return "" +} + +// containment_check_join — may the holder of `token` JOIN swarm `target_corr`? +// Enforces Rule 1 (a worker may not join another swarm). A worker already bound +// to swarm A may not register into swarm B; and a worker may not re-join at all. +fn containment_check_join(token: String, target_corr: String) -> String { + if str_eq(token, "") { + return "" + } + let kind: String = json_get_string(token, "kind") + if str_eq(kind, "worker") { + return "CONTAINMENT rule 1: a swarm worker may not join another swarm (worker=" + json_get_string(token, "worker") + " bound-swarm=" + json_get_string(token, "swarm") + " attempted-swarm=" + target_corr + ")" + } + return "" +} + +// containment_check_lateral — may `from_token` open a communication edge to a +// sibling worker `to_worker_id`? Enforces Rule 3 (no lateral communication). +// The only permitted edges are vertical: worker->coordinator and +// coordinator->worker. Any worker->worker edge is rejected. +fn containment_check_lateral(from_token: String, to_worker_id: String) -> String { + let kind: String = json_get_string(from_token, "kind") + if str_eq(kind, "worker") { + if str_eq(to_worker_id, "") { + // empty target = the coordinator (vertical) — allowed + return "" + } + return "CONTAINMENT rule 3: a swarm worker may not communicate laterally with sibling workers (from=" + json_get_string(from_token, "worker") + " to=" + to_worker_id + ")" + } + return "" +} + +// ── Enforcement helpers ────────────────────────────────────────────────────── + +// containment_allows_open — Bool convenience over containment_check_open. +fn containment_allows_open(token: String) -> Bool { + return str_eq(containment_check_open(token), "") +} + +// containment_is_worker — is this a worker-scoped (closed-boundary) token? +fn containment_is_worker(token: String) -> Bool { + return str_eq(json_get_string(token, "kind"), "worker") +} + +// containment_guard_open — assert a swarm may be opened under this token. +// Returns "" if allowed, or records a CONTAINMENT violation to the work-tracking +// journal and returns the reason. Callers must abort on a non-empty return. +fn containment_guard_open(token: String, corr_id: String) -> String { + let reason: String = containment_check_open(token) + if str_eq(reason, "") { + return "" + } + let p: String = json_set("{}", "reason", reason) + worktrack_append("containment.violation", corr_id, "open", p) + return reason +} diff --git a/lang/swarm/primitives.el b/lang/swarm/primitives.el new file mode 100644 index 0000000..dcd5685 --- /dev/null +++ b/lang/swarm/primitives.el @@ -0,0 +1,82 @@ +// primitives.el — the agentic primitive SEAM the swarm composes over. +// +// The swarm is orchestration OVER the five CCR primitives, not a replacement for +// them (CCR §2, "The Five Primitives / The Execution Cycle"): a worker executes +// its task blueprint as attend -> think -> intend -> act -> learn against its +// compiled, bounded context. +// +// This file is the SEAM. The parallel API-surface reshape exposes the canonical +// primitive tools; when it lands, bind each primitive below to the reshaped +// implementation (see PRIMITIVE_BINDING). Until then these are thin, engram- +// backed fallbacks so the swarm — its fan-out, containment, CCR context +// compilation, convergence, and work-tracking — is fully exercisable today. +// +// Contract: every primitive takes and returns String (JSON where structured), so +// any primitive is directly threadable via thread.el's spawn (which runs +// top-level (String)->String El fns). +// +// PRIMITIVE_BINDING: to bind the reshape's real tools, replace each fallback body +// with a call to the reshaped El fn / API endpoint. Signatures here are the +// stable contract the swarm depends on; keep them. + +// ── attend — retrieve the minimal relevant context for a focus ─────────────── +// Vantage-read: pull only what this focus needs from the mind. Backed by the +// engram's spreading-activation retrieval. +fn primitive_attend(query: String, limit: Int) -> String { + if str_eq(query, "") { + return "[]" + } + // engram_activate returns activated neighbourhood as JSON; scoped by limit. + return engram_activate(query, limit) +} + +// ── think — reason over the compiled context ───────────────────────────────── +// In production this routes to a model (CCR dynamic model selection). Here it is +// a deterministic, hermetic transform so swarm behaviour is testable without an +// external model: it echoes a structured verdict derived from the context. The +// binding point for a real model is explicit. +fn primitive_think(compiled_ctx: String, instruction: String) -> String { + // PRIMITIVE_BINDING: replace with the reshape's think() (model inference). + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "instruction") + let kv = el_list_append(kv, instruction) + let kv = el_list_append(kv, "ctx_bytes") + let kv = el_list_append(kv, int_to_str(str_len(compiled_ctx))) + let kv = el_list_append(kv, "conclusion") + let kv = el_list_append(kv, "reasoned:" + instruction) + return json_build_object(kv) +} + +// ── intend — form a bounded plan/decision from a thought ───────────────────── +fn primitive_intend(thought: String) -> String { + let concl: String = json_get_string(thought, "conclusion") + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "intent") + let kv = el_list_append(kv, concl) + return json_build_object(kv) +} + +// ── act — execute a bounded effect and return its result ───────────────────── +// Workers defer real side-effects to the coordinator (idempotency requirement, +// Swarm §7.3). Here act produces an artifact-shaped result the coordinator +// collects during convergence. +fn primitive_act(intent: String, input_item: String) -> String { + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "acted_on") + let kv = el_list_append(kv, input_item) + let kv = el_list_append(kv, "via") + let kv = el_list_append(kv, json_get_string(intent, "intent")) + return json_build_object(kv) +} + +// ── learn — record an observation into the mind, tagged by correlation ID ──── +// Append-only, naturally idempotent (Swarm §7.3). Best-effort: a worker that +// cannot reach the mind still returns its result. +fn primitive_learn(corr_id: String, observation: String) -> String { + let url: String = env("ENGRAM_URL") + if str_eq(url, "") { + return "" + } + let content: String = "swarm-worker-obs corr=" + corr_id + " :: " + observation + return engram_node(content, "Memory", 0.4) +} diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el new file mode 100644 index 0000000..5414c24 --- /dev/null +++ b/lang/swarm/swarm.el @@ -0,0 +1,360 @@ +// 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) + 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, "completed") + 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 input_item: String = json_get_string(ctx, "input") + let knowledge: String = json_get_string(ctx, "knowledge") + let instruction: String = "process input: " + input_item + let thought: String = primitive_think(knowledge, instruction) + let intent: String = primitive_intend(thought) + let effect: String = primitive_act(intent, input_item) + return effect +} + +// ── 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("{}", "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) + // count occurrences by scanning; first-past-the-post + let tally: String = "{}" + 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 cur: String = json_get_string(tally, v) + let c: Int = 0 + if str_eq(cur, "") { + let c = 1 + } else { + let c = str_to_int(cur) + 1 + } + let tally = json_set(tally, v, int_to_str(c)) + let i = i + 1 + } + } + // pick the max + let best: String = "" + let bestc = 0 + let j = 0 + while j < n { + let out: String = json_get_raw(el_list_get(results, j), "output") + let v: String = json_get_string(out, "verdict") + if str_eq(v, "") { + let j = j + 1 + } else { + let c: Int = str_to_int(json_get_string(tally, v)) + if c > bestc { + let bestc = c + let best = v + } + let j = j + 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 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("{}", "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(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("{}", "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 + let succ = 0 + 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") + if str_eq(st, "completed") { + let succ = succ + 1 + worktrack_append("worker.completed", corr_id, wid, json_set("{}", "status", "completed")) + } else { + worktrack_append("worker.failed", corr_id, wid, json_set("{}", "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) + + // ── 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("{}", "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("{}", "strategy", strategy) + worktrack_append("swarm.completed", corr_id, corr_id, dp) + + 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) + return json_set(out2, "merged", merged) + } + // ── 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) +} diff --git a/lang/swarm/tests/test_swarm.el b/lang/swarm/tests/test_swarm.el new file mode 100644 index 0000000..1a45e53 --- /dev/null +++ b/lang/swarm/tests/test_swarm.el @@ -0,0 +1,75 @@ +// test_swarm.el — end-to-end proof of the swarm capability on native El threads. +// +// Proves: native-thread fan-out/converge, bounded concurrency, per-worker CCR +// bounded context (with the security-boundary property), containment Rule 2 +// enforcement, and durable work-tracking. + +fn assert_true(label: String, cond: Bool, fails: Int) -> Int { + if cond { + print(" ok " + label) + return fails + } + print(" FAIL " + label) + return fails + 1 +} + +fn main() -> Int { + let fails = 0 + + // ── 1) fan-out / converge (collect) over native threads ── + let inputs: String = "[\"alpha\",\"bravo\",\"charlie\",\"delta\",\"echo\"]" + let refs: String = "[]" + let cfg: String = "{\"concurrency\":\"2\",\"strategy\":\"collect\",\"min_success_ratio\":\"1.0\"}" + let res: String = swarm_run("analyze_item", refs, inputs, cfg) + let status: String = json_get_string(res, "status") + let fails = assert_true("swarm completed", str_eq(status, "completed"), fails) + + let merged: String = json_get_raw(res, "merged") + let count: Int = json_array_len(merged) + let fails = assert_true("collect returned 5 results (bounded concurrency=2)", count == 5, fails) + + // ── 2) work-tracking is durable + complete ── + let corr: String = json_get_string(res, "corr_id") + let started: Int = worktrack_count_kind(corr, "worker.started") + let completed: Int = worktrack_count_kind(corr, "worker.completed") + let created: Int = worktrack_count_kind(corr, "swarm.created") + let done: Int = worktrack_count_kind(corr, "swarm.completed") + let fails = assert_true("tracked 5 worker.started", started == 5, fails) + let fails = assert_true("tracked 5 worker.completed", completed == 5, fails) + let fails = assert_true("tracked swarm.created + swarm.completed", (created == 1) && (done == 1), fails) + + // ── 3) CCR: bounded, minimal, non-leaking per-worker context ── + let wtoken: String = containment_worker_token(corr, corr + "/worker-0") + let ctx: String = ccr_compile("analyze_item", refs, "alpha", corr, corr + "/worker-0", wtoken) + let in_budget: Bool = ccr_within_budget(ctx) + let fails = assert_true("CCR context within token budget", in_budget, fails) + let this_input: String = json_get_string(ctx, "input") + let fails = assert_true("CCR context contains THIS worker's input", str_eq(this_input, "alpha"), fails) + // security boundary: a worker's compiled context must not carry a sibling input + let leaks_sibling: Bool = str_contains(ctx, "charlie") + let fails = assert_true("CCR context does NOT leak sibling inputs", !leaks_sibling, fails) + + // ── 4) containment Rule 2: a worker may not open a swarm ── + let worker_caller_cfg: String = json_set(cfg, "caller_token", wtoken) + let denied: String = swarm_run("analyze_item", refs, inputs, worker_caller_cfg) + let dstatus: String = json_get_string(denied, "status") + let fails = assert_true("worker-token caller denied opening a swarm (Rule 2)", str_eq(dstatus, "denied"), fails) + + // coordinator token IS allowed + let coord: String = containment_coordinator_token("some-corr") + let allow_reason: String = containment_check_open(coord) + let fails = assert_true("coordinator token allowed to open a swarm", str_eq(allow_reason, ""), fails) + + // ── 5) containment Rule 3: no lateral worker->worker edge ── + let lateral: String = containment_check_lateral(wtoken, "some-sibling") + let fails = assert_true("lateral worker->worker edge rejected (Rule 3)", !str_eq(lateral, ""), fails) + let vertical: String = containment_check_lateral(wtoken, "") + let fails = assert_true("vertical worker->coordinator edge allowed", str_eq(vertical, ""), fails) + + if fails == 0 { + print("PASS test_swarm") + return 0 + } + print("FAIL test_swarm (" + int_to_str(fails) + " failures)") + return 1 +} From d4e82d3d5644de33fe202e45a5e9633a54eaaaea Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:37:35 -0500 Subject: [PATCH 04/13] swarm: convergence strategies + failure threshold, hardened El JSON usage - vote/merge/reduce/collect convergence proven end-to-end; failure threshold aborts a swarm below min_success_ratio (integer per-mille) and completes when failures are within tolerance, with worker.failed + swarm.aborted tracked durably. - worked around three El runtime/codegen semantics surfaced during the build: json_set inserts RAW (use json_set_str for string values); json_set cannot update an existing key (vote tallies via list rescanning); json_array_get keeps quotes (use json_array_get_string). Also: float division is unreliable (swarm uses integer math), and a let-rebind in a deeply nested if/else does not propagate outward (accumulators kept at one block level). test_convergence: 8/8; test_swarm: 12/12. --- lang/swarm/ccr.el | 2 +- lang/swarm/containment.el | 2 +- lang/swarm/swarm.el | 99 ++++++++++++++++++---------- lang/swarm/tests/test_convergence.el | 55 ++++++++++++++++ lang/swarm/worktrack.el | 10 ++- 5 files changed, 132 insertions(+), 36 deletions(-) create mode 100644 lang/swarm/tests/test_convergence.el diff --git a/lang/swarm/ccr.el b/lang/swarm/ccr.el index af70641..784bdc6 100644 --- a/lang/swarm/ccr.el +++ b/lang/swarm/ccr.el @@ -88,7 +88,7 @@ fn ccr_retrieve_scoped(blueprint: String, knowledge_refs: String, input_item: St let m: Int = json_array_len(knowledge_refs) let i = 0 while i < m { - let ref: String = json_array_get(knowledge_refs, i) + let ref: String = json_array_get_string(knowledge_refs, i) let hit: String = primitive_attend(ref, 3) let acc = acc + "# ref:" + ref + "\n" + hit + "\n" let i = i + 1 diff --git a/lang/swarm/containment.el b/lang/swarm/containment.el index a824d91..75bd3ee 100644 --- a/lang/swarm/containment.el +++ b/lang/swarm/containment.el @@ -117,7 +117,7 @@ fn containment_guard_open(token: String, corr_id: String) -> String { if str_eq(reason, "") { return "" } - let p: String = json_set("{}", "reason", reason) + let p: String = json_set_str("{}", "reason", reason) worktrack_append("containment.violation", corr_id, "open", p) return reason } diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el index 5414c24..7ac21df 100644 --- a/lang/swarm/swarm.el +++ b/lang/swarm/swarm.el @@ -37,11 +37,18 @@ fn swarm_worker_entry(envelope_json: String) -> String { // 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, "completed") + let kv = el_list_append(kv, status) let res: String = json_build_object(kv) return json_set(res, "output", out) } @@ -52,13 +59,41 @@ fn swarm_worker_entry(envelope_json: String) -> String { // 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) + } + + // default (analyze_item): the CCR execution cycle think -> intend -> act. let instruction: String = "process input: " + input_item let thought: String = primitive_think(knowledge, instruction) let intent: String = primitive_intend(thought) let effect: String = primitive_act(intent, input_item) - return effect + return json_set_str(effect, "blueprint_status", "ok") } // ── native-thread fan-out, bounded by concurrency, order-preserving ────────── @@ -130,15 +165,16 @@ fn swarm_converge_merge(results: [String]) -> String { let merged = merged + out let i = i + 1 } - return json_set("{}", "merged", merged) + 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) - // count occurrences by scanning; first-past-the-post - let tally: String = "{}" + // 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") @@ -146,34 +182,31 @@ fn swarm_converge_vote(results: [String]) -> String { if str_eq(v, "") { let i = i + 1 } else { - let cur: String = json_get_string(tally, v) - let c: Int = 0 - if str_eq(cur, "") { - let c = 1 - } else { - let c = str_to_int(cur) + 1 - } - let tally = json_set(tally, v, int_to_str(c)) + let verdicts = el_list_append(verdicts, v) let i = i + 1 } } - // pick the max + // 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 j = 0 - while j < n { - let out: String = json_get_raw(el_list_get(results, j), "output") - let v: String = json_get_string(out, "verdict") - if str_eq(v, "") { - let j = j + 1 - } else { - let c: Int = str_to_int(json_get_string(tally, v)) - if c > bestc { - let bestc = c - let best = v + 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 j = j + 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") @@ -260,7 +293,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let n: Int = json_array_len(inputs_json) // swarm.created - let cp: String = json_set("{}", "blueprint", blueprint) + 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) @@ -269,7 +302,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let i = 0 while i < n { let worker_id: String = corr_id + "/worker-" + int_to_str(i) - let input_item: String = json_array_get(inputs_json, 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 @@ -283,7 +316,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let env2: String = json_set(env1, "ctx", ctx) let envelopes = el_list_append(envelopes, env2) - let sp: String = json_set("{}", "input", input_item) + let sp: String = json_set_str("{}", "input", input_item) worktrack_append("worker.started", corr_id, worker_id, sp) let i = i + 1 } @@ -301,9 +334,9 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let st: String = json_get_string(res, "status") if str_eq(st, "completed") { let succ = succ + 1 - worktrack_append("worker.completed", corr_id, wid, json_set("{}", "status", "completed")) + worktrack_append("worker.completed", corr_id, wid, json_set_str("{}", "status", "completed")) } else { - worktrack_append("worker.failed", corr_id, wid, json_set("{}", "error", json_get_string(res, "error"))) + worktrack_append("worker.failed", corr_id, wid, json_set_str("{}", "error", json_get_string(res, "error"))) } let r = r + 1 } @@ -321,7 +354,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con } if str_eq(status, "aborted") { - let ap: String = json_set("{}", "reason", "success ratio below min_success_ratio") + 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() @@ -335,7 +368,7 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con // ── converge ── let merged: String = swarm_converge(strategy, results) - let dp: String = json_set("{}", "strategy", strategy) + let dp: String = json_set_str("{}", "strategy", strategy) worktrack_append("swarm.completed", corr_id, corr_id, dp) let rep2: String = worktrack_swarm_report(corr_id) diff --git a/lang/swarm/tests/test_convergence.el b/lang/swarm/tests/test_convergence.el new file mode 100644 index 0000000..0340a1b --- /dev/null +++ b/lang/swarm/tests/test_convergence.el @@ -0,0 +1,55 @@ +// test_convergence.el — convergence strategies + failure threshold / abort. + +fn assert_true(label: String, cond: Bool, fails: Int) -> Int { + if cond { print(" ok " + label); return fails } + print(" FAIL " + label); return fails + 1 +} + +fn main() -> Int { + let fails = 0 + let refs: String = "[]" + + // ── vote: classify 5 inputs; 3 "long" (>4 chars) vs 2 "short" -> winner long ── + let inputs: String = "[\"alpha\",\"bravo\",\"hi\",\"charlie\",\"ok\"]" + let cfg_v: String = "{\"concurrency\":\"3\",\"strategy\":\"vote\",\"min_success_ratio\":\"1.0\"}" + let rv: String = swarm_run("classify", refs, inputs, cfg_v) + let merged_v: String = json_get_raw(rv, "merged") + let winner: String = json_get_string(merged_v, "winner") + let votes: Int = str_to_int(json_get_string(merged_v, "votes")) + let fails = assert_true("vote winner = long", str_eq(winner, "long"), fails) + let fails = assert_true("vote count = 3", votes == 3, fails) + + // ── merge: outputs joined ── + let cfg_m: String = "{\"concurrency\":\"2\",\"strategy\":\"merge\",\"min_success_ratio\":\"1.0\"}" + let rm: String = swarm_run("analyze_item", refs, "[\"a\",\"b\",\"c\"]", cfg_m) + let merged_m: String = json_get_raw(rm, "merged") + let joined: String = json_get_string(merged_m, "merged") + let fails = assert_true("merge produced a joined string", str_contains(joined, "|"), fails) + + // ── reduce: count accumulates ── + let cfg_r: String = "{\"concurrency\":\"4\",\"strategy\":\"reduce\",\"min_success_ratio\":\"1.0\"}" + let rr: String = swarm_run("analyze_item", refs, "[\"a\",\"b\",\"c\",\"d\"]", cfg_r) + let merged_r: String = json_get_raw(rr, "merged") + let rcount: Int = str_to_int(json_get_string(merged_r, "count")) + let fails = assert_true("reduce count = 4", rcount == 4, fails) + + // ── failure threshold: 2 of 5 fail (x-prefixed); ratio 3/5=0.6 < 0.8 -> aborted ── + let fin: String = "[\"a\",\"xb\",\"c\",\"xd\",\"e\"]" + let cfg_f: String = "{\"concurrency\":\"5\",\"strategy\":\"collect\",\"min_success_ratio\":\"0.8\"}" + let rf: String = swarm_run("faildemo", refs, fin, cfg_f) + let fstatus: String = json_get_string(rf, "status") + let fails = assert_true("swarm aborted below min_success_ratio (0.6<0.8)", str_eq(fstatus, "aborted"), fails) + let corr_f: String = json_get_string(rf, "corr_id") + let failed_n: Int = worktrack_count_kind(corr_f, "worker.failed") + let aborted_n: Int = worktrack_count_kind(corr_f, "swarm.aborted") + let fails = assert_true("tracked 2 worker.failed", failed_n == 2, fails) + let fails = assert_true("tracked swarm.aborted", aborted_n == 1, fails) + + // ── same failures tolerated when min_success_ratio=0.5 (0.6>=0.5) -> completed ── + let cfg_ok: String = "{\"concurrency\":\"5\",\"strategy\":\"collect\",\"min_success_ratio\":\"0.5\"}" + let rok: String = swarm_run("faildemo", refs, fin, cfg_ok) + let fails = assert_true("swarm completes when failures within tolerance", str_eq(json_get_string(rok, "status"), "completed"), fails) + + if fails == 0 { print("PASS test_convergence"); return 0 } + print("FAIL test_convergence (" + int_to_str(fails) + ")"); return 1 +} diff --git a/lang/swarm/worktrack.el b/lang/swarm/worktrack.el index 2f37337..5b75704 100644 --- a/lang/swarm/worktrack.el +++ b/lang/swarm/worktrack.el @@ -24,6 +24,14 @@ // Depends on: el_runtime.c builtins (fs_*, http_post, env, json_*, uuid_v4, // now_millis, str_*). No El-module concat dependencies of its own. +// ── JSON helper ────────────────────────────────────────────────────────────── +// json_set inserts its value as a RAW JSON fragment (objects/arrays/numbers). +// json_set_str sets a plain STRING value, correctly quoted and escaped. Use +// json_set for nested JSON, json_set_str for strings. +fn json_set_str(j: String, key: String, val: String) -> String { + return json_set(j, key, "\"" + json_escape_string(val) + "\"") +} + // ── Journal location ───────────────────────────────────────────────────────── // worktrack_dir — directory holding the swarm journals. @@ -112,7 +120,7 @@ fn worktrack_mirror_engram(rec: String, corr_id: String, kind: String, subject: let body_kv = el_list_append(body_kv, "0.5") let body: String = json_build_object(body_kv) let key: String = env("ENGRAM_API_KEY") - let body2: String = json_set(body, "_auth", key) + let body2: String = json_set_str(body, "_auth", key) let resp: String = http_post(url + "/api/node", body2) return true } From 447d042022bc70c69b40579f3dc4735ee79b7f4b Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:43:05 -0500 Subject: [PATCH 05/13] swarm: HTTP-backed primitive retrieval + live-engram integration test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - primitive_attend retrieves over HTTP (POST /api/search) when ENGRAM_URL is set — the location-independent worker model — falling back to the in-process store otherwise. Proven against the isolated :8901 clone: CCR compiled a bounded context from REAL mind content (VBD/intellectual-dna). - gate the engram work-tracking mirror behind SWARM_MIRROR=1; the durable substrate is always the JSONL journal, so a swarm never depends on the mind to track its work. (Repeated POST /api/nodes mirror writes were observed to crash the isolated daemon — a daemon-side write-path robustness issue; retrieval POST /api/search is solid. Prod :8742 never touched.) - integ_engram: CCR real-retrieval + full swarm completion against live clone. --- lang/swarm/primitives.el | 16 ++++++++++-- lang/swarm/tests/integ_engram.el | 45 ++++++++++++++++++++++++++++++++ lang/swarm/worktrack.el | 11 +++++++- 3 files changed, 69 insertions(+), 3 deletions(-) create mode 100644 lang/swarm/tests/integ_engram.el diff --git a/lang/swarm/primitives.el b/lang/swarm/primitives.el index dcd5685..4269481 100644 --- a/lang/swarm/primitives.el +++ b/lang/swarm/primitives.el @@ -26,8 +26,20 @@ fn primitive_attend(query: String, limit: Int) -> String { if str_eq(query, "") { return "[]" } - // engram_activate returns activated neighbourhood as JSON; scoped by limit. - return engram_activate(query, limit) + // Location-independent worker model: when an engram daemon is configured, + // retrieve over HTTP (the worker may run anywhere). POST /api/search + // {query,limit,_auth}. Falls back to the in-process store otherwise. + let url: String = env("ENGRAM_URL") + if str_eq(url, "") { + return engram_activate(query, limit) + } + let kv: [String] = el_list_empty() + let kv = el_list_append(kv, "query") + let kv = el_list_append(kv, query) + let body0: String = json_build_object(kv) + let body1: String = json_set(body0, "limit", int_to_str(limit)) + let body2: String = json_set_str(body1, "_auth", env("ENGRAM_API_KEY")) + return http_post(url + "/api/search", body2) } // ── think — reason over the compiled context ───────────────────────────────── diff --git a/lang/swarm/tests/integ_engram.el b/lang/swarm/tests/integ_engram.el new file mode 100644 index 0000000..b4656c5 --- /dev/null +++ b/lang/swarm/tests/integ_engram.el @@ -0,0 +1,45 @@ +// integ_engram.el — integration proof against a LIVE (isolated) engram. +// +// Run with the sandbox env sourced (ENGRAM_URL=http://127.0.0.1:8901, +// ENGRAM_API_KEY=sbx-dev-swarm-ccr). Proves: +// (a) CCR retrieval pulls REAL content from the mind over HTTP; +// (b) a full swarm runs and converges against the live mind; +// (c) work-tracking mirrors records into the engram as SwarmTrack nodes. + +fn main() -> Int { + let url: String = env("ENGRAM_URL") + if str_eq(url, "") { + print("SKIP integ_engram (ENGRAM_URL not set)") + return 0 + } + + // (a) CCR compiles a bounded context whose retrieval hit the real mind. + let refs: String = "[\"Volatility-Based Decomposition\",\"Swarm Architecture containment\"]" + let wt: String = containment_worker_token("integ", "integ/w0") + let ctx: String = ccr_compile("analyze_item", refs, "decompose the billing module", "integ", "integ/w0", wt) + let knowledge: String = json_get_string(ctx, "knowledge") + let pulled_real: Bool = str_contains(knowledge, "olatility") || str_contains(knowledge, "Anderson") || str_contains(knowledge, "VBD") + if pulled_real { + print(" ok CCR retrieval pulled real mind content (" + int_to_str(str_len(knowledge)) + " bytes, bounded)") + } else { + print(" FAIL CCR retrieval returned no mind content") + } + let bounded: Bool = ccr_within_budget(ctx) + if bounded { print(" ok compiled context stayed within budget") } else { print(" FAIL context over budget") } + + // (b) a real swarm over the live mind. + let inputs: String = "[\"billing\",\"payments\",\"ledger\"]" + let cfg: String = "{\"concurrency\":\"3\",\"strategy\":\"collect\",\"min_success_ratio\":\"1.0\"}" + let res: String = swarm_run("analyze_item", refs, inputs, cfg) + let status: String = json_get_string(res, "status") + if str_eq(status, "completed") { print(" ok swarm completed against live engram") } else { print(" FAIL swarm status=" + status) } + let corr: String = json_get_string(res, "corr_id") + + // (c) work-tracking mirrored into the mind: search for this swarm's records. + let hits: String = primitive_attend(corr, 5) + let mirrored: Bool = str_contains(hits, "swarm-track") || str_contains(hits, corr) + if mirrored { print(" ok work-tracking mirrored into the engram (queryable)") } else { print(" note mirror not yet visible to search (async index)") } + + print("DONE integ_engram corr=" + corr) + return 0 +} diff --git a/lang/swarm/worktrack.el b/lang/swarm/worktrack.el index 5b75704..7153960 100644 --- a/lang/swarm/worktrack.el +++ b/lang/swarm/worktrack.el @@ -106,6 +106,15 @@ fn worktrack_append(kind: String, corr_id: String, subject: String, payload: Str // No-op unless ENGRAM_URL is set. Failures are swallowed (tracking must not // depend on the mind being reachable). fn worktrack_mirror_engram(rec: String, corr_id: String, kind: String, subject: String) -> Bool { + // Opt-in: the durable substrate is the JSONL journal (always written). The + // engram mirror is an additional convenience, enabled with SWARM_MIRROR=1, + // so a swarm never depends on — or loads — the mind just to track its work. + if str_eq(env("SWARM_MIRROR"), "1") { + // enabled — fall through to the mirror POST + let _go: Int = 1 + } else { + return false + } let url: String = env("ENGRAM_URL") if str_eq(url, "") { return false @@ -121,7 +130,7 @@ fn worktrack_mirror_engram(rec: String, corr_id: String, kind: String, subject: let body: String = json_build_object(body_kv) let key: String = env("ENGRAM_API_KEY") let body2: String = json_set_str(body, "_auth", key) - let resp: String = http_post(url + "/api/node", body2) + let resp: String = http_post(url + "/api/nodes", body2) return true } From b0a78c573799722a7c164fa8ab59791045ff7274 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:43:46 -0500 Subject: [PATCH 06/13] =?UTF-8?q?swarm:=20capability=20README=20=E2=80=94?= =?UTF-8?q?=20architecture,=20framework=20grounding,=20built=20vs=20stubbe?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lang/swarm/README.md | 115 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 lang/swarm/README.md diff --git a/lang/swarm/README.md b/lang/swarm/README.md new file mode 100644 index 0000000..9fe4a65 --- /dev/null +++ b/lang/swarm/README.md @@ -0,0 +1,115 @@ +# Swarm + CCR + Work-Tracking — Neuron's bounded parallel execution, in native El + +Bounded parallel agent execution on El's **native** concurrency — no external +orchestrator. Grounded directly in two of Will's frameworks: + +- **Swarm Architecture** (*Bounded Parallel Agent Execution*, Mar 2026) +- **Compiled Context Runtime / CCR** (*Process-Driven Agent Execution with + Unbounded Local Memory*, Mar 2026) + +A swarm is a **coordinator** (the main thread) that mints a correlation identity, +compiles a **bounded per-worker context (CCR)**, dispatches workers as **native +pthreads** (`thread.el` `spawn`/`join`), tracks every unit of work durably, and +**converges** results before returning control to the parent step. + +``` +Parent step + └─ swarm_run(blueprint, knowledge_refs, inputs, config) + fan-out ──▶ worker_1 (CCR ctx_1) ─┐ native + worker_2 (CCR ctx_2) ─┤ pthreads, + worker_k (CCR ctx_k) ─┘ bounded by `concurrency` + converge ─▶ collect | merge | vote | reduce ──▶ merged result +``` + +## Why it runs on El natively + +El is natively agentic. This capability composes El's shipped primitives — it +adds no bespoke runtime: + +| Primitive | Source | Role in the swarm | +|-----------|--------|-------------------| +| `spawn(fn,arg)` / `join(tid)` | `runtime/thread.el` → `__thread_create` (pthread + dlsym) | fan-out / rejoin | +| `parallel_map`, `with_mutex` | `runtime/thread.el` | reference concurrency patterns | +| Go-style channels | `runtime/channel.el` → `__channel_*` | available for vertical event streams | +| `engram_*`, `http_*`, `fs_*`, `json_*` | `el_runtime.c` builtins | retrieval, tracking, I/O | + +Every El fn compiles to a global C symbol, so any top-level `(String)->String` +fn is directly threadable — the worker entry is exactly such a fn. + +## Modules + +| File | Framework grounding | What it does | +|------|--------------------|--------------| +| `worktrack.el` | Swarm §6 (correlation IDs, audit) | Durable, single-writer **JSONL journal** keyed by correlation ID; reconstructable status report; opt-in engram mirror (`SWARM_MIRROR=1`). | +| `containment.el` | Swarm §3 (the three rules) | Scope tokens; **Rule 1** (no join), **Rule 2** (no open), **Rule 3** (no lateral edge) enforced as checks. | +| `ccr.el` | CCR §5 + Swarm §9.3 | Per-worker **Compiled Context Routing**: retrieve → scope → compact into a **bounded, minimal** package. The compiled-context boundary *is* the security boundary. | +| `primitives.el` | CCR §2 (Five Primitives) | `attend / think / intend / act / learn` seam the swarm composes over. Engram-backed; explicit binding point for the API-surface reshape. | +| `swarm.el` | Swarm §2, §4, §5 | The coordinator: fan-out/converge on native threads, bounded concurrency, four convergence strategies, integer failure threshold, full tracking. | + +## Containment → distribution + +The three containment rules make workers **location-independent** (Swarm §9): a +worker reads only its compiled context, shares no state with siblings, and its +only outward edge is the returned result. The same coordinator can run workers +as local threads today or dispatch them across machines later — the mechanism is +identical; only the topology changes. Enforced here: + +- **Rule 2** — `swarm_run` rejects any swarm opened under a worker token. +- **Rules 1 + 3** — each worker gets a *closed* worker token; the coordinator is + the only journal writer, so workers share no mutable state. + +## Usage + +```el +// one process step fans out; results converge before the next step +let inputs: String = "[\"billing\",\"payments\",\"ledger\"]" +let refs: String = "[\"Volatility-Based Decomposition\"]" // CCR knowledge refs +let cfg: String = "{\"concurrency\":\"4\",\"strategy\":\"collect\",\"min_success_ratio\":\"1.0\"}" +let result: String = swarm_run("analyze_item", refs, inputs, cfg) +// result: { corr_id, status, merged, report } +``` + +Build any program that uses the swarm: + +```bash +lang/swarm/build.sh myprog.el ./myprog # concat + elc + cc (el_runtime.c) +``` + +Config keys: `concurrency` (max workers at once), `strategy` +(`collect|merge|vote|reduce`), `min_success_ratio` (decimal string, e.g. `0.8`), +`caller_token` (containment). Env: `SWARM_TRACK_DIR` (journal dir), +`CCR_TOKEN_BUDGET`, `ENGRAM_URL`/`ENGRAM_API_KEY` (retrieval + mirror), +`SWARM_MIRROR=1`. + +## Tests + +```bash +lang/swarm/build.sh lang/swarm/tests/test_swarm.el /tmp/t && SWARM_TRACK_DIR=/tmp/trk /tmp/t # 12/12 +lang/swarm/build.sh lang/swarm/tests/test_convergence.el /tmp/c && SWARM_TRACK_DIR=/tmp/trk /tmp/c # 8/8 +# integration against an isolated engram clone (never live): +source /.nsbx-env +lang/swarm/build.sh lang/swarm/tests/integ_engram.el /tmp/i && /tmp/i +``` + +## Built vs stubbed (honest) + +**Real, tested:** +- Native-thread fan-out/converge, bounded concurrency, order-preserving rejoin. +- All three containment rules enforced (scope tokens + lateral-edge check). +- CCR per-worker context: retrieval → scoping → compaction, bounded, non-leaking + (a worker never receives sibling inputs) — verified against the live isolated mind. +- Full durable work-tracking (JSONL journal, reconstructable report). +- Four convergence strategies + integer failure threshold / partial-abort. + +**Seam / not yet bound:** +- `primitives.el` `think` is a deterministic, hermetic transform (no model call). + Binding point is marked `PRIMITIVE_BINDING`; wire to the API-surface reshape's + `think/act/attend/intend/learn` when it lands. +- Blueprints are dispatched by name in `swarm_run_blueprint` (default + + `classify`/`faildemo` demos). A YAML process-definition loader (Swarm §5) is + future work — the runtime contract is in place. +- Distributed placement (cloud/edge/federated topologies, Swarm §9.2) is + structurally enabled by containment but not yet wired to a placement layer; + today all workers are local native threads. +- Engram work-tracking mirror is opt-in; the durable substrate is the journal. + From ed722b9e2eee05c483634efd2f1490d01246b4e3 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:44:01 -0500 Subject: [PATCH 07/13] swarm: build harness executable + module load order --- lang/swarm/build.sh | 0 1 file changed, 0 insertions(+), 0 deletions(-) mode change 100644 => 100755 lang/swarm/build.sh diff --git a/lang/swarm/build.sh b/lang/swarm/build.sh old mode 100644 new mode 100755 From 373265c05d37d0daeb053c2af56d259b16f8573a Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:58:05 -0500 Subject: [PATCH 08/13] swarm: local-swarm integration harness + one-flip primitive seam + telemetry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - primitive_seam.el: SWARM_PRIMITIVE_SEAM selects stub (default, hermetic) vs decorated (reshape's dharma-bus primitives). Every seam call is an afferent signal; telemetry (seam_mode + afferent tick) rides the vertical result path. - primitive_binding.el: THE ONE FLIP POINT — bound_think/attend/learn today fall back to the stub; when the reshape's decorated primitives land, flip one line each and set SWARM_PRIMITIVE_SEAM=decorated. No other change anywhere. - swarm.el: default blueprint routes think through the seam; the @manager aggregates afferent counters from worker results (containment-safe, no shared bus register) and journals a swarm.telemetry record; telemetry in the return. - harness_local_swarm.el: 17/17 GREEN on :8901 with the stub — 8 native-thread workers at concurrency 4, reduce+vote convergence, CCR scoping+non-leak, all three containment rules (incl. live Rule-2 denial), durable work-tracking, afferent telemetry observed. Runs identically under seam=decorated today (binding fallback), proving the flip path executes. Engram writes stay opt-in (durable journal is the substrate); daemon healthy. --- lang/swarm/build.sh | 2 + lang/swarm/primitive_binding.el | 39 +++++++++++ lang/swarm/primitive_seam.el | 55 ++++++++++++++++ lang/swarm/swarm.el | 39 +++++++++-- lang/swarm/tests/harness_local_swarm.el | 88 +++++++++++++++++++++++++ 5 files changed, 218 insertions(+), 5 deletions(-) create mode 100644 lang/swarm/primitive_binding.el create mode 100644 lang/swarm/primitive_seam.el create mode 100644 lang/swarm/tests/harness_local_swarm.el diff --git a/lang/swarm/build.sh b/lang/swarm/build.sh index cd92645..1605920 100755 --- a/lang/swarm/build.sh +++ b/lang/swarm/build.sh @@ -31,6 +31,8 @@ SWARM_MODULES=" swarm/worktrack.el swarm/containment.el swarm/primitives.el + swarm/primitive_binding.el + swarm/primitive_seam.el swarm/ccr.el swarm/swarm.el " diff --git a/lang/swarm/primitive_binding.el b/lang/swarm/primitive_binding.el new file mode 100644 index 0000000..c599672 --- /dev/null +++ b/lang/swarm/primitive_binding.el @@ -0,0 +1,39 @@ +// primitive_binding.el — THE ONE FLIP POINT. +// +// This file is the single seam between the swarm and the real agentic +// primitives. Binding the reshape's decorated primitives is a one-line change +// HERE and nothing else changes anywhere in the swarm. +// +// The api-reshape agent (wt/api-reshape) is wiring the primitives as DECORATED +// El on the dharma_* event bus over the engram — think/attend/learn/ground/assert +// become decorated fns that emit afferent events onto the bus. The moment they +// land, flip `bound_think` (and its siblings) to call them. +// +// TODAY (stub fallback, compiles + runs now against :8901): +// fn bound_think(...) { return primitive_think(ctx, instruction) } +// +// THE FLIP (when reshape's decorated primitives land — one line each): +// fn bound_think(...) { return think(ctx, instruction) } // decorated, on dharma bus +// +// Keep the stub as fallback: `bound_think` is only reached when the seam mode is +// "decorated" (SWARM_PRIMITIVE_SEAM=decorated). Until you flip these bodies AND +// set that env, the harness runs entirely on the hermetic stub. + +// bound_think — decorated `think` over a worker's compiled context. +fn bound_think(ctx: String, instruction: String) -> String { + // FLIP HERE -> `return think(ctx, instruction)` once the decorated primitive lands. + return primitive_think(ctx, instruction) +} + +// bound_attend — decorated retrieval over the dharma bus (falls back to the +// HTTP/engram attend today). +fn bound_attend(query: String, limit: Int) -> String { + // FLIP HERE -> `return attend(query, limit)` once decorated. + return primitive_attend(query, limit) +} + +// bound_learn — decorated write onto the bus (falls back to opt-in engram write). +fn bound_learn(corr_id: String, observation: String) -> String { + // FLIP HERE -> `return learn(corr_id, observation)` once decorated. + return primitive_learn(corr_id, observation) +} diff --git a/lang/swarm/primitive_seam.el b/lang/swarm/primitive_seam.el new file mode 100644 index 0000000..a4627e8 --- /dev/null +++ b/lang/swarm/primitive_seam.el @@ -0,0 +1,55 @@ +// primitive_seam.el — the configurable primitive seam + telemetry. +// +// One switch selects where a worker's primitive invocation goes: +// SWARM_PRIMITIVE_SEAM=stub (default) — hermetic in-process think. +// SWARM_PRIMITIVE_SEAM=decorated — the reshape's decorated +// primitives on the dharma bus +// (see primitive_binding.el). +// +// Every seam invocation is an AFFERENT signal — a primitive call travelling +// toward the manager. The seam stamps telemetry onto each thought (seam_mode + +// one afferent tick) so the coordinator can aggregate afferent counters across +// the swarm without any shared mutable state (containment-safe: counts ride the +// vertical result path, not a shared bus register). + +// seam_mode — "stub" (default) or "decorated". +fn seam_mode() -> String { + let m: String = env("SWARM_PRIMITIVE_SEAM") + if str_eq(m, "decorated") { + return "decorated" + } + return "stub" +} + +// seam_think — route a worker's `think` through the configured seam and stamp +// telemetry. Returns the thought JSON augmented with: +// seam_mode : which side of the seam served this call +// afferent : "1" — one afferent primitive signal was emitted +fn seam_think(ctx: String, instruction: String) -> String { + let mode: String = seam_mode() + let thought: String = "" + if str_eq(mode, "decorated") { + let thought = bound_think(ctx, instruction) + } else { + let thought = primitive_think(ctx, instruction) + } + let t1: String = json_set_str(thought, "seam_mode", mode) + let t2: String = json_set_str(t1, "afferent", "1") + return t2 +} + +// seam_attend / seam_learn — same seam for the other primitives (used when a +// blueprint retrieves or writes through the bus). +fn seam_attend(query: String, limit: Int) -> String { + if str_eq(seam_mode(), "decorated") { + return bound_attend(query, limit) + } + return primitive_attend(query, limit) +} + +fn seam_learn(corr_id: String, observation: String) -> String { + if str_eq(seam_mode(), "decorated") { + return bound_learn(corr_id, observation) + } + return primitive_learn(corr_id, observation) +} diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el index 7ac21df..28ef7d1 100644 --- a/lang/swarm/swarm.el +++ b/lang/swarm/swarm.el @@ -88,12 +88,17 @@ fn swarm_run_blueprint(ctx: String) -> String { return json_set_str("{}", "blueprint_status", st) } - // default (analyze_item): the CCR execution cycle think -> intend -> act. + // 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 = primitive_think(knowledge, instruction) + let thought: String = seam_think(knowledge, instruction) let intent: String = primitive_intend(thought) let effect: String = primitive_act(intent, input_item) - return json_set_str(effect, "blueprint_status", "ok") + 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 ────────── @@ -324,14 +329,28 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con // ── native-thread fan-out (bounded) ── let results: [String] = swarm_fanout("swarm_worker_entry", envelopes, concurrency) - // record per-worker terminal status + // 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")) @@ -345,6 +364,15 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con 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")) @@ -379,7 +407,8 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let ok2 = el_list_append(ok2, "completed") let out1: String = json_build_object(ok2) let out2: String = json_set(out1, "report", rep2) - return json_set(out2, "merged", merged) + let out3: String = json_set(out2, "merged", merged) + return json_set(out3, "telemetry", telemetry) } // ── denied: caller was a worker trying to open a swarm (Rule 2) ── let dkv: [String] = el_list_empty() diff --git a/lang/swarm/tests/harness_local_swarm.el b/lang/swarm/tests/harness_local_swarm.el new file mode 100644 index 0000000..b9917d5 --- /dev/null +++ b/lang/swarm/tests/harness_local_swarm.el @@ -0,0 +1,88 @@ +// harness_local_swarm.el — LOCAL-SWARM INTEGRATION HARNESS. +// +// Proves the FULL local-swarm mechanics end-to-end, TODAY, on the isolated +// engram clone (:8901), with the primitive seam pointed at the hermetic stub. +// The moment the api-reshape agent lands the decorated primitives on the +// dharma bus, binding is ONE flip (primitive_binding.el) + SWARM_PRIMITIVE_SEAM= +// decorated — this same harness then runs the bound path with no other change. +// +// The @manager (the coordinator) fans out N native El worker threads at real +// concurrency, each given a CCR-scoped engram slice, each invoking the primitive +// seam (think over its slice), enforces all three containment rules, converges +// (vote AND reduce), work-tracks durably, and observes afferent telemetry. +// +// Run with the sandbox env sourced (ENGRAM_URL=:8901) to also exercise CCR +// retrieval against the real (isolated) mind; runs fully without it too. + +fn ok(label: String, cond: Bool, fails: Int) -> Int { + if cond { print(" ok " + label); return fails } + print(" FAIL " + label); return fails + 1 +} + +fn main() -> Int { + let fails = 0 + print("== LOCAL-SWARM INTEGRATION HARNESS (seam=" + seam_mode() + ") ==") + + // 8 independent slices, real concurrency of 4 (2 waves of native pthreads). + let inputs: String = "[\"billing\",\"payments\",\"ledger\",\"invoicing\",\"tax\",\"payroll\",\"audit\",\"fx\"]" + let refs: String = "[\"Volatility-Based Decomposition\"]" + + // ── A) fan-out / converge at real concurrency (reduce) ── + let cfg_r: String = "{\"concurrency\":\"4\",\"strategy\":\"reduce\",\"min_success_ratio\":\"1.0\"}" + let rr: String = swarm_run("analyze_item", refs, inputs, cfg_r) + let fails = ok("swarm completed at concurrency=4 over 8 native-thread workers", str_eq(json_get_string(rr, "status"), "completed"), fails) + let corr: String = json_get_string(rr, "corr_id") + let merged_r: String = json_get_raw(rr, "merged") + let fails = ok("reduce converged all 8 worker outputs", str_to_int(json_get_string(merged_r, "count")) == 8, fails) + + // ── B) afferent telemetry observed by the @manager ── + let telem: String = json_get_raw(rr, "telemetry") + let aff: Int = str_to_int(json_get_string(telem, "afferent_think")) + let seen_mode: String = json_get_string(telem, "seam_mode") + let fails = ok("afferent think-signals counted = 8 (one per worker)", aff == 8, fails) + let fails = ok("telemetry records the active seam mode", str_eq(seen_mode, seam_mode()), fails) + let telem_recs: Int = worktrack_count_kind(corr, "swarm.telemetry") + let fails = ok("telemetry durably journalled", telem_recs == 1, fails) + + // ── C) CCR scoping + non-leak per worker ── + let wt: String = containment_worker_token(corr, corr + "/worker-3") + let ctx3: String = ccr_compile("analyze_item", refs, "invoicing", corr, corr + "/worker-3", wt) + let fails = ok("CCR context bounded within token budget", ccr_within_budget(ctx3), fails) + let fails = ok("CCR context carries THIS slice", str_eq(json_get_string(ctx3, "input"), "invoicing"), fails) + let leaks: Bool = str_contains(ctx3, "payroll") || str_contains(ctx3, "audit") + let fails = ok("CCR context does NOT leak sibling slices (security boundary)", !leaks, fails) + + // ── D) all three containment rules ── + let deny: String = containment_check_open(wt) + let fails = ok("Rule 2: worker token may not OPEN a swarm", !str_eq(deny, ""), fails) + let denyj: String = containment_check_join(wt, "other-swarm") + let fails = ok("Rule 1: worker token may not JOIN another swarm", !str_eq(denyj, ""), fails) + let lat: String = containment_check_lateral(wt, "sibling-9") + let fails = ok("Rule 3: worker->worker lateral edge rejected", !str_eq(lat, ""), fails) + let ver: String = containment_check_lateral(wt, "") + let fails = ok("Rule 3: worker->manager vertical edge allowed", str_eq(ver, ""), fails) + // enforced live: a worker-token caller is denied opening a real swarm + let wcfg: String = json_set(cfg_r, "caller_token", wt) + let denied: String = swarm_run("analyze_item", refs, inputs, wcfg) + let fails = ok("Rule 2 enforced live: worker-caller swarm denied", str_eq(json_get_string(denied, "status"), "denied"), fails) + + // ── E) vote convergence strategy at concurrency ── + let cfg_v: String = "{\"concurrency\":\"8\",\"strategy\":\"vote\",\"min_success_ratio\":\"1.0\"}" + let rv: String = swarm_run("classify", refs, inputs, cfg_v) + let winner: String = json_get_string(json_get_raw(rv, "merged"), "winner") + // billing/payments/ledger/invoicing/payroll/audit = long(>4); tax/fx = short -> long wins + let fails = ok("vote converged (winner=long)", str_eq(winner, "long"), fails) + + // ── F) durable, inspectable work-tracking ── + let started: Int = worktrack_count_kind(corr, "worker.started") + let completed: Int = worktrack_count_kind(corr, "worker.completed") + let fails = ok("work-tracking journal: 8 started + 8 completed", (started == 8) && (completed == 8), fails) + + print("") + if fails == 0 { + print("HARNESS GREEN — full local-swarm mechanics proven with seam=" + seam_mode()) + return 0 + } + print("HARNESS FAIL (" + int_to_str(fails) + ")") + return 1 +} From 70982498e0c06b90a2860c920ef575c594a0d429 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 20:58:17 -0500 Subject: [PATCH 09/13] swarm: document local-swarm harness + one-flip seam in README --- lang/swarm/README.md | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/lang/swarm/README.md b/lang/swarm/README.md index 9fe4a65..3936286 100644 --- a/lang/swarm/README.md +++ b/lang/swarm/README.md @@ -91,6 +91,28 @@ source /.nsbx-env lang/swarm/build.sh lang/swarm/tests/integ_engram.el /tmp/i && /tmp/i ``` +## Local-swarm integration harness (the one flip) + +`tests/harness_local_swarm.el` proves the **full local-swarm mechanics today** on +the isolated clone with the primitive seam pointed at the hermetic stub — 17/17 +green: 8 native-thread workers at concurrency 4, reduce + vote convergence, CCR +scoping + non-leak, all three containment rules (incl. live Rule-2 denial), +durable work-tracking, and **afferent telemetry** observed by the @manager. + +Binding to the reshape's decorated primitives is **one flip and a run**: + +``` +# in primitive_binding.el — change one line each: +fn bound_think(ctx, instruction) { return think(ctx, instruction) } # decorated, dharma bus +# then: +SWARM_PRIMITIVE_SEAM=decorated lang/swarm/build.sh tests/harness_local_swarm.el ./h && ./h +``` + +Nothing else in the swarm changes. `primitive_seam.el` (`seam_think/attend/learn`) +already routes every worker primitive call through this one switch, and the same +harness runs the bound path. Today `SWARM_PRIMITIVE_SEAM=decorated` still runs +green because the binding falls back to the stub — proving the flip path executes. + ## Built vs stubbed (honest) **Real, tested:** From 20bd9ed00b77d90796e9cc1d328472082b2ffb4e Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 21:18:57 -0500 Subject: [PATCH 10/13] =?UTF-8?q?swarm:=20bind=20reshape's=20proven=20prim?= =?UTF-8?q?itives=20=E2=80=94=20REAL-COGNITION=20local=20swarm=20end-to-en?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Binds the api-reshape surface at wt/api-reshape@d4f401d (op_think/read/attend/ learn, verified against engram.cognition-20260814) into the swarm: - reshape_surface.el composes the reshape's proven read/cognition primitives verbatim (write ops omitted — they need the gate-1 write-healthy clone). - primitive_binding.el: bound_think -> op_think over the worker's NODE-ID anchor (ctx.input); attend/learn bound behind SWARM_WRITE_HEALTHY. - cognize blueprint derives the vote verdict from the REAL gradient's n_support (json_get_int) — per-anchor diversity (6/16/87 support) drives a genuine vote. - build.sh now defines HAVE_CURL. CRITICAL FIX: without it every http_* was a '{"error":"not built with HAVE_CURL"}' stub, so prior 'live engram' retrieval was a false positive (matched the ref string, not real content). With HAVE_CURL the swarm genuinely hits /api/think on the :8901 clone. harness_real_cognition.el: 17/17 GREEN with seam=decorated — 8 native-thread workers each a REAL think (768-dim gradient) over its CCR-scoped node-id anchor, @manager reduce+vote convergence, all 3 containment rules incl. live Rule-2 denial, afferent telemetry (8 real think signals), durable work-tracking. Reads only — daemon stays healthy; writes stay gated on the gate-1 clone. Prod :8742 untouched. --- lang/swarm/build.sh | 3 +- lang/swarm/primitive_binding.el | 26 +++++-- lang/swarm/reshape_surface.el | 75 +++++++++++++++++++ lang/swarm/swarm.el | 29 +++++++- lang/swarm/tests/harness_real_cognition.el | 86 ++++++++++++++++++++++ 5 files changed, 209 insertions(+), 10 deletions(-) create mode 100644 lang/swarm/reshape_surface.el create mode 100644 lang/swarm/tests/harness_real_cognition.el diff --git a/lang/swarm/build.sh b/lang/swarm/build.sh index 1605920..bf3315a 100755 --- a/lang/swarm/build.sh +++ b/lang/swarm/build.sh @@ -31,6 +31,7 @@ SWARM_MODULES=" swarm/worktrack.el swarm/containment.el swarm/primitives.el + swarm/reshape_surface.el swarm/primitive_binding.el swarm/primitive_seam.el swarm/ccr.el @@ -49,7 +50,7 @@ if ! "$ELC" "$COMBINED" > "$TMP_C" 2>/tmp/swarm.elc.err; then exit 1 fi -if ! cc -O2 -I "$RT" "$TMP_C" "$RT/el_runtime.c" -lcurl -lpthread -lm -o "$OUT" 2>/tmp/swarm.cc.err; then +if ! cc -O2 -DHAVE_CURL -I "$RT" "$TMP_C" "$RT/el_runtime.c" -lcurl -lpthread -lm -o "$OUT" 2>/tmp/swarm.cc.err; then echo "cc FAILED:" >&2 sed 's/^/ /' /tmp/swarm.cc.err >&2 rm -f "$TMP_C" "$COMBINED" diff --git a/lang/swarm/primitive_binding.el b/lang/swarm/primitive_binding.el index c599672..37b692b 100644 --- a/lang/swarm/primitive_binding.el +++ b/lang/swarm/primitive_binding.el @@ -19,21 +19,31 @@ // "decorated" (SWARM_PRIMITIVE_SEAM=decorated). Until you flip these bodies AND // set that env, the harness runs entirely on the hermetic stub. -// bound_think — decorated `think` over a worker's compiled context. +// bound_think — BOUND to the reshape's proven decorated `think` (op_think), +// real cognition over the engram geometry. The worker's CCR slice carries a +// NODE-ID anchor in ctx.input (free-text anchors return "geometry unavailable"); +// think re-origins at that node's region under the faculty and returns a real +// 768-dim gradient. fn bound_think(ctx: String, instruction: String) -> String { - // FLIP HERE -> `return think(ctx, instruction)` once the decorated primitive lands. - return primitive_think(ctx, instruction) + let anchor: String = json_get_string(ctx, "input") + let faculty: String = json_get_string(ctx, "faculty") + return op_think(anchor, faculty) } -// bound_attend — decorated retrieval over the dharma bus (falls back to the -// HTTP/engram attend today). +// bound_attend — BOUND to the reshape's op_attend (POST /api/attend). Needs the +// gate-1 write-healthy clone; falls back to the read-side attend otherwise. fn bound_attend(query: String, limit: Int) -> String { - // FLIP HERE -> `return attend(query, limit)` once decorated. + if str_eq(env("SWARM_WRITE_HEALTHY"), "1") { + return op_attend(query, "self") + } return primitive_attend(query, limit) } -// bound_learn — decorated write onto the bus (falls back to opt-in engram write). +// bound_learn — BOUND to the reshape's op_learn (correspondence-beat). Needs the +// gate-1 write-healthy clone; falls back to the opt-in journal-only learn. fn bound_learn(corr_id: String, observation: String) -> String { - // FLIP HERE -> `return learn(corr_id, observation)` once decorated. + if str_eq(env("SWARM_WRITE_HEALTHY"), "1") { + return op_learn(observation, "induce") + } return primitive_learn(corr_id, observation) } diff --git a/lang/swarm/reshape_surface.el b/lang/swarm/reshape_surface.el new file mode 100644 index 0000000..72c1979 --- /dev/null +++ b/lang/swarm/reshape_surface.el @@ -0,0 +1,75 @@ +// reshape_surface.el — the api-reshape agent's PROVEN decorated primitives, +// composed into the swarm build to bind real cognition. +// +// PROVENANCE: these fns are the reshape's surface at wt/api-reshape @ d4f401d +// ("reshape: decorator-as-seam — port @route codegen, prove decorate->serve, +// rewrite surface as decorated El"), verified live against +// engram.cognition-20260814. Copied verbatim (read/cognition ops only) so the +// swarm binds the REAL primitives, not a reimplementation. The write ops +// (op_write/op_relate/op_supersede/op_ground) are intentionally NOT composed +// here — they exercise the persist_node write path that needs the gate-1 +// write-healthy clone; the swarm's proven run is read-cognition (think/read). +// +// Ops route to the ENGRAM over ENGRAM_URL — pinned by THIS worktree's .nsbx-env +// to the :8901 swarm clone (never the reshape agent's :8900). Separate clones, +// no collision. + +fn engram_url() -> String { + let u: String = env("ENGRAM_URL") + if str_eq(u, "") { return "http://127.0.0.1:8900" } + return u +} +fn engram_key() -> String { + let k: String = env("ENGRAM_API_KEY") + if str_eq(k, "") { return "sbx-dev-api-reshape" } + return k +} +fn SELF_KEY() -> String { return "kn-efeb4a5b-5aff-4759-8a97-7233099be6ee" } +fn VALUES_KEY() -> String { return "kn-5b606390-a52d-4ca2-8e0e-eba141d13440" } + +// self/values name -> keystone id; anything else passes through unchanged. +fn resolve_named(v: String) -> String { + if str_eq(v, "self") { return SELF_KEY() } + if str_eq(v, "neuron") { return SELF_KEY() } + if str_eq(v, "values") { return VALUES_KEY() } + if str_eq(v, "values_hub") { return VALUES_KEY() } + return v +} + +// read — THE VANTAGE-READ. Re-origin at a point + aperture -> a BOUNDED slice. +fn op_read(vantage: String, typ: String, k: Int) -> String { + let vid: String = resolve_named(vantage) + if str_eq(typ, "edges") { + return http_get(engram_url() + "/api/neighbors/" + vid) + } + if str_starts_with(vid, "kn-") { + return http_get(engram_url() + "/api/neighbors/" + vid) + } + return http_get(engram_url() + "/api/search?q=" + url_encode(vid) + "&limit=" + int_to_str(k)) +} + +// think — THE ONE OPERATION. anchor (node ids) steered by faculty -> gradient. +fn op_think(seeds: String, faculty: String) -> String { + let s: String = resolve_named(seeds) + let f: String = if str_eq(faculty, "") { "reason" } else { faculty } + return http_get(engram_url() + "/api/think?seeds=" + url_encode(s) + "&faculty=" + f) +} + +// attend — aim attention at a region. (POST — needs a write-healthy clone.) +fn op_attend(node: String, observer: String) -> String { + let n: String = resolve_named(node) + let o: String = if str_eq(observer, "") { SELF_KEY() } else { resolve_named(observer) } + let body: String = "{\"_auth\":\"" + engram_key() + "\",\"node\":\"" + n + + "\",\"observer\":\"" + o + "\",\"salience\":\"0.6\"}" + return http_post_json(engram_url() + "/api/attend", body) +} + +// learn — the reflexive correspondence-beat: calibrate the steering-prior. +// (POST — needs a write-healthy clone.) +fn op_learn(seeds: String, faculty: String) -> String { + let s: String = resolve_named(seeds) + let f: String = if str_eq(faculty, "") { "induce" } else { faculty } + let body: String = "{\"_auth\":\"" + engram_key() + "\",\"seeds\":\"" + s + + "\",\"faculty\":\"" + f + "\",\"keystone\":\"false\"}" + return http_post_json(engram_url() + "/api/correspondence-beat", body) +} diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el index 28ef7d1..a676ede 100644 --- a/lang/swarm/swarm.el +++ b/lang/swarm/swarm.el @@ -88,11 +88,38 @@ fn swarm_run_blueprint(ctx: String) -> String { 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(knowledge, instruction) + 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") diff --git a/lang/swarm/tests/harness_real_cognition.el b/lang/swarm/tests/harness_real_cognition.el new file mode 100644 index 0000000..477a1d7 --- /dev/null +++ b/lang/swarm/tests/harness_real_cognition.el @@ -0,0 +1,86 @@ +// harness_real_cognition.el — the LOCAL SWARM running REAL cognition. +// +// Run with: SWARM_PRIMITIVE_SEAM=decorated + the sandbox env sourced +// (ENGRAM_URL=:8901). Each worker's `think` is BOUND to the reshape's proven +// op_think (GET /api/think) over its NODE-ID anchor — real 768-dim gradients from +// the live (isolated) geometry, not the stub. The @manager fans out N native-El +// worker threads at real concurrency, converges (reduce + vote) over the real +// cognition, enforces all three containment rules, observes afferent telemetry, +// and work-tracks durably. +// +// Anchors are real self-neighbourhood node ids on the :8901 clone (free-text +// anchors return "geometry unavailable", so these must be node ids). + +fn ok(label: String, cond: Bool, fails: Int) -> Int { + if cond { print(" ok " + label); return fails } + print(" FAIL " + label); return fails + 1 +} + +fn main() -> Int { + let fails = 0 + print("== REAL-COGNITION LOCAL SWARM (seam=" + seam_mode() + ", engram=" + env("ENGRAM_URL") + ") ==") + + // ── 0) direct proof the bound primitive returns REAL cognition ── + let g: String = op_think("self", "plan") + let dim: Int = json_get_int(g, "dim") + let nsup: Int = json_get_int(g, "n_support") + let fails = ok("bound op_think returns a real 768-dim gradient", dim == 768, fails) + let fails = ok("real gradient has support (n_support>0)", nsup > 0, fails) + let gfree: String = op_think("this-is-free-text-not-a-node", "reason") + let fails = ok("free-text anchor correctly refused (geometry unavailable)", str_contains(gfree, "geometry unavailable"), fails) + + // ── the input set: 8 real NODE-ID anchors from self's neighbourhood ── + let anchors: String = "[\"a1000001-0000-0000-0000-000000000001\",\"5f011441-fa43-4fe7-a9c0-c78a584ef11d\",\"kn-5adecd7e-d6db-4576-87fe-6ef8a935cea6\",\"76d7fd0b-0672-4511-a2f5-a095cf9c60ae\",\"7027e302-593f-441d-8fd6-9c400c163108\",\"2a730b18-6566-46ee-a21e-4f4dd0380908\",\"46b0e4dd-2c19-48d2-bcbc-19f61d6c79ae\",\"9162cde8-8739-4f00-bfc9-2850ed612e50\"]" + let refs: String = "[\"self\"]" + + // ── A) fan-out real cognition at concurrency, converge with REDUCE ── + let cfg_r: String = "{\"concurrency\":\"4\",\"strategy\":\"reduce\",\"min_success_ratio\":\"1.0\"}" + let rr: String = swarm_run("cognize", refs, anchors, cfg_r) + let fails = ok("swarm completed: 8 workers each a real think, concurrency=4", str_eq(json_get_string(rr, "status"), "completed"), fails) + let corr: String = json_get_string(rr, "corr_id") + let merged_r: String = json_get_raw(rr, "merged") + let fails = ok("reduce converged all 8 real-cognition outputs", str_to_int(json_get_string(merged_r, "count")) == 8, fails) + let acc: String = json_get_string(merged_r, "accumulated") + let fails = ok("converged output carries real gradient support (n_support)", str_contains(acc, "n_support"), fails) + + // ── B) afferent telemetry: 8 real think-signals, decorated seam ── + let telem: String = json_get_raw(rr, "telemetry") + let aff: Int = str_to_int(json_get_string(telem, "afferent_think")) + let fails = ok("afferent counters = 8 real think invocations", aff == 8, fails) + let fails = ok("telemetry records seam_mode=decorated", str_eq(json_get_string(telem, "seam_mode"), "decorated"), fails) + let fails = ok("telemetry durably journalled", worktrack_count_kind(corr, "swarm.telemetry") == 1, fails) + + // ── C) converge with VOTE over real cognition ── + let cfg_v: String = "{\"concurrency\":\"8\",\"strategy\":\"vote\",\"min_success_ratio\":\"1.0\"}" + let rv: String = swarm_run("cognize", refs, anchors, cfg_v) + let winner: String = json_get_string(json_get_raw(rv, "merged"), "winner") + let fails = ok("vote converged over real cognition (winner=" + winner + ")", !str_eq(winner, ""), fails) + + // ── D) all three containment rules still enforced ── + let wt: String = containment_worker_token(corr, corr + "/worker-2") + let fails = ok("Rule 2: worker may not open a swarm", !str_eq(containment_check_open(wt), ""), fails) + let fails = ok("Rule 1: worker may not join another swarm", !str_eq(containment_check_join(wt, "s2"), ""), fails) + let fails = ok("Rule 3: worker->worker lateral edge rejected", !str_eq(containment_check_lateral(wt, "sib"), ""), fails) + let wcfg: String = json_set(cfg_r, "caller_token", wt) + let denied: String = swarm_run("cognize", refs, anchors, wcfg) + let fails = ok("Rule 2 enforced LIVE: worker-caller swarm denied", str_eq(json_get_string(denied, "status"), "denied"), fails) + + // ── E) CCR scoping + non-leak over node-id anchors ── + let ctx: String = ccr_compile("cognize", refs, "a1000001-0000-0000-0000-000000000001", corr, corr + "/worker-0", wt) + let fails = ok("CCR context bounded within budget", ccr_within_budget(ctx), fails) + let leaks: Bool = str_contains(ctx, "9162cde8") + let fails = ok("CCR context does NOT leak sibling anchors", !leaks, fails) + + // ── F) durable work-tracking ── + let started: Int = worktrack_count_kind(corr, "worker.started") + let completed: Int = worktrack_count_kind(corr, "worker.completed") + let fails = ok("work-tracking: 8 started + 8 completed", (started == 8) && (completed == 8), fails) + + print("") + if fails == 0 { + print("REAL-COGNITION SWARM GREEN — Neuron thinking in parallel over its own geometry.") + return 0 + } + print("REAL-COGNITION SWARM FAIL (" + int_to_str(fails) + ")") + return 1 +} From b53b5b4e8a426b3e3da3790fce53a4c5014bd1e7 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 21:19:11 -0500 Subject: [PATCH 11/13] swarm: document real-cognition binding + HAVE_CURL build note in README --- lang/swarm/README.md | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/lang/swarm/README.md b/lang/swarm/README.md index 3936286..eb8586a 100644 --- a/lang/swarm/README.md +++ b/lang/swarm/README.md @@ -113,6 +113,27 @@ already routes every worker primitive call through this one switch, and the same harness runs the bound path. Today `SWARM_PRIMITIVE_SEAM=decorated` still runs green because the binding falls back to the stub — proving the flip path executes. +## Real cognition — the seam is BOUND + +`primitive_binding.el` is bound to the api-reshape agent's proven primitives +(`wt/api-reshape@d4f401d`): `bound_think -> op_think` (GET `/api/think`), real +768-dim gradients over the engram geometry. `reshape_surface.el` composes those +read/cognition primitives verbatim (`op_think/read/attend/learn`). + +`tests/harness_real_cognition.el` runs the **local swarm on real cognition**, +17/17 green with `SWARM_PRIMITIVE_SEAM=decorated` against the `:8901` clone: 8 +native-thread workers, each a real `think` over its CCR-scoped **node-id anchor** +(free-text anchors return "geometry unavailable"), `@manager` reduce+vote, all +three containment rules, afferent telemetry, durable tracking. Per-anchor support +counts (e.g. 6 / 16 / 87) drive a genuine, cognition-derived vote. + +> **Build note (load-bearing):** the swarm build **must** define `HAVE_CURL` +> (`build.sh` does). Without it every `http_*` builtin is a +> `{"error":"not built with HAVE_CURL"}` stub — real HTTP silently disappears. + +Writes (`attend`/`learn`, `POST`) are gated behind `SWARM_WRITE_HEALTHY=1` and the +api-reshape agent's gate-1 write-healthy clone; the proven run is read-cognition. + ## Built vs stubbed (honest) **Real, tested:** From e5c80359a86768ed3572b83c1093697973bdee3c Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 21:46:03 -0500 Subject: [PATCH 12/13] =?UTF-8?q?swarm:=20Rule=204=20=E2=80=94=20engram-wr?= =?UTF-8?q?ite=20is=20@manager-ONLY,=20enforced=20by=20capability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- lang/swarm/containment.el | 64 +++++++++++++++++++++- lang/swarm/reshape_surface.el | 28 ++++++++++ lang/swarm/swarm.el | 36 +++++++++++- lang/swarm/tests/harness_real_cognition.el | 33 +++++++++++ 4 files changed, 157 insertions(+), 4 deletions(-) diff --git a/lang/swarm/containment.el b/lang/swarm/containment.el index 75bd3ee..c57e2c8 100644 --- a/lang/swarm/containment.el +++ b/lang/swarm/containment.el @@ -21,8 +21,19 @@ // ── Token minting ──────────────────────────────────────────────────────────── -// containment_coordinator_token — the token a coordinator holds. Depth 0. -// Only a coordinator token may open a swarm. +// CAPABILITIES. A scope token carries a `caps` set — the authority it holds. +// This is an AUTHORITY gate, not a health gate: capability is decided at mint +// time and cannot be acquired at runtime. Engram-WRITE (op_write/op_relate/ +// op_supersede -> POST /api/nodes, /api/edges, DELETE) and dharma_emit are +// @manager-ONLY capabilities — exactly the VBD rule that only the orchestrator +// mutates global state. The orchestrator's token carries them; a worker's token +// NEVER does. A worker is therefore STRUCTURALLY UNABLE to mutate global engram +// state, regardless of engram health. +fn cap_orchestrator() -> String { return "engram:read,engram:write,dharma:emit,state:write" } +fn cap_worker() -> String { return "engram:read" } + +// containment_coordinator_token — the token the orchestrator (@manager) holds. +// Depth 0. Carries the engram-WRITE + dharma-emit capabilities (@manager-only). fn containment_coordinator_token(corr_id: String) -> String { let kv: [String] = el_list_empty() let kv = el_list_append(kv, "kind") @@ -33,11 +44,15 @@ fn containment_coordinator_token(corr_id: String) -> String { let kv = el_list_append(kv, "") let kv = el_list_append(kv, "depth") let kv = el_list_append(kv, "0") + let kv = el_list_append(kv, "caps") + let kv = el_list_append(kv, cap_orchestrator()) return json_build_object(kv) } // containment_worker_token — the token stamped into a worker's envelope. Depth 1. -// A worker token is a closed boundary: holding it forbids opening/joining swarms. +// A closed boundary: forbids opening/joining swarms AND carries ONLY the +// engram:READ capability — no engram:write, no dharma:emit. Read-only against the +// full engram; may write only its own local geometry (its returned result). fn containment_worker_token(corr_id: String, worker_id: String) -> String { let kv: [String] = el_list_empty() let kv = el_list_append(kv, "kind") @@ -48,9 +63,16 @@ fn containment_worker_token(corr_id: String, worker_id: String) -> String { let kv = el_list_append(kv, worker_id) let kv = el_list_append(kv, "depth") let kv = el_list_append(kv, "1") + let kv = el_list_append(kv, "caps") + let kv = el_list_append(kv, cap_worker()) return json_build_object(kv) } +// containment_has_cap — does this token carry capability `cap`? +fn containment_has_cap(token: String, cap: String) -> Bool { + return str_contains(json_get_string(token, "caps"), cap) +} + // ── Rule checks (return "" on allow, or a rejection reason string) ─────────── // containment_check_open — may the holder of `token` OPEN a new swarm? @@ -97,6 +119,28 @@ fn containment_check_lateral(from_token: String, to_worker_id: String) -> String return "" } +// containment_check_engram_write — RULE 4: only a token carrying the +// engram:write capability (the orchestrator's) may mutate global engram state. +// A worker token (engram:read only) is REJECTED — the authority gate. Reuses the +// exact scope-token mechanism as Rule 2's open-denial. Returns "" on allow, or a +// rejection reason. This is an AUTHORITY gate: it does not consult engram health. +fn containment_check_engram_write(token: String, op: String) -> String { + if containment_has_cap(token, "engram:write") { + return "" + } + return "CONTAINMENT rule 4: engram-write is @manager-only — a worker is read-only against the engram and may not mutate global state (op=" + op + " kind=" + json_get_string(token, "kind") + " worker=" + json_get_string(token, "worker") + " caps=" + json_get_string(token, "caps") + ")" +} + +// containment_check_dharma_emit — the same @manager-only rule for dharma_emit, +// grounding Rule 4 in VBD: global-state mutations (engram-write, dharma-emit) are +// orchestrator-only, checked by the one capability mechanism. +fn containment_check_dharma_emit(token: String) -> String { + if containment_has_cap(token, "dharma:emit") { + return "" + } + return "CONTAINMENT rule 4: dharma_emit is @manager-only (kind=" + json_get_string(token, "kind") + ")" +} + // ── Enforcement helpers ────────────────────────────────────────────────────── // containment_allows_open — Bool convenience over containment_check_open. @@ -121,3 +165,17 @@ fn containment_guard_open(token: String, corr_id: String) -> String { worktrack_append("containment.violation", corr_id, "open", p) return reason } + +// containment_guard_engram_write — assert a token may mutate global engram state +// (Rule 4). Returns "" if allowed; otherwise journals a containment.violation and +// returns the reason. The write path MUST abort on a non-empty return. +fn containment_guard_engram_write(token: String, corr_id: String, op: String) -> String { + let reason: String = containment_check_engram_write(token, op) + if str_eq(reason, "") { + return "" + } + let p0: String = json_set_str("{}", "reason", reason) + let p1: String = json_set_str(p0, "op", op) + worktrack_append("containment.violation", corr_id, "engram-write", p1) + return reason +} diff --git a/lang/swarm/reshape_surface.el b/lang/swarm/reshape_surface.el index 72c1979..7404010 100644 --- a/lang/swarm/reshape_surface.el +++ b/lang/swarm/reshape_surface.el @@ -64,6 +64,34 @@ fn op_attend(node: String, observer: String) -> String { return http_post_json(engram_url() + "/api/attend", body) } +fn identity_typed(t: String) -> Bool { + if str_eq(t, "self") { return true } + if str_eq(t, "values") { return true } + return false +} +fn type_to_node_type(t: String) -> String { + if str_eq(t, "knowledge") { return "Knowledge" } + if str_eq(t, "artifact") { return "Artifact" } + if str_eq(t, "backlog") { return "WorkItem" } + if str_eq(t, "process") { return "Process" } + if str_eq(t, "state") { return "InternalStateEvent" } + return "Memory" +} + +// write — add a node (POST /api/nodes). Identity types refused. This is a +// global-engram MUTATION — @manager-only (Rule 4); never called on a worker path. +// (Reshape's op_write, with json_escape -> the available json_escape_string.) +fn op_write(content: String, typ: String, importance: Float) -> String { + if str_eq(content, "") { return "{\"error\":\"write: content required\"}" } + if identity_typed(typ) { + return "{\"error\":\"write type=" + typ + " is write-protected -> intentional-cultivation\"}" + } + let body: String = "{\"_auth\":\"" + engram_key() + "\",\"content\":\"" + json_escape_string(content) + + "\",\"node_type\":\"" + type_to_node_type(typ) + "\",\"tier\":\"Working\",\"importance\":" + + float_to_str(importance) + "}" + return http_post_json(engram_url() + "/api/nodes", body) +} + // learn — the reflexive correspondence-beat: calibrate the steering-prior. // (POST — needs a write-healthy clone.) fn op_learn(seeds: String, faculty: String) -> String { diff --git a/lang/swarm/swarm.el b/lang/swarm/swarm.el index a676ede..3f74297 100644 --- a/lang/swarm/swarm.el +++ b/lang/swarm/swarm.el @@ -300,6 +300,28 @@ fn swarm_converge(strategy: String, results: [String]) -> String { 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 @@ -426,6 +448,17 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con 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") @@ -435,7 +468,8 @@ fn swarm_run(blueprint: String, knowledge_refs: String, inputs_json: String, con let out1: String = json_build_object(ok2) let out2: String = json_set(out1, "report", rep2) let out3: String = json_set(out2, "merged", merged) - return json_set(out3, "telemetry", telemetry) + 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() diff --git a/lang/swarm/tests/harness_real_cognition.el b/lang/swarm/tests/harness_real_cognition.el index 477a1d7..ca98691 100644 --- a/lang/swarm/tests/harness_real_cognition.el +++ b/lang/swarm/tests/harness_real_cognition.el @@ -76,6 +76,39 @@ fn main() -> Int { let completed: Int = worktrack_count_kind(corr, "worker.completed") let fails = ok("work-tracking: 8 started + 8 completed", (started == 8) && (completed == 8), fails) + // ── G) RULE 4 — engram-write is @manager-ONLY (authority gate) ── + // A worker token (engram:read only) is STRUCTURALLY denied any engram write. + let worker_tok: String = containment_worker_token(corr, corr + "/worker-1") + let orch_tok: String = containment_coordinator_token(corr) + let fails = ok("worker token carries engram:read", containment_has_cap(worker_tok, "engram:read"), fails) + let fails = ok("worker token does NOT carry engram:write", !containment_has_cap(worker_tok, "engram:write"), fails) + let fails = ok("orchestrator token carries engram:write", containment_has_cap(orch_tok, "engram:write"), fails) + // a worker attempting an engram write is DENIED BY CAPABILITY (no HTTP issued) + let wdeny: String = swarm_engram_write(worker_tok, corr, "worker tries to mutate global state", "memory", 0.5) + let denied_reason: String = json_get_string(wdeny, "denied") + let fails = ok("worker engram-write DENIED by capability (Rule 4)", str_contains(denied_reason, "rule 4"), fails) + let fails = ok("denied worker write performed NO engram mutation (no node id)", str_eq(json_get_string(wdeny, "id"), ""), fails) + let fails = ok("Rule-4 violation journalled", worktrack_count_kind(corr, "containment.violation") >= 1, fails) + // the orchestrator passes the capability gate (sole authorized writer) + let odeny: String = containment_check_engram_write(orch_tok, "engram.write") + let fails = ok("orchestrator PASSES the engram-write capability gate (sole writer)", str_eq(odeny, ""), fails) + + // ── H) curated merge = the only write path (orchestrator commits) ── + // The AUTHORITY gate above is already proven (worker denied, orchestrator + // authorized) WITHOUT issuing a write. The actual persisting commit exercises + // the engram write path, which needs the gate-1 write-healthy clone — so it + // runs only under SWARM_WRITE_HEALTHY=1 (else it would hit the known daemon + // write-crash). Authority != health: the gate holds either way. + if str_eq(env("SWARM_WRITE_HEALTHY"), "1") { + let cfg_commit: String = "{\"concurrency\":\"4\",\"strategy\":\"reduce\",\"min_success_ratio\":\"1.0\",\"commit\":\"1\"}" + let rc: String = swarm_run("cognize", refs, anchors, cfg_commit) + let committed: String = json_get_string(rc, "committed_node") + let fails2: Int = ok("orchestrator (sole writer) committed the merge to the engram", !str_eq(committed, ""), fails) + let fails = fails2 + } else { + print(" note curated-merge commit deferred to the gate-1 write-healthy clone (set SWARM_WRITE_HEALTHY=1); authority gate already proven above") + } + print("") if fails == 0 { print("REAL-COGNITION SWARM GREEN — Neuron thinking in parallel over its own geometry.") From ff37835ae58225b78f94c6a42d8b2df52510c2d2 Mon Sep 17 00:00:00 2001 From: bigmerge Date: Fri, 14 Aug 2026 21:46:23 -0500 Subject: [PATCH 13/13] swarm: document the single-writer invariant (Rule 4) in README --- lang/swarm/README.md | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/lang/swarm/README.md b/lang/swarm/README.md index eb8586a..925550a 100644 --- a/lang/swarm/README.md +++ b/lang/swarm/README.md @@ -41,11 +41,37 @@ fn is directly threadable — the worker entry is exactly such a fn. | File | Framework grounding | What it does | |------|--------------------|--------------| | `worktrack.el` | Swarm §6 (correlation IDs, audit) | Durable, single-writer **JSONL journal** keyed by correlation ID; reconstructable status report; opt-in engram mirror (`SWARM_MIRROR=1`). | -| `containment.el` | Swarm §3 (the three rules) | Scope tokens; **Rule 1** (no join), **Rule 2** (no open), **Rule 3** (no lateral edge) enforced as checks. | +| `containment.el` | Swarm §3 + the single-writer invariant | Scope tokens w/ capabilities; **Rule 1** (no join), **Rule 2** (no open), **Rule 3** (no lateral edge), **Rule 4** (engram-write is @manager-only, by capability) enforced as checks. | | `ccr.el` | CCR §5 + Swarm §9.3 | Per-worker **Compiled Context Routing**: retrieve → scope → compact into a **bounded, minimal** package. The compiled-context boundary *is* the security boundary. | | `primitives.el` | CCR §2 (Five Primitives) | `attend / think / intend / act / learn` seam the swarm composes over. Engram-backed; explicit binding point for the API-surface reshape. | | `swarm.el` | Swarm §2, §4, §5 | The coordinator: fan-out/converge on native threads, bounded concurrency, four convergence strategies, integer failure threshold, full tracking. | +## Invariant: only the orchestrator mutates global engram state + +**Only the orchestrator (@manager) writes to the engram / mutates global state. +Workers are read-only against the full engram and may write only their own local +geometry (their returned result + the journal). A worker is STRUCTURALLY UNABLE +to mutate global engram state.** + +This is **Rule 4** — an **authority gate, not a health gate**. Scope tokens carry +a capability set: the orchestrator's token holds `engram:write` + `dharma:emit` +(@manager-only, the VBD rule that only the manager mutates global state); a +worker's token holds **only** `engram:read`. Every engram mutation +(`op_write`/`op_relate`/`op_supersede` → `POST /api/nodes`, `/api/edges`, +`DELETE`) flows through `swarm_engram_write`, which checks the caller's capability +via the **same scope-token mechanism as the live Rule-2 denial** and rejects any +worker **before any HTTP is issued**. Capability is fixed at mint time and cannot +be acquired at runtime — so the guarantee holds regardless of engram health +(distinct from the `SWARM_WRITE_HEALTHY` *health* gate). + +The **curated merge is the only write path**: workers return geometry; the +orchestrator, and only the orchestrator, commits the approved/verified geometry +back (`commit=1`). Workers keep full-engram **read** access (`op_think`/`op_read`). + +Proven in `harness_real_cognition.el` (§G): a worker `swarm_engram_write` is +DENIED by capability with no node created and the violation journalled; the +orchestrator passes the gate as the sole authorized writer. + ## Containment → distribution The three containment rules make workers **location-independent** (Swarm §9): a