Merge pull request 'swarm: land wt/swarm-ccr onto dev (clean re-merge)' (#122) from merge-swarm-ccr-v2 into dev

lang/runtime/el_runtime.c

@@ -17751,3 +17751,412 @@ el_val_t emit_event(el_val_t name_v, el_val_t duration_ms_v) {
     return trace_span_end(h);
 }
 
+
+/* ── DHARMA runtime additions ────────────────────────────────────────────────
+ *
+ * Functions required by the dharma registry service. Added here so the
+ * released el_runtime.c includes them without requiring dharma to bundle
+ * its own stubs.
+ *
+ * Functions added:
+ *   list_len          — alias for el_list_len (used in handlers.el)
+ *   list_get          — alias for el_list_get  (used in handlers.el)
+ *   json_array_push   — append a pre-encoded JSON element to a JSON array string
+ *   now_millis        — milliseconds since Unix epoch (alias for time_now)
+ *   unix_timestamp_ms — same as now_millis (alias)
+ *   time_now_ms       — same as now_millis (alias)
+ *   log_info          — stderr structured log at INFO level
+ *   log_warn          — stderr structured log at WARN level
+ *   config            — reads a config value from the environment
+ *   http_patch        — HTTP PATCH with JSON Content-Type
+ *   http_post_engram  — HTTP POST with optional X-API-Key header
+ *   http_get_engram   — HTTP GET  with optional X-API-Key header
+ *   str_to_bytes      — encode a string as a JSON array of byte values
+ *   bytes_to_str      — decode a JSON array of byte values back to a string
+ *   hash_sha256       — SHA-256 hex digest of a string
+ */
+
+/* list_len — return the number of elements in a list. */
+el_val_t list_len(el_val_t list) {
+    return el_list_len(list);
+}
+
+/* list_get — return the element at index i in a list. */
+el_val_t list_get(el_val_t list, el_val_t index) {
+    return el_list_get(list, index);
+}
+
+/* json_array_push — append element (a pre-encoded JSON fragment, e.g. "\"foo\""
+ * or "42") to the JSON array string arr. Returns a new JSON array string.
+ * Example: json_array_push("[]", "\"alice\"") -> "[\"alice\"]"
+ *          json_array_push("[\"alice\"]", "\"bob\"") -> "[\"alice\",\"bob\"]" */
+el_val_t json_array_push(el_val_t arr_v, el_val_t elem_v) {
+    const char* arr  = EL_CSTR(arr_v);
+    const char* elem = EL_CSTR(elem_v);
+    if (!arr  || !*arr)  arr  = "[]";
+    if (!elem || !*elem) elem = "null";
+
+    /* Trim whitespace, find the closing ']'. */
+    const char* p = arr;
+    while (*p == ' ' || *p == '\t' || *p == '\n' || *p == '\r') p++;
+    if (*p != '[') {
+        /* Not an array — return a single-element array. */
+        size_t n = strlen(elem) + 4;
+        char* out = el_strbuf(n);
+        snprintf(out, n, "[%s]", elem);
+        return el_wrap_str(out);
+    }
+    size_t arr_len  = strlen(arr);
+    size_t elem_len = strlen(elem);
+
+    /* Walk from the end to find the matching ']'. */
+    const char* end = arr + arr_len - 1;
+    while (end > p && (*end == ' ' || *end == '\t' || *end == '\n' || *end == '\r')) end--;
+    if (*end != ']') {
+        /* Malformed — wrap elem in a new array. */
+        size_t n = elem_len + 4;
+        char* out = el_strbuf(n);
+        snprintf(out, n, "[%s]", elem);
+        return el_wrap_str(out);
+    }
+
+    /* Content between '[' and ']'. */
+    const char* inner_start = p + 1;
+    const char* inner_end   = end;       /* points AT ']' */
+    /* Check if the array is empty (only whitespace between brackets). */
+    const char* q = inner_start;
+    while (q < inner_end && (*q == ' ' || *q == '\t' || *q == '\n' || *q == '\r')) q++;
+    int empty = (q == inner_end);
+
+    /* Build: prefix + (comma if non-empty) + elem + "]" */
+    size_t prefix_len = (size_t)(inner_end - arr); /* up to but not including ']' */
+    size_t sep_len    = empty ? 0 : 1;              /* "," if non-empty */
+    size_t out_len    = prefix_len + sep_len + elem_len + 2; /* +"]" + NUL */
+    char* out = el_strbuf(out_len);
+    memcpy(out, arr, prefix_len);
+    if (!empty) out[prefix_len] = ',';
+    memcpy(out + prefix_len + sep_len, elem, elem_len);
+    out[prefix_len + sep_len + elem_len] = ']';
+    out[prefix_len + sep_len + elem_len + 1] = '\0';
+    return el_wrap_str(out);
+}
+
+/* now_millis — milliseconds since Unix epoch. */
+el_val_t now_millis(void) {
+    return time_now();
+}
+
+/* unix_timestamp_ms — same as now_millis. */
+el_val_t unix_timestamp_ms(void) {
+    return time_now();
+}
+
+/* time_now_ms — same as now_millis. */
+el_val_t time_now_ms(void) {
+    return time_now();
+}
+
+/* log_info — write a structured [INFO] line to stderr. */
+void log_info(el_val_t msg_v) {
+    const char* msg = EL_CSTR(msg_v);
+    fprintf(stderr, "[INFO] %s\n", msg ? msg : "");
+}
+
+/* log_warn — write a structured [WARN] line to stderr. */
+void log_warn(el_val_t msg_v) {
+    const char* msg = EL_CSTR(msg_v);
+    fprintf(stderr, "[WARN] %s\n", msg ? msg : "");
+}
+
+/* config — read a configuration value from the environment.
+ * Returns "" if the variable is not set (same as __env_get). */
+el_val_t config(el_val_t key_v) {
+    const char* key = EL_CSTR(key_v);
+    if (!key || !*key) return EL_STR("");
+    const char* val = getenv(key);
+    if (!val) return EL_STR("");
+    return el_wrap_str(el_strdup(val));
+}
+
+#if !defined(_WIN32) || defined(HAVE_CURL)
+/* http_patch — HTTP PATCH request with Content-Type: application/json.
+ * Returns the response body (same error convention as http_post_json). */
+el_val_t http_patch(el_val_t url_v, el_val_t body_v) {
+    const char* url  = EL_CSTR(url_v);
+    const char* body = EL_CSTR(body_v);
+    if (!url || !*url) return http_error_json("empty url");
+    CURL* c = curl_easy_init();
+    if (!c) return http_error_json("curl_easy_init failed");
+    HttpBuf rb; httpbuf_init(&rb);
+    char errbuf[CURL_ERROR_SIZE]; errbuf[0] = '\0';
+    struct curl_slist* h = NULL;
+    h = curl_slist_append(h, "Content-Type: application/json");
+    curl_easy_setopt(c, CURLOPT_URL, url);
+    curl_easy_setopt(c, CURLOPT_CUSTOMREQUEST, "PATCH");
+    curl_easy_setopt(c, CURLOPT_POSTFIELDS, body ? body : "");
+    curl_easy_setopt(c, CURLOPT_POSTFIELDSIZE, (long)(body ? strlen(body) : 0));
+    curl_easy_setopt(c, CURLOPT_HTTPHEADER, h);
+    curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, http_write_cb);
+    curl_easy_setopt(c, CURLOPT_WRITEDATA, &rb);
+    curl_easy_setopt(c, CURLOPT_FOLLOWLOCATION, 1L);
+    curl_easy_setopt(c, CURLOPT_TIMEOUT_MS, el_http_timeout_ms());
+    curl_easy_setopt(c, CURLOPT_NOSIGNAL, 1L);
+    curl_easy_setopt(c, CURLOPT_ERRORBUFFER, errbuf);
+    curl_easy_setopt(c, CURLOPT_USERAGENT, "el-runtime/1.0");
+    CURLcode rc = curl_easy_perform(c);
+    curl_slist_free_all(h);
+    curl_easy_cleanup(c);
+    if (rc != CURLE_OK) {
+        free(rb.data);
+        const char* m = errbuf[0] ? errbuf : curl_easy_strerror(rc);
+        return http_error_json(m);
+    }
+    return el_wrap_str(rb.data);
+}
+
+/* http_post_engram — HTTP POST with optional X-API-Key header.
+ * If key is "" no authentication header is sent. */
+el_val_t http_post_engram(el_val_t url_v, el_val_t key_v, el_val_t body_v) {
+    const char* url  = EL_CSTR(url_v);
+    const char* key  = EL_CSTR(key_v);
+    const char* body = EL_CSTR(body_v);
+    if (!url || !*url) return http_error_json("empty url");
+    CURL* c = curl_easy_init();
+    if (!c) return http_error_json("curl_easy_init failed");
+    HttpBuf rb; httpbuf_init(&rb);
+    char errbuf[CURL_ERROR_SIZE]; errbuf[0] = '\0';
+    struct curl_slist* h = NULL;
+    h = curl_slist_append(h, "Content-Type: application/json");
+    if (key && *key) {
+        size_t n = strlen(key) + 32;
+        char* hdr = malloc(n);
+        snprintf(hdr, n, "X-API-Key: %s", key);
+        h = curl_slist_append(h, hdr);
+        free(hdr);
+    }
+    curl_easy_setopt(c, CURLOPT_URL, url);
+    curl_easy_setopt(c, CURLOPT_POST, 1L);
+    curl_easy_setopt(c, CURLOPT_POSTFIELDS, body ? body : "");
+    curl_easy_setopt(c, CURLOPT_POSTFIELDSIZE, (long)(body ? strlen(body) : 0));
+    curl_easy_setopt(c, CURLOPT_HTTPHEADER, h);
+    curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, http_write_cb);
+    curl_easy_setopt(c, CURLOPT_WRITEDATA, &rb);
+    curl_easy_setopt(c, CURLOPT_FOLLOWLOCATION, 1L);
+    curl_easy_setopt(c, CURLOPT_TIMEOUT_MS, el_http_timeout_ms());
+    curl_easy_setopt(c, CURLOPT_NOSIGNAL, 1L);
+    curl_easy_setopt(c, CURLOPT_ERRORBUFFER, errbuf);
+    curl_easy_setopt(c, CURLOPT_USERAGENT, "el-runtime/1.0");
+    CURLcode rc = curl_easy_perform(c);
+    curl_slist_free_all(h);
+    curl_easy_cleanup(c);
+    if (rc != CURLE_OK) {
+        free(rb.data);
+        const char* m = errbuf[0] ? errbuf : curl_easy_strerror(rc);
+        return http_error_json(m);
+    }
+    return el_wrap_str(rb.data);
+}
+
+/* http_get_engram — HTTP GET with optional X-API-Key header. */
+el_val_t http_get_engram(el_val_t url_v, el_val_t key_v) {
+    const char* url = EL_CSTR(url_v);
+    const char* key = EL_CSTR(key_v);
+    if (!url || !*url) return http_error_json("empty url");
+    CURL* c = curl_easy_init();
+    if (!c) return http_error_json("curl_easy_init failed");
+    HttpBuf rb; httpbuf_init(&rb);
+    char errbuf[CURL_ERROR_SIZE]; errbuf[0] = '\0';
+    struct curl_slist* h = NULL;
+    if (key && *key) {
+        size_t n = strlen(key) + 32;
+        char* hdr = malloc(n);
+        snprintf(hdr, n, "X-API-Key: %s", key);
+        h = curl_slist_append(h, hdr);
+        free(hdr);
+    }
+    curl_easy_setopt(c, CURLOPT_URL, url);
+    curl_easy_setopt(c, CURLOPT_HTTPGET, 1L);
+    if (h) curl_easy_setopt(c, CURLOPT_HTTPHEADER, h);
+    curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, http_write_cb);
+    curl_easy_setopt(c, CURLOPT_WRITEDATA, &rb);
+    curl_easy_setopt(c, CURLOPT_FOLLOWLOCATION, 1L);
+    curl_easy_setopt(c, CURLOPT_TIMEOUT_MS, el_http_timeout_ms());
+    curl_easy_setopt(c, CURLOPT_NOSIGNAL, 1L);
+    curl_easy_setopt(c, CURLOPT_ERRORBUFFER, errbuf);
+    curl_easy_setopt(c, CURLOPT_USERAGENT, "el-runtime/1.0");
+    CURLcode rc = curl_easy_perform(c);
+    if (h) curl_slist_free_all(h);
+    curl_easy_cleanup(c);
+    if (rc != CURLE_OK) {
+        free(rb.data);
+        const char* m = errbuf[0] ? errbuf : curl_easy_strerror(rc);
+        return http_error_json(m);
+    }
+    return el_wrap_str(rb.data);
+}
+#endif /* HAVE_CURL */
+
+/* str_to_bytes — encode a string as a JSON array of unsigned byte values.
+ * "hello" -> "[104,101,108,108,111]"
+ * Used by db.el to store binary content in Engram JSON nodes. */
+el_val_t str_to_bytes(el_val_t sv) {
+    const char* s = EL_CSTR(sv);
+    if (!s || !*s) return el_wrap_str(el_strdup("[]"));
+    size_t n = strlen(s);
+    /* Worst case: each byte is 3 digits + comma = 4 chars, plus "[]" + NUL. */
+    char* out = el_strbuf(n * 4 + 3);
+    size_t pos = 0;
+    out[pos++] = '[';
+    for (size_t i = 0; i < n; i++) {
+        unsigned char b = (unsigned char)s[i];
+        if (i > 0) out[pos++] = ',';
+        /* Write decimal representation of b. */
+        if (b >= 100) {
+            out[pos++] = (char)('0' + b / 100);
+            out[pos++] = (char)('0' + (b / 10) % 10);
+            out[pos++] = (char)('0' + b % 10);
+        } else if (b >= 10) {
+            out[pos++] = (char)('0' + b / 10);
+            out[pos++] = (char)('0' + b % 10);
+        } else {
+            out[pos++] = (char)('0' + b);
+        }
+    }
+    out[pos++] = ']';
+    out[pos] = '\0';
+    return el_wrap_str(out);
+}
+
+/* bytes_to_str — decode a JSON array of integer byte values back to a string.
+ * "[104,101,108,108,111]" -> "hello"
+ * Inverse of str_to_bytes. */
+el_val_t bytes_to_str(el_val_t arr_v) {
+    const char* s = EL_CSTR(arr_v);
+    if (!s) return el_wrap_str(el_strdup(""));
+    /* Skip whitespace, expect '['. */
+    while (*s == ' ' || *s == '\t' || *s == '\n' || *s == '\r') s++;
+    if (*s != '[') return el_wrap_str(el_strdup(""));
+    s++;
+
+    /* Count elements to size the output buffer. */
+    int64_t n = (int64_t)json_array_len(arr_v);
+    if (n <= 0) return el_wrap_str(el_strdup(""));
+
+    char* out = el_strbuf((size_t)n + 1);
+    size_t pos = 0;
+
+    /* Walk the array, parse each integer, store as a byte. */
+    while (*s) {
+        while (*s == ' ' || *s == '\t' || *s == '\n' || *s == '\r') s++;
+        if (*s == ']' || *s == '\0') break;
+        /* Parse decimal integer. */
+        char* end_ptr;
+        long v = strtol(s, &end_ptr, 10);
+        if (end_ptr == s) break; /* parse failure */
+        s = end_ptr;
+        if (v >= 0 && v <= 255) out[pos++] = (char)(unsigned char)v;
+        while (*s == ' ' || *s == '\t' || *s == '\n' || *s == '\r') s++;
+        if (*s == ',') { s++; continue; }
+        if (*s == ']' || *s == '\0') break;
+    }
+    out[pos] = '\0';
+    return el_wrap_str(out);
+}
+
+/* hash_sha256 — return the SHA-256 hex digest of a string.
+ * Uses the built-in el_sha256_oneshot implementation (no OpenSSL required). */
+el_val_t hash_sha256(el_val_t sv) {
+    const char* s = EL_CSTR(sv);
+    if (!s) s = "";
+    unsigned char digest[32];
+    el_sha256_oneshot((const unsigned char*)s, strlen(s), digest);
+    return el_hex_encode(digest, 32);
+}
+
+
+/* HTTP client aliases — require curl; defined inside #ifdef HAVE_CURL below
+ * with a matching stub in the #ifndef HAVE_CURL block.                       */
+#if !defined(_WIN32) || defined(HAVE_CURL)
+/* __http_do also lives in el_seed.c; marked weak so el_seed.c's definition
+ * wins when both translation units are linked together (the real product build). */
+__attribute__((weak)) el_val_t __http_do(el_val_t method, el_val_t url, el_val_t body,
+                   el_val_t headers_map, el_val_t timeout_ms) {
+    /* timeout_ms is accepted for API compatibility but ignored here;
+     * el_runtime's http_do uses the EL_HTTP_TIMEOUT_MS env var instead. */
+    (void)timeout_ms;
+    struct curl_slist* h = headers_from_map(headers_map);
+    el_val_t r = http_do(EL_CSTR(method), EL_CSTR(url), EL_CSTR(body), h);
+    if (h) curl_slist_free_all(h);
+    return r;
+}
+
+/* __http_do_map — same as __http_do but headers_map arg is a JSON-string
+ * rather than an ElMap. Parse it first, then delegate.                      */
+el_val_t __http_do_map(el_val_t method, el_val_t url, el_val_t body,
+                       el_val_t headers_json, el_val_t timeout_ms) {
+    (void)timeout_ms;
+    /* Build a curl_slist from a JSON object {"Header":"value",...}. */
+    const char* hj = EL_CSTR(headers_json);
+    struct curl_slist* h = NULL;
+    if (hj && *hj && *hj == '{') {
+        /* Walk the JSON pairs with a simple parser reusing json_get_string logic. */
+        /* For correctness we just call the existing json_get iteration path.
+         * We duplicate the key-extraction loop from headers_from_map but driven
+         * by JSON rather than ElMap. Use json_get_raw to iterate is not easy
+         * without knowing keys, so accept the JSON string and build a tmp map. */
+        el_val_t map = json_parse(EL_STR(hj));
+        h = headers_from_map(map);
+    }
+    el_val_t r = http_do(EL_CSTR(method), EL_CSTR(url), EL_CSTR(body), h);
+    if (h) curl_slist_free_all(h);
+    return r;
+}
+
+/* __http_do_map_to_file — same as __http_do_map but streams response body
+ * to a local file path rather than returning it as a string.               */
+el_val_t __http_do_map_to_file(el_val_t method, el_val_t url, el_val_t body,
+                                el_val_t headers_json, el_val_t output_path) {
+    const char* hj = EL_CSTR(headers_json);
+    struct curl_slist* h = NULL;
+    if (hj && *hj && *hj == '{') {
+        el_val_t map = json_parse(EL_STR(hj));
+        h = headers_from_map(map);
+    }
+    el_val_t r = http_do_to_file(EL_CSTR(method), EL_CSTR(url), EL_CSTR(body),
+                                 h, EL_CSTR(output_path));
+    if (h) curl_slist_free_all(h);
+    return r;
+}
+#endif /* HAVE_CURL */
+
+#if defined(_WIN32) && !defined(HAVE_CURL)
+/* ── HAVE_CURL=0 stubs — compile without -lcurl for the elc CLI binary. ───── *
+ * These return a JSON error string so El programs get a clear message if they
+ * call HTTP/LLM functions in a curl-less build.                               */
+static el_val_t _no_curl_err(void) {
+    return el_wrap_str(el_strdup("{\"error\":\"not built with HAVE_CURL\"}"));
+}
+el_val_t http_get(el_val_t url)                          { (void)url; return _no_curl_err(); }
+el_val_t http_post(el_val_t url, el_val_t body)          { (void)url; (void)body; return _no_curl_err(); }
+el_val_t http_post_json(el_val_t url, el_val_t body)     { (void)url; (void)body; return _no_curl_err(); }
+el_val_t http_get_with_headers(el_val_t url, el_val_t h) { (void)url; (void)h; return _no_curl_err(); }
+el_val_t http_post_with_headers(el_val_t url, el_val_t b, el_val_t h) { (void)url; (void)b; (void)h; return _no_curl_err(); }
+el_val_t http_post_json_with_headers(el_val_t url, el_val_t h, el_val_t b) { (void)url; (void)h; (void)b; return _no_curl_err(); }
+el_val_t http_post_form_auth(el_val_t url, el_val_t b, el_val_t a)    { (void)url; (void)b; (void)a; return _no_curl_err(); }
+el_val_t http_delete(el_val_t url)                       { (void)url; return _no_curl_err(); }
+el_val_t http_patch(el_val_t url, el_val_t body)         { (void)url; (void)body; return _no_curl_err(); }
+el_val_t http_get_to_file(el_val_t url, el_val_t h, el_val_t p)       { (void)url; (void)h; (void)p; return _no_curl_err(); }
+el_val_t http_post_to_file(el_val_t url, el_val_t b, el_val_t h, el_val_t p) { (void)url; (void)b; (void)h; (void)p; return _no_curl_err(); }
+el_val_t http_post_engram(el_val_t url, el_val_t k, el_val_t b)       { (void)url; (void)k; (void)b; return _no_curl_err(); }
+el_val_t http_get_engram(el_val_t url, el_val_t k)                    { (void)url; (void)k; return _no_curl_err(); }
+el_val_t llm_call(el_val_t m, el_val_t p)                { (void)m; (void)p; return _no_curl_err(); }
+el_val_t llm_call_system(el_val_t m, el_val_t s, el_val_t u)          { (void)m; (void)s; (void)u; return _no_curl_err(); }
+el_val_t llm_call_agentic(el_val_t m, el_val_t s, el_val_t u, el_val_t t) { (void)m; (void)s; (void)u; (void)t; return _no_curl_err(); }
+el_val_t llm_vision(el_val_t m, el_val_t s, el_val_t p, el_val_t i)  { (void)m; (void)s; (void)p; (void)i; return _no_curl_err(); }
+el_val_t llm_models(void)                                { return el_list_empty(); }
+void     llm_register_tool(el_val_t n, el_val_t f)       { (void)n; (void)f; }
+/* __ HTTP stubs (no-curl build) */
+el_val_t __http_do(el_val_t m, el_val_t u, el_val_t b, el_val_t h, el_val_t t) { (void)m; (void)u; (void)b; (void)h; (void)t; return _no_curl_err(); }
+el_val_t __http_do_map(el_val_t m, el_val_t u, el_val_t b, el_val_t h, el_val_t t) { (void)m; (void)u; (void)b; (void)h; (void)t; return _no_curl_err(); }
+el_val_t __http_do_map_to_file(el_val_t m, el_val_t u, el_val_t b, el_val_t h, el_val_t p) { (void)m; (void)u; (void)b; (void)h; (void)p; return _no_curl_err(); }
+#endif /* !HAVE_CURL */

lang/runtime/el_runtime.h

@@ -275,6 +275,10 @@ el_val_t  json_set(el_val_t json_str, el_val_t key, el_val_t value);
 el_val_t  json_array_len(el_val_t json_str);
 el_val_t  json_array_get(el_val_t json_str, el_val_t index);
 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 ────────────────────────────────────────────────────────────────── */
 
@@ -301,6 +305,8 @@ el_val_t  time_diff(el_val_t ts1, el_val_t ts2, el_val_t unit);
 
 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);
@@ -892,6 +898,111 @@ el_val_t trace_span_start(el_val_t name);
 el_val_t trace_span_end(el_val_t span_handle);
 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.        */
+
+/* I/O */
+el_val_t  __println(el_val_t s);
+el_val_t  __print(el_val_t s);
+el_val_t  __readline(void);
+
+/* String */
+el_val_t  __int_to_str(el_val_t n);
+el_val_t  __str_to_int(el_val_t s);
+el_val_t  __float_to_str(el_val_t f);
+el_val_t  __str_to_float(el_val_t s);
+el_val_t  __str_len(el_val_t s);
+el_val_t  __str_char_at(el_val_t s, el_val_t i);
+el_val_t  __str_cmp(el_val_t a, el_val_t b);
+el_val_t  __str_ncmp(el_val_t a, el_val_t b, el_val_t n);
+el_val_t  __str_concat_raw(el_val_t a, el_val_t b);
+el_val_t  __str_slice_raw(el_val_t s, el_val_t start, el_val_t end);
+el_val_t  __str_alloc(el_val_t n);
+el_val_t  __str_set_char(el_val_t s, el_val_t i, el_val_t c);
+
+/* URL encoding */
+el_val_t  __url_encode(el_val_t s);
+el_val_t  __url_decode(el_val_t s);
+
+/* Environment */
+el_val_t  __env_get(el_val_t key);
+
+/* Subprocess */
+el_val_t  __exec(el_val_t cmd);
+el_val_t  __exec_bg(el_val_t cmd);
+
+/* Process */
+el_val_t  __exit_program(el_val_t code);
+
+/* Filesystem */
+el_val_t  __fs_exists(el_val_t path);
+el_val_t  __fs_mkdir(el_val_t path);
+el_val_t  __fs_read(el_val_t path);
+el_val_t  __fs_write(el_val_t path, el_val_t content);
+el_val_t  __fs_write_bytes(el_val_t path, el_val_t bytes, el_val_t n);
+el_val_t  __fs_list_raw(el_val_t path);
+
+/* HTTP server */
+el_val_t  __http_response(el_val_t status, el_val_t headers_json, el_val_t body);
+el_val_t  __http_serve(el_val_t port, el_val_t handler);
+el_val_t  __http_serve_v2(el_val_t port, el_val_t handler);
+
+/* HTTP conn fd / SSE (weak; overridden by el_seed.c when linked together) */
+el_val_t  __http_conn_fd(void);
+el_val_t  __http_sse_open(el_val_t conn_id);
+el_val_t  __http_sse_send(el_val_t conn_id, el_val_t data);
+el_val_t  __http_sse_close(el_val_t conn_id);
+
+/* HTTP client (requires HAVE_CURL; stubs provided for no-curl builds) */
+el_val_t  __http_do(el_val_t method, el_val_t url, el_val_t body,
+                    el_val_t headers_map, el_val_t timeout_ms);
+el_val_t  __http_do_map(el_val_t method, el_val_t url, el_val_t body,
+                        el_val_t headers_json, el_val_t timeout_ms);
+el_val_t  __http_do_map_to_file(el_val_t method, el_val_t url, el_val_t body,
+                                el_val_t headers_json, el_val_t output_path);
+
+/* JSON */
+el_val_t  __json_array_get(el_val_t json, el_val_t index);
+el_val_t  __json_array_get_string(el_val_t json, el_val_t index);
+el_val_t  __json_array_len(el_val_t json);
+el_val_t  __json_get(el_val_t json, el_val_t key);
+el_val_t  __json_get_raw(el_val_t json, el_val_t key);
+el_val_t  __json_set(el_val_t json, el_val_t key, el_val_t value);
+el_val_t  __json_parse_map(el_val_t json_str);
+el_val_t  __json_stringify_val(el_val_t val);
+
+/* Hashing */
+el_val_t  __sha256_hex(el_val_t s);
+
+/* State K/V */
+el_val_t  __state_del(el_val_t key);
+el_val_t  __state_get(el_val_t key);
+el_val_t  __state_keys(void);
+el_val_t  __state_set(el_val_t key, el_val_t val);
+
+/* UUID */
+el_val_t  __uuid_v4(void);
+
+/* Args */
+el_val_t  __args_json(void);
+
 #ifdef __cplusplus
 }
 #endif

lang/swarm/README.md

@@ -0,0 +1,184 @@
+# 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 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
+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 <sandbox>/.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.
+
+## 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:**
+- 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.
+

lang/swarm/build.sh

@@ -0,0 +1,61 @@
+#!/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 <program.el> <out-binary>
+#
+# 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 <program.el> <out-binary>}"
+OUT="${2:?usage: build.sh <program.el> <out-binary>}"
+
+# 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/reshape_surface.el
+  swarm/primitive_binding.el
+  swarm/primitive_seam.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 -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"
+    exit 1
+fi
+
+rm -f "$TMP_C" "$COMBINED"
+echo "built: $OUT"

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_string(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
+}

lang/swarm/containment.el

@@ -0,0 +1,181 @@
+// 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":"<corr>",
+// "worker":"<id-or-empty>","depth":"<n>"}.
+
+// ── Token minting ────────────────────────────────────────────────────────────
+
+// 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")
+    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")
+    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 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")
+    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")
+    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?
+// 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 ""
+}
+
+// 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.
+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_str("{}", "reason", reason)
+    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
+}

lang/swarm/primitive_binding.el

@@ -0,0 +1,49 @@
+// 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 — 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 {
+    let anchor: String = json_get_string(ctx, "input")
+    let faculty: String = json_get_string(ctx, "faculty")
+    return op_think(anchor, faculty)
+}
+
+// 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 {
+    if str_eq(env("SWARM_WRITE_HEALTHY"), "1") {
+        return op_attend(query, "self")
+    }
+    return primitive_attend(query, limit)
+}
+
+// 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 {
+    if str_eq(env("SWARM_WRITE_HEALTHY"), "1") {
+        return op_learn(observation, "induce")
+    }
+    return primitive_learn(corr_id, observation)
+}

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

lang/swarm/primitives.el

@@ -0,0 +1,94 @@
+// 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 "[]"
+    }
+    // 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 ─────────────────────────────────
+// 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)
+}

lang/swarm/reshape_surface.el

@@ -0,0 +1,103 @@
+// 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)
+}
+
+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 {
+    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)
+}

lang/swarm/swarm.el

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

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
+}

lang/swarm/tests/harness_real_cognition.el

@@ -0,0 +1,119 @@
+// 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)
+
+    // ── 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.")
+        return 0
+    }
+    print("REAL-COGNITION SWARM FAIL (" + int_to_str(fails) + ")")
+    return 1
+}

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
+}

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
+}

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
+}

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
+}

lang/swarm/worktrack.el

@@ -0,0 +1,217 @@
+// 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.
+
+// ── 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.
+//   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 {
+    // 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
+    }
+    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_str(body, "_auth", key)
+    let resp: String = http_post(url + "/api/nodes", 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)
+}