el/ingest/src/ingest.el

// 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() 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() never learns whether a
// chunk is plain text or a base64-encoded raw-byte window; every chunk is
// handled identically either way.
fn transduce(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()'s generic scan already looks for, so
// transduce() 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() 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() 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(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() 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(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(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(). 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)