el/engram/test/run_interoception_p1.sh

#!/usr/bin/env bash
# M-INTEROCEPTION P1 gate: two-threshold consolidation (ENGRAM_CONSOLIDATION).
# Throwaway HOME + /tmp only. Never touches ~/.neuron or :8742.
set -u
HERE="$(cd "$(dirname "$0")" && pwd)"
RT="$HERE/../../lang/runtime/el_runtime.c"
ST="$HERE/../../lang/runtime/engram_store.c"
GEO="$HERE/../../lang/runtime/engram_geometry.c"
VIDX="$HERE/../../lang/runtime/engram_vindex.c"
INC="$HERE/../../lang/runtime"
WORK="$(mktemp -d /tmp/engram-p1-XXXXXX)"
export HOME="$WORK/home"; mkdir -p "$HOME"
unset ENGRAM_STORE ENGRAM_CONSOLIDATION ENGRAM_CONSOL_CONN_MIN ENGRAM_CONSOL_PERM_MIN ENGRAM_CONSOL_WM_TOPK
fail=0

echo "== compile =="
gcc -O1 -std=c11 -I "$INC" "$HERE/test_interoception_p1_consol.c" "$RT" "$ST" "$GEO" "$VIDX" \
    -lcurl -lm -o "$WORK/p1" 2>"$WORK/cc.log" || { echo "COMPILE FAILED"; cat "$WORK/cc.log"; rm -rf "$WORK"; exit 1; }

echo
echo "== (a) HEADLINE: hebb accrual curve over N co-activations (flag OFF, pure trunk) =="
D="$WORK/a"; mkdir -p "$D"
( unset ENGRAM_CONSOLIDATION; "$WORK/p1" accrual "$D" ) >"$WORK/accrual.txt" 2>&1 || { echo "FAIL accrual run"; fail=1; }
python3 - "$WORK/accrual.txt" <<'PY'
import json,sys,re
rows=[]
for line in open(sys.argv[1]):
    m=re.match(r'SAMPLE (\d+) (\{.*\})',line.strip())
    if not m: continue
    n=int(m.group(1)); j=json.loads(m.group(2))
    hm=j.get("hebb_max",0.0); hc=j.get("hebb_cand_max",0.0)
    rows.append((n,hm,hc))
print("    N       hebb_max     1-0.9999^N (predicted EWMA)")
rc=0
for n,hm,hc in rows:
    pred=1-0.9999**n
    print(f"    {n:<7} {hm:<12.6g} {pred:.6g}")
# assertions: monotonic rise, starts near ETA, tracks EWMA prediction
first=rows[0]; last=rows[-1]
def check(c,m):
    global rc; print(("  PASS: " if c else "  FAIL: ")+m);
    if not c: rc=1
check(abs(first[1]-0.0001)<5e-5, f"first sample hebb ~= ETA 0.0001 (got {first[1]:.6g})")
check(all(rows[i][1] <= rows[i+1][1]+1e-9 for i in range(len(rows)-1)), "hebb_max is monotonically non-decreasing over N")
check(last[1] > first[1]*50, f"hebb accrues substantially by N={last[0]} (got {last[1]:.4g} vs {first[1]:.4g})")
# EWMA fit: measured should be within 25% of 1-0.9999^N at the mid samples
mid=[r for r in rows if 100<=r[0]<=2000]
ok=all(abs(hm-(1-0.9999**n))/(1-0.9999**n) < 0.25 for n,hm,hc in mid)
check(ok, "measured curve tracks the 1-0.9999^N EWMA prediction within 25% (co-activation P~1)")
sys.exit(rc)
PY
[ $? -ne 0 ] && fail=1

echo
echo "== (b) CONNECTION threshold: strong ISE wires to wm_top, weak ISE wires nothing (flag ON) =="
D="$WORK/b"; mkdir -p "$D"
( export ENGRAM_CONSOLIDATION=1; "$WORK/p1" connect "$D" ) >"$WORK/connect.txt" 2>&1 || { echo "FAIL connect run"; fail=1; }
cat "$WORK/connect.txt" | sed 's/^/    /'
python3 - "$WORK/connect.txt" "$D/connect.json" <<'PY'
import json,sys,re
txt=open(sys.argv[1]).read()
g=json.load(open(sys.argv[2]))
def field(k):
    m=re.search(rf'{k} (\S+)',txt); return m.group(1) if m else None
sid=field("ISE_STRONG_ID"); wid=field("ISE_WEAK_ID")
m=re.search(r'EDGES before=(\d+) after_strong=(\d+) after_weak=(\d+)',txt)
before,aftS,aftW=int(m.group(1)),int(m.group(2)),int(m.group(3))
rc=0
def check(c,mm):
    global rc; print(("  PASS: " if c else "  FAIL: ")+mm)
    if not c: rc=1
strong_edges=[e for e in g["edges"] if e["from_id"]==sid and e["relation"]=="hebbian-associate"]
weak_edges=[e for e in g["edges"] if e["from_id"]==wid]
check(aftS>before, f"strong ISE formed connection edges ({before} -> {aftS})")
check(aftW==aftS, f"weak ISE formed NO edges ({aftS} -> {aftW})")
check(len(strong_edges)>=1, f"strong ISE has {len(strong_edges)} hebbian-associate edge(s) to wm_top")
check(all('consolidated-from-ISE' in (e.get('metadata') or '') for e in strong_edges),
      "connection edges are provenance-tagged consolidated-from-ISE (reversible)")
check(len(weak_edges)==0, "weak ISE (below connection bar) has zero outgoing edges")
# targets must be the WM-top nodes (hebb-a / hebb-b), not distractors
tgt_labels=set()
byid={n["id"]:n for n in g["nodes"]}
for e in strong_edges:
    t=byid.get(e["to_id"]);
    if t: tgt_labels.add(t.get("label"))
print(f"    connection targets: {sorted(tgt_labels)}")
check(tgt_labels.issubset({"hebb-a","hebb-b"}) and len(tgt_labels)>=1,
      f"connections point at the wm_top nodes {sorted(tgt_labels)}")
sys.exit(rc)
PY
[ $? -ne 0 ] && fail=1

echo
echo "== (c) PERMANENCE threshold: promoted node survives 48h prune, ephemeral is swept (flag ON) =="
D="$WORK/c"; mkdir -p "$D"
( export ENGRAM_CONSOLIDATION=1 ENGRAM_CONSOL_PERM_MIN=-1000; "$WORK/p1" perm "$D" ) >"$WORK/perm.txt" 2>&1 || { echo "FAIL perm run"; fail=1; }
cat "$WORK/perm.txt" | sed 's/^/    /'
python3 - "$WORK/perm.txt" <<'PY'
import sys,re,json
txt=open(sys.argv[1]).read()
rc=0
def check(c,m):
    global rc; print(("  PASS: " if c else "  FAIL: ")+m)
    if not c: rc=1
prom=int(re.search(r'PROMOTED (\d+)',txt).group(1))
m=re.search(r'NODES before=(\d+) after=(\d+) removed=(\d+)',txt)
before,after,removed=int(m.group(1)),int(m.group(2)),int(m.group(3))
dur=re.search(r'DURABLE_NODE (\{.*\})',txt).group(1)
eph=re.search(r'EPHEMERAL_NODE (\{.*\})',txt).group(1)
durj=json.loads(dur); ephj=json.loads(eph)
check(prom==1, "engram_consolidate_permanence promoted the node (returned 1)")
check(before==2 and after==1 and removed==1, f"exactly one node pruned ({before}->{after}, removed={removed})")
check(durj.get("id")=="ise-durable", "durable node SURVIVED the 48h telemetry prune")
check('consolidated-from-ISE' in (durj.get("metadata") or ''), "durable node carries reversible provenance marker")
check(ephj=={} or not ephj.get("id"), "ephemeral (non-permanent) ISE was swept")
sys.exit(rc)
PY
[ $? -ne 0 ] && fail=1

echo
echo "== (d) OFF path byte-identical: ISE creation forms no edges, permanence is a no-op =="
D="$WORK/d"; mkdir -p "$D"
( unset ENGRAM_CONSOLIDATION; "$WORK/p1" offcheck "$D" ) >"$WORK/off.txt" 2>&1
rcoff=$?
cat "$WORK/off.txt" | sed 's/^/    /'
[ $rcoff -eq 0 ] && echo "  PASS: flag OFF — ISE creation added 0 edges and permanence returned 0" \
                 || { echo "  FAIL: OFF path changed behavior"; fail=1; }

echo
echo "== ASan+UBSan (connect + perm + accrual-short) =="
gcc -O1 -g -std=c11 -fsanitize=address,undefined -fno-sanitize-recover=undefined \
    -I "$INC" "$HERE/test_interoception_p1_consol.c" "$RT" "$ST" "$GEO" "$VIDX" \
    -lcurl -lm -o "$WORK/p1.san" 2>"$WORK/san_cc.log" || { echo "SAN COMPILE FAILED"; tail -25 "$WORK/san_cc.log"; fail=1; }
if [ -x "$WORK/p1.san" ]; then
  export ASAN_OPTIONS=detect_leaks=0
  DS="$WORK/san"; mkdir -p "$DS"
  ( export ENGRAM_CONSOLIDATION=1 ENGRAM_CONSOL_PERM_MIN=-1000; "$WORK/p1.san" connect "$DS" ) >/dev/null 2>"$WORK/san_run.log"
  ( export ENGRAM_CONSOLIDATION=1 ENGRAM_CONSOL_PERM_MIN=-1000; "$WORK/p1.san" perm "$DS" ) >/dev/null 2>>"$WORK/san_run.log"
  if grep -qiE 'runtime error|AddressSanitizer|Sanitizer|ERROR: ' "$WORK/san_run.log"; then
    echo "  FAIL: sanitizer findings:"; grep -iE 'runtime error|Sanitizer|ERROR' "$WORK/san_run.log" | head; fail=1
  else echo "  ok: ASan+UBSan clean"; fi
fi

echo
if [ "$fail" -eq 0 ]; then echo "====== P1 CONSOLIDATION GATE: PASS ======"; else echo "====== P1 CONSOLIDATION GATE: FAIL ======"; fi
rm -rf "$WORK"
exit $fail