mempalace-toolkit/scripts/test-device-stamp.sh

#!/usr/bin/env bash
# test-device-stamp.sh — rule tests for bin/mempalace-device-stamp.
#
# The stamper writes directly to the SHARED fleet palace, so its rules must
# never be tried out there: a wrong rule invents device names that then have to
# be un-invented across 30k rows. This builds a THROWAWAY palace under a fake
# $HOME and asserts each rule against it.
#
# Every HOST: fixture below is verbatim from the real palace, collected by
# running `mempalace-device-stamp --dry-run` against it on 2026-08-25 — which is
# how we learned that `HOST:` was already in use with two OTHER grammars:
#   * a composite fingerprint, HOST:<device>.<container>.<image>.<pi-version>
#   * a different SENSE entirely: HOST:exec.via.ssh-controlmaster->alpserv-2,
#     meaning "the box I was executing on", not "the box that wrote this".
# Hence the KNOWN_DEV validation the tests below pin down.
#
# Usage: scripts/test-device-stamp.sh      (exit 0 = all rules behave)
set -euo pipefail

REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
STAMPER="$REPO_ROOT/bin/mempalace-device-stamp"
FAKE="$(mktemp -d)"
trap 'rm -rf "$FAKE"' EXIT

python3 -m py_compile "$STAMPER"

# Devices exist iff they have a feed inbox — that is what KNOWN_DEV reads.
mkdir -p "$FAKE/.mempalace/palace" \
         "$FAKE/mempalace-feed/tor-ms22" \
         "$FAKE/mempalace-feed/emb-7kj4vr4g" \
         "$FAKE/mempalace-feed/mbp-m1-2020"

FAKE="$FAKE" python3 - <<'PYEOF'
import os, sqlite3
db = os.path.join(os.environ['FAKE'], '.mempalace/palace/chroma.sqlite3')
c = sqlite3.connect(db)
c.execute("create table embedding_metadata (id text, key text, string_value text, primary key(id,key))")

docs = {
    # --- the marker the pi bridge writes (extensions/pi/mempalace.ts) --------
    'new_convention':   'HOST:tor-ms22|SESSION:2026-08-25|did.things',
    # --- the older composite grammar already in the palace ------------------
    'composite_v183':   'HOST:emb-7kj4vr4g.f1d3c3f89e3e.v1.8.3.pi0.84.2|SESSION:x',
    'composite_om':     'HOST:emb-7kj4vr4g.f1d3c3f89e3e.pi0.84.2.om3.0.4.bedrock|SESSION:x',
    'composite_devbox': 'HOST:emb-7kj4vr4g.devbox.pi0.84.2.om.699ccc7|SESSION:x',
    # --- must NOT invent a device ------------------------------------------
    'container_only':   'HOST:f1d3c3f89e3e.pi0.84.2|SESSION:x',   # container id, not a device
    'container_bare':   'HOST:2efe2b06f480|SESSION:x',            # ditto, and it changes per recreate
    'garbage_str':      'HOST:str|SESSION:x',
    'other_sense':      'x|HOST:exec.via.ssh-controlmaster->alpserv-2(Alpine,user=joakim)|y',
    'prose':            'we discussed the HOST:tor-ms22 marker convention',  # not at a field boundary
}
rows = []
for k, doc in docs.items():
    rows += [(k, 'chroma:document', doc), (k, 'parent_drawer_id', 'd_' + k), (k, 'added_by', 'pi')]

# Chunk propagation: one write call => one origin. Marker lands in chunk 0 only.
rows += [('chunk0', 'chroma:document', 'HOST:tor-ms22|SESSION:y'), ('chunk0', 'parent_drawer_id', 'multi')]
rows += [('chunk1', 'chroma:document', 'continuation, no marker'), ('chunk1', 'parent_drawer_id', 'multi')]
rows += [('chunk2', 'chroma:document', 'more continuation'),       ('chunk2', 'parent_drawer_id', 'multi')]
# An existing stamp must survive untouched and seed its blank sibling.
rows += [('keep0', 'device', 'mbp-m1-2020'), ('keep0', 'device_source', 'inbox_path'),
         ('keep0', 'parent_drawer_id', 'kept'), ('keep0', 'chroma:document', 'x')]
rows += [('keep1', 'parent_drawer_id', 'kept'), ('keep1', 'chroma:document', 'y')]
# The edge convention on a normal drawer.
rows += [('edge0', 'added_by', 'pi@tor-ms22'), ('edge0', 'chroma:document', 'z'),
         ('edge0', 'parent_drawer_id', 'edged')]
# Defaults that carry no device: must stay blank rather than guess.
rows += [('dflt0', 'added_by', 'checkpoint'), ('dflt0', 'chroma:document', 'q'),
         ('dflt0', 'parent_drawer_id', 'dflt')]

c.executemany("insert into embedding_metadata values (?,?,?)", rows)
c.commit()
PYEOF

# --dry-run must write nothing.
HOME="$FAKE" python3 "$STAMPER" --dry-run >/dev/null
if [ -n "$(FAKE="$FAKE" python3 -c "
import os,sqlite3
db=os.path.join(os.environ['FAKE'],'.mempalace/palace/chroma.sqlite3')
print(''.join(r[0] for r in sqlite3.connect(db).execute(
  \"select id from embedding_metadata where key='device' and id!='keep0'\")))")" ]; then
	echo "FAIL: --dry-run wrote stamps" >&2
	exit 1
fi

HOME="$FAKE" python3 "$STAMPER" >/dev/null
SECOND="$(HOME="$FAKE" python3 "$STAMPER")"   # idempotence: a second pass adds nothing

FAKE="$FAKE" SECOND="$SECOND" python3 - <<'PYEOF'
import os, sqlite3, sys
db = os.path.join(os.environ['FAKE'], '.mempalace/palace/chroma.sqlite3')
c = sqlite3.connect(db)
got = {r[0]: r[1] for r in c.execute("select id,string_value from embedding_metadata where key='device'")}
src = {r[0]: r[1] for r in c.execute("select id,string_value from embedding_metadata where key='device_source'")}

want = {
    'new_convention': 'tor-ms22', 'composite_v183': 'emb-7kj4vr4g',
    'composite_om': 'emb-7kj4vr4g', 'composite_devbox': 'emb-7kj4vr4g',
    'container_only': None, 'container_bare': None, 'garbage_str': None,
    'other_sense': None, 'prose': None,
    'chunk0': 'tor-ms22', 'chunk1': 'tor-ms22', 'chunk2': 'tor-ms22',
    'keep0': 'mbp-m1-2020', 'keep1': 'mbp-m1-2020',
    'edge0': 'tor-ms22', 'dflt0': None,
}
want_src = {
    'new_convention': 'diary_host_marker', 'composite_v183': 'diary_host_marker',
    'chunk1': 'sibling_chunk', 'chunk2': 'sibling_chunk', 'keep1': 'sibling_chunk',
    'keep0': 'inbox_path',            # pre-existing stamp not rewritten
    'edge0': 'agent_at_device',
}
bad = 0
for k, exp in sorted(want.items()):
    act = got.get(k)
    if act != exp:
        print("FAIL %-17s device expected=%s got=%s" % (k, exp, act)); bad += 1
for k, exp in sorted(want_src.items()):
    act = src.get(k)
    if act != exp:
        print("FAIL %-17s source expected=%s got=%s" % (k, exp, act)); bad += 1
if 'device+0' not in os.environ['SECOND']:
    print("FAIL not idempotent, second pass said: %s" % os.environ['SECOND'].strip()); bad += 1
print("device-stamp rules: %s (%d checks)" % ("PASS" if not bad else "%d FAILURES" % bad,
                                              len(want) + len(want_src) + 1))
sys.exit(1 if bad else 0)
PYEOF