Compare commits

..

2 Commits

Author SHA1 Message Date
will.anderson c2d8a07c7b transduce: name the invisible mechanism, fix a silent-failure bug, drop a CRUD verb
El SDK CI - dev / build-and-test (pull_request) Failing after 14m6s
ingest and transduce are complements, not synonyms: ingest is the conscious,
deliberate act of pointing at a source (ingest_file/dir/url/llm/stream stay
named exactly that); transduce is the automatic, invisible mechanism inside
it that converts extracted surface content into geometry (renamed
build_prose/build_structured -> transduce_prose/transduce_structured, the
functions that actually turn raw text into a node+edge manifold).

Real bug found and fixed along the way: the final /api/load-merge response
was never checked for an error. A total failure (bad auth, network down,
anything) silently reported nodes_added:0/edges_added:0 — indistinguishable
from a benign 'everything was already known' outcome. Verified live: with a
wrong key, the tool now honestly returns {"error":"load-merge failed:
unauthorized",...} instead of a misleading zero.

Also dropped a CRUD-verb smell: the per-decision println said CREATE (a
database-log verb for something that hasn't actually been written to the
server yet — it's a local, tentative decision pending the batch merge).
Renamed to FORM. The dead-code eg_create_node (defined, never called)
renamed to eg_crystallize_node and annotated honestly as unused, since if
it's ever wired up it represents the real server-confirmed write, unlike
the local FORM guess.

Not yet re-verified end-to-end against a real successful write: the
ingest-test sandbox (nsbx up ingest-test) is itself currently broken —
it prints a green "ready" banner after its own readiness check fails,
and nothing is actually listening. Filed separately; not in scope here.
2026-08-15 14:44:18 -05:00
bigmerge 710bea174d Add native EL afferent ingest organ
El SDK CI - dev / build-and-test (pull_request) Successful in 6m29s
Source-polymorphic ingest(source) primitive: extracts content faithfully
from a directory/file/url/llm-query/structured-primitive-set/stream,
decomposes it into a discrete multi-node graph manifold (nodes + internal
edges, never a single blob), and merges it into the engram geometry with
dedup (search + exact/cosine match), provenance, grounding-level, and
stewardship-class tagging from the moment of entry.

Pure HTTP client of the engram server (links only el_runtime.c, never
el_seed.c/the engine directly). Tested against a live nsbx sandbox engram
clone (127.0.0.1:8903) with real writes confirmed via /api/stats
(node_count 3201 / edge_count 6601).

Excludes ingest/build/ — local compiler scratch output (binaries, .c
codegen, .err logs), not source.
2026-08-15 14:22:07 -05:00
4 changed files with 764 additions and 505 deletions
+764
View File
@@ -0,0 +1,764 @@
// ingest.el the native EL AFFERENT INGEST ORGAN
//
// The source-polymorphic ingest(source) primitive: point it at a directory,
// file, url, llm-query, structured-primitive set, or stream; it EXTRACTS the
// real content faithfully (no invention), TRANSDUCES it into a DISCRETE
// MANIFOLD (multiple nodes + internal edges meaning-structure, never a
// single blob; the conversion from extracted surface content into geometry
// is automatic and invisible to the caller, the way digestion is invisible
// to the one who chose to eat ingest is the conscious act, transduce is
// the mechanism underneath it, and it is no less real for being unseen),
// and MERGES that manifold into the engram geometry: shared
// meanings DEDUP onto existing nodes (search + exact/cosine match), genuinely
// new meanings add nodes, relations add edges. Every node enters with
// PROVENANCE + grounding-level + stewardship class from the moment of entry.
//
// It is a pure HTTP CLIENT of the engram server it links only el_runtime.c
// via fs/http/json/string builtins; it never links el_seed.c or the engram
// engine. This is the general afferent metabolism the migration / reseed /
// fetch_fact / conversation / multimodal-learning all ride on.
//
// Build (canonical runtime):
// ELC=lang/dist/platform/elc ; RT=lang/releases/v1.0.0-20260501
// $ELC ingest/src/ingest.el > ingest/build/ingest.c
// cc -std=c11 -O2 -I $RT -o ingest/build/ingest ingest/build/ingest.c $RT/el_runtime.c -lcurl -lpthread
//
// Run (against an nsbx sandbox clone NEVER the live :8742):
// ENGRAM_URL=http://127.0.0.1:8902 ENGRAM_KEY=sbx-ingest-test \
// INGEST_KIND=file INGEST_ARG=/abs/path.md ./ingest/build/ingest
//
// SECTION A JSON helpers (self-defined; canonical runtime does not export
// json_build_object / json_escape_string, so we own them here)
//
fn j_esc(s: String) -> String {
let a: String = str_replace(s, "\\", "\\\\")
let b: String = str_replace(a, "\"", "\\\"")
let c: String = str_replace(b, "\n", "\\n")
let d: String = str_replace(c, "\r", "\\r")
let e: String = str_replace(d, "\t", "\\t")
return e
}
// a quoted, escaped JSON string literal
fn j_q(s: String) -> String {
return "\"" + j_esc(s) + "\""
}
// Extract the top-level keys of a JSON object string. A thin, self-contained
// scanner (FLAGGED: the one non-trivial parser in this organ everything else
// is faithful text handling). Tracks string state + brace/bracket depth; a key
// is a string at object-interior depth 1 immediately followed by ':'.
fn json_object_keys(obj: String) -> [String] {
let keys: [String] = el_list_empty()
let n: Int = str_len(obj)
let i: Int = 0
let depth: Int = 0
let in_str: Bool = false
let esc: Bool = false
let str_start: Int = -1
let cur: String = ""
let have_key: Bool = false
while i < n {
let c: String = str_char_at(obj, i)
if in_str {
if esc {
esc = false
} else {
if str_eq(c, "\\") {
esc = true
} else {
if str_eq(c, "\"") {
in_str = false
cur = str_slice(obj, str_start + 1, i)
have_key = true
}
}
}
} else {
if str_eq(c, "\"") {
in_str = true
str_start = i
}
if str_eq(c, "{") { depth = depth + 1 }
if str_eq(c, "}") { depth = depth - 1 }
if str_eq(c, "[") { depth = depth + 1 }
if str_eq(c, "]") { depth = depth - 1 }
if str_eq(c, ":") {
if have_key {
if depth == 1 {
keys = el_list_append(keys, cur)
}
}
have_key = false
}
if str_eq(c, ",") { have_key = false }
}
i = i + 1
}
return keys
}
//
// SECTION B engram HTTP client (provenance-carrying afferent LOAD)
//
fn eg_base() -> String {
let u: String = env("ENGRAM_URL")
if !str_eq(u, "") { return u }
let s: String = env("SBX_URL")
if !str_eq(s, "") { return s }
return "http://127.0.0.1:8902"
}
fn eg_key() -> String {
let k: String = env("ENGRAM_KEY")
if !str_eq(k, "") { return k }
let s: String = env("SBX_KEY")
if !str_eq(s, "") { return s }
return ""
}
// POST a JSON body (auth _auth injected) to an engram path.
fn eg_post(path: String, body_inner: String) -> String {
let key: String = eg_key()
let auth: String = if str_eq(key, "") { "" } else { ",\"_auth\":" + j_q(key) }
let body: String = "{" + body_inner + auth + "}"
return http_post_json(eg_base() + path, body)
}
fn eg_get(path: String) -> String {
return http_get(eg_base() + path)
}
// crystallize a node with full provenance-bearing metadata (server-confirmed
// write unlike the local FORM decision in merge_manifold, this is real);
// returns the new node id. Not currently called by any live path (dead code,
// kept for a future single-node ad-hoc write use case) 2026-08-15.
fn eg_crystallize_node(content: String, ntype: String, tier: String,
sal: String, imp: String, conf: String, tags: String) -> String {
let inner: String =
"\"content\":" + j_q(content) +
",\"node_type\":" + j_q(ntype) +
",\"label\":" + j_q(str_slice(content, 0, 80)) +
",\"tier\":" + j_q(tier) +
",\"salience\":" + sal +
",\"importance\":" + imp +
",\"confidence\":" + conf +
",\"tags\":" + j_q(tags)
let resp: String = eg_post("/api/nodes", inner)
return json_get_string(resp, "id")
}
// search the existing geometry (lexical token-overlap rank); returns JSON array
fn eg_search(query: String, limit: Int) -> String {
let inner: String = "\"query\":" + j_q(query) + ",\"limit\":" + int_to_str(limit)
return eg_post("/api/search", inner)
}
// cosine similarity between two existing (embedded) nodes; -2 if not comparable
fn eg_similarity(a: String, b: String) -> Float {
let resp: String = eg_get("/api/similarity?a=" + a + "&b=" + b)
return json_get_float(resp, "cosine")
}
fn eg_embed_backfill(n: Int) -> String {
return eg_get("/api/embed-backfill?n=" + int_to_str(n))
}
fn eg_forget(id: String) -> String {
return http_delete(eg_base() + "/api/nodes/" + id)
}
// DEDUP probe: is this meaning already in the graph?
// TIER 1 (deterministic, no embedding needed): search by content tokens, then
// exact normalized-content match among the candidates. Returns the existing
// node id, or "" if the meaning is genuinely new.
fn find_existing_by_content(content: String) -> String {
let want: String = str_trim(content)
if str_eq(want, "") { return "" }
let arr: String = eg_search(content, 8)
let n: Int = json_array_len(arr)
let i: Int = 0
while i < n {
let hit: String = json_array_get(arr, i)
let hc: String = str_trim(json_get_string(hit, "content"))
if str_eq(hc, want) {
return json_get_string(hit, "id")
}
i = i + 1
}
return ""
}
//
// SECTION C manifold representation (nodes + internal edges, in memory)
// A NODE is a JSON obj {lid, content, ntype, tier, sal, imp, conf, tags}.
// An EDGE is a JSON obj {from, rel, to}. lid = local id within this manifold.
//
fn mk_node(lid: String, content: String, ntype: String, tier: String,
sal: String, imp: String, conf: String, tags: String) -> String {
return "{\"lid\":" + j_q(lid) +
",\"content\":" + j_q(content) +
",\"ntype\":" + j_q(ntype) +
",\"tier\":" + j_q(tier) +
",\"sal\":" + j_q(sal) +
",\"imp\":" + j_q(imp) +
",\"conf\":" + j_q(conf) +
",\"tags\":" + j_q(tags) + "}"
}
fn mk_edge(ef: String, rel: String, et: String) -> String {
return "{\"from\":" + j_q(ef) + ",\"rel\":" + j_q(rel) + ",\"to\":" + j_q(et) + "}"
}
// linear lookup in parallel lid/real lists
fn lid_lookup(lids: [String], reals: [String], lid: String) -> String {
let n: Int = el_list_len(lids)
let i: Int = 0
while i < n {
if str_eq(el_list_get(lids, i), lid) {
return el_list_get(reals, i)
}
i = i + 1
}
return ""
}
//
// SECTION D the MERGE: resolve each manifold node (dedup or create), then
// wire the internal edges onto the resolved real ids. This is the
// merge boundary: shared meanings collapse onto existing nodes;
// genuinely-new meanings add nodes; relations add edges. Structure
// grows, size saturates.
//
// within-run content dedup: has this exact meaning already been resolved in
// THIS manifold? returns its real id, or "".
fn lookup_content(contents: [String], reals: [String], content: String) -> String {
let n: Int = el_list_len(contents)
let i: Int = 0
while i < n {
if str_eq(el_list_get(contents, i), content) {
return el_list_get(reals, i)
}
i = i + 1
}
return ""
}
fn ingest_snap_path() -> String {
let p: String = env("INGEST_SNAP")
if !str_eq(p, "") { return p }
return "/tmp/ingest-organ-snap.json"
}
// The MERGE. Resolve every manifold node against (1) already-resolved nodes in
// this run and (2) the existing graph (search + exact content match). Shared
// meanings collapse onto an existing id (DEDUP); genuinely-new meanings get a
// fresh id and go into the snapshot (CREATE). Then wire the internal edges onto
// resolved ids. LOAD is ONE snapshot merged via /api/load-merge a single
// write (scales to the migration), the sanctioned rail. Structure grows, size
// saturates: re-ingesting adds ~0 nodes, only edges/strengthening.
fn merge_manifold(nodes: [String], edges: [String]) -> String {
let nn: Int = el_list_len(nodes)
let lids: [String] = el_list_empty()
let reals: [String] = el_list_empty()
let contents: [String] = el_list_empty()
let snap_nodes: String = "["
let sn_count: Int = 0
let created: Int = 0
let deduped: Int = 0
let i: Int = 0
while i < nn {
let node: String = el_list_get(nodes, i)
let lid: String = json_get_string(node, "lid")
let content: String = json_get_string(node, "content")
let ntype: String = json_get_string(node, "ntype")
let tier: String = json_get_string(node, "tier")
let sal: String = json_get_string(node, "sal")
let imp: String = json_get_string(node, "imp")
let conf: String = json_get_string(node, "conf")
let tags: String = json_get_string(node, "tags")
let real: String = ""
let prior: String = lookup_content(contents, reals, content)
if !str_eq(prior, "") {
real = prior
deduped = deduped + 1
println(" DEDUP* " + real + " :: " + head80(content))
} else {
let existing: String = find_existing_by_content(content)
if !str_eq(existing, "") {
real = existing
deduped = deduped + 1
println(" DEDUP " + real + " :: " + head80(content))
} else {
real = uuid_v4()
// provenance + grounding + stewardship: searchable in tags,
// structured in metadata carried from the moment of entry.
let meta: String = "{\"provenance\":" + j_q(tags) + ",\"ingest_organ\":\"native-el\"}"
let njson: String = "{\"id\":" + j_q(real) +
",\"content\":" + j_q(content) +
",\"node_type\":" + j_q(ntype) +
",\"label\":" + j_q(head80(content)) +
",\"tier\":" + j_q(tier) +
",\"tags\":" + j_q(tags) +
",\"metadata\":" + j_q(meta) +
",\"salience\":" + sal +
",\"importance\":" + imp +
",\"confidence\":" + conf + "}"
let sep: String = if sn_count == 0 { "" } else { "," }
snap_nodes = snap_nodes + sep + njson
sn_count = sn_count + 1
created = created + 1
println(" FORM " + real + " :: " + head80(content))
}
}
lids = el_list_append(lids, lid)
reals = el_list_append(reals, real)
contents = el_list_append(contents, content)
i = i + 1
}
snap_nodes = snap_nodes + "]"
// resolve internal edges onto real ids
let ne: Int = el_list_len(edges)
let snap_edges: String = "["
let ec: Int = 0
let j: Int = 0
while j < ne {
let edge: String = el_list_get(edges, j)
let flid: String = json_get_string(edge, "from")
let tlid: String = json_get_string(edge, "to")
let rel: String = json_get_string(edge, "rel")
let fr: String = lid_lookup(lids, reals, flid)
let tr: String = lid_lookup(lids, reals, tlid)
if !str_eq(fr, "") {
if !str_eq(tr, "") {
let eid: String = uuid_v4()
let ejson: String = "{\"id\":" + j_q(eid) +
",\"from_id\":" + j_q(fr) + ",\"to_id\":" + j_q(tr) +
",\"relation\":" + j_q(rel) + ",\"weight\":0.6}"
let sep: String = if ec == 0 { "" } else { "," }
snap_edges = snap_edges + sep + ejson
ec = ec + 1
println(" EDGE " + fr + " -" + rel + "-> " + tr)
}
}
j = j + 1
}
snap_edges = snap_edges + "]"
// LOAD: one snapshot, one merge (single write).
let snap: String = "{\"nodes\":" + snap_nodes + ",\"edges\":" + snap_edges + "}"
let path: String = ingest_snap_path()
fs_write(path, snap)
let resp: String = eg_post("/api/load-merge", "\"path\":" + j_q(path))
// HONESTY GATE: the local FORM/DEDUP/EDGE decisions above are real (they
// describe what this manifold contains), but they are NOT confirmation of
// a server write only this response is. If the server returned an error
// (bad auth, network failure, anything), nodes_added/edges_added silently
// default to 0 via json_get_int, which reads identically to "everything
// was already known" a real failure and a benign no-op must never look
// the same. Surface the distinction explicitly rather than let a caller
// (or a human) infer success from a quiet zero.
let srv_err: String = json_get_string(resp, "error")
if !str_eq(srv_err, "") {
return "{\"error\":" + j_q("load-merge failed: " + srv_err) +
",\"nodes_formed_locally\":" + int_to_str(created) +
",\"nodes_deduped_locally\":" + int_to_str(deduped) +
",\"manifold_nodes\":" + int_to_str(nn) +
",\"manifold_edges\":" + int_to_str(ne) +
",\"note\":" + j_q("nothing below this manifold was confirmed persisted by the server") + "}"
}
let nadd: Int = json_get_int(resp, "nodes_added")
let eadd: Int = json_get_int(resp, "edges_added")
return "{\"nodes_created\":" + int_to_str(created) +
",\"nodes_deduped\":" + int_to_str(deduped) +
",\"new_in_snapshot\":" + int_to_str(sn_count) +
",\"nodes_added\":" + int_to_str(nadd) +
",\"edges_resolved\":" + int_to_str(ec) +
",\"edges_added\":" + int_to_str(eadd) +
",\"manifold_nodes\":" + int_to_str(nn) +
",\"manifold_edges\":" + int_to_str(ne) + "}"
}
fn head80(s: String) -> String {
let t: String = str_trim(s)
if str_len(t) <= 80 { return t }
return str_slice(t, 0, 80) + "..."
}
//
// SECTION E EXTRACTORS (faithful; no invention). Each returns a manifold by
// APPENDING to the nodes/edges accumulators via a returned struct.
// We accumulate into module-level lists carried by the caller.
//
// PROSE: chunk text into a discrete manifold. Split on blank lines into
// paragraphs; every non-empty paragraph is its own node (NEVER one blob).
// Edges: doc-root -contains-> chunk; chunk -precedes-> next chunk;
// most-recent-heading -section_of-> chunk. Content is verbatim (substring of
// the source) pure extraction of ground truth.
fn transduce_prose(nodes: [String], edges: [String], text: String,
prov: String, ground: String, steward: String,
root_lid: String, root_title: String) -> [String] {
// returns [nodes_json_list_encoded, edges_json_list_encoded] is awkward in
// EL; instead we mutate by returning a 2-list. We package results as a
// single JSON array string carrying {nodes:[...],edges:[...]} additions.
// (Kept simple: caller passes empty lists and receives the packaged pair.)
let tagbase: String = "prov:" + prov + " ground:" + ground + " steward:" + steward
// root node
nodes = el_list_append(nodes, mk_node(root_lid, "document: " + root_title,
"Concept", "Semantic", "0.6", "0.6", "0.9", tagbase + " kind:document"))
let paras: [String] = str_split(text, "\n\n")
let np: Int = el_list_len(paras)
let idx: Int = 0
let last_chunk: String = ""
let last_heading: String = ""
let ci: Int = 0
while idx < np {
let raw: String = str_trim(el_list_get(paras, idx))
if !str_eq(raw, "") {
let lid: String = root_lid + ":c" + int_to_str(ci)
let is_heading: Bool = str_starts_with(raw, "#")
let kind: String = if is_heading { "kind:heading" } else { "kind:doc-chunk" }
nodes = el_list_append(nodes, mk_node(lid, raw,
"Knowledge", "Semantic", "0.55", "0.55", "0.9", tagbase + " " + kind))
// containment: document root -contains-> chunk
edges = el_list_append(edges, mk_edge(root_lid, "contains", lid))
// sequence: previous chunk -precedes-> this chunk
if !str_eq(last_chunk, "") {
edges = el_list_append(edges, mk_edge(last_chunk, "precedes", lid))
}
// sectioning: most-recent heading -section_of-> this chunk
if is_heading {
last_heading = lid
} else {
if !str_eq(last_heading, "") {
edges = el_list_append(edges, mk_edge(last_heading, "section_of", lid))
}
}
last_chunk = lid
ci = ci + 1
}
idx = idx + 1
}
// package: we return the two lists concatenated via a sentinel; but EL
// lists can't nest heterogeneously here, so we instead return nodes and
// rely on the caller holding edges by reference is not possible so we
// encode both into one list: [ "N" + nodejson ... , "E" + edgejson ... ].
let packed: [String] = el_list_empty()
let a: Int = 0
let an: Int = el_list_len(nodes)
while a < an { packed = el_list_append(packed, "N" + el_list_get(nodes, a)) a = a + 1 }
let b: Int = 0
let bn: Int = el_list_len(edges)
while b < bn { packed = el_list_append(packed, "E" + el_list_get(edges, b)) b = b + 1 }
return packed
}
// STRUCTURED / RAW-GEOMETRY: ingest structured primitives (phonetics/formants,
// instrument signatures, scene primitives) as GEOMETRY, faithfully. Normalized
// input shape:
// {"dataset":"<name>","primitive_type":"<t>",
// "records":[{"key":"<id>","features":{...categorical...},"attributes":{...}}]}
// Each record -> a primitive node; each categorical feature -> a SHARED feature
// node (deduped across records: many primitives -> one feature node = real
// connective geometry, meaning saturates); numeric attributes fold into the
// primitive's content (unique values, no dedup benefit). This is knowledge
// represented as geometry, not prose the path speech/music/image ingest on.
fn transduce_structured(nodes: [String], edges: [String], js: String,
prov: String, ground: String, steward: String,
root_lid: String) -> [String] {
// grounding integrity: the SOURCE may declare its own epistemic grounding
// (measured / derived / convention / ...) via a top-level "grounding" field;
// honor it faithfully over the ingest-time default. This keeps the per-node
// ground: facet consistent with the source's honest self-description.
let src_ground: String = json_get_string(js, "grounding")
let use_ground: String = if str_eq(src_ground, "") { ground } else { src_ground }
let tagbase: String = "prov:" + prov + " ground:" + use_ground + " steward:" + steward
let dsname: String = json_get_string(js, "dataset")
let ptype: String = json_get_string(js, "primitive_type")
// capture the source's own scholarly provenance citation (verbatim) onto
// the dataset root faithful attribution, retrievable, reachable from every
// primitive via its -contains- edge back to the root.
let src_cite: String = json_get_string(js, "provenance")
let root_content: String = "dataset: " + dsname + " (" + ptype + ")"
if !str_eq(src_cite, "") { root_content = root_content + " | provenance: " + src_cite }
nodes = el_list_append(nodes, mk_node(root_lid, root_content,
"Concept", "Semantic", "0.6", "0.6", "0.9", tagbase + " kind:dataset"))
let recs: String = json_get_raw(js, "records")
let nr: Int = json_array_len(recs)
let r: Int = 0
while r < nr {
let rec: String = json_array_get(recs, r)
let rkey: String = json_get_string(rec, "key")
let attrs: String = json_get_raw(rec, "attributes")
// faithful compact serialization of the primitive's numeric signature
let attr_str: String = flatten_pairs(attrs)
let content: String = ptype + " " + rkey
if !str_eq(attr_str, "") { content = content + " | " + attr_str }
let plid: String = root_lid + ":" + rkey
nodes = el_list_append(nodes, mk_node(plid, content,
"Concept", "Semantic", "0.6", "0.6", "0.92",
tagbase + " kind:primitive primitive:" + ptype + " key:" + rkey))
edges = el_list_append(edges, mk_edge(root_lid, "contains", plid))
// categorical features -> SHARED (deduped) feature nodes + labelled edges
let feats: String = json_get_raw(rec, "features")
let fkeys: [String] = json_object_keys(feats)
let fk: Int = el_list_len(fkeys)
let k: Int = 0
while k < fk {
let fname: String = el_list_get(fkeys, k)
let fval: String = json_get_string(feats, fname)
// shared feature node: content is the feature=value pair; identical
// pairs across records dedup onto ONE node (the geometry).
let flid: String = "feat:" + fname + "=" + fval
let fcontent: String = fname + "=" + fval
nodes = el_list_append(nodes, mk_node(flid, fcontent,
"Concept", "Semantic", "0.5", "0.5", "0.9",
tagbase + " kind:feature feature:" + fname))
edges = el_list_append(edges, mk_edge(plid, fname, flid))
k = k + 1
}
r = r + 1
}
let packed: [String] = el_list_empty()
let a: Int = 0
let an: Int = el_list_len(nodes)
while a < an { packed = el_list_append(packed, "N" + el_list_get(nodes, a)) a = a + 1 }
let b: Int = 0
let bn: Int = el_list_len(edges)
while b < bn { packed = el_list_append(packed, "E" + el_list_get(edges, b)) b = b + 1 }
return packed
}
// flatten a flat JSON object of scalar fields into "k=v k=v" (faithful; values
// verbatim). Used for numeric attribute signatures.
fn flatten_pairs(obj: String) -> String {
if str_eq(obj, "") { return "" }
let keys: [String] = json_object_keys(obj)
let n: Int = el_list_len(keys)
let out: String = ""
let i: Int = 0
while i < n {
let k: String = el_list_get(keys, i)
// json_get_raw returns the raw token works for NUMBERS (bare, e.g.
// "270") where json_get_string yields "" for non-string values. Strip
// surrounding quotes if the value happens to be a string token.
let raw: String = json_get_raw(obj, k)
let v: String = str_replace(raw, "\"", "")
let sep: String = if i == 0 { "" } else { " " }
out = out + sep + k + "=" + v
i = i + 1
}
return out
}
// unpack the "N"/"E"-prefixed packed list back into two lists, then merge
fn merge_packed(packed: [String]) -> String {
let nodes: [String] = el_list_empty()
let edges: [String] = el_list_empty()
let n: Int = el_list_len(packed)
let i: Int = 0
while i < n {
let item: String = el_list_get(packed, i)
let tag: String = str_slice(item, 0, 1)
let rest: String = str_slice(item, 1, str_len(item))
if str_eq(tag, "N") { nodes = el_list_append(nodes, rest) }
if str_eq(tag, "E") { edges = el_list_append(edges, rest) }
i = i + 1
}
return merge_manifold(nodes, edges)
}
//
// SECTION F DISPATCH on source kind
//
fn basename(path: String) -> String {
let parts: [String] = str_split(path, "/")
let n: Int = el_list_len(parts)
if n == 0 { return path }
return el_list_get(parts, n - 1)
}
fn ends_with_ci(s: String, suf: String) -> Bool {
return str_ends_with(str_to_lower(s), suf)
}
fn is_text_file(path: String) -> Bool {
return ends_with_ci(path, ".md") || ends_with_ci(path, ".txt")
|| ends_with_ci(path, ".markdown") || ends_with_ci(path, ".text")
}
// default ingestion grounding; overridable per-invocation via INGEST_GROUND.
// Note: a source's OWN top-level "grounding" field (structured) takes precedence
// over this the author's honest self-description wins.
fn default_ground() -> String {
let g: String = env("INGEST_GROUND")
if str_eq(g, "") { return "extracted" }
return g
}
fn default_steward() -> String {
let s: String = env("INGEST_STEWARD")
if str_eq(s, "") { return "local-private" }
return s
}
// ingest one file -> report JSON
fn ingest_file(path: String) -> String {
let text: String = fs_read(path)
if str_eq(text, "") {
return "{\"error\":\"empty or unreadable\",\"path\":" + j_q(path) + "}"
}
let prov: String = "file:" + path
if ends_with_ci(path, ".json") {
let packed: [String] = transduce_structured(el_list_empty(), el_list_empty(),
text, prov, default_ground(), default_steward(), "ds:" + basename(path))
return merge_packed(packed)
}
let packed: [String] = transduce_prose(el_list_empty(), el_list_empty(),
text, prov, default_ground(), default_steward(),
"doc:" + basename(path), basename(path))
return merge_packed(packed)
}
// ingest a directory: walk one level, ingest each supported file, aggregate
fn ingest_dir(path: String) -> String {
let entries: [String] = fs_list(path)
let n: Int = el_list_len(entries)
let tot_created: Int = 0
let tot_deduped: Int = 0
let tot_edges: Int = 0
let files: Int = 0
let i: Int = 0
while i < n {
let name: String = str_trim(el_list_get(entries, i))
if !str_eq(name, "") {
let full: String = path + "/" + name
if is_text_file(full) || ends_with_ci(full, ".json") {
println("FILE " + full)
let rep: String = ingest_file(full)
tot_created = tot_created + json_get_int(rep, "nodes_created")
tot_deduped = tot_deduped + json_get_int(rep, "nodes_deduped")
tot_edges = tot_edges + json_get_int(rep, "edges_added")
files = files + 1
}
}
i = i + 1
}
return "{\"kind\":\"directory\",\"path\":" + j_q(path) +
",\"files_ingested\":" + int_to_str(files) +
",\"nodes_created\":" + int_to_str(tot_created) +
",\"nodes_deduped\":" + int_to_str(tot_deduped) +
",\"edges_accepted\":" + int_to_str(tot_edges) + "}"
}
// ingest a url: fetch, treat body as prose (faithful extraction of what's there)
fn ingest_url(url: String) -> String {
let body: String = http_get(url)
if str_eq(body, "") { return "{\"error\":\"empty fetch\",\"url\":" + j_q(url) + "}" }
let packed: [String] = transduce_prose(el_list_empty(), el_list_empty(),
body, "url:" + url, "extracted", "public-web",
"url:" + url, url)
return merge_packed(packed)
}
// ingest an llm-query: pose the query to the local guide model, take the answer
// as a CANDIDATE (provisional, guide-sourced grounding) never believe-the-
// model. The answer is ingested faithfully as what the model said, marked.
fn ingest_llm(query: String) -> String {
let model: String = if str_eq(env("INGEST_MODEL"), "") { "qwen3:1.7b" } else { env("INGEST_MODEL") }
let body: String = "{\"model\":" + j_q(model) + ",\"prompt\":" + j_q(query) + ",\"stream\":false}"
let resp: String = http_post_json("http://127.0.0.1:11434/api/generate", body)
let answer: String = json_get_string(resp, "response")
if str_eq(answer, "") { return "{\"error\":\"no model response\"}" }
let packed: [String] = transduce_prose(el_list_empty(), el_list_empty(),
answer, "llm:" + model + ":" + query, "candidate-provisional", "guide-provisional",
"llm:" + query, "guide answer: " + query)
return merge_packed(packed)
}
// ingest a stream: a file whose lines are turns; each line a node, sequence
// edges the conversational-manifold degenerate case (continuous metabolism).
fn ingest_stream(path: String) -> String {
let text: String = fs_read(path)
if str_eq(text, "") { return "{\"error\":\"empty stream\"}" }
let lines: [String] = str_split(text, "\n")
let nodes: [String] = el_list_empty()
let edges: [String] = el_list_empty()
let prov: String = "stream:" + path
let tagbase: String = "prov:" + prov + " ground:extracted steward:local-private"
nodes = el_list_append(nodes, mk_node("stream", "stream: " + basename(path),
"Concept", "Semantic", "0.6", "0.6", "0.9", tagbase + " kind:stream"))
let n: Int = el_list_len(lines)
let i: Int = 0
let prev: String = ""
let ci: Int = 0
while i < n {
let ln: String = str_trim(el_list_get(lines, i))
if !str_eq(ln, "") {
let lid: String = "stream:t" + int_to_str(ci)
nodes = el_list_append(nodes, mk_node(lid, ln,
"Memory", "Episodic", "0.5", "0.5", "0.85", tagbase + " kind:turn"))
edges = el_list_append(edges, mk_edge("stream", "contains", lid))
if !str_eq(prev, "") { edges = el_list_append(edges, mk_edge(prev, "precedes", lid)) }
prev = lid
ci = ci + 1
}
i = i + 1
}
return merge_manifold(nodes, edges)
}
//
// SECTION G ENTRY
//
let kind: String = env("INGEST_KIND")
let arg: String = env("INGEST_ARG")
println("[ingest] organ online — engram=" + eg_base() + " kind=" + kind)
println("[ingest] source=" + arg)
let report: String = ""
if str_eq(kind, "dir") {
report = ingest_dir(arg)
} else {
if str_eq(kind, "file") {
report = ingest_file(arg)
} else {
if str_eq(kind, "structured") {
report = ingest_file(arg)
} else {
if str_eq(kind, "url") {
report = ingest_url(arg)
} else {
if str_eq(kind, "llm") {
report = ingest_llm(arg)
} else {
if str_eq(kind, "stream") {
report = ingest_stream(arg)
} else {
report = "{\"error\":\"unknown INGEST_KIND: " + kind + "\"}"
}
}
}
}
}
}
println("REPORT " + report)
-269
View File
@@ -1,269 +0,0 @@
// swarm.el Native interruptibility for Neuron-dispatched agents.
//
// CANON: Neuron memory 1ca4d3e9 (native-interruptibility). A dispatched worker
// must be re-steerable OR killable the INSTANT a correction / new signal arrives,
// MID-TASK like a human who stops the moment they're told "that's wrong", not
// one who finishes the wrong workflow first. Claude's sub-agents cannot do this:
// reset off a wrong path, they finish the current workflow before absorbing the
// correction, wasting work on a KNOWN-WRONG thing. Neuron's must never.
//
// MECHANISM (Will's steer, 2026-08-15 "not a hard problem, don't over-engineer"):
// CancellationToken + always-on input.
// (1) INPUT ALWAYS ON every worker holds a CONTROL CHANNEL (the token). The
// coordinator can push a signal at ANY time; the channel is NEVER gated by
// whether the worker is busy. Like a person who keeps hearing while talking.
// (2) COOPERATIVE CANCELLATION the worker CHECKS the token at every step
// boundary (a non-blocking poll, __channel_try_recv). On a signal it STOPS,
// ABSORBS the correction (re-plans) or TERMINATES cleanly with ZERO wasted
// continuation of known-wrong work.
//
// The SAME mechanism serves the conversational speech loop: a mic barge-in simply
// invokes the token on the running render (see peripheral loop 66155ba9).
//
// SAFETY (Rule-4 single-writer / bounded purview): a worker holds its task as a
// PURVIEW (a meaning-plan). It writes ONLY to its own out-channel and its own
// local purview it holds NO global write-lock. So it can be interrupted or
// killed mid-step with NO half-committed global state; the coordinator just
// signals stop. Bounded purviews are what make live interruption clean.
//
// Built on the runtime concurrency primitives:
// __channel_new / __channel_send / __channel_recv / __channel_try_recv (channels)
// __thread_create / __thread_join (workers)
//
// Contrast: the pre-existing channel.el `_channel_worker` loop spawns+joins the
// WHOLE task with no signal check the exact broken pattern. This module checks
// the token BETWEEN every bounded step, so the interrupt latency is one step, not
// one whole workflow.
// low-level worker/thread wrappers (call seed prims directly; no collision) ──
fn _swarm_spawn(fn_name: String, arg: String) -> Int {
return __thread_create(fn_name, arg)
}
fn _swarm_join(tid: Int) -> String {
return __thread_join(tid)
}
// Cancellation token the always-on control channel
// token_new create a cancellation/pause/redirect token for one worker.
// Unbounded so the coordinator NEVER blocks when it signals (always-on input).
fn token_new() -> Int {
return __channel_new(0)
}
// Coordinator ops INVOKE the token (interrupt / re-steer / kill)
// token_signal send a raw signal JSON. Low-level; prefer the named helpers.
fn token_signal(tok: Int, sig_json: String) {
__channel_send(tok, sig_json)
}
// token_interrupt ask the worker to stop at the next step boundary and hold.
fn token_interrupt(tok: Int) {
__channel_send(tok, "{\"sig\":\"PAUSE\"}")
}
// token_pause alias for interrupt: stop stepping, hold state, wait for RESUME.
fn token_pause(tok: Int) {
__channel_send(tok, "{\"sig\":\"PAUSE\"}")
}
// token_resume resume a paused worker on its held plan.
fn token_resume(tok: Int) {
__channel_send(tok, "{\"sig\":\"RESUME\"}")
}
// token_kill terminate the worker cleanly at the next step boundary.
fn token_kill(tok: Int) {
__channel_send(tok, "{\"sig\":\"KILL\"}")
}
// token_redirect re-steer the worker onto a NEW plan MID-TASK. Progress already
// made (completed steps, emitted outputs) is PRESERVED; only the remaining plan is
// replaced. new_steps is a JSON array string, e.g. "[\"step-a\",\"step-b\"]".
fn token_redirect(tok: Int, new_goal: String, new_steps: String) {
let msg: String = "{\"sig\":\"REDIRECT\",\"goal\":\"" +
json_escape_string(new_goal) + "\",\"steps\":" + new_steps + "}"
__channel_send(tok, msg)
}
// Worker ops CHECK the token (cooperative cancellation point)
// token_check non-blocking poll of the token. Returns the pending signal JSON,
// or "" when there is no signal. Called at every step boundary. This is the
// always-on check: it never blocks the worker, so work proceeds at full speed
// until and only until a signal actually arrives.
fn token_check(tok: Int) -> String {
return __channel_try_recv(tok)
}
// token_wait BLOCK until the next signal. Used only while PAUSED (the worker is
// idle, so blocking is correct it is not spinning).
fn token_wait(tok: Int) -> String {
return __channel_recv(tok)
}
// Purview the worker's meaning-plan (held task, resumable)
// purview_new build a bounded meaning-plan the worker carries.
// id purview id
// goal what the worker is trying to achieve (steerable)
// steps JSON array string of step descriptors (the remaining plan)
// A redirect UPDATES this plan; it is not lost.
fn purview_new(id: String, goal: String, steps: String) -> String {
return "{\"id\":\"" + json_escape_string(id) +
"\",\"goal\":\"" + json_escape_string(goal) +
"\",\"steps\":" + steps +
",\"idx\":0,\"completed\":0,\"status\":\"ready\"}"
}
// The interruptible worker loop
// swarm_worker_run the generic natively-interruptible worker.
//
// arg is a JSON object carrying:
// "tok" the cancellation token (control channel handle)
// "out" the worker's own output channel handle (its ONLY write surface)
// "step_fn" name of an El fn (String)->String that performs ONE bounded step
// "purview" the meaning-plan (from purview_new)
//
// The loop: at EVERY iteration it first polls the token (always-on check). Only
// then does it execute ONE bounded step. So a KILL/REDIRECT/PAUSE is absorbed
// within a single step never after finishing the whole (possibly wrong) plan.
//
// step_fn is invoked as a child thread joined immediately: exactly ONE step of
// work is in flight, so the interrupt latency is bounded by a single step.
//
// Returns the final purview JSON (status: complete | killed | redirected).
fn swarm_worker_run(arg: String) -> String {
let tok: Int = str_to_int(json_get(arg, "tok"))
let out_ch: Int = str_to_int(json_get(arg, "out"))
let step_fn: String = json_get(arg, "step_fn")
let purview: String = json_get_raw(arg, "purview")
let pid: String = json_get(purview, "id")
let goal: String = json_get(purview, "goal")
let steps: String = json_get_raw(purview, "steps")
let n: Int = json_array_len(steps)
let idx: Int = 0
let completed: Int = 0
let status: String = "running"
let stop: Int = 0
while stop == 0 {
// ALWAYS-ON INTERRUPT CHECK (the cooperative cancellation point)
let sig: String = token_check(tok)
if str_eq(sig, "") {
// no signal proceed with one bounded step (or finish)
if idx >= n {
let status = "complete"
let stop = 1
} else {
let step: String = json_array_get(steps, idx)
let step_arg: String = "{\"goal\":\"" + json_escape_string(goal) +
"\",\"idx\":" + int_to_str(idx) +
",\"step\":" + step + "}"
let tid: Int = _swarm_spawn(step_fn, step_arg)
let result: String = _swarm_join(tid)
__channel_send(out_ch, result)
let idx = idx + 1
let completed = completed + 1
}
} else {
// a signal arrived ABSORB it immediately, before any more work
let kind: String = json_get(sig, "sig")
if str_eq(kind, "KILL") {
// terminate cleanly commit nothing further. Bounded purview =
// no half-committed global state to unwind.
__channel_send(out_ch, "[KILL absorbed @step " + int_to_str(idx) +
" — stopped, no wasted continuation]")
let status = "killed"
let stop = 1
} else {
if str_eq(kind, "REDIRECT") {
// ABSORB the correction: re-plan onto the new goal/steps.
// Completed steps + emitted outputs are PRESERVED; only the
// remaining plan is replaced.
let kept: Int = completed
let goal = json_get(sig, "goal")
let steps = json_get_raw(sig, "steps")
let n = json_array_len(steps)
let idx = 0
let status = "redirected"
__channel_send(out_ch, "[REDIRECT absorbed @step " +
int_to_str(kept) + " — kept " +
int_to_str(kept) +
" done, re-planned to goal=" + goal + "]")
} else {
if str_eq(kind, "PAUSE") {
// hold state; idle-wait for the next signal
__channel_send(out_ch, "[PAUSE absorbed @step " +
int_to_str(idx) + " — holding plan]")
let status = "paused"
let resumed: Int = 0
while resumed == 0 {
let s2: String = token_wait(tok)
let k2: String = json_get(s2, "sig")
if str_eq(k2, "RESUME") {
__channel_send(out_ch, "[RESUME @step " +
int_to_str(idx) + "]")
let status = "running"
let resumed = 1
} else {
if str_eq(k2, "KILL") {
__channel_send(out_ch, "[KILL absorbed while paused]")
let status = "killed"
let stop = 1
let resumed = 1
} else {
if str_eq(k2, "REDIRECT") {
let goal = json_get(s2, "goal")
let steps = json_get_raw(s2, "steps")
let n = json_array_len(steps)
let idx = 0
__channel_send(out_ch, "[REDIRECT absorbed while paused — goal=" + goal + "]")
let status = "running"
let resumed = 1
}
}
}
}
}
}
}
}
}
// final purview snapshot
let final: String = "{\"id\":\"" + pid +
"\",\"goal\":\"" + json_escape_string(goal) +
"\",\"idx\":" + int_to_str(idx) +
",\"completed\":" + int_to_str(completed) +
",\"planned\":" + int_to_str(n) +
",\"status\":\"" + status + "\"}"
__channel_send(out_ch, "[DONE status=" + status +
" completed=" + int_to_str(completed) + "]")
return final
}
// Coordinator convenience dispatch an interruptible worker
// swarm_dispatch spawn a natively-interruptible worker on a purview.
// step_fn name of the per-step executor (String)->String
// purview the meaning-plan (from purview_new)
// Returns a handle JSON: {"tok":T,"out":O,"tid":D} the coordinator holds `tok`
// to interrupt/redirect/kill live, and drains `out` for results.
fn swarm_dispatch(step_fn: String, purview: String) -> String {
let tok: Int = token_new()
let out_ch: Int = __channel_new(0)
let arg: String = "{\"tok\":" + int_to_str(tok) +
",\"out\":" + int_to_str(out_ch) +
",\"step_fn\":\"" + step_fn +
"\",\"purview\":" + purview + "}"
let tid: Int = _swarm_spawn("swarm_worker_run", arg)
return "{\"tok\":" + int_to_str(tok) +
",\"out\":" + int_to_str(out_ch) +
",\"tid\":" + int_to_str(tid) + "}"
}
-180
View File
@@ -1,180 +0,0 @@
// proof.el Proof of native interruptibility for Neuron-dispatched agents.
//
// Concatenated after runtime/swarm.el (see run.sh). Demonstrates, with raw
// before/after counts, that a dispatched worker interrupted MID-TASK stops
// instantly (not after finishing the wrong workflow), absorbs a redirect
// (re-plans, keeping progress), or terminates cleanly.
//
// Three scenarios on the SAME 12-step plan, SAME per-step cost, SAME signal
// timing (sent ~50ms in after step 3):
// A. BASELINE the broken Claude-style worker: no mid-loop signal check.
// A kill sent at step ~3 is ignored until the whole plan finishes all 12
// steps run = wasted work on a known-wrong task.
// B. INTERRUPTIBLE KILL swarm_worker_run: the kill is absorbed within one
// step stops at ~3, status=killed, ~9 steps of waste AVOIDED.
// C. INTERRUPTIBLE REDIRECT mirrors the live incident (build-python-synth
// native-fetch-render): the correction is absorbed mid-task; the 3 done
// steps are kept; the worker re-plans onto the new goal and finishes it.
// per-step work one bounded unit (~15ms)
fn demo_step(arg: String) -> String {
let goal: String = json_get(arg, "goal")
let idx: String = json_get(arg, "idx")
let step: String = json_get(arg, "step")
sleep_ms(15)
return "did[" + goal + "] step#" + idx + " (" + step + ")"
}
// BASELINE: the broken, non-interruptible worker (Claude-style)
// Same shape as swarm_worker_run BUT it never polls the token during the loop.
// It "finishes the workflow" and only notices the signal at the very end the
// exact pattern Will called out. Reports how many steps it wasted post-signal.
fn broken_worker_run(arg: String) -> String {
let tok: Int = str_to_int(json_get(arg, "tok"))
let out_ch: Int = str_to_int(json_get(arg, "out"))
let step_fn: String = json_get(arg, "step_fn")
let purview: String = json_get_raw(arg, "purview")
let goal: String = json_get(purview, "goal")
let steps: String = json_get_raw(purview, "steps")
let n: Int = json_array_len(steps)
let idx: Int = 0
// NO token check inside the loop this is the bug.
while idx < n {
let step: String = json_array_get(steps, idx)
let step_arg: String = "{\"goal\":\"" + goal + "\",\"idx\":" +
int_to_str(idx) + ",\"step\":" + step + "}"
let tid: Int = __thread_create(step_fn, step_arg)
let result: String = __thread_join(tid)
__channel_send(out_ch, result)
let idx = idx + 1
}
// Only NOW does it look at the control signal too late.
let late: String = token_check(tok)
if str_eq(late, "") {
__channel_send(out_ch, "[no signal]")
} else {
__channel_send(out_ch, "[TOO LATE: absorbed " + json_get(late, "sig") +
" only after running ALL " + int_to_str(n) +
" steps — wasted work]")
}
return "{\"status\":\"ran-to-completion\",\"completed\":" + int_to_str(idx) + "}"
}
// helpers
// drain_count drain out_ch, print each line, return count of real step outputs
// (lines beginning with "did").
fn drain_and_report(out_ch: Int) -> Int {
let done: Int = 0
let did: Int = 0
while done == 0 {
let m: String = __channel_try_recv(out_ch)
if str_eq(m, "") {
let done = 1
} else {
println(" | " + m)
if str_starts_with(m, "did") {
let did = did + 1
}
}
}
return did
}
fn twelve_steps() -> String {
return "[\"s0\",\"s1\",\"s2\",\"s3\",\"s4\",\"s5\",\"s6\",\"s7\",\"s8\",\"s9\",\"s10\",\"s11\"]"
}
fn main() -> Void {
println("=====================================================================")
println(" NATIVE INTERRUPTIBILITY — PROOF (canon 1ca4d3e9)")
println(" plan: 12 bounded steps @ ~15ms; signal sent ~50ms in (≈ after step 3)")
println("=====================================================================")
// A. BASELINE broken, non-interruptible
println("")
println("[A] BASELINE broken worker (no mid-loop signal check) — Claude-style")
let tokA: Int = token_new()
let outA: Int = __channel_new(0)
let pvA: String = purview_new("A", "build-wrong-thing", twelve_steps())
let argA: String = "{\"tok\":" + int_to_str(tokA) + ",\"out\":" + int_to_str(outA) +
",\"step_fn\":\"demo_step\",\"purview\":" + pvA + "}"
let t0A: Int = time_now()
let tidA: Int = __thread_create("broken_worker_run", argA)
sleep_ms(50)
println(" -> coordinator sends KILL at +" + int_to_str(time_now() - t0A) + "ms (≈step 3)")
token_kill(tokA)
let finA: String = __thread_join(tidA)
let elapA: Int = time_now() - t0A
let didA: Int = drain_and_report(outA)
println(" RESULT: executed " + int_to_str(didA) + "/12 steps, wall=" +
int_to_str(elapA) + "ms, final=" + finA)
println(" >> ignored the kill, RAN ALL 12 — ~9 steps of KNOWN-WRONG waste")
// B. INTERRUPTIBLE KILL
println("")
println("[B] INTERRUPTIBLE swarm_worker_run + token_kill")
let pvB: String = purview_new("B", "build-wrong-thing", twelve_steps())
let hB: String = swarm_dispatch("demo_step", pvB)
let tokB: Int = str_to_int(json_get(hB, "tok"))
let outB: Int = str_to_int(json_get(hB, "out"))
let tidB: Int = str_to_int(json_get(hB, "tid"))
let t0B: Int = time_now()
sleep_ms(50)
println(" -> coordinator sends KILL at +" + int_to_str(time_now() - t0B) + "ms (≈step 3)")
token_kill(tokB)
let finB: String = __thread_join(tidB)
let elapB: Int = time_now() - t0B
let didB: Int = drain_and_report(outB)
println(" RESULT: executed " + int_to_str(didB) + "/12 steps, wall=" +
int_to_str(elapB) + "ms, final=" + finB)
println(" >> STOPPED within one step of the signal — no wasted continuation")
// C. INTERRUPTIBLE REDIRECT (the live incident)
println("")
println("[C] INTERRUPTIBLE redirect mid-task: build-python-synth -> native-fetch-render")
let pvC: String = purview_new("C", "build-python-synth-renderer", twelve_steps())
let hC: String = swarm_dispatch("demo_step", pvC)
let tokC: Int = str_to_int(json_get(hC, "tok"))
let outC: Int = str_to_int(json_get(hC, "out"))
let tidC: Int = str_to_int(json_get(hC, "tid"))
let t0C: Int = time_now()
sleep_ms(50)
println(" -> coordinator REDIRECTS at +" + int_to_str(time_now() - t0C) + "ms (≈step 3)")
token_redirect(tokC, "native-fetch-render", "[\"fetch\",\"realize\",\"cohere\"]")
let finC: String = __thread_join(tidC)
let elapC: Int = time_now() - t0C
let didC: Int = drain_and_report(outC)
println(" RESULT: executed " + int_to_str(didC) + " steps total, wall=" +
int_to_str(elapC) + "ms, final=" + finC)
println(" >> absorbed the correction mid-task: kept early progress,")
println(" re-planned onto native-fetch-render, finished the RIGHT plan")
// D. INTERRUPTIBLE PAUSE -> RESUME (hold state, then continue)
println("")
println("[D] INTERRUPTIBLE pause mid-task, hold state, then resume to finish")
let pvD: String = purview_new("D", "long-render", twelve_steps())
let hD: String = swarm_dispatch("demo_step", pvD)
let tokD: Int = str_to_int(json_get(hD, "tok"))
let outD: Int = str_to_int(json_get(hD, "out"))
let tidD: Int = str_to_int(json_get(hD, "tid"))
let t0D: Int = time_now()
sleep_ms(50)
println(" -> coordinator PAUSES at +" + int_to_str(time_now() - t0D) + "ms (≈step 3)")
token_pause(tokD)
sleep_ms(60)
println(" -> worker held idle for ~60ms; coordinator RESUMES at +" +
int_to_str(time_now() - t0D) + "ms")
token_resume(tokD)
let finD: String = __thread_join(tidD)
let didD: Int = drain_and_report(outD)
println(" RESULT: executed " + int_to_str(didD) + "/12 steps, final=" + finD)
println(" >> paused on the spot, held its plan, resumed and finished it")
println("")
println("=====================================================================")
println(" A ran all 12 wrong steps. B stopped at ~3. C re-planned at ~3.")
println(" Same plan, same timing, same signal — only the interruptible worker")
println(" stops the instant it's told, like a human. QED.")
println("=====================================================================")
}
-56
View File
@@ -1,56 +0,0 @@
#!/usr/bin/env bash
# run.sh — build and run the native-interruptibility proof.
#
# Concatenates runtime/swarm.el + proof.el into one translation unit (the El
# multi-file strategy — elc does not resolve cross-directory imports), compiles
# via the canonical elc, links against el_runtime.c, and runs.
#
# A tiny forward-declaration prelude is prepended to the generated C for the
# __channel_* seed primitives (they are defined in el_runtime.c but not declared
# in el_runtime.h). This touches nothing shared — it is local to this build.
set -uo pipefail
cd "$(dirname "$0")"
EL_HOME="${EL_HOME:-$(cd ../.. && pwd)}"
ELC="${EL_HOME}/dist/platform/elc"
RT="${EL_HOME}/el-compiler/runtime"
SWARM="${EL_HOME}/runtime/swarm.el"
OSSL="$(brew --prefix openssl@3 2>/dev/null || brew --prefix openssl 2>/dev/null || echo /usr/local)"
LDF=(); [ -d "${OSSL}/lib" ] && LDF=(-L"${OSSL}/lib")
BUILD="$(mktemp -d -t swarmproof.XXXXXX)"
trap 'rm -rf "${BUILD}"' EXIT
if [ ! -x "${ELC}" ]; then echo "elc not found at ${ELC}" >&2; exit 1; fi
# 1. Concatenate library + proof into one .el
cat "${SWARM}" proof.el > "${BUILD}/combined.el"
# 2. elc emit -> C
if ! "${ELC}" "${BUILD}/combined.el" > "${BUILD}/body.c" 2>"${BUILD}/elc.err"; then
echo "elc FAILED:"; sed 's/^/ /' "${BUILD}/elc.err"; exit 1
fi
# 3. Prepend forward-decl prelude for the __channel_* seed primitives
cat > "${BUILD}/prog.c" <<'PRELUDE'
#include <stdint.h>
typedef int64_t el_val_t;
el_val_t __channel_new(el_val_t);
el_val_t __channel_send(el_val_t, el_val_t);
el_val_t __channel_recv(el_val_t);
el_val_t __channel_try_recv(el_val_t);
el_val_t __channel_close(el_val_t);
PRELUDE
cat "${BUILD}/body.c" >> "${BUILD}/prog.c"
# 4. cc link
if ! cc -O2 -Wno-implicit-function-declaration -I "${RT}" "${LDF[@]}" \
"${BUILD}/prog.c" "${RT}/el_runtime.c" \
-lcurl -lssl -lcrypto -lpthread -lm -o "${BUILD}/proof" 2>"${BUILD}/cc.err"; then
echo "cc FAILED:"; tail -20 "${BUILD}/cc.err"; exit 1
fi
# 5. run
"${BUILD}/proof"