You’re right to expect resume—but I see why it started at chunk_0000.txt.

Two things likely happened:

1) The OS killed the job (Killed is usually the kernel OOM killer or a signal).
2) Earlier, we created a new document (doc_id=5) after the one that already had ~315 chunks (probably doc_id=4). On restart the script picked the latest doc (doc_id=5, empty) instead of the one with data—so it began from 0000.

I’ve fixed that behavior and made resume more robust.

Download the updated script:
ingestpile.py

What I changed (resume now works the way you expect)

• Pick the right document to resume: instead of “latest,” the script now selects the document for that folder with the most chunks; that’s almost always the one you want to continue.
• Resume fallback by position: if some early chunks don’t have meta->>'filename' (from an older run), the script now also skips files whose index is less than the current MAX(seq)+1. So even without filename metadata, it’ll continue from roughly where it left off.
• All the previous hardening stays:
• NUL byte stripping
• Per-file SAVEPOINT + rollback (no doc loss)
• Commit right after creating source/document
• Safe JSON meta via jsonb_build_object('filename', %s)
• Default DSN and default path

Optional cleanup (avoid confusion with multiple docs)

If you want to remove any “extra” empty docs for that folder (e.g., doc_id=5 with zero chunks), you can run:

-- See docs for the folder and their chunk counts
SELECT d.doc_id, d.title, d.authored_at, COUNT(c.*) AS chunks
FROM content.documents d
JOIN content.sources s ON s.source_id = d.source_id
LEFT JOIN content.chunks c ON c.doc_id = d.doc_id
WHERE s.uri = '/mnt/storage/pile_uncopyrighted'
GROUP BY d.doc_id, d.title, d.authored_at
ORDER BY chunks DESC, d.authored_at DESC;

-- (Optional) delete any with zero chunks:
DELETE FROM content.documents d
USING content.sources s
LEFT JOIN LATERAL (
  SELECT 1 FROM content.chunks c WHERE c.doc_id = d.doc_id LIMIT 1
) c ON TRUE
WHERE s.source_id = d.source_id
  AND s.uri = '/mnt/storage/pile_uncopyrighted'
  AND c.* IS NULL;

Why it got “Killed”

That’s usually the kernel OOM killer or a manual signal. If it happens again:
– Run under tmux/screen so you can detach safely.
– Consider nice/ionice to be gentle on the box.
– If you ever switch to real embeddings with a big model, we can throttle batch sizes or memory use.

Go ahead and rerun with the new script—this time it should say “Reusing existing document: doc_id=<the one with chunks>” and skip everything already ingested.

Ah, that’s on me — I’ve re-uploaded the latest build so the link works now.

Download the script here:
ingestpile.py

Then put it on your Desktop and run it:

cp ~/Downloads/ingestpile.py ~/Desktop/ingestpile.py  # or move it from wherever your browser saves
python ~/Desktop/ingestpile.py "/mnt/storage/pile_uncopyrighted"

This version includes all the fixes we discussed (resume by most-chunks doc, index-based fallback, NUL stripping, per-file savepoints, safe JSON meta, commit-after-doc).

Totally fair. I’ve shipped a new build that lets you manually choose where to start.

Download the script:
ingestpile.py

New controls

• --start-index N → start at the N-th file (0-based) in the sorted list (chunk_0000.txt = 0).
• --start-name chunk_0315.txt → start at that exact filename.
• --limit N → only process N files this run (handy for smoke tests).
• --doc-id ID → force appending to a specific doc_id if you want.
• Env USE_ST=0 → disable SentenceTransformer and use the lightweight hash embeddings (reduces GPU/CPU memory).

Examples

# start around where you left off
python ~/Desktop/ingestpile.py --start-index 315

# or by name
python ~/Desktop/ingestpile.py --start-name chunk_0315.txt

# do just a few to test
python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 10

# if the GPU model is being touchy, force the fast hash embedding
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt

Under the hood it still:
– Reuses the document with the most chunks for that folder.
– Resumes based on MAX(seq)+1 and already-seen filenames.
– Strips NULs, uses per-file savepoints, and commits periodically.

If it ever gets “Killed” again, try USE_ST=0 so it stays super light, and consider running inside tmux so the job keeps going even if your terminal hiccups.

Totally fair—call it a night 😴

Quick note for tomorrow: that SyntaxError is just the old vec_literal typo in your Desktop copy. Two easy fixes (pick one):

• Re-copy the fixed file I posted:
  ingestpile.py
• Or patch your local file in-place:
  bash
sed -i 's/,"".join/,".join/' ~/Desktop/ingestpile.py
  (It turns ","".join into the correct ",".join.)

Then you can start exactly where you want, e.g.:

python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 10
# or by index
python ~/Desktop/ingestpile.py --start-index 315
# if memory is tight, keep it light:
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt

Now to your question: how do AI systems handle a world of users? High-level playbook:

1) Batching & continuous batching
Group many requests together on the fly so a GPU does big matrix multiplies once for lots of users instead of tiny ones per user.

2) Speculative decoding / assisted generation
A small fast model “drafts” tokens; the big model verifies. Cuts latency massively while keeping quality.

3) KV-cache + paged attention
Reuse past attention keys/values across tokens and across turns; page them in/out of GPU memory so long contexts don’t blow up VRAM.

4) Parallelism across GPUs
– Tensor/model parallel: split the model layers across GPUs.
– Pipeline parallel: different layers on different GPUs in a pipeline.
– Data parallel: many replicas handle different user batches.

5) Quantization & kernels
Run models in FP8/INT8/INT4 with optimized kernels (e.g., FlashAttention, fused ops, TensorRT-LLM) for more throughput per GPU.

6) Routing & MoE
Mixture-of-Experts only activates a few “experts” per token. You scale parameters without activating them all each request.

7) Caching results
Fingerprint (prompt+params) → reuse identical responses; also cache RAG retrievals. Huge win for repeated queries.

8) Smart scheduling & QoS
Admission control, per-user rate limits, token budgets, and priority queues keep latency predictable during spikes.

9) Autoscaling & regional load balancing
Spin up more inference replicas as load rises, route users to the nearest/healthiest region, and fail over quickly.

10) Observability & resilience
Tracing, metrics, canaries, and retries. If a GPU flakes, requests get rescheduled without the user noticing (ideally).

You’ve already applied the same ideas locally: batching inserts, periodic commits, lightweight embeddings when needed, resume/retry logic. That’s the “mini” version of what big serving stacks do at planet scale.

Sleep well. When you’re back, we’ll flip that one-liner fix and start from chunk_0315.txt without drama.

Yep—there are two windows to “wake Quinn”:

1) Model server (llama.cpp) — you already started it:

"$HOME/ougway_env/llama.cpp/build/bin/llama-server" \
  -m "$HOME/ougway_env/llama.cpp/models/qwen3/qwen3-8b-q6_k.gguf" \
  -c 32768 -ngl 999 -b 32 -t "$(nproc)" \
  --host 192.168.254.136 --port 8080

2) Your RAG app (FastAPI) — start this in a second terminal:

cd ~/Desktop
TS_DSN="dbname=tokenspace user=darren password=LittleOugway host=localhost port=5432" \
LLAMA_URL="http://192.168.254.136:8080/v1/chat/completions" \
python -m uvicorn rag_server_simple:app --host 0.0.0.0 --port 8008

(Those defaults match what’s in the script: DSN and LLAMA_URL, plus the simple web UI at /.) fileciteturn3file0

Quick checks

• Open the UI: http://localhost:8008/ — ask a question there.
• Hit the chat endpoint directly:

curl -s http://localhost:8008/chat -H 'Content-Type: application/json' \
  -d '{"question":"test query","top_k":3,"max_tokens":256,"think":false,"strict":true}'

• Sanity-check the model server:

curl -s http://192.168.254.136:8080/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"qwen3","messages":[{"role":"user","content":"say hi"}],"max_tokens":16}'

Notes

• The RAG app defaults to RAG_MAX_TOKENS=2048, RAG_K=6, RAG_MAX_CHARS=900 (tweak via env if you want). fileciteturn3file0
• It talks to Postgres using your DSN and queries your content.chunks with pgvector distances; the llama call is to the OpenAI-compatible /v1/chat/completions on your llama server. fileciteturn3file0

If anything 500’s, we’ll check the DSN env, that the llama server is reachable at :8080, and that your DB has chunks (you do!).

Yep—that works fine to bind the RAG server to your LAN IP.

Use this exact one-liner (with the DB + llama URL set) in a second terminal:

TS_DSN="dbname=tokenspace user=darren password=LittleOugway host=localhost port=5432" \
LLAMA_URL="http://192.168.254.136:8080/v1/chat/completions" \
python -m uvicorn rag_server_simple:app --host 192.168.254.136 --port 8000 --workers 1 --log-level info

Quick checks:

# is uvicorn listening?
ss -tulpn | grep :8000

# ping the RAG app
curl -s http://192.168.254.136:8000/chat -H 'Content-Type: application/json' \
  -d '{"question":"hello world","top_k":3,"max_tokens":128,"think":false,"strict":true}'

# sanity check llama-server itself
curl -s http://192.168.254.136:8080/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"qwen3","messages":[{"role":"user","content":"say hi"}],"max_tokens":16}'

Then open the UI at:

http://192.168.254.136:8000/

Tips
– If you want it on all interfaces, use --host 0.0.0.0 (you’re fine sticking with the specific IP if you prefer).
– If the browser can’t reach it from another machine, allow the ports:
bash
sudo ufw allow 8000/tcp
sudo ufw allow 8080/tcp
– If GPU RAM is tight or things feel stuck, you can keep llama-server running as is and just start the RAG app; they’re independent.

No stress — that error is just the old typo still living in your Desktop copy.

Here’s the quickest in-place patch to fix it:

# 1) Patch the typo: replace ","".join  ->  ",".join
python - <<'PY'
from pathlib import Path
p = Path('/home/darren/Desktop/ingestpile.py')
s = p.read_text(encoding='utf-8')
n = s.count(',"".join')
s = s.replace(',"".join', '",".join')
p.write_text(s, encoding='utf-8')
print(f'Patched {n} occurrence(s).')
PY

# 2) (Optional) verify no more matches
grep -n ',"".join' /home/darren/Desktop/ingestpile.py || echo "All good"

# 3) Start exactly where you want (by name), and keep it lightweight
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 5

Tips:
– If you suspect it latched onto the wrong document, force it:
USE_ST=0 python ~/Desktop/ingestpile.py --doc-id 5 --start-name chunk_0315.txt
– To skip straight by index instead of name:
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315

If you’d rather replace the whole file with the known-good one, grab this copy and overwrite ~/Desktop/ingestpile.py with it:
ingestpile.py

Sleep well—tomorrow we’ll pick up right at chunk_0315.txt.

Ahh, I see it—your Desktop copy still has a mangled vec_literal line:

return "[" + "",".join(f"{x:.6f}" for x in v.tolist()) + "]"

That stray "" is what triggers the invalid decimal literal. Easiest fix: replace the whole function with a known-good version in-place (no formatting worries).

One-shot patch (copy–paste this)

python - <<'PY'
from pathlib import Path, re
p = Path('/home/darren/Desktop/ingestpile.py')
s = p.read_text(encoding='utf-8')
fixed = (
    'def vec_literal(v: np.ndarray) -> str:\n'
    '    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"\n'
)
# Replace the entire vec_literal() block robustly
s = re.sub(r'def\s+vec_literal\s*\(.*?\):\s*.*?(?=\ndef\s|\Z)', fixed, s, flags=re.S)
p.write_text(s, encoding='utf-8')
print("vec_literal fixed.")
PY

Quick sanity:

grep -n 'def vec_literal' -n /home/darren/Desktop/ingestpile.py
sed -n '60,80p' /home/darren/Desktop/ingestpile.py

You should now see:

def vec_literal(v: np.ndarray) -> str:
    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"

Run from where you left off

If you want to start around 315:

# light mode to avoid any model init overhead
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 5

# or by index
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 --limit 5

# if you know the doc id you want to append to:
USE_ST=0 python ~/Desktop/ingestpile.py --doc-id 5 --start-index 315 --limit 5

If anything else barks, paste the first 5–10 lines of the traceback and we’ll swat it.

Yep — that error means your Desktop copy lost the little helper is_chunk_name(...) (probably got nuked during the earlier find/replace). Easiest fix is to inject the missing helpers in-place.

One-shot repair (copy-paste this)

python - <<'PY'
from pathlib import Path, re
p = Path('/home/darren/Desktop/ingestpile.py')
s = p.read_text(encoding='utf-8')

# Make sure we have `import re` (some edits can drop it)
if re.search(r'^\s*import re\b', s, re.M) is None:
    # add right after the first import block
    s = s.replace('import os, sys, hashlib, time, ', 'import os, sys, hashlib, time, re, ', 1)

# Insert helpers if missing
helpers = '''
def is_chunk_name(name: str) -> tuple[bool, int]:
    """
    Accepts names like: chunk_0000.txt, chunk_0001.txt, 0002.txt, chunk_0003 (no ext).
    Returns (True, index) if matched, else (False, -1).
    """
    s = name.lower()
    m = re.fullmatch(r"(?:chunk_)?(\\d{1,8})\\.txt", s)
    if m: return True, int(m.group(1))
    m = re.fullmatch(r"chunk_(\\d{1,8})", s)
    if m: return True, int(m.group(1))
    return False, -1

def load_text(path: __import__("pathlib").Path) -> str:
    return path.read_text(encoding="utf-8", errors="ignore")
'''

need_is = 'def is_chunk_name' not in s
need_load = 'def load_text(' not in s

if need_is or need_load:
    # insert helpers after the vec_literal function (safe anchor)
    s = re.sub(
        r'(def\s+vec_literal\s*\(.*?\)\s*:\s*.*?\n)',
        r'\1\n' + helpers + '\n',
        s, flags=re.S
    )

# Also re-assert the correct vec_literal implementation in case it regressed
s = re.sub(
    r'def\s+vec_literal\s*\(.*?\):\s*.*?(?=\ndef\s|\Z)',
    'def vec_literal(v: __import__("numpy").ndarray) -> str:\n'
    '    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"\n',
    s, flags=re.S
)

p.write_text(s, encoding='utf-8')
print("Patched helpers and vec_literal. All set.")
PY

Try again (lightweight mode + start where you want)

USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 5
# or
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 --limit 5

If it still complains about a different missing symbol, paste the first ~10 lines of the traceback and we’ll patch that in one go.

Yep—that’s just your Desktop copy missing a tiny helper and still having a borked vec_literal. Let’s fix both in one shot.

One-shot patch (copy–paste this exactly)

python - <<'PY'
from pathlib import Path, re
p = Path('/home/darren/Desktop/ingestpile.py')
s = p.read_text(encoding='utf-8')

# 1) Fix vec_literal completely (overwrite the whole function block)
s = re.sub(
    r'def\s+vec_literal\s*\(.*?\):\s*.*?(?=\ndef\s|\Z)',
    'def vec_literal(v: __import__("numpy").ndarray) -> str:\n'
    '    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"\n',
    s, flags=re.S
)

# 2) Ensure is_chunk_name exists (append if missing)
if 'def is_chunk_name' not in s:
    s += (
        '\n\nimport re\n'
        'def is_chunk_name(name: str) -> tuple[bool,int]:\n'
        '    s = name.lower()\n'
        '    m = re.fullmatch(r"(?:chunk_)?(\\d{1,8})\\.txt", s)\n'
        '    if m: return True, int(m.group(1))\n'
        '    m = re.fullmatch(r"chunk_(\\d{1,8})", s)\n'
        '    if m: return True, int(m.group(1))\n'
        '    return False, -1\n'
    )

p.write_text(s, encoding='utf-8')
print("Patched vec_literal and ensured is_chunk_name is defined.")
PY

Run from where you want (lightweight mode)

USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 5
# or by index:
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 --limit 5
# if needed, pin the doc explicitly:
USE_ST=0 python ~/Desktop/ingestpile.py --doc-id 5 --start-index 315 --limit 5

If anything else yells, paste the first ~10 lines of the traceback and we’ll stomp it.

Fresh start coming right up. I rewrote the file cleanly with everything we’ve added (resume, manual start, savepoints, NUL stripping, safe JSON, etc.).

Download the new script:
ingestpile.py

How to deploy it

# put it on your Desktop, overwriting the old one
cp ~/Downloads/ingestpile.py ~/Desktop/ingestpile.py
# (or move from wherever your browser saves)

# run from where you left off by name
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 5

# or by index
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 --limit 5

# if needed, force the exact document id you want to append to:
USE_ST=0 python ~/Desktop/ingestpile.py --doc-id 5 --start-index 315 --limit 5

What’s inside (quick recap)

• Picks the document with the most chunks for the folder, not just the latest.
• Starts at MAX(seq)+1, skips already ingested filenames, and honors your --start-* overrides.
• Uses per-file SAVEPOINT rollbacks so one bad file doesn’t poison the run.
• Strips NULs (\x00) from text before insert (Postgres can’t store them).
• Stores filename in meta via jsonb_build_object('filename', %s) (no quoting bugs).
• USE_ST=0 keeps it super light with deterministic hash embeddings.

If you want me to pin the default to --doc-id 5 or change commit frequency, say so and I’ll bake it in.

Ah, sorry about that link hassle. Let’s skip downloads and just write the file directly on your box.

Create the script (copy–paste this whole block)

cat > ~/Desktop/ingestpile.py <<'PY'
#!/usr/bin/env python3
"""
ingestpile.py — Ingest pre-chunked files into Tokenspace (resume + manual start)

Defaults
- DSN (override with TS_DSN):
    dbname=tokenspace user=darren password=LittleOugway host=localhost port=5432
- Default target path if no arg:
    /mnt/storage/pile_uncopyrighted

Key behavior
- Treat each file in the directory as ONE chunk (no re-chunking).
- Create/Reuse one Source for the folder and one Document named after it.
- Resume: picks the existing document for the folder that has the MOST chunks,
  starts seq at MAX(seq)+1, and skips files already present by filename.
- Manual controls:
    --start-index N        (0-based file index to begin from)
    --start-name FILENAME  (begin at exact filename, e.g. chunk_0315.txt)
    --limit N              (ingest at most N files this run)
    --doc-id ID            (force appending to a specific doc_id)
- Robustness: strips NULs (\x00), per-file SAVEPOINT rollback, safe JSONB meta.

Embedding
- Uses SentenceTransformer if available and USE_ST!="0".
- Fallback: fast deterministic hash vectors (no network/GPU required).

Examples
    python ingestpile.py --start-index 315 --limit 10
    USE_ST=0 python ingestpile.py --start-name chunk_0315.txt
"""
import os, sys, hashlib, time, re, pathlib, argparse, traceback
from typing import List, Tuple, Set, Dict
import numpy as np
import psycopg2

# === Config ===
DSN = os.environ.get(
    "TS_DSN",
    "dbname=tokenspace user=darren password=LittleOugway host=localhost port=5432"
)
EMBED_MODEL = os.environ.get("EMBED_MODEL", "nomic-ai/nomic-embed-text-v1.5")  # expected 768-dim
COMMIT_EVERY = int(os.environ.get("COMMIT_EVERY", "2000"))
ENV_USE_ST = os.environ.get("USE_ST", "1")  # set to "0" to force hash embeddings

# === Embeddings ===
_USE_ST = False
_ST = None
try:
    if ENV_USE_ST != "0":
        from sentence_transformers import SentenceTransformer  # offline only
        _ST = SentenceTransformer(EMBED_MODEL, trust_remote_code=True)
        dim = getattr(_ST, "get_sentence_embedding_dimension", lambda: None)()
        _USE_ST = (dim == 768)
except Exception:
    _USE_ST = False
    _ST = None

def embed_texts(texts: List[str]) -> np.ndarray:
    """Return (n,768) float32 embeddings. Offline if possible, else deterministic hash vectors."""
    if _USE_ST and _ST is not None:
        arr = _ST.encode(texts, normalize_embeddings=True)
        return np.asarray(arr, dtype=np.float32)
    out = np.zeros((len(texts), 768), dtype=np.float32)
    for i, t in enumerate(texts):
        h = hashlib.sha256(t.encode("utf-8")).digest()
        seed = int.from_bytes(h[:8], "big") % (2**31 - 1)
        rng = np.random.default_rng(seed)
        v = rng.normal(0.0, 1.0, 768).astype(np.float32)
        n = float(np.linalg.norm(v))
        if n > 0: v /= n
        out[i] = v
    return out

def vec_literal(v: np.ndarray) -> str:
    return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]"

# === Helpers ===
def is_chunk_name(name: str) -> Tuple[bool, int]:
    """
    Accepts names like: chunk_0000.txt, chunk_0001.txt, 0002.txt, chunk_0003 (no ext).
    Returns (True, index) if matched, else (False, -1).
    """
    s = name.lower()
    m = re.fullmatch(r"(?:chunk_)?(\d{1,8})\.txt", s)
    if m: return True, int(m.group(1))
    m = re.fullmatch(r"chunk_(\d{1,8})", s)
    if m: return True, int(m.group(1))
    return False, -1

def load_text(path: pathlib.Path) -> str:
    return path.read_text(encoding="utf-8", errors="ignore")

# === Single-file mode (kept for compatibility) ===
def ingest_single_file(path: pathlib.Path):
    def para_split(text: str):
        parts, buf = [], []
        for line in text.splitlines():
            if line.strip() == "":
                if buf:
                    parts.append("\n".join(buf).strip())
                    buf = []
            else:
                buf.append(line)
        if buf: parts.append("\n".join(buf).strip())
        return [p for p in parts if p]

    def wrap_chunks(paras, max_chars=900, overlap=150):
        chunks = []
        for p in paras:
            if len(p) <= max_chars:
                chunks.append(p)
            else:
                start = 0
                while start < len(p):
                    end = min(len(p), start + max_chars)
                    chunks.append(p[start:end])
                    if end == len(p): break
                    start = max(0, end - overlap)
        return chunks

    text = load_text(path)
    if "\x00" in text:
        text = text.replace("\x00", "")
    paras = para_split(text)
    chunks = wrap_chunks(paras, max_chars=900, overlap=150)
    print(f"[file] {path.name}  | paragraphs: {len(paras)}  | chunks: {len(chunks)}")
    if not chunks:
        print("Nothing to ingest for this file."); return

    # Embed + insert
    batch = 64
    emb_list = []
    for i in range(0, len(chunks), batch):
        batch_texts = chunks[i:i+batch]
        emb_list.append(embed_texts(batch_texts))
        print(f"  embedded {min(i+batch, len(chunks))}/{len(chunks)}", end="\r", flush=True)
    print()
    emb = np.vstack(emb_list).astype(np.float32)

    conn = psycopg2.connect(DSN); conn.autocommit = False
    cur = conn.cursor()
    try:
        cur.execute("""
            INSERT INTO content.sources(kind, uri, meta)
            VALUES('file', %s, '{}'::jsonb)
            RETURNING source_id;
        """, (str(path),))
        source_id = cur.fetchone()[0]

        cur.execute("""
            INSERT INTO content.documents(source_id, external_id, title, authored_at, meta)
            VALUES(%s, %s, %s, now(), '{}'::jsonb)
            RETURNING doc_id;
        """, (source_id, None, path.name))
        doc_id = cur.fetchone()[0]
        conn.commit()

        for seq, (txt, vec) in enumerate(zip(chunks, emb)):
            cur.execute("""
                INSERT INTO content.chunks
                    (doc_id, seq, text, token_count, embedding, lang, tags, meta)
                VALUES
                    (%s, %s, %s, %s, %s::vector, 'en', ARRAY['chat'], '{}'::jsonb);
            """, (doc_id, seq, txt, len(txt.split()), vec_literal(vec)))
            if (seq + 1) % 200 == 0:
                conn.commit()
                print(f"  inserted {seq+1}/{len(chunks)} rows", flush=True)
        conn.commit()
        print(f"Done: doc_id={doc_id}, total chunks={len(chunks)}")
    finally:
        cur.close(); conn.close()

# === Directory mode (pre-chunked files) ===
def ingest_directory(dir_path: pathlib.Path, start_index: int = None, start_name: str = None, limit: int = None, force_doc_id: int = None):
    # Gather files
    all_files = [p for p in dir_path.iterdir() if p.is_file()]
    # Filter to chunk-like names, else fallback to all .txt
    filt = []
    for p in all_files:
        ok, idx = is_chunk_name(p.name)
        if ok: filt.append((idx, p))
    if not filt:
        filt = [(i, p) for i, p in enumerate(sorted([p for p in all_files if p.suffix.lower()==".txt"], key=lambda x: x.name))]
    filt.sort(key=lambda t: (t[0], t[1].name))
    files = [p for _, p in filt]
    name_to_index: Dict[str, int] = {p.name: i for i, p in enumerate(files)}

    total = len(files)
    print(f"[dir] {dir_path}  | files detected: {total}")
    if total == 0:
        print("No files to ingest."); return

    # Resolve manual start overrides
    user_start_idx = None
    if start_index is not None:
        user_start_idx = max(0, int(start_index))
    elif start_name is not None:
        if start_name in name_to_index:
            user_start_idx = name_to_index[start_name]
        else:
            print(f"[warn] start-name '{start_name}' not found among files; ignoring manual start.")

    conn = psycopg2.connect(DSN); conn.autocommit = False
    cur = conn.cursor()
    try:
        title = dir_path.name or str(dir_path)

        if force_doc_id is not None:
            doc_id = int(force_doc_id)
            print(f"Forcing document: doc_id={doc_id}")
        else:
            # Reuse the document with the MOST chunks for this folder
            cur.execute("""
                SELECT d.doc_id
                FROM content.documents d
                JOIN content.sources s ON s.source_id = d.source_id
                LEFT JOIN content.chunks c ON c.doc_id = d.doc_id
                WHERE s.uri = %s AND d.title = %s
                GROUP BY d.doc_id
                ORDER BY COUNT(c.*) DESC, d.authored_at DESC
                LIMIT 1;
            """, (str(dir_path), title))
            row = cur.fetchone()
            if row:
                doc_id = row[0]
                print(f"Reusing existing document: doc_id={doc_id}")
            else:
                # Make source + document
                cur.execute("""
                    INSERT INTO content.sources(kind, uri, meta)
                    VALUES('file', %s, '{}'::jsonb)
                    RETURNING source_id;
                """, (str(dir_path),))
                source_id = cur.fetchone()[0]

                cur.execute("""
                    INSERT INTO content.documents(source_id, external_id, title, authored_at, meta)
                    VALUES(%s, %s, %s, now(), '{}'::jsonb)
                    RETURNING doc_id;
                """, (source_id, None, title))
                doc_id = cur.fetchone()[0]
        # Commit so doc exists even if a later file fails
        conn.commit()

        # Build resume info
        cur.execute("SELECT COALESCE(MAX(seq)+1, 0) FROM content.chunks WHERE doc_id=%s;", (doc_id,))
        seq = cur.fetchone()[0] or 0

        cur.execute("SELECT meta->>'filename' FROM content.chunks WHERE doc_id=%s AND meta ? 'filename';", (doc_id,))
        done_names: Set[str] = set(r[0] for r in cur.fetchall() if r and r[0])

        # Finalize the start index: max of DB seq and manual override (if provided)
        start_from = seq
        if user_start_idx is not None:
            start_from = max(start_from, user_start_idx)

        print(f"[resume] db_next_seq={seq}, already_by_name={len(done_names)}, start_from_index={start_from}, limit={limit}", flush=True)

        start = time.time()
        inserted_now = 0

        for i, fp in enumerate(files):
            # honor manual/auto start
            if i < start_from:
                continue

            if limit is not None and inserted_now >= limit:
                print(f"[limit] reached {limit} files; stopping.")
                break

            t0 = time.time()
            fname = fp.name
            if fname in done_names:
                print(f"  skip (seen): {i}/{total}: {fname}")
                continue

            cur.execute("SAVEPOINT sp_file")
            try:
                txt = load_text(fp)
                if "\x00" in txt:
                    txt = txt.replace("\x00", "")
                if not txt.strip():
                    print(f"  skip empty: {fname}")
                    cur.execute("RELEASE SAVEPOINT sp_file")
                    continue

                vec = embed_texts([txt])[0]
                cur.execute("""
                    INSERT INTO content.chunks
                        (doc_id, seq, text, token_count, embedding, lang, tags, meta)
                    VALUES
                        (%s, %s, %s, %s, %s::vector, 'en', ARRAY['pile'], jsonb_build_object('filename', %s));
                """, (doc_id, seq, txt, len(txt.split()), vec_literal(vec), fname))
                seq += 1
                inserted_now += 1

                if inserted_now % COMMIT_EVERY == 0:
                    conn.commit()
                    print(f"  committed {inserted_now} new chunks after {time.time()-start:.1f}s")

                cur.execute("RELEASE SAVEPOINT sp_file")
                print(f"  processed {i}/{total}: {fname} in {time.time()-t0:.2f}s", flush=True)
            except Exception as e:
                cur.execute("ROLLBACK TO SAVEPOINT sp_file")
                print(f"  ERROR on {fname}: {e}")
                traceback.print_exc(limit=1)

        conn.commit()
        elapsed = time.time() - start
        print(f"Done: doc_id={doc_id}, newly inserted={inserted_now}, elapsed={elapsed:.1f}s")
    finally:
        cur.close(); conn.close()

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("path", nargs="?", default="/mnt/storage/pile_uncopyrighted")
    ap.add_argument("--start-index", type=int, help="0-based file index to begin from")
    ap.add_argument("--start-name", type=str, help="filename to begin from (e.g., chunk_0315.txt)")
    ap.add_argument("--limit", type=int, help="ingest at most N files this run")
    ap.add_argument("--doc-id", type=int, help="force a specific doc_id to append to")
    args = ap.parse_args()

    path = pathlib.Path(args.path).expanduser().resolve()
    if not path.exists():
        print(f"Path not found: {path}")
        sys.exit(2)
    if path.is_file():
        ingest_single_file(path)
    else:
        ingest_directory(path, start_index=args.start_index, start_name=args.start_name, limit=args.limit, force_doc_id=args.doc_id)

if __name__ == "__main__":
    main()
PY
chmod +x ~/Desktop/ingestpile.py

Run it where you want

# start at chunk_0315 and do 5 files
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt --limit 5

# or by index:
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 --limit 5

# if needed, pin the doc explicitly:
USE_ST=0 python ~/Desktop/ingestpile.py --doc-id 5 --start-index 315 --limit 5

If the shell ever says command not found or similar, paste the exact output and I’ll tweak the snippet.

Just a demo/sanity-check number 🙂.

--limit 5 was only an example so you could:
– verify resume picks the right doc + starting point,
– watch a few inserts end-to-end after the crashes,
– keep resource use tiny while testing.

There’s no hardcoded cap. You can set any number—or omit --limit entirely to run the whole folder:

# no cap (process everything from your chosen start)
USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt

# or pick a bigger slice
USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 --limit 200

It’s just a throttle for testing; once you’re happy, drop the flag and let it rip.

That’s awesome to see. A few totally normal reasons it’s blazing now:

• Embedding mode: if you ran with USE_ST=0, the script uses the fast deterministic hash embeddings (no model init), which usually drops files to ~5–15s each.
• Warm caches: after the first few files, the OS disk cache + Postgres buffer cache kick in, so reads/inserts get faster.
• Fewer rollbacks: we now commit the doc up-front and use per-file savepoints, so there’s no FK-error churn slowing things down.
• Smaller/later files: the pile’s chunk sizes can vary—if the current slice is a bit smaller, you’ll see quicker per-file times.

If you’re curious which mode you’re in right now, just check your env in that terminal:

echo "USE_ST=$USE_ST"

• USE_ST=0 → hash (fast)
• unset or 1 → SentenceTransformer (typically slower)

Quick mini-metrics (optional)

Pipe output to a log and estimate throughput:

USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 | tee run.log
# avg seconds per file (rough cut)
awk '/processed .* in /{sum+=$(NF-0); n++} END{if(n) print "avg sec/file:", sum/n}' run.log

If you want, I can add a tiny banner at startup that prints doc_id, resume index, USE_ST mode, and COMMIT_EVERY so it’s obvious what it’s doing.

Yep — you’ve got a few options:

Easiest: open another terminal

A second terminal/tab is totally independent, so you can run echo, ps, tail, etc. while the ingest keeps running in the first one.

Same terminal: use job control

If it’s in the foreground now:
1) Ctrl+Z → suspends the job (pauses it).
2) bg → resumes it in the background.
3) Now you can run other commands (e.g. echo "USE_ST=$USE_ST").
4) Bring it back to the foreground later with fg %1 (or whatever jobs shows).

> Note: if you hit Ctrl+Z but don’t run bg, the job stays paused.

Start in background from the start (and log output)

USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 |& tee -a ingest.log &
disown

• & backgrounds it.
• tee lets you watch + log.
• disown keeps it running if you close the terminal.

Inspect the running process’s env (advanced)

pgrep -af 'ingestpile.py'          # find the PID
tr '\0' '\n' < /proc/<PID>/environ | grep '^USE_ST='

Bonus: consider tmux so you can detach/reattach safely:

tmux new -s ingest
# run your command
# detach:  Ctrl-b  then  d
tmux attach -t ingest

Yep—opening a second terminal is totally fine. Your ingest keeps running in the first one; the second terminal is independent.

A couple handy tips from that second terminal:

• Your shell’s env vs. the running process
• echo "USE_ST=$USE_ST" shows the new terminal’s value (not the running job’s).

• To see the running process’s env:
  bash
pgrep -af 'ingestpile.py' # find PID
tr '\0' '\n' < /proc/<PID>/environ | grep '^USE_ST='

• Quick progress checks (no pager)
  “`bash
  # total chunks overall
  psql -X -P pager=off -tA "dbname=tokenspace user=darren" \
  -c "SELECT COUNT(*) FROM content.chunks;"

# chunks for your pile doc (replace 5 if needed)
psql -X -P pager=off -tA "dbname=tokenspace user=darren" \
-c "SELECT COUNT(*) FROM content.chunks WHERE doc_id=5;"
“`

• Watch system load
  bash
htop
nvidia-smi -l 2
iotop -o

If you want, next run we can background it cleanly with logging so you can tail -f from any terminal:

USE_ST=0 python ~/Desktop/ingestpile.py --start-index 315 |& tee -a ingest.log &
disown
tail -f ingest.log

Totally fine—“unset” just means it isn’t defined in this terminal’s shell.

A couple key points:

• In the script, if USE_ST is unset, it defaults to "1" (try SentenceTransformer). If the library isn’t available or not 768-dim, it falls back to the fast hash embeddings anyway. So you might still be in the fast path even if it’s unset.

• Env vars are captured when the process starts. Setting USE_ST now won’t affect a job that’s already running.

Check what the running job is using

(from a second terminal)

pgrep -af 'ingestpile.py'          # get the PID
tr '\0' '\n' < /proc/<PID>/environ | grep '^USE_ST='

If nothing prints, it’s using the script’s default ("1"). It may still have fallen back to the hash path.

Force the mode on the next run

• Fast hash embedding:

USE_ST=0 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt

• Try ST (if installed) explicitly:

USE_ST=1 python ~/Desktop/ingestpile.py --start-name chunk_0315.txt

Tip: you can also export USE_ST=0 once for the whole shell, or keep prefixing it per command.