Neuron
688f24b4c1
El SDK CI - dev / build-and-test (pull_request) Failing after 14m6s
ingest.el's transduce() was renamed to transduce_manifold() earlier the same day on the reasoning that it 'was never signal->geometry -- it chunks already-extracted content and PACKS it into a node+edge manifold, one layer up, and it had taken the name that belongs to the primitive underneath it.' That reasoning was backwards. Producing a node+edge manifold is not a layer above transduction, it IS transduction. Signal -> one vector is the operation underneath, and its name is geometry. The layer doing it right was renamed out of the way so the layer doing it wrong could have the name. With the primitive corrected to return a Manifold, the two layers do the same kind of thing and the inversion dissolves. What is left is a real distinction about MODALITY, not layering: transduce() dispatches to a realizer that knows its modality and can name its components; transduce_bytes() is the opaque-bytes realizer, the decomposition available to a reader that knows nothing about what it is reading. It still yields components and relations, which is why it is transduction and not packing -- it just cuts on byte boundaries, so its components are positional rather than meaningful. That is a limitation of this realizer, not the definition of the operation. Renamed by modality rather than demoted by layer. A distinct symbol is still mechanically required: reusing transduce here is a conflicting-types error the moment ingest.c links el_runtime.c. lang/examples/transduce.el asserted #144's contract and would now fail, so it is replaced by the decomposition worked example: transduce a chord, persist the five components and six relations as real nodes and edges, read each part's geometry back off its own node, and ground one part while its sibling is demonstrably untouched.
762 lines
36 KiB
EmacsLisp
762 lines
36 KiB
EmacsLisp
// ingest.el — the native EL AFFERENT INGEST ORGAN
|
|
//
|
|
// The source-polymorphic ingest(source) primitive: point it at a directory,
|
|
// file, url, llm-query, 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.
|
|
//
|
|
// transduce_bytes() is THE single mechanism — one function, polymorphic, with no
|
|
// content-type branch inside it. It does not ask whether a payload is
|
|
// prose, structured data, or raw/opaque bytes (audio, or anything else);
|
|
// it runs one boundary-scan-with-fixed-window-fallback chunking algorithm
|
|
// and one dedup mechanism on whatever bytes it's handed, unconditionally.
|
|
// Any deeper structure a payload might have (shared fields, relationships,
|
|
// what a chunk of audio "means") is NOT interpreted here — that's left
|
|
// entirely to the engram's own mechanisms (embedding, spreading activation,
|
|
// dedup) acting on this real geometry over time. This organ claims zero
|
|
// semantic understanding of any payload it transduces.
|
|
//
|
|
// 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) + "\""
|
|
}
|
|
|
|
// ═══════════════════════════════════════════════════════════════════════════
|
|
// 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.
|
|
// ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
// TRANSDUCE — the single mechanism. Takes ANY payload (prose, structured
|
|
// data, raw/opaque bytes — audio, whatever) as one opaque string and turns
|
|
// it into a discrete manifold: nodes + internal edges. There is no
|
|
// content-type branch anywhere in this function. It never asks "is this
|
|
// text," "is this JSON," "is this audio" — it runs ONE algorithm on the
|
|
// bytes it is given, unconditionally:
|
|
//
|
|
// 1. BOUNDARY SCAN — split on "\n\n". This is a property of the bytes
|
|
// (does a blank-line-style marker occur in them, yes or no), not a
|
|
// classification of what the content IS. Prose paragraphs split on it
|
|
// because that's how prose is typically written; that's a fact about
|
|
// the bytes, not a rule this function knows about prose. Anything else
|
|
// that happens to contain the same marker splits on it too, and
|
|
// anything that doesn't, doesn't — same code path either way.
|
|
// 2. FIXED-WINDOW FALLBACK — if step 1 found no boundary (0 or 1 non-empty
|
|
// piece), the payload is cut into fixed-size windows instead. Same
|
|
// chunk-per-node, edge-per-adjacency structure as step 1 produces; only
|
|
// the source of the cut point differs.
|
|
//
|
|
// Every resulting chunk becomes its own node (never one blob), wired with
|
|
// the same edges regardless of what's inside a chunk: root -contains->
|
|
// chunk, chunk -precedes-> next chunk, and — if a chunk happens to start
|
|
// with "#" — most-recent-heading -section_of-> chunk. That "#" check is a
|
|
// structural marker (a fact about a chunk's first byte), not a decision
|
|
// about whether this run is "the text case": chunks from any payload that
|
|
// never happen to start with "#" simply never trigger it.
|
|
//
|
|
// Dedup is the existing, fully generic mechanism (find_existing_by_content,
|
|
// via merge_manifold downstream of merge_packed) applied uniformly to every
|
|
// chunk from every payload — there is no separate "structured" dedup path.
|
|
// Any deeper structure that might exist inside a payload (shared fields,
|
|
// repeated records, relationships) is NOT pre-computed here; that's left to
|
|
// the engram's own mechanisms (embedding, spreading activation, dedup)
|
|
// acting on this geometry over time, which is the whole point of handing it
|
|
// raw bytes instead of a hand-coded interpretation of them.
|
|
//
|
|
// Byte-safety note: `source` must already be a string this function can
|
|
// safely str_split/str_slice. Protecting it from silent truncation (El
|
|
// strings are NUL-unsafe under strlen-based ops; fs_read()'s result
|
|
// truncates at the first embedded NUL, which is routine in real binary
|
|
// bytes) is a MECHANICAL fidelity concern that belongs to whatever produced
|
|
// `source` (see ingest_file's file_source_string below) — not a
|
|
// content-type judgment made in here. transduce_bytes() never learns whether a
|
|
// chunk is plain text or a base64-encoded raw-byte window; every chunk is
|
|
// handled identically either way.
|
|
// NAMING, CORRECTED 2026-08-16 (second pass). This function was renamed
|
|
// `transduce` -> `transduce_bytes` earlier the same day, on the reasoning
|
|
// that it "was never signal->geometry — it chunks already-extracted content
|
|
// and PACKS it into a node+edge manifold, one layer up, and it had taken the
|
|
// name that belongs to the primitive underneath it."
|
|
//
|
|
// THAT REASONING WAS BACKWARDS, and it is worth recording why rather than
|
|
// quietly re-renaming. Producing a node+edge manifold is not a layer above
|
|
// transduction — it IS transduction. Transduction is not conversion. When you
|
|
// take in music you do not store the song as one discrete geometry; you break
|
|
// it into its component parts and store the geometry of each along with the
|
|
// relations between them. The song is the structure of those relations.
|
|
// Signal -> one vector is the operation UNDERNEATH transduction, and its name
|
|
// is encoding, or geometry. So the layer that was doing it right got renamed
|
|
// out of the way so the layer doing it wrong could have the name.
|
|
//
|
|
// The primitive has since been corrected: `transduce(signal, modality)` now
|
|
// returns a Manifold — components plus relations — not a Geometry
|
|
// (el_runtime.c, "Manifold"). The two layers are therefore doing the SAME KIND
|
|
// of thing, and the inversion dissolves rather than needing to be re-argued.
|
|
//
|
|
// What is left is a real distinction, and it is about MODALITY, not layering:
|
|
//
|
|
// * `transduce(signal, modality)` dispatches to a realizer that KNOWS the
|
|
// modality and can name its components — for audio: pitch, interval,
|
|
// rhythm, harmonic function.
|
|
// * `transduce_bytes` below is the OPAQUE-BYTES realizer: the decomposition
|
|
// available to a reader that knows nothing about what it is reading. It
|
|
// still yields components and relations (chunk nodes; contains / precedes
|
|
// / section_of edges), which is why it is transduction and not packing. It
|
|
// just cuts on the only structure visible without understanding — byte
|
|
// boundaries — so its components are positional rather than meaningful.
|
|
// That is a LIMITATION of this realizer, not the definition of the
|
|
// operation.
|
|
//
|
|
// The name is suffixed by its modality, not demoted to a lesser layer. Keeping
|
|
// a distinct symbol is also still mechanically required: every El `fn name`
|
|
// compiles to a global C symbol, so reusing `transduce` here is a hard
|
|
// `conflicting types` error the moment ingest.c links el_runtime.c.
|
|
//
|
|
// WHERE THIS SHOULD GO: this function should become a registered realizer
|
|
// returning a real Manifold, so ingest rides the same primitive as every other
|
|
// modality instead of carrying a parallel implementation. Not done here.
|
|
// Nothing about this function's behaviour changed in this pass.
|
|
fn transduce_bytes(nodes: [String], edges: [String], source: String,
|
|
prov: String, ground: String, steward: String,
|
|
root_lid: String, root_title: String) -> [String] {
|
|
let tagbase: String = "prov:" + prov + " ground:" + ground + " steward:" + steward
|
|
nodes = el_list_append(nodes, mk_node(root_lid, "source: " + root_title,
|
|
"Concept", "Semantic", "0.6", "0.6", "0.9", tagbase + " kind:source"))
|
|
|
|
// step 1: universal boundary scan
|
|
let boundary_parts: [String] = str_split(source, "\n\n")
|
|
let chunks: [String] = el_list_empty()
|
|
let bp_n: Int = el_list_len(boundary_parts)
|
|
let bp_i: Int = 0
|
|
while bp_i < bp_n {
|
|
let piece: String = str_trim(el_list_get(boundary_parts, bp_i))
|
|
if !str_eq(piece, "") { chunks = el_list_append(chunks, piece) }
|
|
bp_i = bp_i + 1
|
|
}
|
|
|
|
// step 2: no boundary found -> fixed-size windows over the whole
|
|
// payload. 4096 chars/node: low kilobytes — big enough to keep node
|
|
// count sane on a large unbroken payload, small enough that each node
|
|
// stays a legible, individually embeddable/dedupable unit rather than
|
|
// one giant blob.
|
|
if el_list_len(chunks) <= 1 {
|
|
chunks = el_list_empty()
|
|
let total: Int = str_len(source)
|
|
let win: Int = 4096
|
|
let off: Int = 0
|
|
while off < total {
|
|
let endp: Int = if off + win < total { off + win } else { total }
|
|
let piece: String = str_slice(source, off, endp)
|
|
if !str_eq(piece, "") { chunks = el_list_append(chunks, piece) }
|
|
off = off + win
|
|
}
|
|
}
|
|
|
|
let nc: Int = el_list_len(chunks)
|
|
let ci: Int = 0
|
|
let last_chunk: String = ""
|
|
let last_heading: String = ""
|
|
while ci < nc {
|
|
let raw: String = el_list_get(chunks, ci)
|
|
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:chunk" }
|
|
nodes = el_list_append(nodes, mk_node(lid, raw,
|
|
"Knowledge", "Semantic", "0.55", "0.55", "0.9", tagbase + " " + kind))
|
|
// containment: 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
|
|
}
|
|
|
|
// package both lists into one, "N"/"E"-prefixed (see merge_packed).
|
|
let packed: [String] = el_list_empty()
|
|
let pn_i: Int = 0
|
|
let pn_n: Int = el_list_len(nodes)
|
|
while pn_i < pn_n { packed = el_list_append(packed, "N" + el_list_get(nodes, pn_i)) pn_i = pn_i + 1 }
|
|
let pe_i: Int = 0
|
|
let pe_n: Int = el_list_len(edges)
|
|
while pe_i < pe_n { packed = el_list_append(packed, "E" + el_list_get(edges, pe_i)) pe_i = pe_i + 1 }
|
|
return packed
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
// default ingestion grounding; overridable per-invocation via INGEST_GROUND.
|
|
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
|
|
}
|
|
|
|
// Mechanical fidelity guard — NOT a content-type test. fs_read()'s el_val_t
|
|
// result truncates at the first embedded NUL byte under El's strlen-based
|
|
// string ops (see fs_size's doc comment in runtime/el_runtime.h); comparing
|
|
// its length against fs_size() (a real stat()-based byte count) is a
|
|
// technical fact about whether the string channel captured the file intact
|
|
// — computed the same way for a poem, a JSON file, or a WAV, and saying
|
|
// nothing about what the file IS. When the counts agree, `text` is
|
|
// trustworthy verbatim. When they don't (silent truncation happened),
|
|
// rebuild the payload as base64-encoded fixed-size windows read directly
|
|
// off disk (fs_read_b64_chunk — binary-safe in C), joined with the same
|
|
// "\n\n" boundary marker transduce_bytes()'s generic scan already looks for, so
|
|
// transduce_bytes() sees one ordinary boundary-delimited payload and runs its one
|
|
// algorithm on it exactly as it would on prose — it never learns that a
|
|
// fidelity problem occurred upstream, let alone why.
|
|
fn file_source_string(path: String, text: String, real_size: Int) -> String {
|
|
if real_size <= 0 { return text }
|
|
if str_len(text) == real_size { return text }
|
|
// 3072 raw bytes -> 4096 base64 chars (3 divides evenly into base64's
|
|
// 3-byte/4-char ratio); keeps each resulting node's content a clean,
|
|
// bounded, low-kilobytes unit, same order of magnitude as the fixed
|
|
// fallback window in transduce_bytes() itself.
|
|
let win: Int = 3072
|
|
let out: String = ""
|
|
let off: Int = 0
|
|
let first: Bool = true
|
|
while off < real_size {
|
|
let chunk_b64: String = fs_read_b64_chunk(path, off, win)
|
|
if str_eq(chunk_b64, "") {
|
|
off = real_size
|
|
} else {
|
|
let sep: String = if first { "" } else { "\n\n" }
|
|
out = out + sep + chunk_b64
|
|
first = false
|
|
off = off + win
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// ingest one file -> report JSON. Uniform for every file regardless of
|
|
// extension or content — transduce_bytes() decides nothing about content-type, so
|
|
// neither does this function; it only decides whether the raw bytes made it
|
|
// through the read intact (file_source_string), which is a fidelity
|
|
// question, not a format one.
|
|
fn ingest_file(path: String) -> String {
|
|
let real_size: Int = fs_size(path)
|
|
let text: String = fs_read(path)
|
|
let source: String = file_source_string(path, text, real_size)
|
|
if str_eq(source, "") {
|
|
return "{\"error\":\"empty or unreadable\",\"path\":" + j_q(path) + "}"
|
|
}
|
|
let prov: String = "file:" + path
|
|
let packed: [String] = transduce_bytes(el_list_empty(), el_list_empty(),
|
|
source, prov, default_ground(), default_steward(),
|
|
"doc:" + basename(path), basename(path))
|
|
return merge_packed(packed)
|
|
}
|
|
|
|
// ingest a directory: walk one level, ingest every file found, aggregate.
|
|
// No extension filter — transduce_bytes() handles any payload uniformly now, so
|
|
// there is no content-type gate at the directory boundary either.
|
|
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
|
|
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, hand the body straight to transduce (faithful
|
|
// extraction of what's there — no interpretation of what it is)
|
|
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_bytes(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_bytes(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
|
|
// ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
// INGEST_KIND selects an ACQUISITION mechanism only — dir/file/url/llm/
|
|
// stream — i.e. which RPC shape to use to go get the bytes (walk a
|
|
// directory, open a file, fetch a URL, query an LLM, read a turn-stream).
|
|
// That is a genuinely unavoidable choice at the process-entry boundary
|
|
// (nothing about the string "/tmp/x" tells you whether it's a file to read
|
|
// or a stream to read line-by-line, or distinguishes an LLM query from a
|
|
// path), so it cannot be dropped the way content-type dispatch was.
|
|
// It is NOT a content-type flag: it says nothing about what's inside the
|
|
// bytes once fetched, and none of the five ingest_* functions it selects
|
|
// among interpret their payload differently by content shape anymore —
|
|
// they all hand off to the single, format-agnostic transduce_bytes(). The old
|
|
// "structured" value (a caller-declared alias for "file", used only to hint
|
|
// the now-removed JSON-vs-prose branch) is gone along with that branch.
|
|
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, "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)
|