From 13d23ad3fdf699ae5f36e9b75de92f403f8746ac Mon Sep 17 00:00:00 2001 From: william arzt Date: Sat, 18 Jul 2026 21:24:20 -0400 Subject: [PATCH 1/4] fix(openclaw): keep lens retries out of the inbound dedupe tracker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Retries reuse the original message id, and the SDK inbound dedupe tracker keys on MessageSid and remembers failed runs — so a manual retry of a failed run was silently swallowed. Suffix retry dispatches with the lens id to bust dedupe, keeping the canonical id in MessageSidFull. Co-Authored-By: Claude Opus 4.7 --- packages/openclaw/src/monitor/index.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/packages/openclaw/src/monitor/index.ts b/packages/openclaw/src/monitor/index.ts index b8596ce2ab..a05baa70a2 100644 --- a/packages/openclaw/src/monitor/index.ts +++ b/packages/openclaw/src/monitor/index.ts @@ -2908,7 +2908,14 @@ export async function monitorTlonProvider( CommandSource: 'text' as const, Provider: 'tlon', Surface: 'tlon', - MessageSid: messageId, + // Retries reuse the original message id; the SDK inbound dedupe + // tracker keys on MessageSid and remembers failed runs, so a bare + // retry would be silently swallowed. Suffix retries with the lens id + // to bust dedupe while keeping the canonical id in MessageSidFull. + MessageSid: params.retryOf + ? `${messageId}#retry:${lens.lensId}` + : messageId, + MessageSidFull: messageId, // Include downloaded media attachments (MediaPaths/MediaUrls/MediaTypes for OpenClaw media pipeline) ...(attachments.length > 0 && { MediaPaths: attachments.map((a) => a.path), From 23003baf6349cd919426ab4ffa35a47136acf013 Mon Sep 17 00:00:00 2001 From: william arzt Date: Sat, 18 Jul 2026 21:24:25 -0400 Subject: [PATCH 2/4] feat(openclaw): recover interrupted lens runs as aborted on boot Persist non-terminal run snapshots to an inflight checkpoint sidecar and, on the first store init per process, mark any runs the gateway abandoned (SIGTERM/SIGKILL) as aborted so they finalize and become retryable instead of hanging in a non-terminal state forever. Terminal events save the durable record before removing the checkpoint entry (and keep the entry when the save fails), and boot recovery removes checkpoint entries individually, so a crash or write failure between the two writes can no longer lose the run entirely. Co-Authored-By: Claude Opus 4.7 --- .../openclaw/src/context-lens-routes.test.ts | 1 + .../openclaw/src/context-lens-store.test.ts | 275 +++++++++++++++++- packages/openclaw/src/context-lens-store.ts | 259 ++++++++++++++++- 3 files changed, 529 insertions(+), 6 deletions(-) diff --git a/packages/openclaw/src/context-lens-routes.test.ts b/packages/openclaw/src/context-lens-routes.test.ts index 6e8af287d9..b7e59a4c35 100644 --- a/packages/openclaw/src/context-lens-routes.test.ts +++ b/packages/openclaw/src/context-lens-routes.test.ts @@ -269,6 +269,7 @@ describe('context lens run route', () => { save: () => {}, size: () => 1, get: (lensId) => (lensId === stored.lensId ? stored : null), + list: () => [stored], }); try { const { routes } = setupRoutes(); diff --git a/packages/openclaw/src/context-lens-store.test.ts b/packages/openclaw/src/context-lens-store.test.ts index 7f377a9397..cd3f98df8a 100644 --- a/packages/openclaw/src/context-lens-store.test.ts +++ b/packages/openclaw/src/context-lens-store.test.ts @@ -7,11 +7,20 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { publishContextLensEvent } from './context-lens-events.js'; import { createContextLensStore, + createInflightCheckpoint, getContextLensStore, + inflightCheckpointPath, initContextLensStore, + markLensAborted, + recoverInterruptedRuns, setContextLensStore, } from './context-lens-store.js'; -import { type ContextLens, createContextLensRegistry } from './context-lens.js'; +import { + type ContextLens, + type ContextLensStatus, + createContextLensRegistry, +} from './context-lens.js'; +import { sharedSlot } from './shared-state.js'; let tmpDir: string; let filePath: string; @@ -43,6 +52,18 @@ function makeLens( }; } +function makeInflightLens( + overrides: { messageId?: string; status?: ContextLensStatus } = {} +): ContextLens { + const registry = createContextLensRegistry({ ttlMs: 60_000 }); + const lens = registry.create({ + messageId: overrides.messageId ?? 'inflight-1', + chatType: 'dm', + trigger: 'dm', + }); + return { ...lens, status: overrides.status ?? 'dispatching' }; +} + describe('createContextLensStore', () => { it('round-trips a saved run', () => { const store = createContextLensStore({ filePath }); @@ -154,6 +175,209 @@ describe('createContextLensStore', () => { }); }); +describe('createInflightCheckpoint', () => { + it('upserts, removes, and survives reload', () => { + const checkpointPath = inflightCheckpointPath(filePath); + const checkpoint = createInflightCheckpoint({ filePath: checkpointPath }); + const a = makeInflightLens({ messageId: 'ckpt-a' }); + const b = makeInflightLens({ messageId: 'ckpt-b' }); + + checkpoint.upsert(a); + checkpoint.upsert(b); + expect( + checkpoint + .list() + .map((l) => l.lensId) + .toSorted() + ).toEqual([a.lensId, b.lensId].toSorted()); + + checkpoint.remove(a.lensId); + const reloaded = createInflightCheckpoint({ filePath: checkpointPath }); + expect(reloaded.list().map((l) => l.lensId)).toEqual([b.lensId]); + }); + + it('keeps the latest snapshot per lensId', () => { + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(filePath), + }); + const lens = makeInflightLens({ + messageId: 'ckpt-dup', + status: 'assembling', + }); + checkpoint.upsert(lens); + checkpoint.upsert({ ...lens, status: 'tool_running' }); + expect(checkpoint.list()).toHaveLength(1); + expect(checkpoint.list()[0].status).toBe('tool_running'); + }); + + it('clear empties the file', () => { + const checkpointPath = inflightCheckpointPath(filePath); + const checkpoint = createInflightCheckpoint({ filePath: checkpointPath }); + checkpoint.upsert(makeInflightLens()); + checkpoint.clear(); + expect(checkpoint.list()).toHaveLength(0); + expect(fs.readFileSync(checkpointPath, 'utf8')).toBe(''); + }); + + it('evicts the oldest entry past maxInflight and warns about it', () => { + const warnings: string[] = []; + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(filePath), + maxInflight: 2, + logger: { info: () => {}, warn: (message) => warnings.push(message) }, + }); + const a = makeInflightLens({ messageId: 'evict-a' }); + const b = makeInflightLens({ messageId: 'evict-b' }); + const c = makeInflightLens({ messageId: 'evict-c' }); + checkpoint.upsert(a); + checkpoint.upsert(b); + checkpoint.upsert(c); + expect(checkpoint.list().map((l) => l.lensId)).toEqual([ + b.lensId, + c.lensId, + ]); + expect(warnings.join('\n')).toContain(a.lensId); + expect(warnings.join('\n')).toContain('will not be recovered'); + }); +}); + +describe('markLensAborted', () => { + it('finalizes status, timings, and open tool runs', () => { + const now = 10_000; + const base = makeInflightLens({ status: 'tool_running' }); + const lens: ContextLens = { + ...base, + createdAt: now - 5_000, + tools: { + ...base.tools, + runs: [ + { + id: 't1', + callIndex: 1, + name: 'search', + startedAt: now - 3_000, + completedAt: null, + durationMs: null, + status: 'running', + }, + { + id: 't2', + callIndex: 2, + name: 'fetch', + startedAt: now - 8_000, + completedAt: now - 7_000, + durationMs: 1_000, + status: 'completed', + }, + ], + }, + }; + + const aborted = markLensAborted(lens, now); + + expect(aborted.status).toBe('aborted'); + expect(aborted.error).toContain('Gateway stopped'); + expect(aborted.lifecycle.completedAt).toBe(now); + expect(aborted.lifecycle.durationMs).toBe(5_000); + expect(aborted.tools.runs[0]).toMatchObject({ + status: 'error', + completedAt: now, + durationMs: 3_000, + }); + // A tool run that already finished is left untouched. + expect(aborted.tools.runs[1]).toMatchObject({ + status: 'completed', + durationMs: 1_000, + }); + }); + + it('preserves an existing terminal timestamp and error', () => { + const base = makeInflightLens(); + const lens: ContextLens = { + ...base, + error: 'original error', + lifecycle: { ...base.lifecycle, completedAt: 42, durationMs: 7 }, + }; + const aborted = markLensAborted(lens, 99); + expect(aborted.lifecycle.completedAt).toBe(42); + expect(aborted.lifecycle.durationMs).toBe(7); + expect(aborted.error).toBe('original error'); + }); +}); + +describe('recoverInterruptedRuns', () => { + it('aborts leftover in-flight runs into the store and clears the checkpoint', () => { + const store = createContextLensStore({ filePath }); + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(filePath), + }); + const lens = makeInflightLens({ messageId: 'killed' }); + checkpoint.upsert(lens); + + const recovered = recoverInterruptedRuns({ store, checkpoint }); + + expect(recovered).toBe(1); + expect(store.get(lens.lensId)?.status).toBe('aborted'); + expect(checkpoint.list()).toHaveLength(0); + }); + + it('skips runs already terminal on disk (stale checkpoint line)', () => { + const store = createContextLensStore({ filePath }); + const finished = makeLens({ messageId: 'finished' }); + store.save(finished); + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(filePath), + }); + // Same lensId, but the checkpoint still holds the pre-terminal snapshot. + checkpoint.upsert({ ...finished, status: 'dispatching' }); + + const recovered = recoverInterruptedRuns({ store, checkpoint }); + + expect(recovered).toBe(0); + expect(store.get(finished.lensId)?.status).toBe('completed'); + }); + + it('does no work when the checkpoint is empty', () => { + const store = createContextLensStore({ filePath }); + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(filePath), + }); + expect(recoverInterruptedRuns({ store, checkpoint })).toBe(0); + }); + + it('keeps entries whose store save fails checkpointed for the next boot', () => { + const store = createContextLensStore({ filePath }); + const failing = makeInflightLens({ messageId: 'fails' }); + const ok = makeInflightLens({ messageId: 'ok' }); + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(filePath), + }); + checkpoint.upsert(failing); + checkpoint.upsert(ok); + + const flakyStore = { + ...store, + save: (lens: ContextLens) => { + if (lens.lensId === failing.lensId) { + throw new Error('disk full'); + } + store.save(lens); + }, + }; + const recovered = recoverInterruptedRuns({ + store: flakyStore, + checkpoint, + logger: { info: () => {}, warn: () => {} }, + }); + + expect(recovered).toBe(1); + expect(store.get(ok.lensId)?.status).toBe('aborted'); + expect(store.get(failing.lensId)).toBeNull(); + // The failed entry survives for the next boot's recovery attempt. + expect(checkpoint.list().map((l) => l.lensId)).toEqual([failing.lensId]); + }); +}); + // Keep this block last: initContextLensStore subscribes to the global lens // event stream, and that subscription persists for the rest of the file. describe('initContextLensStore', () => { @@ -199,8 +423,13 @@ describe('initContextLensStore', () => { publishContextLensEvent('final', finalized); expect(store?.get(finalized.lensId)?.messageId).toBe('finalized'); + const aborted = makeLens({ messageId: 'aborted-run' }); + publishContextLensEvent('final', { ...aborted, status: 'aborted' }); + expect(store?.get(aborted.lensId)?.messageId).toBe('aborted-run'); + const reloaded = createContextLensStore({ filePath }); expect(reloaded.get(finalized.lensId)?.messageId).toBe('finalized'); + expect(reloaded.get(aborted.lensId)?.status).toBe('aborted'); }); it('replaces the event subscription on re-init instead of stacking writers', () => { @@ -229,4 +458,48 @@ describe('initContextLensStore', () => { : ''; expect(staleContents).not.toContain(finalized.lensId); }); + + it('checkpoints in-flight events and clears them on terminal', () => { + initContextLensStore( + makeApi({ + enabled: true, + authToken: 'a-token-of-sufficient-length', + store: { path: filePath }, + }) + ); + const checkpointPath = inflightCheckpointPath(filePath); + + const lens = makeInflightLens({ messageId: 'checkpointed' }); + publishContextLensEvent('created', { ...lens, status: 'dispatching' }); + expect(fs.readFileSync(checkpointPath, 'utf8')).toContain(lens.lensId); + + publishContextLensEvent('final', { ...lens, status: 'completed' }); + expect(fs.readFileSync(checkpointPath, 'utf8')).not.toContain(lens.lensId); + }); + + it('recovers an interrupted run as aborted on boot', () => { + // Simulate a prior process that died mid-run: a checkpoint entry exists + // with no terminal record in the store. + const stranded = makeInflightLens({ + messageId: 'stranded', + status: 'tool_running', + }); + fs.writeFileSync( + inflightCheckpointPath(filePath), + `${JSON.stringify(stranded)}\n` + ); + // Recovery is once-per-process; reset the guard so this boot runs it. + sharedSlot('contextLens.inflight.recovered').set(false); + + const store = initContextLensStore( + makeApi({ + enabled: true, + authToken: 'a-token-of-sufficient-length', + store: { path: filePath }, + }) + ); + + expect(store?.get(stranded.lensId)?.status).toBe('aborted'); + expect(fs.readFileSync(inflightCheckpointPath(filePath), 'utf8')).toBe(''); + }); }); diff --git a/packages/openclaw/src/context-lens-store.ts b/packages/openclaw/src/context-lens-store.ts index d7e6e94fd9..c0efb70498 100644 --- a/packages/openclaw/src/context-lens-store.ts +++ b/packages/openclaw/src/context-lens-store.ts @@ -10,19 +10,25 @@ import { resolveTlonAccount } from './types.js'; export const DEFAULT_STORE_RETAIN_DAYS = 30; export const DEFAULT_STORE_MAX_STORED = 500; +export const DEFAULT_MAX_INFLIGHT = 200; const DAY_MS = 24 * 60 * 60 * 1000; +const ABORTED_BY_RECOVERY_ERROR = 'Gateway stopped before this run finished'; + const TERMINAL_STATUSES: ReadonlySet = new Set([ 'completed', 'no_reply', 'timed_out', + 'aborted', 'error', ]); export type ContextLensStore = { save: (lens: ContextLens) => void; get: (lensId: string) => ContextLens | null; + /** Retained runs, oldest→newest by finalization time. */ + list: () => ContextLens[]; size: () => number; filePath: string; }; @@ -40,6 +46,11 @@ const storeUnsubscribeSlot = sharedSlot<() => void>( 'contextLens.store.unsubscribe' ); +// Recovery of interrupted runs is a once-per-process boot operation: a plugin +// re-init mid-run must not re-read the checkpoint and abort runs that are +// still live in the current process. +const recoveredSlot = sharedSlot('contextLens.inflight.recovered'); + export function getContextLensStore(): ContextLensStore | null { return storeSlot.get(); } @@ -52,7 +63,7 @@ export function defaultContextLensStorePath(): string { return path.join(resolveStateDir(), 'tlon', 'context-lens-runs.jsonl'); } -function lensFinalizedAt(lens: ContextLens): number { +export function lensFinalizedAt(lens: ContextLens): number { return lens.lifecycle.completedAt ?? lens.updatedAt ?? lens.createdAt; } @@ -175,14 +186,217 @@ export function createContextLensStore(opts: { } return lens; }, + list: () => { + const now = Date.now(); + return [...runs.values()].filter((lens) => isRetained(lens, now)); + }, size: () => runs.size, }; } +export function inflightCheckpointPath(storeFilePath: string): string { + return `${storeFilePath}.inflight.jsonl`; +} + +/** + * Sidecar checkpoint of runs that are currently in flight. A run only reaches + * the durable store on a terminal event, so a SIGKILL/crash mid-run (or a + * SIGTERM that abandons the run without finalizing) would otherwise leave no + * trace on the gateway and a frozen in-flight copy on the ship. This file + * keeps the latest non-terminal snapshot per run; on the next boot + * `recoverInterruptedRuns` finalizes whatever is left as `aborted`. + * + * Backed by an in-memory map with whole-file atomic rewrites — the file only + * ever holds concurrently-running runs (terminal events remove them), so it + * stays tiny. + */ +export type InflightCheckpoint = { + upsert: (lens: ContextLens) => void; + remove: (lensId: string) => void; + list: () => ContextLens[]; + clear: () => void; + filePath: string; +}; + +export function createInflightCheckpoint(opts: { + filePath: string; + maxInflight?: number | null; + logger?: StoreLogger; +}): InflightCheckpoint { + const filePath = opts.filePath; + const maxInflight = opts.maxInflight ?? DEFAULT_MAX_INFLIGHT; + const logger = opts.logger; + const pending = new Map(); + + const flush = () => { + const lines = [...pending.values()] + .map((lens) => JSON.stringify(lens)) + .join('\n'); + const tmpPath = `${filePath}.tmp`; + fs.writeFileSync(tmpPath, lines.length > 0 ? `${lines}\n` : '', { + mode: 0o600, + }); + fs.renameSync(tmpPath, filePath); + }; + + const load = () => { + fs.mkdirSync(path.dirname(filePath), { recursive: true }); + if (!fs.existsSync(filePath)) { + return; + } + const raw = fs.readFileSync(filePath, 'utf8'); + for (const line of raw.split('\n')) { + const trimmed = line.trim(); + if (!trimmed) { + continue; + } + try { + const lens = JSON.parse(trimmed) as ContextLens; + if (typeof lens?.lensId === 'string' && lens.lensId) { + pending.delete(lens.lensId); + pending.set(lens.lensId, lens); + } + } catch { + // Drop malformed checkpoint lines; a partial run record is not worth + // failing boot over. + } + } + }; + + try { + load(); + } catch (error) { + logger?.warn( + `[tlon] Context lens in-flight checkpoint load failed: ${String(error)}` + ); + } + + return { + filePath, + upsert: (lens) => { + pending.delete(lens.lensId); + pending.set(lens.lensId, lens); + while (pending.size > maxInflight) { + const oldest = pending.keys().next().value; + if (!oldest) { + break; + } + pending.delete(oldest); + logger?.warn( + `[tlon] Context lens in-flight checkpoint full (${maxInflight}); evicting ${oldest} — it will not be recovered if the gateway crashes` + ); + } + flush(); + }, + remove: (lensId) => { + if (pending.delete(lensId)) { + flush(); + } + }, + list: () => [...pending.values()], + clear: () => { + if (pending.size === 0 && !fs.existsSync(filePath)) { + return; + } + pending.clear(); + flush(); + }, + }; +} + +/** + * Finalize an interrupted run as `aborted`: stamp completion timings and close + * any tool runs left open when the process died, so the inspector shows a + * coherent terminal record rather than a run frozen mid-flight. + */ +export function markLensAborted( + lens: ContextLens, + now = Date.now() +): ContextLens { + const completedAt = lens.lifecycle.completedAt ?? now; + return { + ...lens, + status: 'aborted', + error: lens.error ?? ABORTED_BY_RECOVERY_ERROR, + tools: { + ...lens.tools, + runs: lens.tools.runs.map((run) => + run.completedAt + ? run + : { + ...run, + completedAt: now, + durationMs: now - run.startedAt, + status: 'error' as const, + error: run.error ?? ABORTED_BY_RECOVERY_ERROR, + } + ), + }, + lifecycle: { + ...lens.lifecycle, + completedAt, + durationMs: lens.lifecycle.durationMs ?? completedAt - lens.createdAt, + }, + updatedAt: now, + }; +} + +/** + * On boot, finalize runs that a previous process left in flight: each leftover + * checkpoint snapshot is marked `aborted` and saved to the durable store, + * where ship-sync's boot replay picks it up and overwrites the ship's frozen + * in-flight copy. Skips runs already terminal on disk (a stale checkpoint line + * whose terminal removal was lost to the crash). Checkpoint entries are + * removed individually as they are handled; an entry whose store save fails + * stays checkpointed so the next boot can retry it. + */ +export function recoverInterruptedRuns(opts: { + store: ContextLensStore; + checkpoint: InflightCheckpoint; + logger?: StoreLogger; + now?: number; +}): number { + const { store, checkpoint, logger } = opts; + const now = opts.now ?? Date.now(); + const pending = checkpoint.list(); + if (pending.length === 0) { + return 0; + } + let recovered = 0; + for (const lens of pending) { + if (TERMINAL_STATUSES.has(lens.status)) { + checkpoint.remove(lens.lensId); + continue; + } + const existing = store.get(lens.lensId); + if (existing && TERMINAL_STATUSES.has(existing.status)) { + checkpoint.remove(lens.lensId); + continue; + } + try { + store.save(markLensAborted(lens, now)); + recovered += 1; + checkpoint.remove(lens.lensId); + } catch (error) { + logger?.warn( + `[tlon] Context lens recovery failed for ${lens.lensId}: ${String(error)}` + ); + } + } + if (recovered > 0) { + logger?.info( + `[tlon] Context lens recovered ${recovered} interrupted run(s) as aborted` + ); + } + return recovered; +} + /** * Wire the disk store to the lens event stream: every event whose lens has * reached a terminal status is persisted (last write wins, so a later - * "final" snapshot for the same lensId replaces the earlier one). + * "final" snapshot for the same lensId replaces the earlier one). Non-terminal + * events are checkpointed to a sidecar so an interrupted run can be recovered + * as `aborted` on the next boot. * * Returns the store, or null when the lens or its store is disabled. */ @@ -209,17 +423,52 @@ export function initContextLensStore(api: { return null; } setContextLensStore(store); + + const checkpoint = createInflightCheckpoint({ + filePath: inflightCheckpointPath(store.filePath), + logger: api.logger, + }); + // Once per process: a mid-run re-init must not re-read the checkpoint and + // abort runs still live in this process. + if (!recoveredSlot.get()) { + recoveredSlot.set(true); + try { + recoverInterruptedRuns({ store, checkpoint, logger: api.logger }); + } catch (error) { + api.logger.warn(`[tlon] Context lens recovery failed: ${String(error)}`); + } + } + storeUnsubscribeSlot.get()?.(); storeUnsubscribeSlot.set( subscribeToContextLensEvents((event) => { - if (!TERMINAL_STATUSES.has(event.lens.status)) { + const lens = event.lens; + if (TERMINAL_STATUSES.has(lens.status)) { + // Durable save first: if the save fails (or the process dies between + // the two writes), the checkpoint entry survives and boot recovery + // finalizes the run instead of losing it entirely. + try { + store.save(lens); + } catch (error) { + api.logger.warn( + `[tlon] Context lens store write failed for ${lens.lensId}: ${String(error)}` + ); + return; + } + try { + checkpoint.remove(lens.lensId); + } catch (error) { + api.logger.warn( + `[tlon] Context lens checkpoint clear failed for ${lens.lensId}: ${String(error)}` + ); + } return; } try { - store.save(event.lens); + checkpoint.upsert(lens); } catch (error) { api.logger.warn( - `[tlon] Context lens store write failed for ${event.lens.lensId}: ${String(error)}` + `[tlon] Context lens checkpoint write failed for ${lens.lensId}: ${String(error)}` ); } }) From bdcba744fd513e6b1e770170f9b5ecd8e63392fa Mon Sep 17 00:00:00 2001 From: william arzt Date: Sat, 18 Jul 2026 21:24:32 -0400 Subject: [PATCH 3/4] feat(openclaw): replay and retry unsynced lens run-finals A run-final poke that failed (ship unreachable, monitor disconnected, or the process died with the poke in flight) left the ship's copy of the run frozen at its last milestone forever. Track synced finals in an atomically-written ledger, replay unsynced terminal runs on boot (24h window, 50-run cap, idempotent ship-side), and arm a single exponential-backoff retry (30s base, 10min cap) on mid-session final failures so the replay re-runs once connectivity returns. The ledger write happens before the backoff reset so a failed write leaves the retry armed to repair it. Co-Authored-By: Claude Opus 4.7 --- .../src/context-lens-ship-sync.test.ts | 346 +++++++++++++++++- .../openclaw/src/context-lens-ship-sync.ts | 271 +++++++++++++- 2 files changed, 606 insertions(+), 11 deletions(-) diff --git a/packages/openclaw/src/context-lens-ship-sync.test.ts b/packages/openclaw/src/context-lens-ship-sync.test.ts index 819bf8ba8f..07a81300af 100644 --- a/packages/openclaw/src/context-lens-ship-sync.test.ts +++ b/packages/openclaw/src/context-lens-ship-sync.test.ts @@ -1,5 +1,8 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; import type { OpenClawConfig } from 'openclaw/plugin-sdk/core'; -import { describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { type ContextLensEvent, @@ -8,10 +11,19 @@ import { import { buildLensRunPayload, createContextLensShipSync, + createLensSyncRetry, initContextLensShipSync, isContextLensEffectivelyEnabled, + loadSyncedLensIds, + recordSyncedLensId, + replayUnsyncedFinalRuns, resolveLensOwner, + syncedLensIdsPath, } from './context-lens-ship-sync.js'; +import { + createContextLensStore, + setContextLensStore, +} from './context-lens-store.js'; import { type ContextLens, createContextLensRegistry } from './context-lens.js'; import { API_CLIENT_PARAMS_SLOT, @@ -365,6 +377,286 @@ describe('createContextLensShipSync', () => { // Keep this block last: initContextLensShipSync subscribes to the global lens // event stream, and the final subscription persists for the rest of the file. +describe('synced lens id bookkeeping', () => { + it('round-trips ids and survives a missing or garbage file', () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lens-synced-')); + const file = syncedLensIdsPath(path.join(dir, 'runs.jsonl')); + try { + expect(loadSyncedLensIds(file).size).toBe(0); + + recordSyncedLensId(file, 'a'); + recordSyncedLensId(file, 'b'); + recordSyncedLensId(file, 'a'); + expect([...loadSyncedLensIds(file)]).toEqual(['b', 'a']); + + fs.writeFileSync(file, 'not json'); + expect(loadSyncedLensIds(file).size).toBe(0); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + + it('reports run finals via onRunFinal only after a successful poke', async () => { + const pokes: RecordedPoke[] = []; + const params = makeParams(pokes); + let connected = false; + const finals: string[] = []; + const sync = createContextLensShipSync({ + owner: '~bus', + logger: silentLogger, + getParams: () => (connected ? params : null), + onRunFinal: (lens) => finals.push(lens.lensId), + }); + + // Dropped (no params): poke never sent, so no bookkeeping. + sync.handleEvent(makeEvent(makeLens({ status: 'completed' }))); + await sync.flush(); + expect(finals).toHaveLength(0); + + connected = true; + // Milestone events are not tracked, only finals. + const lens = makeLens({ status: 'tool_running' }); + sync.handleEvent(makeEvent(lens)); + sync.handleEvent(makeEvent({ ...lens, status: 'aborted' })); + await sync.flush(); + expect(finals).toEqual([lens.lensId]); + }); +}); + +describe('replayUnsyncedFinalRuns', () => { + function makeFinalLens( + overrides: Partial & { completedAt?: number } + ): ContextLens { + const { completedAt, ...rest } = overrides; + const lens = makeLens({ status: 'completed', ...rest }); + return { + ...lens, + lifecycle: { ...lens.lifecycle, completedAt: completedAt ?? Date.now() }, + }; + } + + it('re-pokes only unsynced, recent, non-internal terminal runs', async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lens-replay-')); + const filePath = path.join(dir, 'runs.jsonl'); + try { + const store = createContextLensStore({ filePath }); + const missed = makeFinalLens({ status: 'aborted' }); + const alreadySynced = makeFinalLens({}); + const internal = makeFinalLens({ visibility: 'internal' }); + const stale = makeFinalLens({ + completedAt: Date.now() - 25 * 60 * 60 * 1_000, + }); + for (const lens of [missed, alreadySynced, internal]) { + store.save(lens); + } + // Bypass save()'s own retention so the stale lens is present in memory. + fs.appendFileSync(filePath, `${JSON.stringify(stale)}\n`); + const reloaded = createContextLensStore({ filePath }); + recordSyncedLensId(syncedLensIdsPath(filePath), alreadySynced.lensId); + + const pokes: RecordedPoke[] = []; + const params = makeParams(pokes); + const finals: string[] = []; + const sync = createContextLensShipSync({ + owner: '~bus', + logger: silentLogger, + getParams: () => params, + onRunFinal: (lens) => { + finals.push(lens.lensId); + recordSyncedLensId(syncedLensIdsPath(filePath), lens.lensId); + }, + }); + + expect(replayUnsyncedFinalRuns(reloaded, sync, silentLogger)).toBe(1); + await sync.flush(); + expect(finals).toEqual([missed.lensId]); + + // Second boot: the replayed run is now recorded as synced. + expect(replayUnsyncedFinalRuns(reloaded, sync, silentLogger)).toBe(0); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); +}); + +describe('createLensSyncRetry', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('fires run after the base delay and backs off exponentially', () => { + vi.useFakeTimers(); + let runs = 0; + const retry = createLensSyncRetry({ + run: () => { + runs += 1; + }, + logger: silentLogger, + baseMs: 1_000, + maxMs: 8_000, + }); + + retry.schedule(); + vi.advanceTimersByTime(999); + expect(runs).toBe(0); + vi.advanceTimersByTime(1); + expect(runs).toBe(1); + + // Attempts 2..4 double each time, then cap at maxMs. + retry.schedule(); + vi.advanceTimersByTime(2_000); + expect(runs).toBe(2); + retry.schedule(); + vi.advanceTimersByTime(4_000); + expect(runs).toBe(3); + retry.schedule(); + vi.advanceTimersByTime(8_000); + expect(runs).toBe(4); + retry.schedule(); + vi.advanceTimersByTime(8_000); + expect(runs).toBe(5); + }); + + it('dedupes schedule calls while a retry is pending', () => { + vi.useFakeTimers(); + let runs = 0; + const retry = createLensSyncRetry({ + run: () => { + runs += 1; + }, + logger: silentLogger, + baseMs: 1_000, + }); + + retry.schedule(); + retry.schedule(); + retry.schedule(); + vi.advanceTimersByTime(10_000); + expect(runs).toBe(1); + }); + + it('reset returns to the base delay; cancel drops the pending timer', () => { + vi.useFakeTimers(); + let runs = 0; + const retry = createLensSyncRetry({ + run: () => { + runs += 1; + }, + logger: silentLogger, + baseMs: 1_000, + }); + + retry.schedule(); + vi.advanceTimersByTime(1_000); + retry.reset(); + retry.schedule(); + vi.advanceTimersByTime(1_000); + expect(runs).toBe(2); + + retry.schedule(); + retry.cancel(); + vi.advanceTimersByTime(60_000); + expect(runs).toBe(2); + }); +}); + +describe('onRunFinalFailure', () => { + it('fires when the run-final poke rejects, but not for milestone failures', async () => { + const failures: string[] = []; + const failing: SharedApiClientParams = { + poke: () => Promise.reject(new Error('ship offline')), + shipName: '~zod', + shipUrl: 'http://localhost:8080', + }; + const sync = createContextLensShipSync({ + owner: '~bus', + logger: silentLogger, + getParams: () => failing, + onRunFinalFailure: (lens) => failures.push(lens.lensId), + }); + + const lens = makeLens({ status: 'tool_running' }); + sync.handleEvent(makeEvent(lens)); + await sync.flush(); + expect(failures).toHaveLength(0); + + sync.handleEvent(makeEvent({ ...lens, status: 'completed' })); + await sync.flush(); + expect(failures).toEqual([lens.lensId]); + }); + + it('fires when the final is dropped because no params are published', async () => { + const failures: string[] = []; + const sync = createContextLensShipSync({ + owner: '~bus', + logger: silentLogger, + getParams: () => null, + onRunFinalFailure: (lens) => failures.push(lens.lensId), + }); + + const lens = makeLens({ status: 'completed' }); + sync.handleEvent(makeEvent(lens)); + await sync.flush(); + expect(failures).toEqual([lens.lensId]); + }); + + it('heals a failed final through replay once the ship is reachable again', async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lens-retry-')); + const filePath = path.join(dir, 'runs.jsonl'); + try { + const store = createContextLensStore({ filePath }); + const pokes: RecordedPoke[] = []; + let online = false; + const flaky: SharedApiClientParams = { + poke: (params) => { + if (!online) { + return Promise.reject(new Error('ship offline')); + } + pokes.push(params as RecordedPoke); + return Promise.resolve(undefined); + }, + shipName: '~zod', + shipUrl: 'http://localhost:8080', + }; + const failures: string[] = []; + const sync = createContextLensShipSync({ + owner: '~bus', + logger: silentLogger, + getParams: () => flaky, + onRunFinal: (lens) => { + recordSyncedLensId(syncedLensIdsPath(filePath), lens.lensId); + }, + onRunFinalFailure: (lens) => failures.push(lens.lensId), + }); + + // Run finishes during an outage: store has it, ship poke fails. + const lens = makeLens({ status: 'completed' }); + store.save({ + ...lens, + lifecycle: { ...lens.lifecycle, completedAt: Date.now() }, + }); + sync.handleEvent(makeEvent(lens)); + await sync.flush(); + expect(failures).toEqual([lens.lensId]); + expect(pokes).toHaveLength(0); + + // Connectivity returns; the retry-driven replay finalizes the run. + online = true; + expect(replayUnsyncedFinalRuns(store, sync, silentLogger)).toBe(1); + await sync.flush(); + expect(pokes.map(pokeKind)).toEqual(['configure', 'lens']); + expect( + loadSyncedLensIds(syncedLensIdsPath(filePath)).has(lens.lensId) + ).toBe(true); + + // Nothing left to heal on the next pass. + expect(replayUnsyncedFinalRuns(store, sync, silentLogger)).toBe(0); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); +}); + describe('initContextLensShipSync', () => { it('replaces the event subscription on re-init instead of stacking pokes', async () => { const pokes: RecordedPoke[] = []; @@ -399,4 +691,56 @@ describe('initContextLensShipSync', () => { slot.set(previousParams); } }); + + it('arms the sync retry when the ledger write fails after a successful poke', async () => { + const pokes: RecordedPoke[] = []; + const slot = sharedSlot(API_CLIENT_PARAMS_SLOT); + const previousParams = slot.get(); + slot.set(makeParams(pokes)); + const messages: string[] = []; + const logger = { + info: (message: string) => messages.push(message), + warn: (message: string) => messages.push(message), + }; + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lens-ledger-failure-')); + // recordSyncedLensId writes next to the store file; a store whose + // directory does not exist makes that write throw. + setContextLensStore({ + filePath: path.join(dir, 'missing', 'runs.jsonl'), + save: () => {}, + size: () => 0, + get: () => null, + list: () => [], + }); + const api = { + config: { + channels: { + tlon: { + ship: '~zod', + contextLens: { + enabled: true, + authToken: 'a-token-of-sufficient-length', + owner: '~bus', + }, + }, + }, + } as OpenClawConfig, + logger, + }; + + try { + expect(initContextLensShipSync(api)).toBe(true); + + publishContextLensEvent('final', makeLens({ status: 'completed' })); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(pokes.map(pokeKind)).toEqual(['configure', 'lens']); + expect(messages.join('\n')).toContain('ledger write failed'); + expect(messages.join('\n')).toContain('retrying unsynced finals'); + } finally { + slot.set(previousParams); + setContextLensStore(null); + fs.rmSync(dir, { recursive: true, force: true }); + } + }); }); diff --git a/packages/openclaw/src/context-lens-ship-sync.ts b/packages/openclaw/src/context-lens-ship-sync.ts index 3b6610f7f3..52389add64 100644 --- a/packages/openclaw/src/context-lens-ship-sync.ts +++ b/packages/openclaw/src/context-lens-ship-sync.ts @@ -1,9 +1,15 @@ +import fs from 'node:fs'; import type { OpenClawConfig } from 'openclaw/plugin-sdk/core'; import { type ContextLensEvent, subscribeToContextLensEvents, } from './context-lens-events.js'; +import { + type ContextLensStore, + getContextLensStore, + lensFinalizedAt, +} from './context-lens-store.js'; import type { ContextLens, ContextLensStatus } from './context-lens.js'; import { API_CLIENT_PARAMS_SLOT, @@ -17,6 +23,13 @@ const PAYLOAD_SCHEMA_VERSION = 1; const MAX_SUMMARY_CHARS = 4_096; const MAX_PAYLOAD_CHARS = 50 * 1_024; const MAX_TRACKED_RUNS = 1_000; +const MAX_SYNCED_IDS = 1_000; +const REPLAY_POLL_MS = 2_000; +const REPLAY_GIVE_UP_MS = 10 * 60_000; +const REPLAY_WINDOW_MS = 24 * 60 * 60 * 1_000; +const REPLAY_MAX_RUNS = 50; +const RETRY_BASE_MS = 30_000; +const RETRY_MAX_MS = 10 * 60_000; const TERMINAL_STATUSES: ReadonlySet = new Set([ 'completed', @@ -42,6 +55,41 @@ const apiClientParamsSlot = sharedSlot( const shipSyncUnsubscribeSlot = sharedSlot<() => void>( 'contextLens.shipSync.unsubscribe' ); +const replayCancelSlot = sharedSlot<() => void>( + 'contextLens.shipSync.replayCancel' +); +const finalRetryCancelSlot = sharedSlot<() => void>( + 'contextLens.shipSync.finalRetryCancel' +); + +export function syncedLensIdsPath(storeFilePath: string): string { + return `${storeFilePath}.synced.json`; +} + +export function loadSyncedLensIds(filePath: string): Set { + try { + const parsed: unknown = JSON.parse(fs.readFileSync(filePath, 'utf8')); + if (!Array.isArray(parsed)) { + return new Set(); + } + return new Set(parsed.filter((id): id is string => typeof id === 'string')); + } catch { + return new Set(); + } +} + +export function recordSyncedLensId(filePath: string, lensId: string): void { + const ids = loadSyncedLensIds(filePath); + ids.delete(lensId); + ids.add(lensId); + const trimmed = [...ids].slice(-MAX_SYNCED_IDS); + // Atomic rewrite: a crash mid-write must not leave invalid JSON, which + // would read back as an empty ledger and trigger a full (if idempotent) + // replay on the next boot. + const tmpPath = `${filePath}.tmp`; + fs.writeFileSync(tmpPath, JSON.stringify(trimmed), { mode: 0o600 }); + fs.renameSync(tmpPath, filePath); +} function truncateSummary(value: string | undefined): string | undefined { if (value === undefined || value.length <= MAX_SUMMARY_CHARS) { @@ -172,22 +220,36 @@ export function createContextLensShipSync(opts: { owner: string; logger: SyncLogger; getParams?: () => SharedApiClientParams | null; + /** Called after a run-final poke is acked — basis for boot-time replay. */ + onRunFinal?: (lens: ContextLens) => void; + /** + * Called when a run-final poke is dropped (monitor not connected) or + * rejected (ship unreachable) — basis for retry-until-synced. Milestone + * failures do not fire this: the eventual run-final supersedes them. + */ + onRunFinalFailure?: (lens: ContextLens) => void; }): ContextLensShipSync { - const { owner, logger } = opts; + const { owner, logger, onRunFinal, onRunFinalFailure } = opts; const getParams = opts.getParams ?? (() => apiClientParamsSlot.get() ?? null); const lastStatusByLensId = new Map(); let configuredFor: SharedApiClientParams | null = null; let queue: Promise = Promise.resolve(); - const enqueuePoke = (label: string, json: unknown) => { + const enqueuePoke = ( + label: string, + json: unknown, + onSuccess?: () => void, + onFailure?: () => void + ) => { queue = queue .then(async () => { const params = getParams(); if (!params) { // Monitor not connected yet (or shut down); drop rather than // buffer — the ship store is bounded and the gateway store keeps - // the full run. + // the full run. Finals report the drop so replay can heal it. + onFailure?.(); return; } if (params !== configuredFor) { @@ -203,6 +265,13 @@ export function createContextLensShipSync(opts: { mark: 'steward-lens-action-1', json, }); + try { + onSuccess?.(); + } catch (error) { + logger.warn( + `[tlon] Context lens ship sync bookkeeping failed (${label}): ${String(error)}` + ); + } }) .catch((error) => { // A failed %configure must retry before the next run poke. @@ -210,6 +279,13 @@ export function createContextLensShipSync(opts: { logger.warn( `[tlon] Context lens ship sync poke failed (${label}): ${String(error)}` ); + try { + onFailure?.(); + } catch (callbackError) { + logger.warn( + `[tlon] Context lens ship sync failure bookkeeping failed (${label}): ${String(callbackError)}` + ); + } }); }; @@ -220,13 +296,18 @@ export function createContextLensShipSync(opts: { } if (TERMINAL_STATUSES.has(lens.status)) { lastStatusByLensId.delete(lens.lensId); - enqueuePoke(`run-final ${lens.lensId}`, { - entry: { - id: lens.lensId, - payload: buildLensRunPayload(lens), - final: true, + enqueuePoke( + `run-final ${lens.lensId}`, + { + entry: { + id: lens.lensId, + payload: buildLensRunPayload(lens), + final: true, + }, }, - }); + onRunFinal ? () => onRunFinal(lens) : undefined, + onRunFinalFailure ? () => onRunFinalFailure(lens) : undefined + ); return; } if (lens.status === lastStatusByLensId.get(lens.lensId)) { @@ -256,6 +337,129 @@ export function createContextLensShipSync(opts: { }; } +/** + * Re-poke terminal runs whose run-final never reached the ship — e.g. runs + * finalized during a SIGTERM shutdown, where the store write (sync fs) + * landed but the poke died with the process. Without this the ship copy + * stays frozen at the last in-flight status forever. Idempotent ship-side + * (last write wins per run id); bounded by a recency window and a count cap + * so a first boot without a synced-ids file cannot flood the ship. + */ +export function replayUnsyncedFinalRuns( + store: ContextLensStore, + sync: ContextLensShipSync, + logger: SyncLogger, + now = Date.now() +): number { + const synced = loadSyncedLensIds(syncedLensIdsPath(store.filePath)); + const cutoff = now - REPLAY_WINDOW_MS; + const missed = store + .list() + .filter( + (lens) => + TERMINAL_STATUSES.has(lens.status) && + lens.visibility !== 'internal' && + !synced.has(lens.lensId) && + lensFinalizedAt(lens) > cutoff + ) + .slice(-REPLAY_MAX_RUNS); + if (missed.length === 0) { + return 0; + } + logger.info( + `[tlon] Context lens ship sync replaying ${missed.length} unsynced terminal run(s)` + ); + for (const lens of missed) { + sync.handleEvent({ seq: 0, at: now, phase: 'replay', lens }); + } + return missed.length; +} + +export type LensSyncRetry = { + /** Arm (or keep armed) a retry; no-op while one is already pending. */ + schedule: () => void; + /** Reset the backoff after a successful sync. */ + reset: () => void; + cancel: () => void; +}; + +/** + * Exponential-backoff retry used to re-run the unsynced-final replay after a + * mid-session failure (ship unreachable, monitor disconnected). Without this + * a run whose run-final poke fails stays frozen at its last milestone on the + * ship until the next gateway restart. One timer at most: replay covers all + * unsynced finals at once, so per-run timers would just multiply pokes. + */ +export function createLensSyncRetry(opts: { + run: () => void; + logger: SyncLogger; + baseMs?: number; + maxMs?: number; +}): LensSyncRetry { + const baseMs = opts.baseMs ?? RETRY_BASE_MS; + const maxMs = opts.maxMs ?? RETRY_MAX_MS; + let attempt = 0; + let timer: ReturnType | null = null; + const schedule = () => { + if (timer) { + return; + } + const delay = Math.min(baseMs * 2 ** attempt, maxMs); + attempt += 1; + opts.logger.info( + `[tlon] Context lens ship sync retrying unsynced finals in ${Math.round(delay / 1_000)}s (attempt ${attempt})` + ); + timer = setTimeout(() => { + timer = null; + opts.run(); + }, delay); + timer.unref?.(); + }; + return { + schedule, + reset: () => { + attempt = 0; + }, + cancel: () => { + if (timer) { + clearTimeout(timer); + timer = null; + } + }, + }; +} + +function startShipSyncReplay( + sync: ContextLensShipSync, + logger: SyncLogger +): void { + replayCancelSlot.get()?.(); + const startedAt = Date.now(); + // Poll: at init time neither the api-client params (monitor not connected) + // nor the disk store (initialized after ship sync) exist yet. + const timer = setInterval(() => { + if (Date.now() - startedAt > REPLAY_GIVE_UP_MS) { + stop(); + return; + } + const store = getContextLensStore(); + if (!store || !apiClientParamsSlot.get()) { + return; + } + stop(); + try { + replayUnsyncedFinalRuns(store, sync, logger); + } catch (error) { + logger.warn( + `[tlon] Context lens ship sync replay failed: ${String(error)}` + ); + } + }, REPLAY_POLL_MS); + timer.unref?.(); + const stop = () => clearInterval(timer); + replayCancelSlot.set(stop); +} + /** * Wire ship sync to the lens event stream. Returns true when active, false * when the lens is disabled or no owner resolves (no contextLens.owner and @@ -275,9 +479,56 @@ export function initContextLensShipSync(api: { ); return false; } - const sync = createContextLensShipSync({ owner, logger: api.logger }); + // Forward reference: the retry replays through `sync`, which is created + // below with callbacks that arm/reset the retry. `run` only fires from a + // timer, well after both bindings exist. + const retry = createLensSyncRetry({ + logger: api.logger, + run: () => { + const store = getContextLensStore(); + if (!store || !apiClientParamsSlot.get()) { + // Still disconnected (or store gone): keep backing off. + retry.schedule(); + return; + } + try { + replayUnsyncedFinalRuns(store, sync, api.logger); + } catch (error) { + api.logger.warn( + `[tlon] Context lens ship sync retry failed: ${String(error)}` + ); + } + }, + }); + const sync = createContextLensShipSync({ + owner, + logger: api.logger, + onRunFinal: (lens) => { + // The poke succeeded but the ledger write can still fail; without a + // scheduled retry the run would stay marked unsynced until the next + // boot replay. Arm the retry so the ledger write is re-attempted (the + // replayed poke is idempotent ship-side). + const store = getContextLensStore(); + if (store) { + try { + recordSyncedLensId(syncedLensIdsPath(store.filePath), lens.lensId); + } catch (error) { + api.logger.warn( + `[tlon] Context lens sync ledger write failed: ${String(error)}` + ); + retry.schedule(); + return; + } + } + retry.reset(); + }, + onRunFinalFailure: () => retry.schedule(), + }); shipSyncUnsubscribeSlot.get()?.(); shipSyncUnsubscribeSlot.set(subscribeToContextLensEvents(sync.handleEvent)); + finalRetryCancelSlot.get()?.(); + finalRetryCancelSlot.set(retry.cancel); + startShipSyncReplay(sync, api.logger); api.logger.info( `[tlon] Context lens ship sync enabled, fanning out to ${owner}` ); From 636e755cc648fddc17cbd4d01e2a561fc7a3d20d Mon Sep 17 00:00:00 2001 From: william arzt Date: Sat, 18 Jul 2026 21:24:32 -0400 Subject: [PATCH 4/4] chore(openclaw): gitignore local lens panel snapshot artifact Keeps .lens-panel-snapshot.txt (may contain private conversation data) out of commits. Co-Authored-By: Claude Opus 4.7 --- packages/openclaw/.gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/openclaw/.gitignore b/packages/openclaw/.gitignore index 9ab4d35cc4..b304094b17 100644 --- a/packages/openclaw/.gitignore +++ b/packages/openclaw/.gitignore @@ -11,3 +11,6 @@ WORKLOG.md # Publish staging .publish/ *.tgz + +# Local debug artifacts (may contain private conversation data) +.lens-panel-snapshot.txt