my-vault/.scripts/memory/incremental_ingest.py

#!/usr/bin/env python3

import argparse
from pathlib import Path

from index_common import (
    VAULT_ROOT,
    embed_text,
    get_conn,
    index_lock,
    is_excluded,
    sha256_text,
    vector_literal,
)

MIN_TEXT_LEN = 50


def parse_changes(changes_file: Path) -> list[dict]:
    events: list[dict] = []
    for raw_line in changes_file.read_text(encoding='utf-8', errors='ignore').splitlines():
        line = raw_line.strip()
        if not line:
            continue

        parts = line.split('\t')
        code = parts[0][0]
        if code in {'A', 'M', 'T'} and len(parts) >= 2:
            events.append({'code': code, 'path': parts[1]})
        elif code == 'D' and len(parts) >= 2:
            events.append({'code': 'D', 'old': parts[1]})
        elif code == 'R' and len(parts) >= 3:
            events.append({'code': 'R', 'old': parts[1], 'new': parts[2]})
    return events


def upsert_file(cur, rel_path: str) -> None:
    abs_path = VAULT_ROOT / rel_path
    if not abs_path.exists() or abs_path.suffix.lower() != '.md':
        return

    if is_excluded(abs_path):
        cur.execute('DELETE FROM memory_primary WHERE id=%s', (rel_path,))
        cur.execute(
            '''
            INSERT INTO memory_secure_audit (id, source, risk)
            VALUES (%s, %s, %s)
            ON CONFLICT (id) DO UPDATE SET
              risk = EXCLUDED.risk,
              updated_at = now()
            ''',
            (rel_path, rel_path, 'excluded_or_sensitive'),
        )
        print(f'[QUARANTINED] {rel_path}')
        return

    text = abs_path.read_text(encoding='utf-8', errors='ignore')
    if len(text.strip()) < MIN_TEXT_LEN:
        cur.execute('DELETE FROM memory_primary WHERE id=%s', (rel_path,))
        cur.execute('DELETE FROM memory_secure_audit WHERE id=%s', (rel_path,))
        print(f'[PRUNED] {rel_path}')
        return

    try:
        embedding = embed_text(text)
    except Exception as error:
        print(f'[EMBED-ERROR] {rel_path} :: {error}')
        return
    cur.execute(
        '''
        INSERT INTO memory_primary (id, source, content, content_hash, embedding)
        VALUES (%s, %s, %s, %s, %s::vector)
        ON CONFLICT (id) DO UPDATE SET
          source = EXCLUDED.source,
          content = EXCLUDED.content,
          content_hash = EXCLUDED.content_hash,
          embedding = EXCLUDED.embedding,
          updated_at = now()
        ''',
        (rel_path, rel_path, text, sha256_text(text), vector_literal(embedding)),
    )
    cur.execute('DELETE FROM memory_secure_audit WHERE id=%s', (rel_path,))
    print(f'[UPSERTED] {rel_path}')


def run(changes_file: Path) -> None:
    events = parse_changes(changes_file)
    if not events:
        return

    with index_lock():
        with get_conn() as conn:
            with conn.cursor() as cur:
                for event in events:
                    code = event['code']
                    if code == 'D':
                        old_path = event['old']
                        cur.execute('DELETE FROM memory_primary WHERE id=%s', (old_path,))
                        cur.execute('DELETE FROM memory_secure_audit WHERE id=%s', (old_path,))
                        print(f'[DELETED] {old_path}')
                        continue

                    if code == 'R':
                        old_path = event['old']
                        new_path = event['new']
                        cur.execute('DELETE FROM memory_primary WHERE id=%s', (old_path,))
                        cur.execute('DELETE FROM memory_secure_audit WHERE id=%s', (old_path,))
                        print(f'[RENAMED-OLD-DELETED] {old_path}')
                        upsert_file(cur, new_path)
                        continue

                    upsert_file(cur, event['path'])
            conn.commit()


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument('--changes-file', required=True)
    args = parser.parse_args()
    run(Path(args.changes_file))


if __name__ == '__main__':
    main()