diff --git a/hindsight-integrations/openclaw/openclaw.plugin.json b/hindsight-integrations/openclaw/openclaw.plugin.json index 4fc6b610..3c65579e 100644 --- a/hindsight-integrations/openclaw/openclaw.plugin.json +++ b/hindsight-integrations/openclaw/openclaw.plugin.json @@ -215,6 +215,20 @@ "type": "number", "description": "Interval in ms to batch retain/recall log summaries. 0 = log every event individually. Default: 300000 (5 min).", "default": 300000 + }, + "retainQueuePath": { + "type": "string", + "description": "Path to JSONL file for buffering failed retains (external API mode only). Default: ~/.openclaw/data/hindsight-retain-queue.jsonl" + }, + "retainQueueMaxAgeMs": { + "type": "number", + "description": "Max age in ms for queued retain items. -1 = keep forever.", + "default": -1 + }, + "retainQueueFlushIntervalMs": { + "type": "number", + "description": "How often to attempt flushing queued retains in ms. Default: 60000 (1 min).", + "default": 60000 } }, "additionalProperties": false @@ -349,6 +363,18 @@ "logSummaryIntervalMs": { "label": "Log Summary Interval (ms)", "placeholder": "300000" + }, + "retainQueuePath": { + "label": "Retain Queue File Path", + "placeholder": "~/.openclaw/data/hindsight-retain-queue.jsonl" + }, + "retainQueueMaxAgeMs": { + "label": "Retain Queue Max Age (ms)", + "placeholder": "-1 (forever)" + }, + "retainQueueFlushIntervalMs": { + "label": "Retain Queue Flush Interval (ms)", + "placeholder": "60000" } } } diff --git a/hindsight-integrations/openclaw/src/index.ts b/hindsight-integrations/openclaw/src/index.ts index fdbcd43e..e2b48bab 100644 --- a/hindsight-integrations/openclaw/src/index.ts +++ b/hindsight-integrations/openclaw/src/index.ts @@ -1,11 +1,14 @@ import type { MoltbotPluginAPI, PluginConfig, PluginHookAgentContext, MemoryResult } from './types.js'; import { HindsightEmbedManager } from './embed-manager.js'; import { HindsightClient, type HindsightClientOptions } from './client.js'; +import { RetainQueue } from './retain-queue.js'; import { createHash } from 'crypto'; -import { dirname } from 'path'; +import { dirname, join } from 'path'; import { fileURLToPath } from 'url'; import * as log from './logger.js'; import { configureLogger, setApiLogger, stopLogger } from './logger.js'; +import { mkdirSync } from 'fs'; +import { homedir } from 'os'; // Debug logging: silent by default, enable with debug: true or logLevel: 'debug' let debugEnabled = false; @@ -50,6 +53,69 @@ let lastReinitAttempt = 0; let isReinitInProgress = false; const REINIT_COOLDOWN_MS = 30_000; +// Retain queue (external API mode only) +let retainQueue: RetainQueue | null = null; +let retainQueueFlushTimer: ReturnType | null = null; +let isFlushInProgress = false; +const DEFAULT_FLUSH_INTERVAL_MS = 60_000; // 1 min + +/** + * Attempt to flush pending retains from the queue. + * Each item is sent exactly as it would have been originally — same bank, payload, metadata. + */ +async function flushRetainQueue(): Promise { + if (!retainQueue || isFlushInProgress) return; + const pending = retainQueue.size(); + if (pending === 0) return; + + isFlushInProgress = true; + let flushed = 0; + let failed = 0; + + try { + if (!clientOptions) return; // no client config — can't flush + + // Cleanup expired items first + retainQueue.cleanup(); + + const items = retainQueue.peek(50); + const flushedIds: string[] = []; + for (const item of items) { + try { + let bankClient = clientsByBankId.get(item.bankId); + if (!bankClient) { + bankClient = new HindsightClient(clientOptions); + bankClient.setBankId(item.bankId); + clientsByBankId.set(item.bankId, bankClient); + } + + await bankClient.retain({ + content: item.content, + document_id: item.documentId, + metadata: item.metadata, + }); + + flushedIds.push(item.id); + flushed++; + } catch { + // API still down — stop trying this batch + failed++; + break; + } + } + + if (flushedIds.length > 0) retainQueue.removeMany(flushedIds); + const remaining = retainQueue.size(); + if (flushed > 0) { + log.info(`queue flush: ${flushed} queued retains delivered${remaining > 0 ? `, ${remaining} still pending` : ', queue empty'}`); + } else if (failed > 0) { + debug(`[Hindsight] Queue flush: API still unreachable, ${remaining} retains pending`); + } + } finally { + isFlushInProgress = false; + } +} + const DEFAULT_RECALL_PROMPT_PREAMBLE = 'Relevant memories from past conversations (prioritize recent when conflicting). Only use memories that are directly useful to continue this conversation; ignore the rest:'; @@ -808,6 +874,31 @@ export default function (api: MoltbotPluginAPI) { usingExternalApi = true; debug(`[Hindsight] ✓ Using external API: ${externalApi.apiUrl}`); + // Initialize retain queue (external API mode only) + try { + const queueDir = pluginConfig.retainQueuePath + ? dirname(pluginConfig.retainQueuePath) + : join(homedir(), '.openclaw', 'data'); + mkdirSync(queueDir, { recursive: true }); + const queuePath = pluginConfig.retainQueuePath || join(queueDir, 'hindsight-retain-queue.jsonl'); + const queueFlushInterval = pluginConfig.retainQueueFlushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS; + const queueMaxAge = pluginConfig.retainQueueMaxAgeMs ?? -1; + retainQueue = new RetainQueue({ filePath: queuePath, maxAgeMs: queueMaxAge }); + const pending = retainQueue.size(); + if (pending > 0) { + log.info(`retain queue: ${pending} items pending from previous session, will flush shortly`); + } + debug(`[Hindsight] Retain queue initialized: ${queuePath}`); + + // Periodic flush timer + if (queueFlushInterval > 0) { + retainQueueFlushTimer = setInterval(flushRetainQueue, queueFlushInterval); + retainQueueFlushTimer.unref?.(); + } + } catch (error) { + log.warn(`could not initialize retain queue: ${error}`); + } + // Set env vars so CLI commands (uvx hindsight-embed) use external API process.env.HINDSIGHT_EMBED_API_URL = externalApi.apiUrl; if (externalApi.apiToken) { @@ -1027,6 +1118,20 @@ export default function (api: MoltbotPluginAPI) { embedManager = null; } + // Close retain queue + if (retainQueueFlushTimer) { + clearInterval(retainQueueFlushTimer); + retainQueueFlushTimer = null; + } + if (retainQueue) { + const pending = retainQueue.size(); + if (pending > 0) { + debug(`[Hindsight] Service stopping with ${pending} queued retains (will resume on next start)`); + } + retainQueue.close(); + retainQueue = null; + } + client = null; clientOptions = null; clientsByBankId.clear(); @@ -1303,7 +1408,7 @@ ${memoriesFormatted} // Retain to Hindsight debug(`[Hindsight] Retaining to bank ${bankId}, document: ${documentId}, chars: ${transcript.length}\n---\n${transcript.substring(0, 500)}${transcript.length > 500 ? '\n...(truncated)' : ''}\n---`); - await client.retain({ + const retainRequest = { content: transcript, document_id: documentId, metadata: { @@ -1313,10 +1418,27 @@ ${memoriesFormatted} channel_id: effectiveCtx?.channelId, sender_id: effectiveCtx?.senderId, }, - }); + }; - log.trackRetain(bankId, messageCount); - debug(`[Hindsight] Retained ${messageCount} messages to bank ${bankId} for session ${documentId}`); + try { + await client.retain(retainRequest); + log.trackRetain(bankId, messageCount); + debug(`[Hindsight] Retained ${messageCount} messages to bank ${bankId} for session ${documentId}`); + + // After a successful retain, try flushing any queued items + if (retainQueue && retainQueue.size() > 0) { + flushRetainQueue().catch(() => {}); + } + } catch (retainError) { + // Queue the failed retain for later delivery (external API mode only) + if (retainQueue) { + retainQueue.enqueue(bankId, retainRequest, retainRequest.metadata); + const pending = retainQueue.size(); + log.warn(`API unreachable — retain queued (${pending} pending, bank: ${bankId}): ${retainError instanceof Error ? retainError.message : retainError}`); + } else { + log.error('error retaining messages', retainError); + } + } } catch (error) { log.error('error retaining messages', error); } diff --git a/hindsight-integrations/openclaw/src/retain-queue.ts b/hindsight-integrations/openclaw/src/retain-queue.ts new file mode 100644 index 00000000..de0813c1 --- /dev/null +++ b/hindsight-integrations/openclaw/src/retain-queue.ts @@ -0,0 +1,122 @@ +/** + * JSONL-backed retain queue for buffering failed HTTP retains. + * + * When the external Hindsight API is unreachable, retain requests are stored + * as JSON lines in a local file and flushed once connectivity is restored. + * Only used in external API mode — local daemon mode handles its own persistence. + * + * Zero dependencies — uses only Node built-ins. + */ + +import { readFileSync, writeFileSync, appendFileSync, existsSync, renameSync, unlinkSync } from 'fs'; +import { randomBytes } from 'crypto'; +import type { RetainRequest } from './types.js'; + +export interface QueuedRetain { + id: string; + bankId: string; + content: string; + documentId: string; + metadata: Record; + createdAt: string; // ISO 8601 +} + +export interface RetainQueueOptions { + /** Path to the JSONL queue file */ + filePath: string; + /** Max age in ms for queued items. -1 = keep forever (default) */ + maxAgeMs?: number; +} + +export class RetainQueue { + private filePath: string; + private maxAgeMs: number; + private cachedSize: number; + + constructor(opts: RetainQueueOptions) { + this.filePath = opts.filePath; + this.maxAgeMs = opts.maxAgeMs ?? -1; + // Initialize cached size from file + this.cachedSize = this.readAll().length; + } + + /** Store a failed retain for later delivery — exact same payload as the HTTP request */ + enqueue(bankId: string, request: RetainRequest, metadata?: Record): void { + const item: QueuedRetain = { + id: `${Date.now()}-${randomBytes(4).toString('hex')}`, + bankId, + content: request.content, + documentId: request.document_id || 'conversation', + metadata: metadata || request.metadata || {}, + createdAt: new Date().toISOString(), + }; + appendFileSync(this.filePath, JSON.stringify(item) + '\n', 'utf8'); + this.cachedSize++; + } + + /** Read all pending items from file */ + private readAll(): QueuedRetain[] { + if (!existsSync(this.filePath)) return []; + const content = readFileSync(this.filePath, 'utf8').trim(); + if (!content) return []; + const items: QueuedRetain[] = []; + for (const line of content.split('\n')) { + try { + items.push(JSON.parse(line) as QueuedRetain); + } catch { + // skip malformed lines + } + } + return items; + } + + /** Atomically rewrite the file with the given items */ + private writeAll(items: QueuedRetain[]): void { + if (items.length === 0) { + try { unlinkSync(this.filePath); } catch { /* already gone */ } + this.cachedSize = 0; + return; + } + const tmpPath = this.filePath + '.tmp'; + writeFileSync(tmpPath, items.map(i => JSON.stringify(i)).join('\n') + '\n', 'utf8'); + renameSync(tmpPath, this.filePath); + this.cachedSize = items.length; + } + + /** Get oldest pending items (FIFO) */ + peek(limit = 50): QueuedRetain[] { + return this.readAll().slice(0, limit); + } + + /** Remove a single item by id */ + remove(id: string): void { + const items = this.readAll().filter(i => i.id !== id); + this.writeAll(items); + } + + /** Remove multiple items by id in a single file rewrite */ + removeMany(ids: string[]): void { + const idSet = new Set(ids); + const items = this.readAll().filter(i => !idSet.has(i.id)); + this.writeAll(items); + } + + /** Number of items waiting (cached, O(1)) */ + size(): number { + return this.cachedSize; + } + + /** Remove items older than maxAgeMs (no-op when maxAgeMs is -1) */ + cleanup(): number { + if (this.maxAgeMs < 0) return 0; + const cutoff = Date.now() - this.maxAgeMs; + const items = this.readAll(); + const kept = items.filter(i => new Date(i.createdAt).getTime() >= cutoff); + const removed = items.length - kept.length; + if (removed > 0) this.writeAll(kept); + return removed; + } + + /** No-op for JSONL (no connection to close), kept for API compatibility */ + close(): void {} +} diff --git a/hindsight-integrations/openclaw/src/types.ts b/hindsight-integrations/openclaw/src/types.ts index 48a6af19..9ee903fb 100644 --- a/hindsight-integrations/openclaw/src/types.ts +++ b/hindsight-integrations/openclaw/src/types.ts @@ -83,6 +83,9 @@ export interface PluginConfig { debug?: boolean; // Enable debug logging (default: false) logLevel?: 'off' | 'error' | 'warning' | 'info' | 'debug'; // Console log verbosity (default: 'info'). logSummaryIntervalMs?: number; // Batch retain/recall log summaries over this interval in ms. 0 = log every event. Default: 300000 (5 min). + retainQueuePath?: string; // Path to JSONL file for buffering failed retains. Default: ~/.openclaw/data/hindsight-retain-queue.jsonl + retainQueueMaxAgeMs?: number; // Max age in ms for queued items. -1 = keep forever (default: -1) + retainQueueFlushIntervalMs?: number; // How often to attempt flushing the queue in ms. Default: 60000 (1 min) } export interface ServiceConfig {