neuron-technologies/el
2
0
Fork 0
Code Issues 1 Pull Requests 6 Actions Packages Projects Releases 5 Wiki Activity

feat(engram): wire ISE emission into core processing paths (checkpoint, high-importance writes, WM promotion)

Browse Source
This commit is contained in:
Will Anderson
2026-05-13 15:45:16 -05:00
parent a3ead6552e
commit 0c2ff6957e
4 changed files with 652 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lang/releases/v1.0.0-20260501/el_runtime.c
+161 -2
View File
@@ -1526,6 +1526,83 @@ void http_serve(el_val_t port, el_val_t handler) {
close(sock);
}
/* ── http_serve_async — non-blocking HTTP server ─────────────────────────── */
/* Runs the accept loop in a background pthread, returns immediately so the
* calling EL script can continue (e.g. to run an awareness loop).
*
* El signature: http_serve_async(port, handler) -> Void */
typedef struct { int sock; } HttpServeAsyncArg;
static void* _http_serve_async_loop(void* raw) {
HttpServeAsyncArg* a = (HttpServeAsyncArg*)raw;
int sock = a->sock;
free(a);
while (1) {
struct sockaddr_in6 cli;
socklen_t clen = sizeof(cli);
int cfd = accept(sock, (struct sockaddr*)&cli, &clen);
if (cfd < 0) {
if (errno == EINTR) continue;
perror("accept"); break;
}
pthread_mutex_lock(&_http_conn_mu);
while (_http_conn_active >= HTTP_MAX_CONNS) {
pthread_cond_wait(&_http_conn_cv, &_http_conn_mu);
}
_http_conn_active++;
pthread_mutex_unlock(&_http_conn_mu);
HttpWorkerArg* arg = malloc(sizeof(HttpWorkerArg));
if (!arg) { close(cfd); continue; }
arg->fd = cfd;
pthread_t tid;
if (pthread_create(&tid, NULL, http_worker, arg) != 0) {
close(cfd); free(arg);
pthread_mutex_lock(&_http_conn_mu);
_http_conn_active--;
pthread_cond_signal(&_http_conn_cv);
pthread_mutex_unlock(&_http_conn_mu);
continue;
}
pthread_detach(tid);
}
close(sock);
return NULL;
}
void http_serve_async(el_val_t port, el_val_t handler) {
const char* hname = EL_CSTR(handler);
if (hname && looks_like_string(handler)) {
http_set_handler(handler);
}
int p = (int)port;
if (p <= 0 || p > 65535) { fprintf(stderr, "http_serve_async: invalid port %d\n", p); return; }
int sock = socket(AF_INET6, SOCK_STREAM, 0);
if (sock < 0) { perror("socket"); return; }
int yes = 1; int no = 0;
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, &no, sizeof(no));
struct sockaddr_in6 addr;
memset(&addr, 0, sizeof(addr));
addr.sin6_family = AF_INET6;
addr.sin6_addr = in6addr_any;
addr.sin6_port = htons((uint16_t)p);
if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
perror("bind"); close(sock); return;
}
if (listen(sock, 64) < 0) { perror("listen"); close(sock); return; }
fprintf(stderr, "[http] async listening on [::]:%d (dual-stack)\n", p);
HttpServeAsyncArg* a = malloc(sizeof(HttpServeAsyncArg));
if (!a) { close(sock); return; }
a->sock = sock;
pthread_t tid;
if (pthread_create(&tid, NULL, _http_serve_async_loop, a) != 0) {
perror("pthread_create"); free(a); close(sock); return;
}
pthread_detach(tid);
/* Returns immediately — caller can now run awareness_run() or any loop. */
}
/* ── HTTP server v2 — request headers + structured response ──────────────── */
/*
* v2 widens the handler signature from
@@ -5884,6 +5961,7 @@ static int engram_load_binary(const char* path);
static void engram_embed_node(EngramNode* n);
static float engram_cosine_sim(const float* a, const float* b, uint32_t dim);
static void engram_checkpoint(void);
static void engram_emit_ise_internal(const char* content, const char* label);
/* Salience may arrive either as a float bit-pattern or as a small integer
* (e.g. 1, meaning 1.0). Heuristic: if interpreted as double it's in
@@ -5977,6 +6055,14 @@ el_val_t engram_node_full(el_val_t content, el_val_t node_type, el_val_t label,
engram_checkpoint();
/* engram_persist_node is now a no-op — binary checkpoint handles persistence */
engram_persist_node(engram_data_dir(), n);
/* ISE: high-importance node written */
if (n->importance >= 0.8f) {
char ise_buf[512];
snprintf(ise_buf, sizeof(ise_buf),
"{\"event\":\"high_importance_node\",\"node_id\":\"%s\",\"node_type\":\"%s\",\"importance\":%.2f,\"ts\":%lld}",
n->id, n->node_type, n->importance, (long long)(time(NULL)*1000LL));
engram_emit_ise_internal(ise_buf, "high-importance-node");
}
return el_wrap_str(el_strdup(n->id));
}
@@ -6844,9 +6930,35 @@ el_val_t engram_activate(el_val_t query, el_val_t depth) {
n->suppression_count = 0;
}
/* Persist working_memory_weight (post Pass 3) to node store. */
/* Persist working_memory_weight (post Pass 3) to node store.
* Also emit ISE when a node transitions into working memory (was 0, now > 0.5),
* excluding InternalStateEvent nodes to prevent ISEs about ISEs. */
for (int64_t i = 0; i < g->node_count; i++) {
g->nodes[i].working_memory_weight = wm_weights[i];
EngramNode* wn = &g->nodes[i];
double prev_wm = wn->working_memory_weight;
wn->working_memory_weight = wm_weights[i];
/* ISE: working memory promotion — only on 0→>0.5 transition, non-ISE nodes */
if (prev_wm <= 0.0 && wm_weights[i] > 0.5 &&
wn->node_type && strcmp(wn->node_type, "InternalStateEvent") != 0) {
char ise_buf[512];
char label_safe[64];
const char* lbl = wn->label ? wn->label : "";
size_t llen = strlen(lbl);
if (llen > 60) llen = 60;
memcpy(label_safe, lbl, llen);
/* Escape any double-quotes in label for JSON safety */
size_t out = 0;
for (size_t k = 0; k < llen && out < 62; k++) {
if (lbl[k] == '"' || lbl[k] == '\\') label_safe[out++] = '\\';
label_safe[out++] = lbl[k];
}
label_safe[out] = '\0';
snprintf(ise_buf, sizeof(ise_buf),
"{\"event\":\"wm_promotion\",\"node_id\":\"%s\",\"label\":\"%s\",\"salience\":%.3f,\"wm_weight\":%.3f,\"ts\":%lld}",
wn->id, label_safe, wn->salience, wm_weights[i],
(long long)(time(NULL)*1000LL));
engram_emit_ise_internal(ise_buf, "wm-promotion");
}
}
/* ── HEBBIAN STRENGTHENING: fire together, wire together ─────────────
@@ -7661,6 +7773,43 @@ static int engram_load_binary(const char* path) {
return 1;
}
/* ── Internal ISE emission ────────────────────────────────────────────────── */
/* Fires an InternalStateEvent node directly into the in-memory graph from
* within core processing paths. Guarded against recursion so it's safe to
* call from node-write and activation paths. */
static int g_in_ise_emit = 0;
static void engram_emit_ise_internal(const char* content, const char* label) {
if (g_in_ise_emit) return; /* recursion guard */
if (!content || !*content) return;
g_in_ise_emit = 1;
EngramStore* g = engram_get();
engram_grow_nodes();
EngramNode* n = &g->nodes[g->node_count];
memset(n, 0, sizeof(*n));
n->id = engram_new_id();
n->content = strdup(content);
n->node_type = strdup("InternalStateEvent");
n->label = strdup(label && *label ? label : "ise");
n->tier = strdup("Working");
n->tags = strdup("internal-state,ise-internal");
n->metadata = strdup("{}");
n->salience = 0.3f;
n->importance = 0.5f;
n->confidence = 1.0f;
n->temporal_decay_rate = 0.0;
n->activation_count = 0;
int64_t now = engram_now_ms();
n->last_activated = now;
n->created_at = now;
n->updated_at = now;
n->layer_id = ENGRAM_LAYER_DEFAULT;
g->node_count++;
/* Skip engram_embed_node and engram_checkpoint to avoid recursion */
g_in_ise_emit = 0;
}
/* Checkpoint: save binary every ENGRAM_CHECKPOINT_INTERVAL writes. */
static void engram_checkpoint(void) {
g_writes_since_checkpoint++;
@@ -7671,6 +7820,16 @@ static void engram_checkpoint(void) {
snprintf(db_path, sizeof(db_path), "%s/engram.db", ddir);
if (engram_write_binary(db_path)) {
fprintf(stderr, "[engram] checkpoint saved: %s\n", db_path);
/* ISE: graph state at checkpoint */
EngramStore* g = engram_get();
char ise_buf[512];
struct stat st; long db_bytes = 0;
if (stat(db_path, &st) == 0) db_bytes = (long)st.st_size;
snprintf(ise_buf, sizeof(ise_buf),
"{\"event\":\"checkpoint\",\"nodes\":%lld,\"edges\":%lld,\"db_bytes\":%ld,\"ts\":%lld}",
(long long)g->node_count, (long long)g->edge_count,
db_bytes, (long long)(time(NULL)*1000LL));
engram_emit_ise_internal(ise_buf, "checkpoint");
}
}
lang/releases/v1.0.0-20260501/el_runtime.h
+1
View File
@@ -143,6 +143,7 @@ el_val_t http_post_with_headers(el_val_t url, el_val_t body, el_val_t headers_m
el_val_t http_post_form_auth(el_val_t url, el_val_t form_body, el_val_t auth_header);
el_val_t http_delete(el_val_t url);
void http_serve(el_val_t port, el_val_t handler);
void http_serve_async(el_val_t port, el_val_t handler);
void http_set_handler(el_val_t name);
/* HTTP server v2 ─────────────────────────────────────────────────────────────