133 lines
3.7 KiB
JavaScript
133 lines
3.7 KiB
JavaScript
import fs from 'node:fs';
|
|
import path from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
import Database from 'better-sqlite3';
|
|
|
|
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
|
import os from 'node:os';
|
|
const SESSIONS_DIR = process.env.SESSIONS_DIR || path.join(os.homedir(), '.openclaw/agents/main/sessions');
|
|
const DB_FILE = path.join(__dirname, 'sessions.db');
|
|
|
|
const db = new Database(DB_FILE);
|
|
|
|
// Schema
|
|
db.exec(`
|
|
CREATE TABLE IF NOT EXISTS entries (
|
|
id TEXT PRIMARY KEY,
|
|
session TEXT NOT NULL,
|
|
type TEXT,
|
|
timestamp TEXT,
|
|
data TEXT NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_session ON entries (session);
|
|
CREATE INDEX IF NOT EXISTS idx_type ON entries (type);
|
|
CREATE INDEX IF NOT EXISTS idx_timestamp ON entries (timestamp);
|
|
|
|
`);
|
|
|
|
// Add chainId column if missing (migration)
|
|
try { db.exec('ALTER TABLE entries ADD COLUMN chainId TEXT'); } catch {}
|
|
try { db.exec('CREATE INDEX IF NOT EXISTS idx_chainId ON entries (chainId)'); } catch {}
|
|
|
|
const insertEntry = db.prepare(`
|
|
INSERT OR IGNORE INTO entries (id, session, type, timestamp, data)
|
|
VALUES (@id, @session, @type, @timestamp, @data)
|
|
`);
|
|
const getEntry = db.prepare('SELECT id, data FROM entries WHERE id = ?');
|
|
const setChainId = db.prepare('UPDATE entries SET chainId = @chainId WHERE id = @id');
|
|
const getUnchained = db.prepare(`
|
|
SELECT id, data FROM entries
|
|
WHERE chainId IS NULL
|
|
ORDER BY timestamp ASC
|
|
`);
|
|
|
|
// --- Step 1: Import entries (full scan, INSERT OR IGNORE deduplicates) ---
|
|
const jsonlFiles = fs.readdirSync(SESSIONS_DIR)
|
|
.filter(f => f.endsWith('.jsonl') && !f.includes('.deleted'));
|
|
|
|
let exported = 0;
|
|
|
|
const runImport = db.transaction(() => {
|
|
for (const file of jsonlFiles) {
|
|
const filePath = path.join(SESSIONS_DIR, file);
|
|
const sessionName = file.replace('.jsonl', '');
|
|
const lines = fs.readFileSync(filePath, 'utf-8').split('\n').filter(l => l.trim());
|
|
|
|
for (let i = 0; i < lines.length; i++) {
|
|
try {
|
|
const entry = JSON.parse(lines[i]);
|
|
if (!entry.timestamp) continue;
|
|
const id = entry.id || `${sessionName}-${i}`;
|
|
const res = insertEntry.run({
|
|
id,
|
|
session: sessionName,
|
|
type: entry.type ?? 'unknown',
|
|
timestamp: entry.timestamp,
|
|
data: JSON.stringify(entry),
|
|
});
|
|
if (res.changes > 0) exported++;
|
|
} catch { /* skip */ }
|
|
}
|
|
}
|
|
});
|
|
|
|
runImport();
|
|
|
|
// --- Step 2: Compute chainId for entries that don't have one ---
|
|
// chainId = the nearest user message when walking up the parentId chain.
|
|
// If no user message found, chainId = own id.
|
|
|
|
let chained = 0;
|
|
const chainCache = new Map(); // id -> chainId
|
|
|
|
function resolveChainId(id) {
|
|
if (chainCache.has(id)) return chainCache.get(id);
|
|
|
|
const visited = new Set();
|
|
let currentId = id;
|
|
|
|
while (currentId && !visited.has(currentId)) {
|
|
visited.add(currentId);
|
|
|
|
if (chainCache.has(currentId) && currentId !== id) {
|
|
// Already resolved - reuse
|
|
const result = chainCache.get(currentId);
|
|
chainCache.set(id, result);
|
|
return result;
|
|
}
|
|
|
|
const row = getEntry.get(currentId);
|
|
if (!row) break;
|
|
|
|
const d = JSON.parse(row.data);
|
|
|
|
// If this is a user message, it's the chain root
|
|
if (d.message?.role === 'user') {
|
|
chainCache.set(id, currentId);
|
|
return currentId;
|
|
}
|
|
|
|
currentId = d.parentId || null;
|
|
}
|
|
|
|
// No user message found - self is root
|
|
chainCache.set(id, id);
|
|
return id;
|
|
}
|
|
|
|
const runChaining = db.transaction(() => {
|
|
const unchained = getUnchained.all();
|
|
|
|
for (const row of unchained) {
|
|
const chainId = resolveChainId(row.id);
|
|
setChainId.run({ id: row.id, chainId });
|
|
chained++;
|
|
}
|
|
});
|
|
|
|
runChaining();
|
|
db.close();
|
|
|
|
console.log(`${exported} neue, ${chained} verkettet`);
|