diff --git a/state_10-40-29-186.db b/state_10-40-29-186.db index 10c2c83..3fc8bc6 100644 Binary files a/state_10-40-29-186.db and b/state_10-40-29-186.db differ diff --git a/state_merged.db b/state_merged.db new file mode 100644 index 0000000..33286f4 Binary files /dev/null and b/state_merged.db differ diff --git a/sync.sh b/sync.sh index be69c1d..3ed8b7d 100755 --- a/sync.sh +++ b/sync.sh @@ -8,24 +8,32 @@ cd "$SYNC_DIR" echo "[$(date '+%H:%M:%S')] Sync from $HOSTNAME..." -# ── Step 1: Export local state.db → state_.db ────────────────── +# ── Step 1: Export local state.db (via temp dir to avoid lock) ──────────── python3 << PYEOF -import sqlite3, os, shutil +import sqlite3, os, shutil, tempfile local_db = os.path.join(os.environ['HERMES_HOME'], 'state.db') export_db = os.path.join(os.environ['SYNC_DIR'], f"state_{os.environ['HOSTNAME']}.db") +tmpdir = tempfile.mkdtemp(prefix='hs_exp_') -# Checkpoint WAL try: conn = sqlite3.connect(local_db) conn.execute('PRAGMA wal_checkpoint(TRUNCATE)') conn.execute('PRAGMA optimize') conn.close() -except Exception as e: - print(f'Checkpoint: {e}', file=__import__('sys').stderr) -shutil.copy2(local_db, export_db) -print(f'Exported: {export_db}') + tmp_db = os.path.join(tmpdir, 'db') + shutil.copy2(local_db, tmp_db) + + test = sqlite3.connect(tmp_db) + r = test.execute('SELECT COUNT(*) FROM sessions').fetchone()[0] + m = test.execute('SELECT COUNT(*) FROM messages').fetchone()[0] + test.close() + + shutil.copy2(tmp_db, export_db) + print(f'Exported: {r}s/{m}m') +finally: + shutil.rmtree(tmpdir, ignore_errors=True) PYEOF # ── Step 2: Git stage ──────────────────────────────────────────────────── @@ -65,42 +73,26 @@ fi # ── Step 4: Merge all state_*.db → state_merged.db ────────────────────── python3 << 'PYEOF' -import sqlite3, os, glob, shutil +import sqlite3, os, glob, shutil, tempfile sync_dir = os.environ['SYNC_DIR'] -hermes_home = os.environ['HERMES_HOME'] merged_path = os.path.join(sync_dir, 'state_merged.db') db_files = sorted(glob.glob(os.path.join(sync_dir, 'state_*.db'))) db_files = [f for f in db_files if not f.endswith('_merged.db')] -print(f'DBs to merge: {[os.path.basename(f) for f in db_files]}') +print(f'Merging {len(db_files)} DBs') -if os.path.exists(merged_path): - os.remove(merged_path) +tmpdir = tempfile.mkdtemp(prefix='hs_merge_') +tmp_merged = os.path.join(tmpdir, 'merged.db') -def get_schema(conn, table): - cols = conn.execute(f'PRAGMA table_info("{table}")').fetchall() - return [c[1] for c in cols] # column names +try: + # Create merged DB with proper schema + conn = sqlite3.connect(tmp_merged) + conn.execute('PRAGMA journal_mode=DELETE') + conn.execute('PRAGMA locking_mode=NORMAL') + conn.execute('PRAGMA synchronous=FULL') -def copy_table(src_conn, dst_conn, table): - cols = get_schema(src_conn, table) - placeholders = ','.join(['?'] * len(cols)) - col_names = ','.join(f'"{c}"' for c in cols) - - rows = src_conn.execute(f'SELECT {col_names} FROM "{table}"').fetchall() - for row in rows: - dst_conn.execute(f'INSERT OR REPLACE INTO "{table}" ({col_names}) VALUES ({placeholders})', row) - return len(rows) - -# Open merged DB and rebuild from scratch -conn_merged = sqlite3.connect(merged_path) -conn_merged.execute('PRAGMA journal_mode=WAL') -conn_merged.execute('PRAGMA synchronous=NORMAL') - -# We need at least one source to initialize schema -if not db_files: - print('No state DBs found, creating empty merged DB') - conn_merged.execute(''' - CREATE TABLE IF NOT EXISTS sessions ( + conn.execute(''' + CREATE TABLE sessions ( id TEXT PRIMARY KEY, source TEXT NOT NULL, user_id TEXT, model TEXT, model_config TEXT, system_prompt TEXT, parent_session_id TEXT, started_at REAL, ended_at REAL, @@ -112,8 +104,9 @@ if not db_files: estimated_cost_usd REAL, actual_cost_usd REAL, cost_status TEXT, cost_source TEXT, pricing_version TEXT, title TEXT )''') - conn_merged.execute(''' - CREATE TABLE IF NOT EXISTS messages ( + + conn.execute(''' + CREATE TABLE messages ( id INTEGER PRIMARY KEY, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT, tool_call_id TEXT, tool_calls TEXT, tool_name TEXT, @@ -121,40 +114,50 @@ if not db_files: finish_reason TEXT, reasoning TEXT, reasoning_details TEXT, codex_reasoning_items TEXT )''') - conn_merged.commit() -else: - first_db = db_files[0] - src_conn = sqlite3.connect(first_db) - src_conn.execute('PRAGMA wal_checkpoint(TRUNCATE)') - - for table in ['sessions', 'messages']: - cols = get_schema(src_conn, table) - col_defs = [] - for c in cols: - if c == 'id' and table == 'messages': - col_defs.append('id INTEGER PRIMARY KEY') - elif c == 'id': - col_defs.append(f'{c} TEXT PRIMARY KEY') - else: - col_defs.append(f'{c} TEXT') - - conn_merged.execute(f'DROP TABLE IF EXISTS "{table}"') - conn_merged.execute(f'CREATE TABLE "{table}" ({", ".join(col_defs)})') - - placeholders = ','.join(['?'] * len(cols)) - col_names = ','.join(f'"{c}"' for c in cols) - rows = src_conn.execute(f'SELECT {col_names} FROM "{table}"').fetchall() - for row in rows: - conn_merged.execute(f'INSERT OR REPLACE INTO "{table}" ({col_names}) VALUES ({placeholders})', row) - print(f' {os.path.basename(first_db)}.{table}: {len(rows)} rows') - src_conn.close() + conn.execute('CREATE INDEX idx_msg_session ON messages(session_id)') + conn.execute('CREATE INDEX idx_msg_ts ON messages(timestamp)') -conn_merged.execute('PRAGMA wal_checkpoint(TRUNCATE)') -conn_merged.commit() -conn_merged.close() + for db_file in db_files: + name = os.path.basename(db_file) + # Copy to temp file first (avoids WAL lock issues with open connections) + tmp_copy = os.path.join(tmpdir, name) + shutil.copy2(db_file, tmp_copy) + # Also copy WAL if exists + if os.path.exists(db_file + '-wal'): + shutil.copy2(db_file + '-wal', tmp_copy + '-wal') -print(f'Merged: {merged_path} ({os.path.getsize(merged_path)/1024:.0f} KB)') + src = sqlite3.connect(tmp_copy) + + s_cnt = src.execute('SELECT COUNT(*) FROM sessions').fetchone()[0] + m_cnt = src.execute('SELECT COUNT(*) FROM messages').fetchone()[0] + print(f' {name}: {s_cnt}s/{m_cnt}m') + + sess_rows = src.execute('SELECT * FROM sessions').fetchall() + sess_cols = len(src.execute('PRAGMA table_info(sessions)').fetchall()) + for row in sess_rows: + conn.execute(f'INSERT OR REPLACE INTO sessions VALUES ({",".join(["?"]*sess_cols)})', row) + + msg_rows = src.execute('SELECT * FROM messages').fetchall() + msg_cols = len(src.execute('PRAGMA table_info(messages)').fetchall()) + for row in msg_rows: + conn.execute(f'INSERT OR IGNORE INTO messages VALUES ({",".join(["?"]*msg_cols)})', row) + + src.close() + os.remove(tmp_copy) + if os.path.exists(tmp_copy + '-wal'): + os.remove(tmp_copy + '-wal') + + conn.commit() + conn.close() + + if os.path.exists(merged_path): + os.remove(merged_path) + shutil.copy2(tmp_merged, merged_path) + print(f'Merged: {os.path.getsize(merged_path)/1024:.0f} KB') + +finally: + shutil.rmtree(tmpdir, ignore_errors=True) PYEOF # ── Step 5: Push ───────────────────────────────────────────────────────── @@ -171,36 +174,36 @@ fi # ── Step 6: Restore merged state to local hermes ───────────────────────── python3 << 'PYEOF' -import sqlite3, os, shutil +import sqlite3, os, shutil, tempfile -sync_dir = os.environ['SYNC_DIR'] hermes_home = os.environ['HERMES_HOME'] -merged_path = os.path.join(sync_dir, 'state_merged.db') +merged_path = os.path.join(os.environ['SYNC_DIR'], 'state_merged.db') local_db = os.path.join(hermes_home, 'state.db') if not os.path.exists(merged_path): print('No merged DB, skipping restore') else: - # Checkpoint + close local + tmpdir = tempfile.mkdtemp(prefix='hs_rest_') try: conn = sqlite3.connect(local_db) conn.execute('PRAGMA wal_checkpoint(TRUNCATE)') conn.execute('PRAGMA optimize') conn.close() - except Exception as e: - print(f'Local checkpoint: {e}') - - # Backup - shutil.copy2(local_db, local_db + '.pre_sync_bak') - - # Replace with merged - shutil.copy2(merged_path, local_db) - print(f'Restored merged state to {local_db}') - - # Remove backup after successful restore - bak = local_db + '.pre_sync_bak' - if os.path.exists(bak): - os.remove(bak) + + tmp_db = os.path.join(tmpdir, 'db') + shutil.copy2(merged_path, tmp_db) + + test = sqlite3.connect(tmp_db) + r = test.execute('SELECT COUNT(*) FROM sessions').fetchone()[0] + m = test.execute('SELECT COUNT(*) FROM messages').fetchone()[0] + test.close() + + shutil.copy2(local_db, local_db + '.bak') + shutil.copy2(tmp_db, local_db) + os.remove(local_db + '.bak') + print(f'Restored: {r}s/{m}m') + finally: + shutil.rmtree(tmpdir, ignore_errors=True) PYEOF # ── Step 7: Sync memories + skills (additive) ────────────────────────────