scripts/forge.py

#!/usr/bin/env python3
"""Forge SZL-MiniEmbed — a REAL trained word-embedding suite for SZLHOLDINGS/szl-kernels.

No gensim. Embeddings come from a term–term co-occurrence matrix (sliding window,
PPMI-weighted) reduced with sklearn TruncatedSVD — a classic, fully reproducible
distributional-semantics pipeline. The corpus is the SZL text estate:

  * SZLHOLDINGS/doctrine-v10-v11   (doctrine .md)
  * SZLHOLDINGS/rag-corpus-v1      (corpus.jsonl, ~948KB)
  * SZLHOLDINGS/thesis-corpus-v18  (thesis .tex chapters)
  * kernel-family READMEs          (this repo's own README + build metadata)

Ships vectors.npz + vocab.json + config.json. Evaluation is INTRINSIC ONLY:
nearest-neighbour sanity on ~15 doctrine terms (neighbour lists receipted).
No downstream benchmark is claimed. Seeded, receipted, reproducible.

Self-contained: resolves corpora from the repo's own dir when shipped in-repo
(corpus/ subdir), else from /tmp/corpus + /tmp/kernel-probe (forge-dev run)."""
import json, os, re, sys, time, hashlib, platform, glob
from collections import Counter, defaultdict
import numpy as np
from scipy.sparse import csr_matrix
from sklearn.decomposition import TruncatedSVD

SEED = 20260721
np.random.seed(SEED)
T0 = time.time()

_here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def corpus_root(name, subdirs):
    """Prefer an in-repo bundled copy; else the forge-dev download location."""
    in_repo = os.path.join(_here, "corpus", name)
    if os.path.isdir(in_repo):
        return in_repo
    for sd in subdirs:
        if os.path.isdir(sd):
            return sd
    return None


DOCTRINE_DIR = corpus_root("doctrine-v10-v11", ["/tmp/corpus/doctrine-v10-v11"])
RAG_DIR = corpus_root("rag-corpus-v1", ["/tmp/corpus/rag-corpus-v1"])
THESIS_DIR = corpus_root("thesis-corpus-v18", ["/tmp/corpus/thesis-corpus-v18"])
KERNEL_DIR = corpus_root("kernels", ["/tmp/kernel-probe/szl-kernels", _here])

assert DOCTRINE_DIR and RAG_DIR, "doctrine + rag corpora required — refuse"

# ---------------------------------------------------------------------------
# 1. Load raw text from every source. Record byte sha256 of each source file.
# ---------------------------------------------------------------------------
docs = []            # list of raw text strings
source_files = []    # (path, sha256, n_chars)


def add_text(path, text):
    if not text or not text.strip():
        return
    docs.append(text)
    source_files.append((os.path.relpath(path, "/tmp"),
                         hashlib.sha256(text.encode("utf-8", "ignore")).hexdigest(),
                         len(text)))


# doctrine markdown
for p in sorted(glob.glob(os.path.join(DOCTRINE_DIR, "*.md")) +
                glob.glob(os.path.join(DOCTRINE_DIR, "**", "*.md"), recursive=True)):
    try:
        add_text(p, open(p, encoding="utf-8", errors="ignore").read())
    except Exception:
        pass
# rag corpus.jsonl -> text field
rag_jsonl = os.path.join(RAG_DIR, "corpus.jsonl")
if os.path.isfile(rag_jsonl):
    parts = []
    for line in open(rag_jsonl, encoding="utf-8", errors="ignore"):
        line = line.strip()
        if not line:
            continue
        try:
            parts.append(json.loads(line).get("text", ""))
        except Exception:
            pass
    add_text(rag_jsonl, "\n".join(parts))
# thesis .tex chapters
if THESIS_DIR:
    for p in sorted(glob.glob(os.path.join(THESIS_DIR, "**", "*.tex"), recursive=True)):
        try:
            add_text(p, open(p, encoding="utf-8", errors="ignore").read())
        except Exception:
            pass
# kernel-family READMEs (this repo + its build metadata)
if KERNEL_DIR:
    for p in (sorted(glob.glob(os.path.join(KERNEL_DIR, "README.md"))) +
              sorted(glob.glob(os.path.join(KERNEL_DIR, "**", "*.md"), recursive=True)) +
              sorted(glob.glob(os.path.join(KERNEL_DIR, "**", "*.py"), recursive=True))):
        try:
            add_text(p, open(p, encoding="utf-8", errors="ignore").read())
        except Exception:
            pass

assert len(docs) >= 5, f"insufficient corpus ({len(docs)} docs) — refuse"

# ---------------------------------------------------------------------------
# 2. Tokenize. Keep alphabetic tokens + a few doctrine glyphs; lowercase; strip
#    latex/markdown control noise. Deterministic.
# ---------------------------------------------------------------------------
TOKEN_RE = re.compile(r"[a-z][a-z0-9\-]+|λ|ouroboros")
STOP = set("""the a an and or of to in is are was were be been being for on at by with as it its
this that these those from into than then so such not no nor but if while do does did done has have
had can could should would may might will shall must we you they he she i our your their his her them
us me my mine ours yours theirs which who whom whose what when where why how all any both each few more
most other some only own same too very s t just about above after again against because before below
between down during further here off out over under up once""".split())


def tokenize(text):
    text = re.sub(r"\\[a-zA-Z]+\{?|[{}$%#&_^~\\]", " ", text)  # strip latex/md control
    return [w for w in TOKEN_RE.findall(text.lower())
            if w not in STOP and len(w) > 2]


tokenized_docs = [tokenize(d) for d in docs]
all_tokens = [t for doc in tokenized_docs for t in doc]
freq = Counter(all_tokens)

MIN_COUNT = 5
vocab_terms = sorted([w for w, c in freq.items() if c >= MIN_COUNT])
MAX_VOCAB = 6000
if len(vocab_terms) > MAX_VOCAB:
    vocab_terms = [w for w, _ in Counter({w: freq[w] for w in vocab_terms}).most_common(MAX_VOCAB)]
    vocab_terms = sorted(vocab_terms)
vocab = {w: i for i, w in enumerate(vocab_terms)}
V = len(vocab)
assert V >= 200, f"vocab too small ({V}) — refuse"

# ---------------------------------------------------------------------------
# 3. Term–term co-occurrence (symmetric sliding window). PPMI weighting.
# ---------------------------------------------------------------------------
WINDOW = 5
cooc = defaultdict(float)
tok_total = 0
for doc in tokenized_docs:
    idx = [vocab[t] for t in doc if t in vocab]
    tok_total += len(idx)
    for i, wi in enumerate(idx):
        lo = max(0, i - WINDOW)
        hi = min(len(idx), i + WINDOW + 1)
        for j in range(lo, hi):
            if j == i:
                continue
            wj = idx[j]
            cooc[(wi, wj)] += 1.0 / abs(j - i)  # distance-weighted

rows = np.fromiter((k[0] for k in cooc), dtype=np.int32, count=len(cooc))
cols = np.fromiter((k[1] for k in cooc), dtype=np.int32, count=len(cooc))
vals = np.fromiter((cooc[k] for k in cooc), dtype=np.float64, count=len(cooc))
C = csr_matrix((vals, (rows, cols)), shape=(V, V))

# PPMI: log( P(i,j) / (P(i)P(j)) ), clipped at 0
total = C.sum()
row_sum = np.asarray(C.sum(axis=1)).ravel()
col_sum = np.asarray(C.sum(axis=0)).ravel()
C = C.tocoo()
pmi_vals = np.log((C.data * total) / (row_sum[C.row] * col_sum[C.col]) + 1e-12)
pmi_vals = np.maximum(pmi_vals, 0.0)
PPMI = csr_matrix((pmi_vals, (C.row, C.col)), shape=(V, V))

# ---------------------------------------------------------------------------
# 4. TruncatedSVD -> dense L2-normalized embeddings.
# ---------------------------------------------------------------------------
DIM = 128
svd = TruncatedSVD(n_components=DIM, random_state=SEED, n_iter=10)
E = svd.fit_transform(PPMI)              # (V, DIM)
norms = np.linalg.norm(E, axis=1, keepdims=True)
norms[norms == 0] = 1.0
E = (E / norms).astype(np.float32)

# ---------------------------------------------------------------------------
# 5. INTRINSIC nearest-neighbour sanity on doctrine terms (present in vocab).
# ---------------------------------------------------------------------------
PROBE_TERMS = ["ouroboros", "governance", "governed", "receipt", "provenance",
               "invariant", "kernel", "energy", "lambda", "conjecture",
               "doctrine", "honest", "tamper", "verify", "chain",
               "loop", "signature", "attestation", "measured", "trust"]


def neighbors(term, k=6):
    if term not in vocab:
        return None
    v = E[vocab[term]]
    sims = E @ v
    order = np.argsort(-sims)
    out = []
    for idx in order:
        if idx == vocab[term]:
            continue
        out.append((vocab_terms[idx], round(float(sims[idx]), 4)))
        if len(out) >= k:
            break
    return out


neighbor_lists = {}
present = 0
for t in PROBE_TERMS:
    nb = neighbors(t)
    if nb is not None:
        neighbor_lists[t] = nb
        present += 1
assert present >= 12, f"only {present} probe terms in vocab — corpus too thin"

# ---------------------------------------------------------------------------
# 6. Save artifacts: vectors.npz + vocab.json + config.json.
# ---------------------------------------------------------------------------
out = _here
np.savez_compressed(os.path.join(out, "vectors.npz"), vectors=E)
with open(os.path.join(out, "vocab.json"), "w") as f:
    json.dump({"vocab": vocab_terms, "index": vocab}, f)
config = {
    "model": "SZL-MiniEmbed",
    "method": "term-term co-occurrence (distance-weighted, window=%d) -> PPMI -> TruncatedSVD" % WINDOW,
    "dim": DIM, "vocab_size": V, "min_count": MIN_COUNT, "window": WINDOW,
    "seed": SEED, "normalization": "L2 row-normalized",
    "files": {"vectors": "vectors.npz (key 'vectors', float32 [V,dim])",
              "vocab": "vocab.json ({'vocab':[term...], 'index':{term:i}})"},
}
with open(os.path.join(out, "config.json"), "w") as f:
    json.dump(config, f, indent=2)

vec_sha = hashlib.sha256(open(os.path.join(out, "vectors.npz"), "rb").read()).hexdigest()
vocab_sha = hashlib.sha256(open(os.path.join(out, "vocab.json"), "rb").read()).hexdigest()

# explained variance is a MEASURED intrinsic property of the SVD fit
explained = float(svd.explained_variance_ratio_.sum())

receipt = {
    "artifact": "SZLHOLDINGS/szl-kernels — SZL-MiniEmbed v1",
    "role": "intrinsic word-embedding suite over the SZL text estate — the kernel suite remains the primary artifact; embeddings are a companion, NOT a benchmark claim",
    "generator": {"script": "scripts/forge.py", "seed": SEED,
                  "method": config["method"], "no_gensim": True},
    "data": {
        "sources": ["SZLHOLDINGS/doctrine-v10-v11", "SZLHOLDINGS/rag-corpus-v1 (corpus.jsonl)",
                    "SZLHOLDINGS/thesis-corpus-v18", "szl-kernels family READMEs + build/*.py"],
        "n_documents": len(docs),
        "n_source_files": len(source_files),
        "total_tokens_in_window": int(tok_total),
        "vocab_size": V, "min_count": MIN_COUNT,
        "source_file_sha256": [{"path": p, "sha256": s, "n_chars": n} for p, s, n in source_files],
    },
    "model": {"type": "sklearn.TruncatedSVD over PPMI co-occurrence", "dim": DIM,
              "window": WINDOW, "params": {"n_components": DIM, "n_iter": 10, "random_state": SEED},
              "files": {"vectors.npz": vec_sha, "vocab.json": vocab_sha},
              "config": "config.json"},
    "metrics_MEASURED": {
        "vocab_size": V,
        "embedding_dim": DIM,
        "svd_explained_variance_ratio": round(explained, 4),
        "probe_terms_in_vocab": present,
        "intrinsic_nearest_neighbors": neighbor_lists,
        "claim_scope": "INTRINSIC SANITY ONLY — nearest-neighbour lists are the measured evidence; NO downstream/analogy benchmark score is claimed",
    },
    "environment": {"python": platform.python_version(),
                    "sklearn": __import__("sklearn").__version__,
                    "scipy": __import__("scipy").__version__,
                    "numpy": np.__version__, "host": "replit 2-vCPU container",
                    "wall_seconds": round(time.time() - T0, 1)},
    "honesty": "Every number above is MEASURED by this run. Embeddings are an intrinsic distributional artifact over the SZL corpus; no benchmark skill is claimed. The kernel suite stays the primary artifact. Λ untouched = Conjecture 1 (open).",
    "trained_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
with open(os.path.join(out, "TRAINING_RECEIPT.json"), "w") as f:
    json.dump(receipt, f, indent=2)

print(json.dumps({k: v for k, v in receipt["metrics_MEASURED"].items()
                  if k != "intrinsic_nearest_neighbors"}, indent=2))
for t in ["ouroboros", "governance", "receipt", "lambda"]:
    if t in neighbor_lists:
        print(f"  {t:12s} -> {[w for w,_ in neighbor_lists[t]]}")
print(f"docs={len(docs)} vocab={V} tokens={tok_total} wall={receipt['environment']['wall_seconds']}s")