feat(openclaw): JSONL-backed retain queue for external API resilience (#740)

When the external Hindsight API is unreachable, retain requests are
buffered as JSON lines in a local file and automatically flushed once
connectivity is restored. Queue survives process restarts.

- Only active in external API mode (local daemon handles its own persistence)
- Zero dependencies — uses only Node built-ins (fs, crypto)
- Bulk removal via removeMany() for O(1) file rewrites during flush
- Cached item count so size() is O(1)
- Configurable: retainQueuePath, retainQueueMaxAgeMs (-1 = forever),
  retainQueueFlushIntervalMs (default 60s)
- Flushes on successful retain and on a periodic timer
- All logging routed through structured logger (api.logger)

Co-authored-by: billy <billy@oclaw.local>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Antoine Khater <ak@ptgroup.eu>

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"
     }
   }
 }

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<typeof setInterval> | 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<void> {
+  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);
+        try {
-        debug(`[Hindsight] Retained ${messageCount} messages to bank ${bankId} for session ${documentId}`);
+          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);
       }

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<string, unknown>;
+  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<string, unknown>): 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 {}
+}

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 {