mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 04:47:03 -06:00
merge(main): refresh realtime continuity repair
* commit '81e392b7ff43bcdb7a3be50a99adf65c3b0141b5': fix(channels): preserve filters in offline status diagnostics (#117570) perf(gateway): separate startup maintenance timing (#117577) refactor(tests): deduplicate database-first guard fixtures (#117594) fix(cli): align sessions cleanup label summary by visible width (#117459) perf(gateway): skip empty session reconciliation (#117589) refactor(ui): consolidate cron form control rendering (#117576) fix(memory): recover restored session freshness (#117548) fix(ui): preserve Talk transcript surrogate bounds fix(ui): harden Talk transcript marker bounds test(ui): cover bounded realtime Talk entries fix(ui): bound realtime Talk conversation text
This commit is contained in:
Vincent Koc
@@ -106,6 +106,12 @@ describe("memory session sync state", () => {
|
||||
mtimeMs: 250,
|
||||
size: 20,
|
||||
},
|
||||
{
|
||||
absPath: "/tmp/sessions/rolled-back.jsonl",
|
||||
path: "sessions/rolled-back.jsonl",
|
||||
mtimeMs: 150,
|
||||
size: 20,
|
||||
},
|
||||
{
|
||||
absPath: "/tmp/sessions/resized.jsonl",
|
||||
path: "sessions/resized.jsonl",
|
||||
@@ -124,6 +130,7 @@ describe("memory session sync state", () => {
|
||||
{ path: "sessions/sub-ms-newer.jsonl", hash: "hash-sub-ms", mtime: 100.25, size: 10 },
|
||||
{ path: "sessions/invalidated.jsonl", hash: "", mtime: 200, size: 20 },
|
||||
{ path: "sessions/newer.jsonl", hash: "hash-newer", mtime: 200, size: 20 },
|
||||
{ path: "sessions/rolled-back.jsonl", hash: "hash-rolled-back", mtime: 200, size: 20 },
|
||||
{ path: "sessions/resized.jsonl", hash: "hash-resized", mtime: 300, size: 30 },
|
||||
],
|
||||
});
|
||||
@@ -132,6 +139,7 @@ describe("memory session sync state", () => {
|
||||
"/tmp/sessions/sub-ms-newer.jsonl",
|
||||
"/tmp/sessions/invalidated.jsonl",
|
||||
"/tmp/sessions/newer.jsonl",
|
||||
"/tmp/sessions/rolled-back.jsonl",
|
||||
"/tmp/sessions/resized.jsonl",
|
||||
"/tmp/sessions/missing.jsonl",
|
||||
]);
|
||||
|
||||
@@ -26,7 +26,9 @@ export function resolveMemorySessionStartupDirtyFiles(params: {
|
||||
dirtyFiles.push(file.absPath);
|
||||
continue;
|
||||
}
|
||||
if (file.size !== indexedSize || file.mtimeMs > indexedMtimeMs) {
|
||||
// File mtimes and SQLite session updatedAt values can move backward after
|
||||
// restore/reset. The downstream content-hash gate suppresses unchanged rewrites.
|
||||
if (file.size !== indexedSize || file.mtimeMs !== indexedMtimeMs) {
|
||||
dirtyFiles.push(file.absPath);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -174,6 +174,29 @@ export abstract class MemoryManagerSourceSyncOps extends MemoryManagerSessionSyn
|
||||
const deleteChunksByPathAndSource = this.db.prepare(
|
||||
`DELETE FROM memory_index_chunks WHERE path = ? AND source = ?`,
|
||||
);
|
||||
const updateUnchangedSessionSourceMetadata = this.db.prepare(
|
||||
`UPDATE memory_index_sources
|
||||
SET mtime = ?, size = ?
|
||||
WHERE path = ? AND source = 'sessions' AND hash = ?`,
|
||||
);
|
||||
const refreshUnchangedSessionSourceMetadata = (entry: MemoryIndexEntry): boolean => {
|
||||
// Hash equality preserves chunks and embeddings; only converge the source
|
||||
// fingerprint so restored sessions do not repeat catch-up on every startup.
|
||||
return (
|
||||
updateUnchangedSessionSourceMetadata.run(entry.mtimeMs, entry.size, entry.path, entry.hash)
|
||||
.changes === 1
|
||||
);
|
||||
};
|
||||
const canSkipUnchangedSessionEntry = (
|
||||
entry: MemoryIndexEntry,
|
||||
absPath: string,
|
||||
existingHash: string | undefined,
|
||||
): boolean => {
|
||||
if (params.needsFullReindex || existingHash !== entry.hash) {
|
||||
return false;
|
||||
}
|
||||
return !this.sessionsDirtyFiles.has(absPath) || refreshUnchangedSessionSourceMetadata(entry);
|
||||
};
|
||||
const deleteFtsRowsByPathAndSource =
|
||||
this.fts.enabled && this.fts.available
|
||||
? this.db.prepare(`DELETE FROM ${FTS_TABLE} WHERE path = ? AND source = ?`)
|
||||
@@ -340,7 +363,7 @@ export abstract class MemoryManagerSourceSyncOps extends MemoryManagerSessionSyn
|
||||
path: entry.path,
|
||||
existingHashes,
|
||||
});
|
||||
if (!params.needsFullReindex && existingHash === entry.hash) {
|
||||
if (canSkipUnchangedSessionEntry(entry, absPath, existingHash)) {
|
||||
if (params.progress) {
|
||||
params.progress.completed += 1;
|
||||
params.progress.report({
|
||||
@@ -412,7 +435,7 @@ export abstract class MemoryManagerSourceSyncOps extends MemoryManagerSessionSyn
|
||||
path: entry.path,
|
||||
existingHashes,
|
||||
});
|
||||
if (!params.needsFullReindex && existingHash === entry.hash) {
|
||||
if (canSkipUnchangedSessionEntry(entry, absPath, existingHash)) {
|
||||
if (params.progress) {
|
||||
params.progress.completed += 1;
|
||||
params.progress.report({
|
||||
|
||||
@@ -2,13 +2,16 @@
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import {
|
||||
resolveSessionTranscriptsDirForAgent,
|
||||
type OpenClawConfig,
|
||||
type ResolvedMemorySearchConfig,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
|
||||
import { statSessionEntrySync } from "openclaw/plugin-sdk/memory-core-host-engine-qmd";
|
||||
import {
|
||||
buildSessionEntry,
|
||||
statSessionEntrySync,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-qmd";
|
||||
import {
|
||||
MEMORY_CHUNKING_VERSION,
|
||||
type MemorySource,
|
||||
@@ -64,8 +67,43 @@ type MemorySessionTranscriptUpdate = {
|
||||
const originalStartupStateDir = process.env.OPENCLAW_STATE_DIR;
|
||||
const originalStartupConfigPath = process.env.OPENCLAW_CONFIG_PATH;
|
||||
let transcriptUpdateListener: ((update: MemorySessionTranscriptUpdate) => void) | undefined;
|
||||
const startupHarnessDatabases = new Set<DatabaseSync>();
|
||||
|
||||
type SourceStateRow = { path: string; hash: string; mtime: number; size: number };
|
||||
|
||||
function createStartupHarnessDatabase(sourceRows: SourceStateRow[]): DatabaseSync {
|
||||
const db = new DatabaseSync(":memory:");
|
||||
db.exec(`
|
||||
CREATE TABLE memory_index_sources (
|
||||
path TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
mtime REAL NOT NULL,
|
||||
size INTEGER NOT NULL,
|
||||
UNIQUE(path, source)
|
||||
);
|
||||
CREATE TABLE memory_index_chunks (
|
||||
id TEXT PRIMARY KEY,
|
||||
path TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
model TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE memory_index_source_update_audit (path TEXT NOT NULL);
|
||||
CREATE TRIGGER memory_index_source_update_audit_trigger
|
||||
AFTER UPDATE ON memory_index_sources
|
||||
BEGIN
|
||||
INSERT INTO memory_index_source_update_audit (path) VALUES (NEW.path);
|
||||
END;
|
||||
`);
|
||||
const insert = db.prepare(
|
||||
`INSERT INTO memory_index_sources (path, source, hash, mtime, size) VALUES (?, 'sessions', ?, ?, ?)`,
|
||||
);
|
||||
for (const row of sourceRows) {
|
||||
insert.run(row.path, row.hash, row.mtime, row.size);
|
||||
}
|
||||
startupHarnessDatabases.add(db);
|
||||
return db;
|
||||
}
|
||||
function setStartupStateDir(stateDir: string): void {
|
||||
Reflect.set(process.env, "OPENCLAW_STATE_DIR", stateDir);
|
||||
}
|
||||
@@ -148,16 +186,37 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
|
||||
sourceRows: SourceStateRow[],
|
||||
private readonly indexSessionUpdates = false,
|
||||
private readonly subscribeToRealEvents = false,
|
||||
private readonly deferSessionIndex = false,
|
||||
database?: DatabaseSync,
|
||||
) {
|
||||
super();
|
||||
this.sources.add("sessions");
|
||||
this.db = {
|
||||
prepare: () => ({
|
||||
all: () => sourceRows,
|
||||
get: () => undefined,
|
||||
run: () => undefined,
|
||||
}),
|
||||
} as unknown as DatabaseSync;
|
||||
this.db = database ?? createStartupHarnessDatabase(sourceRows);
|
||||
}
|
||||
|
||||
restartForStartup(): SessionStartupCatchupHarness {
|
||||
return new SessionStartupCatchupHarness(
|
||||
[],
|
||||
this.indexSessionUpdates,
|
||||
false,
|
||||
this.deferSessionIndex,
|
||||
this.db,
|
||||
);
|
||||
}
|
||||
|
||||
getIndexedSourceState(pathname: string): SourceStateRow | undefined {
|
||||
return this.db
|
||||
.prepare(
|
||||
`SELECT path, hash, mtime, size FROM memory_index_sources WHERE path = ? AND source = 'sessions'`,
|
||||
)
|
||||
.get(pathname) as SourceStateRow | undefined;
|
||||
}
|
||||
|
||||
getSourceMetadataUpdateCount(): number {
|
||||
const row = this.db
|
||||
.prepare(`SELECT COUNT(*) AS count FROM memory_index_source_update_audit`)
|
||||
.get() as { count: number };
|
||||
return row.count;
|
||||
}
|
||||
|
||||
async catchUp(): Promise<string[]> {
|
||||
@@ -172,6 +231,13 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
|
||||
await this.runSync(params);
|
||||
}
|
||||
|
||||
async runArchiveSyncForTest(): Promise<void> {
|
||||
await this.syncArchiveFiles({
|
||||
needsFullReindex: false,
|
||||
deferIndex: this.deferSessionIndex,
|
||||
});
|
||||
}
|
||||
|
||||
getDirtyArchiveFiles(): string[] {
|
||||
return Array.from(this.sessionsDirtyFiles);
|
||||
}
|
||||
@@ -273,7 +339,10 @@ class SessionStartupCatchupHarness extends MemoryManagerSyncOps {
|
||||
protected async sync(params?: MemorySyncParams): Promise<void> {
|
||||
this.syncCalls.push(params ?? {});
|
||||
this.pendingSyncWork = this.indexSessionUpdates
|
||||
? this.syncArchiveFiles({ needsFullReindex: false }).then(() => undefined)
|
||||
? this.syncArchiveFiles({
|
||||
needsFullReindex: false,
|
||||
deferIndex: this.deferSessionIndex,
|
||||
}).then(() => undefined)
|
||||
: Promise.resolve();
|
||||
await this.pendingSyncWork;
|
||||
}
|
||||
@@ -333,20 +402,29 @@ describe("session startup catch-up", () => {
|
||||
restoreStartupEnv();
|
||||
clearRuntimeConfigSnapshot();
|
||||
clearConfigCache();
|
||||
for (const database of startupHarnessDatabases) {
|
||||
database.close();
|
||||
}
|
||||
startupHarnessDatabases.clear();
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
await fs.rm(stateDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
async function writeSessionFile(
|
||||
name: string,
|
||||
content = "startup catchup",
|
||||
timestamp?: string,
|
||||
): Promise<{ filePath: string; size: number; mtimeMs: number }> {
|
||||
const sessionsDir = resolveSessionTranscriptsDirForAgent("main");
|
||||
await fs.mkdir(sessionsDir, { recursive: true });
|
||||
const filePath = path.join(sessionsDir, name);
|
||||
await fs.writeFile(
|
||||
filePath,
|
||||
JSON.stringify({ type: "message", message: { role: "user", content: "startup catchup" } }) +
|
||||
"\n",
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
...(timestamp ? { timestamp } : {}),
|
||||
message: { role: "user", content },
|
||||
}) + "\n",
|
||||
"utf-8",
|
||||
);
|
||||
const stat = await fs.stat(filePath);
|
||||
@@ -533,6 +611,169 @@ describe("session startup catch-up", () => {
|
||||
expect(harness.syncCalls).toEqual([]);
|
||||
});
|
||||
|
||||
it("indexes a same-size file transcript whose mtime rolled back", async () => {
|
||||
const archiveName = "thread.jsonl.deleted.2026-08-01T10-00-00.000Z";
|
||||
const original = await writeSessionFile(archiveName, "version before");
|
||||
const originalEntry = await buildSessionEntry(original.filePath);
|
||||
if (!originalEntry) {
|
||||
throw new Error("expected original file transcript entry");
|
||||
}
|
||||
const replacement = await writeSessionFile(archiveName, "version after!");
|
||||
expect(replacement.size).toBe(original.size);
|
||||
const rolledBackMtime = new Date(Math.max(1, original.mtimeMs - 60_000));
|
||||
await fs.utimes(replacement.filePath, rolledBackMtime, rolledBackMtime);
|
||||
const rolledBack = await fs.stat(replacement.filePath);
|
||||
expect(rolledBack.mtimeMs).toBeLessThan(original.mtimeMs);
|
||||
|
||||
const harness = new SessionStartupCatchupHarness(
|
||||
[
|
||||
{
|
||||
path: originalEntry.path,
|
||||
hash: originalEntry.hash,
|
||||
mtime: original.mtimeMs,
|
||||
size: original.size,
|
||||
},
|
||||
],
|
||||
true,
|
||||
);
|
||||
|
||||
await expect(harness.catchUp()).resolves.toEqual([replacement.filePath]);
|
||||
await harness.waitForSessionSync();
|
||||
|
||||
expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]);
|
||||
expect(harness.indexedPaths).toEqual([`sessions/main/${archiveName}`]);
|
||||
expect(harness.indexedContents).toEqual(["User: version after!"]);
|
||||
});
|
||||
|
||||
it("converges an unchanged file mtime rollback after direct session sync", async () => {
|
||||
const archiveName = "thread.jsonl.deleted.2026-08-01T11-00-00.000Z";
|
||||
const messageTimestamp = "2026-08-01T10:30:00.000Z";
|
||||
const original = await writeSessionFile(archiveName, "unchanged content", messageTimestamp);
|
||||
const originalEntry = await buildSessionEntry(original.filePath);
|
||||
if (!originalEntry) {
|
||||
throw new Error("expected original file transcript entry");
|
||||
}
|
||||
const rolledBackMtime = new Date(Math.max(1, original.mtimeMs - 60_000));
|
||||
await fs.utimes(original.filePath, rolledBackMtime, rolledBackMtime);
|
||||
const restoredEntry = await buildSessionEntry(original.filePath);
|
||||
if (!restoredEntry) {
|
||||
throw new Error("expected restored file transcript entry");
|
||||
}
|
||||
expect(restoredEntry.hash).toBe(originalEntry.hash);
|
||||
|
||||
const harness = new SessionStartupCatchupHarness(
|
||||
[
|
||||
{
|
||||
path: originalEntry.path,
|
||||
hash: originalEntry.hash,
|
||||
mtime: original.mtimeMs,
|
||||
size: original.size,
|
||||
},
|
||||
],
|
||||
true,
|
||||
);
|
||||
|
||||
await expect(harness.catchUp()).resolves.toEqual([original.filePath]);
|
||||
await harness.waitForSessionSync();
|
||||
expect(harness.indexedPaths).toEqual([]);
|
||||
expect(harness.getIndexedSourceState(originalEntry.path)).toEqual({
|
||||
path: originalEntry.path,
|
||||
hash: originalEntry.hash,
|
||||
mtime: restoredEntry.mtimeMs,
|
||||
size: restoredEntry.size,
|
||||
});
|
||||
expect(harness.getSourceMetadataUpdateCount()).toBe(1);
|
||||
|
||||
const restarted = harness.restartForStartup();
|
||||
await expect(restarted.catchUp()).resolves.toEqual([]);
|
||||
expect(restarted.syncCalls).toEqual([]);
|
||||
expect(restarted.indexedPaths).toEqual([]);
|
||||
});
|
||||
|
||||
it("indexes a SQLite transcript whose updatedAt rolled back", async () => {
|
||||
const session = await writeSqliteSession({
|
||||
content: "SQLite rollback",
|
||||
updatedAt: 10,
|
||||
});
|
||||
const state = statSessionEntrySync(session.sessionKey, {
|
||||
agentId: "main",
|
||||
sessionId: session.sessionId,
|
||||
storePath: session.storePath,
|
||||
sessionKey: session.sessionKey,
|
||||
updatedAtMs: 10,
|
||||
});
|
||||
if (!state) {
|
||||
throw new Error("expected SQLite transcript state");
|
||||
}
|
||||
const harness = new SessionStartupCatchupHarness(
|
||||
[
|
||||
{
|
||||
path: state.path,
|
||||
hash: "previous-hash",
|
||||
mtime: 20,
|
||||
size: state.size,
|
||||
},
|
||||
],
|
||||
true,
|
||||
);
|
||||
|
||||
await expect(harness.catchUp()).resolves.toEqual([session.sessionKey]);
|
||||
await harness.waitForSessionSync();
|
||||
|
||||
expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]);
|
||||
expect(harness.indexedPaths).toEqual([session.corpusPath]);
|
||||
expect(harness.indexedContents).toEqual(["User: SQLite rollback"]);
|
||||
});
|
||||
|
||||
it("converges an unchanged SQLite updatedAt rollback after deferred session sync", async () => {
|
||||
const session = await writeSqliteSession({ updatedAt: 10 });
|
||||
const entry = await buildSessionEntry(session.sessionKey, {
|
||||
agentId: "main",
|
||||
sessionId: session.sessionId,
|
||||
storePath: session.storePath,
|
||||
sessionKey: session.sessionKey,
|
||||
updatedAtMs: 10,
|
||||
sessionKind: "interactive",
|
||||
});
|
||||
if (!entry) {
|
||||
throw new Error("expected SQLite transcript entry");
|
||||
}
|
||||
const harness = new SessionStartupCatchupHarness(
|
||||
[
|
||||
{
|
||||
path: entry.path,
|
||||
hash: entry.hash,
|
||||
mtime: 20,
|
||||
size: entry.size,
|
||||
},
|
||||
],
|
||||
true,
|
||||
false,
|
||||
true,
|
||||
);
|
||||
|
||||
await expect(harness.catchUp()).resolves.toEqual([session.sessionKey]);
|
||||
await harness.waitForSessionSync();
|
||||
|
||||
expect(harness.syncCalls).toEqual([{ reason: "session-startup-catchup" }]);
|
||||
expect(harness.indexedPaths).toEqual([]);
|
||||
expect(harness.indexedContents).toEqual([]);
|
||||
expect(harness.getIndexedSourceState(entry.path)).toEqual({
|
||||
path: entry.path,
|
||||
hash: entry.hash,
|
||||
mtime: entry.mtimeMs,
|
||||
size: entry.size,
|
||||
});
|
||||
expect(harness.getSourceMetadataUpdateCount()).toBe(1);
|
||||
|
||||
const restarted = harness.restartForStartup();
|
||||
await expect(restarted.catchUp()).resolves.toEqual([]);
|
||||
expect(restarted.syncCalls).toEqual([]);
|
||||
expect(restarted.indexedPaths).toEqual([]);
|
||||
await restarted.runArchiveSyncForTest();
|
||||
expect(restarted.getSourceMetadataUpdateCount()).toBe(1);
|
||||
});
|
||||
|
||||
it("does not fall back to full session sync when identity targets normalize away", async () => {
|
||||
await writeSessionFile("thread.jsonl");
|
||||
const harness = new SessionStartupCatchupHarness([]);
|
||||
|
||||
@@ -11,6 +11,10 @@ vi.mock("../channels/plugins/index.js", () => ({
|
||||
listChannelPlugins: () => activeChannelPlugins,
|
||||
getLoadedChannelPlugin: (id: string) => activeChannelPlugins.find((plugin) => plugin.id === id),
|
||||
getChannelPlugin: (id: string) => activeChannelPlugins.find((plugin) => plugin.id === id),
|
||||
normalizeChannelId: (value: string) => {
|
||||
const normalized = value.trim().toLowerCase();
|
||||
return normalized === "wa" || normalized === "whatsapp" ? "whatsapp" : null;
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("../channels/plugins/read-only.js", () => ({
|
||||
@@ -25,12 +29,18 @@ async function formatLocalStatusSummary(
|
||||
cfg: unknown,
|
||||
options?: {
|
||||
sourceConfig?: unknown;
|
||||
channel?: string;
|
||||
},
|
||||
) {
|
||||
const lines = await formatConfigChannelsStatusLines(
|
||||
cfg as never,
|
||||
{ mode: "local" },
|
||||
options?.sourceConfig ? { sourceConfig: options.sourceConfig as never } : undefined,
|
||||
options
|
||||
? {
|
||||
...(options.sourceConfig ? { sourceConfig: options.sourceConfig as never } : {}),
|
||||
...(options.channel !== undefined ? { channel: options.channel } : {}),
|
||||
}
|
||||
: undefined,
|
||||
);
|
||||
return lines.join("\n");
|
||||
}
|
||||
@@ -193,6 +203,68 @@ function requireReadOnlyPluginListCall(): unknown[] {
|
||||
}
|
||||
|
||||
describe("config-only channels status output", () => {
|
||||
it.each([
|
||||
{
|
||||
label: "unregistered external channels",
|
||||
channel: "token-only",
|
||||
included: ["TokenOnly"],
|
||||
excluded: ["WhatsApp"],
|
||||
},
|
||||
{
|
||||
label: "case-insensitive external channels",
|
||||
channel: "TOKEN-ONLY",
|
||||
included: ["TokenOnly"],
|
||||
excluded: ["WhatsApp"],
|
||||
},
|
||||
{
|
||||
label: "bundled channel aliases",
|
||||
channel: "wa",
|
||||
included: ["WhatsApp"],
|
||||
excluded: ["TokenOnly"],
|
||||
},
|
||||
{
|
||||
label: "unknown channels",
|
||||
channel: "missing-channel",
|
||||
included: [],
|
||||
excluded: ["TokenOnly", "WhatsApp"],
|
||||
},
|
||||
{
|
||||
label: "no channel filter",
|
||||
channel: undefined,
|
||||
included: ["TokenOnly", "WhatsApp"],
|
||||
excluded: [],
|
||||
},
|
||||
{
|
||||
label: "blank channel filters",
|
||||
channel: " ",
|
||||
included: ["TokenOnly", "WhatsApp"],
|
||||
excluded: [],
|
||||
},
|
||||
])(
|
||||
"preserves exact config-only status filtering for $label",
|
||||
async ({ channel, included, excluded }) => {
|
||||
activeChannelPlugins.splice(
|
||||
0,
|
||||
activeChannelPlugins.length,
|
||||
makeUnavailableTokenPlugin(),
|
||||
makeIndeterminateLinkPlugin(),
|
||||
);
|
||||
|
||||
const summary = await formatLocalStatusSummary(
|
||||
{ channels: { "token-only": {}, whatsapp: {} } },
|
||||
{ channel },
|
||||
);
|
||||
|
||||
expect(summary).toContain("Gateway not reachable; showing config-only status.");
|
||||
for (const label of included) {
|
||||
expect(summary).toContain(label);
|
||||
}
|
||||
for (const label of excluded) {
|
||||
expect(summary).not.toContain(label);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("uses setup fallback plugins so configured external channels can be shown", async () => {
|
||||
registerSingleTestPlugin("token-only", makeUnavailableTokenPlugin());
|
||||
listReadOnlyChannelPluginsForConfig.mockClear();
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
// Config-only channel status formatter used when the gateway is unreachable.
|
||||
import { normalizeOptionalLowercaseString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { formatDocsLink } from "../../../packages/terminal-core/src/links.js";
|
||||
import { theme } from "../../../packages/terminal-core/src/theme.js";
|
||||
import {
|
||||
@@ -68,7 +69,9 @@ export async function formatConfigChannelsStatusLines(
|
||||
});
|
||||
|
||||
const sourceConfig = opts?.sourceConfig ?? cfg;
|
||||
const requestedChannel = opts?.channel ? normalizeChannelId(opts.channel) : null;
|
||||
const requestedChannel = opts?.channel
|
||||
? (normalizeChannelId(opts.channel) ?? normalizeOptionalLowercaseString(opts.channel))
|
||||
: null;
|
||||
const plugins = listReadOnlyChannelPluginsForConfig(cfg, {
|
||||
activationSourceConfig: sourceConfig,
|
||||
includeSetupFallbackPlugins: true,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
// Sessions cleanup tests cover stale session cleanup and runtime output.
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { visibleWidth } from "../../packages/terminal-core/src/ansi.js";
|
||||
import type { SessionEntry } from "../config/sessions.js";
|
||||
import type { RuntimeEnv } from "../runtime.js";
|
||||
|
||||
@@ -643,6 +644,76 @@ describe("sessionsCleanupCommand", () => {
|
||||
expectLogsToInclude(logs, "Total: 3 kept, 4 pruned");
|
||||
});
|
||||
|
||||
it("aligns the label summary columns for emoji and CJK labels", async () => {
|
||||
mocks.enforceSessionDiskBudget.mockResolvedValue(null);
|
||||
mocks.runSessionsCleanup.mockResolvedValue({
|
||||
mode: "warn",
|
||||
previewResults: [
|
||||
{
|
||||
summary: {
|
||||
agentId: "main",
|
||||
storePath: "/resolved/sessions.json",
|
||||
mode: "warn",
|
||||
dryRun: true,
|
||||
beforeCount: 2,
|
||||
afterCount: 2,
|
||||
missing: 0,
|
||||
dmScopeRetired: 0,
|
||||
pruned: 0,
|
||||
capped: 0,
|
||||
unreferencedArtifacts: {
|
||||
scannedFiles: 0,
|
||||
removedFiles: 0,
|
||||
freedBytes: 0,
|
||||
olderThanMs: 604800000,
|
||||
},
|
||||
diskBudget: null,
|
||||
wouldMutate: true,
|
||||
},
|
||||
beforeStore: {
|
||||
emojiKept: {
|
||||
sessionId: "emoji-kept",
|
||||
updatedAt: 2,
|
||||
model: "test:opus",
|
||||
label: "🔥修复",
|
||||
},
|
||||
plainKept: {
|
||||
sessionId: "plain-kept",
|
||||
updatedAt: 1,
|
||||
model: "test:opus",
|
||||
label: "plain",
|
||||
},
|
||||
},
|
||||
missingKeys: new Set<string>(),
|
||||
staleKeys: new Set<string>(),
|
||||
cappedKeys: new Set<string>(),
|
||||
budgetEvictedKeys: new Set<string>(),
|
||||
dmScopeRetiredKeys: new Set<string>(),
|
||||
},
|
||||
],
|
||||
appliedSummaries: [],
|
||||
});
|
||||
|
||||
const { runtime, logs } = makeRuntime();
|
||||
await sessionsCleanupCommand(
|
||||
{
|
||||
dryRun: true,
|
||||
},
|
||||
runtime,
|
||||
);
|
||||
|
||||
expectLogsToInclude(logs, "Summary by Label:");
|
||||
const summaryLogs = logs.slice(logs.indexOf("Summary by Label:") + 1);
|
||||
const emojiLine = summaryLogs.find((line) => line.includes("🔥修复"));
|
||||
const plainLine = summaryLogs.find((line) => line.includes("plain"));
|
||||
expect(emojiLine).toBeDefined();
|
||||
expect(plainLine).toBeDefined();
|
||||
// "🔥修复" is 6 visible columns (wide emoji + 2 CJK) but only 5 UTF-16 code
|
||||
// units; padding by code-unit length would shift the counts column left.
|
||||
const keptColumn = (line: string) => visibleWidth(line.slice(0, line.indexOf("1 kept")));
|
||||
expect(keptColumn(emojiLine ?? "")).toBe(keptColumn(plainLine ?? ""));
|
||||
});
|
||||
|
||||
it("returns grouped JSON for --all-agents dry-runs", async () => {
|
||||
mocks.resolveSessionStoreTargets.mockReturnValue([
|
||||
{ agentId: "main", storePath: "/resolved/main-sessions.json" },
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
* It can delegate cleanup to a live gateway or run local store maintenance,
|
||||
* with dry-run tables that explain every planned pruning action.
|
||||
*/
|
||||
import { visibleWidth } from "../../packages/terminal-core/src/ansi.js";
|
||||
import { sanitizeTerminalText } from "../../packages/terminal-core/src/safe-text.js";
|
||||
import { isRich, theme } from "../../packages/terminal-core/src/theme.js";
|
||||
import { getRuntimeConfig } from "../config/config.js";
|
||||
@@ -129,15 +130,15 @@ function renderLabelSummaries(params: {
|
||||
if (summaries.length === 0) {
|
||||
return;
|
||||
}
|
||||
const labelPad = Math.max(...summaries.map((summary) => summary.label.length));
|
||||
const labelPad = Math.max(...summaries.map((summary) => visibleWidth(summary.label)));
|
||||
const totalKept = summaries.reduce((total, summary) => total + summary.kept, 0);
|
||||
const totalPruned = summaries.reduce((total, summary) => total + summary.pruned, 0);
|
||||
params.runtime.log("");
|
||||
params.runtime.log("Summary by Label:");
|
||||
for (const summary of summaries) {
|
||||
params.runtime.log(
|
||||
`${summary.label.padEnd(labelPad)} ${summary.kept} kept, ${summary.pruned} pruned`,
|
||||
);
|
||||
const remaining = labelPad - visibleWidth(summary.label);
|
||||
const paddedLabel = remaining > 0 ? `${summary.label}${" ".repeat(remaining)}` : summary.label;
|
||||
params.runtime.log(`${paddedLabel} ${summary.kept} kept, ${summary.pruned} pruned`);
|
||||
}
|
||||
params.runtime.log(`Total: ${totalKept} kept, ${totalPruned} pruned`);
|
||||
}
|
||||
|
||||
@@ -475,12 +475,20 @@ export async function prepareGatewayServerBootstrap(input: {
|
||||
const workerModule = await loadWorkerEnvironmentStartupModule();
|
||||
return await workerModule.loadGatewayWorkerEnvironmentStartupState();
|
||||
});
|
||||
const { prepareGatewayPluginBootstrap } = await loadStartupPluginsModule();
|
||||
const { prepareGatewayPluginBootstrap, runGatewayStartupMaintenance } =
|
||||
await loadStartupPluginsModule();
|
||||
await startupTrace.measure("startup.maintenance", () =>
|
||||
runGatewayStartupMaintenance({
|
||||
cfgAtStart,
|
||||
startupRuntimeConfig,
|
||||
minimalTestGateway,
|
||||
log,
|
||||
}),
|
||||
);
|
||||
const pluginBootstrap = await startupTrace.measure("plugins.bootstrap", () =>
|
||||
prepareGatewayPluginBootstrap({
|
||||
cfgAtStart,
|
||||
activationSourceConfig: startupActivationSourceConfig,
|
||||
startupRuntimeConfig,
|
||||
pluginMetadataSnapshot: startupConfigLoad.pluginMetadataSnapshot,
|
||||
workerProviderIds: workerEnvironmentStartup?.durableProviderIds ?? [],
|
||||
minimalTestGateway,
|
||||
|
||||
@@ -108,6 +108,12 @@ const listAmbientOnlyConfiguredChannelIds = vi.hoisted(() =>
|
||||
vi.fn((_params: unknown) => [] as string[]),
|
||||
);
|
||||
const runStartupSessionMigration = vi.hoisted(() => vi.fn(async (_params: unknown) => undefined));
|
||||
const migrateLegacyDevicePairingStore = vi.hoisted(() =>
|
||||
vi.fn(async (_params: unknown) => undefined),
|
||||
);
|
||||
const migrateLegacyNodePairingStore = vi.hoisted(() =>
|
||||
vi.fn(async (_params: unknown) => undefined),
|
||||
);
|
||||
vi.mock("../agents/agent-scope.js", () => ({
|
||||
resolveAgentWorkspaceDir: () => "/workspace",
|
||||
resolveDefaultAgentId: () => "default",
|
||||
@@ -130,6 +136,14 @@ vi.mock("../infra/openclaw-root.js", () => ({
|
||||
resolveOpenClawPackageRootSync: (params: unknown) => resolveOpenClawPackageRootSync(params),
|
||||
}));
|
||||
|
||||
vi.mock("../infra/device-pairing-migration.js", () => ({
|
||||
migrateLegacyDevicePairingStore: (params: unknown) => migrateLegacyDevicePairingStore(params),
|
||||
}));
|
||||
|
||||
vi.mock("../infra/node-pairing-migration.js", () => ({
|
||||
migrateLegacyNodePairingStore: (params: unknown) => migrateLegacyNodePairingStore(params),
|
||||
}));
|
||||
|
||||
vi.mock("../plugins/channel-presence-policy.js", () => ({
|
||||
listAmbientOnlyConfiguredChannelIds: (params: unknown) =>
|
||||
listAmbientOnlyConfiguredChannelIds(params),
|
||||
@@ -222,7 +236,6 @@ async function prepareBootstrapWithRuntimeConfig(
|
||||
|
||||
return await prepareGatewayPluginBootstrap({
|
||||
cfgAtStart: cfg,
|
||||
startupRuntimeConfig: cfg,
|
||||
minimalTestGateway: false,
|
||||
log,
|
||||
...options,
|
||||
@@ -246,6 +259,83 @@ function expectStartupPluginLoad(params: {
|
||||
expect(startupInput.suppressPluginInfoLogs).toBe(params.suppressPluginInfoLogs);
|
||||
}
|
||||
|
||||
describe("runGatewayStartupMaintenance", () => {
|
||||
beforeEach(() => {
|
||||
runChannelPluginStartupMaintenance.mockClear();
|
||||
runStartupSessionMigration.mockClear();
|
||||
migrateLegacyDevicePairingStore.mockClear();
|
||||
migrateLegacyNodePairingStore.mockClear();
|
||||
});
|
||||
|
||||
it("runs channel, session, and ordered pairing maintenance for a normal gateway", async () => {
|
||||
const log = createLog();
|
||||
const { runGatewayStartupMaintenance } = await import("./server-startup-plugins.js");
|
||||
|
||||
await runGatewayStartupMaintenance({
|
||||
cfgAtStart: {},
|
||||
startupRuntimeConfig: {},
|
||||
minimalTestGateway: false,
|
||||
log,
|
||||
});
|
||||
|
||||
expect(runChannelPluginStartupMaintenance).toHaveBeenCalledWith({
|
||||
cfg: {},
|
||||
env: process.env,
|
||||
log,
|
||||
});
|
||||
expect(runStartupSessionMigration).toHaveBeenCalledWith({
|
||||
cfg: {},
|
||||
env: process.env,
|
||||
log,
|
||||
});
|
||||
expect(migrateLegacyDevicePairingStore).toHaveBeenCalledWith({ log });
|
||||
expect(migrateLegacyNodePairingStore).toHaveBeenCalledWith({ log });
|
||||
const deviceMigrationOrder = migrateLegacyDevicePairingStore.mock.invocationCallOrder[0];
|
||||
const nodeMigrationOrder = migrateLegacyNodePairingStore.mock.invocationCallOrder[0];
|
||||
expect(deviceMigrationOrder).toBeDefined();
|
||||
expect(nodeMigrationOrder).toBeDefined();
|
||||
expect(deviceMigrationOrder!).toBeLessThan(nodeMigrationOrder!);
|
||||
});
|
||||
|
||||
it("skips maintenance for a minimal gateway without channel config", async () => {
|
||||
const { runGatewayStartupMaintenance } = await import("./server-startup-plugins.js");
|
||||
|
||||
await runGatewayStartupMaintenance({
|
||||
cfgAtStart: {},
|
||||
startupRuntimeConfig: {},
|
||||
minimalTestGateway: true,
|
||||
log: createLog(),
|
||||
});
|
||||
|
||||
expect(runChannelPluginStartupMaintenance).not.toHaveBeenCalled();
|
||||
expect(runStartupSessionMigration).not.toHaveBeenCalled();
|
||||
expect(migrateLegacyDevicePairingStore).not.toHaveBeenCalled();
|
||||
expect(migrateLegacyNodePairingStore).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("runs only channel maintenance for a minimal gateway with recovered channel config", async () => {
|
||||
const log = createLog();
|
||||
const recoveredConfig = slackConfig();
|
||||
const { runGatewayStartupMaintenance } = await import("./server-startup-plugins.js");
|
||||
|
||||
await runGatewayStartupMaintenance({
|
||||
cfgAtStart: {},
|
||||
startupRuntimeConfig: recoveredConfig,
|
||||
minimalTestGateway: true,
|
||||
log,
|
||||
});
|
||||
|
||||
expect(runChannelPluginStartupMaintenance).toHaveBeenCalledWith({
|
||||
cfg: recoveredConfig,
|
||||
env: process.env,
|
||||
log,
|
||||
});
|
||||
expect(runStartupSessionMigration).not.toHaveBeenCalled();
|
||||
expect(migrateLegacyDevicePairingStore).not.toHaveBeenCalled();
|
||||
expect(migrateLegacyNodePairingStore).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("prepareGatewayPluginBootstrap startup plugins", () => {
|
||||
beforeEach(() => {
|
||||
applyPluginAutoEnable.mockClear();
|
||||
@@ -263,7 +353,18 @@ describe("prepareGatewayPluginBootstrap startup plugins", () => {
|
||||
resolveOpenClawPackageRootSync.mockClear().mockReturnValue("/package");
|
||||
runChannelPluginStartupMaintenance.mockClear();
|
||||
runStartupSessionMigration.mockClear();
|
||||
migrateLegacyDevicePairingStore.mockClear();
|
||||
migrateLegacyNodePairingStore.mockClear();
|
||||
});
|
||||
it("does not run startup maintenance", async () => {
|
||||
await prepareBootstrapWithRuntimeConfig({});
|
||||
|
||||
expect(runChannelPluginStartupMaintenance).not.toHaveBeenCalled();
|
||||
expect(runStartupSessionMigration).not.toHaveBeenCalled();
|
||||
expect(migrateLegacyDevicePairingStore).not.toHaveBeenCalled();
|
||||
expect(migrateLegacyNodePairingStore).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("derives startup activation from source config instead of runtime plugin defaults", async () => {
|
||||
const sourceConfig = {
|
||||
channels: {
|
||||
@@ -328,7 +429,6 @@ describe("prepareGatewayPluginBootstrap startup plugins", () => {
|
||||
await prepareGatewayPluginBootstrap({
|
||||
cfgAtStart: runtimeConfig,
|
||||
activationSourceConfig: sourceConfig,
|
||||
startupRuntimeConfig: runtimeConfig,
|
||||
pluginMetadataSnapshot,
|
||||
minimalTestGateway: false,
|
||||
log,
|
||||
@@ -435,7 +535,6 @@ describe("prepareGatewayPluginBootstrap startup plugins", () => {
|
||||
const { prepareGatewayPluginBootstrap } = await import("./server-startup-plugins.js");
|
||||
const result = await prepareGatewayPluginBootstrap({
|
||||
cfgAtStart: { channels: {} },
|
||||
startupRuntimeConfig: { channels: {} },
|
||||
minimalTestGateway: false,
|
||||
ambientEnvTriggers: "suppress",
|
||||
log,
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
// Gateway plugin startup bootstrap.
|
||||
// Runs startup maintenance, loads plugin runtime, and prepares advertised methods.
|
||||
// Gateway plugin startup bootstrap and adjacent startup maintenance.
|
||||
import { resolveAgentWorkspaceDir, resolveDefaultAgentId } from "../agents/agent-scope.js";
|
||||
import { initSubagentRegistry } from "../agents/subagent-registry.js";
|
||||
import type { AmbientEnvTriggerPolicy } from "../channels/config-presence.js";
|
||||
@@ -45,20 +44,13 @@ export function resolveGatewayStartupMaintenanceConfig(params: {
|
||||
: params.cfgAtStart;
|
||||
}
|
||||
|
||||
/** Builds plugin startup state and gateway method lists before the server binds. */
|
||||
export async function prepareGatewayPluginBootstrap(params: {
|
||||
/** Runs channel, session, and pairing maintenance before plugin bootstrap. */
|
||||
export async function runGatewayStartupMaintenance(params: {
|
||||
cfgAtStart: OpenClawConfig;
|
||||
activationSourceConfig?: OpenClawConfig;
|
||||
startupRuntimeConfig: OpenClawConfig;
|
||||
pluginMetadataSnapshot?: PluginMetadataSnapshot;
|
||||
workerProviderIds?: readonly string[];
|
||||
minimalTestGateway: boolean;
|
||||
log: GatewayPluginBootstrapLog;
|
||||
loadRuntimePlugins?: boolean;
|
||||
loadSetupRuntimePlugins?: boolean;
|
||||
ambientEnvTriggers?: AmbientEnvTriggerPolicy;
|
||||
}) {
|
||||
const activationSourceConfig = params.activationSourceConfig ?? params.cfgAtStart;
|
||||
}): Promise<void> {
|
||||
const startupMaintenanceConfig = resolveGatewayStartupMaintenanceConfig({
|
||||
cfgAtStart: params.cfgAtStart,
|
||||
startupRuntimeConfig: params.startupRuntimeConfig,
|
||||
@@ -110,7 +102,21 @@ export async function prepareGatewayPluginBootstrap(params: {
|
||||
}
|
||||
await Promise.all(startupTasks);
|
||||
}
|
||||
}
|
||||
|
||||
/** Builds plugin startup state and gateway method lists before the server binds. */
|
||||
export async function prepareGatewayPluginBootstrap(params: {
|
||||
cfgAtStart: OpenClawConfig;
|
||||
activationSourceConfig?: OpenClawConfig;
|
||||
pluginMetadataSnapshot?: PluginMetadataSnapshot;
|
||||
workerProviderIds?: readonly string[];
|
||||
minimalTestGateway: boolean;
|
||||
log: GatewayPluginBootstrapLog;
|
||||
loadRuntimePlugins?: boolean;
|
||||
loadSetupRuntimePlugins?: boolean;
|
||||
ambientEnvTriggers?: AmbientEnvTriggerPolicy;
|
||||
}) {
|
||||
const activationSourceConfig = params.activationSourceConfig ?? params.cfgAtStart;
|
||||
initSubagentRegistry();
|
||||
|
||||
// Activation uses the pre-runtime source so auto-enable policy cannot be skewed by
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
/**
|
||||
* Gateway startup session migration tests.
|
||||
*/
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { resolveOpenClawAgentSqlitePath } from "../state/openclaw-agent-db.paths.js";
|
||||
import { runStartupSessionMigration } from "./server-startup-session-migration.js";
|
||||
|
||||
type StartupMigrationDeps = NonNullable<Parameters<typeof runStartupSessionMigration>[0]["deps"]>;
|
||||
@@ -14,6 +18,7 @@ type RunDoctorSessionSqlite = NonNullable<StartupMigrationDeps["runDoctorSession
|
||||
type ReconcileSessionTranscriptIndexes = NonNullable<
|
||||
StartupMigrationDeps["reconcileSessionTranscriptIndexes"]
|
||||
>;
|
||||
type SessionSqliteDatabaseExists = NonNullable<StartupMigrationDeps["sessionSqliteDatabaseExists"]>;
|
||||
|
||||
function makeLog() {
|
||||
return {
|
||||
@@ -50,9 +55,16 @@ function makeDeps(
|
||||
.mockResolvedValue(0),
|
||||
runDoctorSessionSqlite,
|
||||
reconcileSessionTranscriptIndexes,
|
||||
sessionSqliteDatabaseExists: vi.fn<SessionSqliteDatabaseExists>().mockReturnValue(true),
|
||||
};
|
||||
}
|
||||
|
||||
function useDefaultDatabaseExists(deps: ReturnType<typeof makeDeps>): StartupMigrationDeps {
|
||||
const defaultDeps: StartupMigrationDeps = { ...deps };
|
||||
delete defaultDeps.sessionSqliteDatabaseExists;
|
||||
return defaultDeps;
|
||||
}
|
||||
|
||||
function firstLogMessage(log: ReturnType<typeof vi.fn>, label: string): string {
|
||||
const [message] = log.mock.calls[0] ?? [];
|
||||
if (typeof message !== "string") {
|
||||
@@ -231,6 +243,105 @@ describe("runStartupSessionMigration", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("skips transcript reconciliation when configured agents have no SQLite database", async () => {
|
||||
const log = makeLog();
|
||||
const migrate = vi.fn<MigrateSessionKeys>().mockResolvedValue({ changes: [], warnings: [] });
|
||||
const reconcile = vi
|
||||
.fn<ReconcileSessionTranscriptIndexes>()
|
||||
.mockResolvedValue({ reconciledSessions: 0 });
|
||||
const deps = makeDeps(migrate, 0, makeSessionSqliteImport(), reconcile);
|
||||
const defaultDeps = useDefaultDatabaseExists(deps);
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-session-startup-"));
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir };
|
||||
try {
|
||||
await runStartupSessionMigration({
|
||||
cfg: {
|
||||
agents: { defaults: {}, list: [{ id: "main" }, { id: "ops" }] },
|
||||
session: {},
|
||||
} as Parameters<typeof runStartupSessionMigration>[0]["cfg"],
|
||||
env,
|
||||
log,
|
||||
deps: defaultDeps,
|
||||
});
|
||||
|
||||
expect(reconcile).not.toHaveBeenCalled();
|
||||
expect(fs.existsSync(resolveOpenClawAgentSqlitePath({ agentId: "main", env }))).toBe(false);
|
||||
expect(fs.existsSync(resolveOpenClawAgentSqlitePath({ agentId: "ops", env }))).toBe(false);
|
||||
} finally {
|
||||
fs.rmSync(stateDir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reconciles configured agents with an existing SQLite database", async () => {
|
||||
const log = makeLog();
|
||||
const migrate = vi.fn<MigrateSessionKeys>().mockResolvedValue({ changes: [], warnings: [] });
|
||||
const reconcile = vi
|
||||
.fn<ReconcileSessionTranscriptIndexes>()
|
||||
.mockResolvedValue({ reconciledSessions: 1 });
|
||||
const deps = makeDeps(migrate, 0, makeSessionSqliteImport(), reconcile);
|
||||
const defaultDeps = useDefaultDatabaseExists(deps);
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-session-startup-"));
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir };
|
||||
const databasePath = resolveOpenClawAgentSqlitePath({ agentId: "ops", env });
|
||||
fs.mkdirSync(path.dirname(databasePath), { recursive: true });
|
||||
fs.writeFileSync(databasePath, "");
|
||||
try {
|
||||
await runStartupSessionMigration({
|
||||
cfg: {
|
||||
agents: { defaults: {}, list: [{ id: "main" }, { id: "ops" }] },
|
||||
session: {},
|
||||
} as Parameters<typeof runStartupSessionMigration>[0]["cfg"],
|
||||
env,
|
||||
log,
|
||||
deps: defaultDeps,
|
||||
});
|
||||
|
||||
expect(reconcile).toHaveBeenCalledOnce();
|
||||
expect(reconcile).toHaveBeenCalledWith({ agentId: "ops", env });
|
||||
} finally {
|
||||
fs.rmSync(stateDir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reconciles a SQLite database created by the startup import", async () => {
|
||||
const log = makeLog();
|
||||
const migrate = vi.fn<MigrateSessionKeys>().mockResolvedValue({ changes: [], warnings: [] });
|
||||
const events: string[] = [];
|
||||
const importSessionSqlite = makeSessionSqliteImport();
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-session-startup-"));
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir };
|
||||
const databasePath = resolveOpenClawAgentSqlitePath({ agentId: "main", env });
|
||||
const runDoctorSessionSqlite = vi.fn<RunDoctorSessionSqlite>().mockImplementation(async () => {
|
||||
events.push("import");
|
||||
fs.mkdirSync(path.dirname(databasePath), { recursive: true });
|
||||
fs.writeFileSync(databasePath, "");
|
||||
return await importSessionSqlite({
|
||||
allAgents: true,
|
||||
cfg: makeCfg(),
|
||||
env,
|
||||
mode: "import",
|
||||
});
|
||||
});
|
||||
const reconcile = vi
|
||||
.fn<ReconcileSessionTranscriptIndexes>()
|
||||
.mockResolvedValue({ reconciledSessions: 1 });
|
||||
const deps = makeDeps(migrate, 0, runDoctorSessionSqlite, reconcile);
|
||||
const defaultDeps = useDefaultDatabaseExists(deps);
|
||||
const cfg = makeCfg();
|
||||
reconcile.mockImplementation(async () => {
|
||||
events.push("reconcile");
|
||||
return { reconciledSessions: 1 };
|
||||
});
|
||||
try {
|
||||
await runStartupSessionMigration({ cfg, env, log, deps: defaultDeps });
|
||||
|
||||
expect(events).toEqual(["import", "reconcile"]);
|
||||
expect(reconcile).toHaveBeenCalledWith({ agentId: "main", env });
|
||||
} finally {
|
||||
fs.rmSync(stateDir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("warns without blocking when hot legacy session SQLite import reports legacy file issues", async () => {
|
||||
const log = makeLog();
|
||||
const migrate = vi.fn<MigrateSessionKeys>().mockResolvedValue({ changes: [], warnings: [] });
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import fs from "node:fs";
|
||||
import { listAgentIds } from "../agents/agent-scope.js";
|
||||
import {
|
||||
isSessionSqliteMigrationWarning,
|
||||
@@ -10,6 +11,7 @@ import {
|
||||
type SessionStartupMigrationLogger,
|
||||
} from "../config/sessions/startup-migration.js";
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import { resolveOpenClawAgentSqlitePath } from "../state/openclaw-agent-db.paths.js";
|
||||
|
||||
type SessionSqliteStartupImportRunner = (params: {
|
||||
allAgents: true;
|
||||
@@ -28,10 +30,16 @@ type SessionSqliteStartupFailureReportWriter = (
|
||||
params: { reason: string },
|
||||
) => { jsonPath: string; markdownPath: string };
|
||||
|
||||
type SessionSqliteDatabaseExists = (params: {
|
||||
agentId: string;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
}) => boolean;
|
||||
|
||||
type SessionMigrationDeps = Parameters<typeof runSessionStartupMigration>[0]["deps"] & {
|
||||
reconcileSessionTranscriptIndexes?: typeof import("../config/sessions/session-transcript-reconcile.js").reconcileSessionTranscriptIndexes;
|
||||
restoreSessionSqliteMigrationRun?: SessionSqliteStartupRestoreRunner;
|
||||
runDoctorSessionSqlite?: SessionSqliteStartupImportRunner;
|
||||
sessionSqliteDatabaseExists?: SessionSqliteDatabaseExists;
|
||||
writeSessionSqliteMigrationFailureReports?: SessionSqliteStartupFailureReportWriter;
|
||||
};
|
||||
|
||||
@@ -58,12 +66,27 @@ async function reconcileStartupSessionTranscriptIndexes(params: {
|
||||
log: SessionStartupMigrationLogger;
|
||||
deps?: SessionMigrationDeps;
|
||||
}): Promise<void> {
|
||||
const databaseExists =
|
||||
params.deps?.sessionSqliteDatabaseExists ??
|
||||
((input: Parameters<SessionSqliteDatabaseExists>[0]) =>
|
||||
fs.existsSync(resolveOpenClawAgentSqlitePath(input)));
|
||||
const agentIds = listAgentIds(params.cfg).filter((agentId) =>
|
||||
databaseExists({
|
||||
agentId,
|
||||
...(params.env ? { env: params.env } : {}),
|
||||
}),
|
||||
);
|
||||
if (agentIds.length === 0) {
|
||||
// No durable session rows can need projection repair when the agent DB is absent.
|
||||
// Avoid creating and schema-registering one empty DB per configured agent at startup.
|
||||
return;
|
||||
}
|
||||
const reconcile =
|
||||
params.deps?.reconcileSessionTranscriptIndexes ??
|
||||
(await import("../config/sessions/session-transcript-reconcile.js"))
|
||||
.reconcileSessionTranscriptIndexes;
|
||||
let reconciledSessions = 0;
|
||||
for (const agentId of listAgentIds(params.cfg)) {
|
||||
for (const agentId of agentIds) {
|
||||
const result = await reconcile({
|
||||
agentId,
|
||||
...(params.env ? { env: params.env } : {}),
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -128,6 +128,137 @@ describe("realtime Talk conversation", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it("bounds streamed assistant delta growth while retaining useful context", () => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
const opening = "Opening context stays visible. ";
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: `${opening}${"a".repeat(7_900)}`,
|
||||
final: false,
|
||||
nowMs: 1,
|
||||
});
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: "b".repeat(500),
|
||||
final: false,
|
||||
nowMs: 2,
|
||||
});
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: `${"c".repeat(500)}NEWEST`,
|
||||
final: false,
|
||||
nowMs: 3,
|
||||
});
|
||||
|
||||
expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000);
|
||||
expect(state.entries[0]?.text.startsWith(opening)).toBe(true);
|
||||
expect(state.entries[0]?.text).toContain("\n…\n");
|
||||
expect(state.entries[0]?.text.split("\n…\n")).toHaveLength(2);
|
||||
expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true);
|
||||
});
|
||||
|
||||
it("replaces a bounded assistant stream with the authoritative final transcript", () => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
const opening = "Original opening context. ";
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: `${opening}${"draft ".repeat(1_600)}`,
|
||||
final: false,
|
||||
nowMs: 1,
|
||||
});
|
||||
expect(state.entries[0]?.text).toContain("\n…\n");
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: `${opening}corrected ${"final ".repeat(1_600)}DONE`,
|
||||
final: true,
|
||||
nowMs: 2,
|
||||
});
|
||||
|
||||
expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000);
|
||||
expect(state.entries[0]?.text.startsWith(`${opening}corrected `)).toBe(true);
|
||||
expect(state.entries[0]?.text).not.toContain("draft ");
|
||||
expect(state.entries[0]?.text.endsWith("DONE")).toBe(true);
|
||||
expect(state.entries[0]?.isStreaming).toBe(false);
|
||||
});
|
||||
|
||||
it("does not expose dangling surrogates at a bounded transcript edge", () => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
const transcript = `${"a".repeat(8_000)}🚀${"b".repeat(7_740)}`;
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: transcript,
|
||||
final: true,
|
||||
nowMs: 1,
|
||||
});
|
||||
|
||||
const text = state.entries[0]?.text ?? "";
|
||||
expect(text.length).toBeLessThanOrEqual(8_000);
|
||||
expect(text).not.toMatch(
|
||||
/(?:[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF])/,
|
||||
);
|
||||
});
|
||||
|
||||
it("does not trust a natural truncation marker outside the bounded prefix", () => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: `${"a".repeat(7_998)}\n…\n${"b".repeat(500)}NEWEST`,
|
||||
final: true,
|
||||
nowMs: 1,
|
||||
});
|
||||
|
||||
expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000);
|
||||
expect(state.entries[0]?.text.startsWith("a".repeat(256))).toBe(true);
|
||||
expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true);
|
||||
});
|
||||
|
||||
it.each([255, 256])(
|
||||
"does not retain a lone high surrogate before a natural marker at offset %i",
|
||||
(markerOffset) => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
const retainedText = "a".repeat(markerOffset - 1);
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role: "assistant",
|
||||
text: `${retainedText}\uD800\n…\n${"b".repeat(8_000)}NEWEST`,
|
||||
final: true,
|
||||
nowMs: 1,
|
||||
});
|
||||
|
||||
const text = state.entries[0]?.text ?? "";
|
||||
expect(text.length).toBeLessThanOrEqual(8_000);
|
||||
expect(text.startsWith(`${retainedText}\n…\n`)).toBe(true);
|
||||
expect(text.endsWith("NEWEST")).toBe(true);
|
||||
expect(text).not.toMatch(
|
||||
/(?:[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF])/,
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["user", "assistant"] as const)(
|
||||
"bounds oversized final %s entries while retaining the newest text",
|
||||
(role) => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
|
||||
state = updateRealtimeTalkConversation(state, {
|
||||
role,
|
||||
text: `Useful opening. ${"x".repeat(9_000)}NEWEST`,
|
||||
final: true,
|
||||
nowMs: 1,
|
||||
});
|
||||
|
||||
expect(state.entries[0]?.text.length).toBeLessThanOrEqual(8_000);
|
||||
expect(state.entries[0]?.text.startsWith("Useful opening. ")).toBe(true);
|
||||
expect(state.entries[0]?.text.endsWith("NEWEST")).toBe(true);
|
||||
expect(state.entries[0]?.isStreaming).toBe(false);
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps alternating realtime turns as separate bubbles", () => {
|
||||
let state = createRealtimeTalkConversationState();
|
||||
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
// Control UI chat module implements realtime talk conversation behavior.
|
||||
import { sliceUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
|
||||
|
||||
type RealtimeTalkConversationRole = "user" | "assistant";
|
||||
|
||||
export type RealtimeTalkConversationEntry = {
|
||||
@@ -25,6 +27,9 @@ type RealtimeTalkTranscriptUpdate = {
|
||||
};
|
||||
|
||||
const MAX_CONVERSATION_ENTRIES = 60;
|
||||
const MAX_CONVERSATION_ENTRY_CHARS = 8_000;
|
||||
const CONVERSATION_ENTRY_PREFIX_CHARS = 256;
|
||||
const CONVERSATION_ENTRY_TRUNCATION_MARKER = "\n…\n";
|
||||
const USER_FINAL_REWRITE_GRACE_MS = 1_500;
|
||||
|
||||
export function createRealtimeTalkConversationState(): RealtimeTalkConversationState {
|
||||
@@ -96,7 +101,12 @@ function upsertRealtimeConversationEntry(
|
||||
const id = `rt-${state.nextEntryId}`;
|
||||
const entries = [
|
||||
...state.entries,
|
||||
{ id, role, text: text.trimStart(), isStreaming: !isFinal },
|
||||
{
|
||||
id,
|
||||
role,
|
||||
text: boundRealtimeConversationText(text.trimStart()),
|
||||
isStreaming: !isFinal,
|
||||
},
|
||||
].slice(-MAX_CONVERSATION_ENTRIES);
|
||||
return rememberRealtimeConversationEntry(
|
||||
{ ...state, entries, nextEntryId: state.nextEntryId + 1 },
|
||||
@@ -115,10 +125,11 @@ function upsertRealtimeConversationEntry(
|
||||
if (!entry) {
|
||||
return upsertRealtimeConversationEntry(state, role, null, text, isFinal, nowMs);
|
||||
}
|
||||
const updatedText =
|
||||
const mergedText =
|
||||
role === "assistant"
|
||||
? mergeAssistantTranscriptText(entry.text, text, isFinal)
|
||||
: mergeRealtimeTranscriptText(entry.text, text, isFinal);
|
||||
const updatedText = boundRealtimeConversationText(mergedText);
|
||||
const entries =
|
||||
entry.text === updatedText && entry.isStreaming === !isFinal
|
||||
? state.entries
|
||||
@@ -263,6 +274,27 @@ function mergeRealtimeTranscriptText(existing: string, incoming: string, isFinal
|
||||
return `${existing}${separator}${suffix}`;
|
||||
}
|
||||
|
||||
function boundRealtimeConversationText(text: string): string {
|
||||
if (text.length <= MAX_CONVERSATION_ENTRY_CHARS) {
|
||||
return text;
|
||||
}
|
||||
// Keep the opening context for late full-final replacement detection and
|
||||
// the newest tail for the visible conversation. Reuse the original prefix
|
||||
// so repeated streaming deltas do not move the truncation boundary.
|
||||
const markerIndex = text.indexOf(CONVERSATION_ENTRY_TRUNCATION_MARKER);
|
||||
const hasBoundedPrefix =
|
||||
markerIndex >= CONVERSATION_ENTRY_PREFIX_CHARS - 1 &&
|
||||
markerIndex <= CONVERSATION_ENTRY_PREFIX_CHARS;
|
||||
const prefixEnd = hasBoundedPrefix ? markerIndex : CONVERSATION_ENTRY_PREFIX_CHARS;
|
||||
// A natural marker can follow malformed provider text ending in a lone high
|
||||
// surrogate. Keep that code unit out of the retained truncation boundary.
|
||||
const prefix = sliceUtf16Safe(text, 0, prefixEnd).replace(/[\uD800-\uDBFF]$/, "");
|
||||
const tailChars =
|
||||
MAX_CONVERSATION_ENTRY_CHARS - prefix.length - CONVERSATION_ENTRY_TRUNCATION_MARKER.length;
|
||||
const tail = sliceUtf16Safe(text, -tailChars);
|
||||
return `${prefix}${CONVERSATION_ENTRY_TRUNCATION_MARKER}${tail}`;
|
||||
}
|
||||
|
||||
function looksLikeTranscriptReplacement(existing: string, incoming: string): boolean {
|
||||
const existingWords = transcriptWords(existing);
|
||||
const incomingWords = transcriptWords(incoming);
|
||||
|
||||
@@ -602,19 +602,38 @@ describe("cron view editor", () => {
|
||||
expect(onClosePanel).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("wires form changes from prompt and name inputs", () => {
|
||||
it("wires shared text and select controls without changing their field ownership", () => {
|
||||
const onFormChange = vi.fn();
|
||||
const container = renderView({ createOpen: true, onFormChange });
|
||||
const container = renderView({
|
||||
createOpen: true,
|
||||
channels: ["telegram"],
|
||||
channelMeta: [{ id: "telegram", label: "", detailLabel: "Telegram" }],
|
||||
channelLabels: { telegram: "Telegram fallback" },
|
||||
form: { ...DEFAULT_CRON_FORM, scheduleKind: "cron", failureAlertMode: "custom" },
|
||||
onFormChange,
|
||||
});
|
||||
|
||||
const prompt = getElement(container, "#cron-payload-text", HTMLTextAreaElement);
|
||||
prompt.value = "do the thing";
|
||||
prompt.dispatchEvent(new Event("input", { bubbles: true }));
|
||||
expect(onFormChange).toHaveBeenCalledWith({ payloadText: "do the thing" });
|
||||
|
||||
const name = getElement(container, "#cron-name", HTMLInputElement);
|
||||
name.value = "Thing";
|
||||
name.dispatchEvent(new Event("input", { bubbles: true }));
|
||||
expect(onFormChange).toHaveBeenCalledWith({ name: "Thing" });
|
||||
for (const field of ["name", "sessionKey", "deliveryAccountId", "payloadModel"] as const) {
|
||||
const id = `cron-${field.replace(/[A-Z]/g, (letter) => `-${letter.toLowerCase()}`)}`;
|
||||
const input = getElement(container, `#${id}`, HTMLInputElement);
|
||||
if (field === "sessionKey" || field === "deliveryAccountId") {
|
||||
expect(input.placeholder).toBe(field === "sessionKey" ? "agent:main:main" : "default");
|
||||
}
|
||||
input.value = field;
|
||||
input.dispatchEvent(new Event("input", { bubbles: true }));
|
||||
expect(onFormChange).toHaveBeenLastCalledWith({ [field]: field });
|
||||
}
|
||||
|
||||
const channel = getElement(container, "#cron-failure-alert-channel", HTMLSelectElement);
|
||||
channel.value = "telegram";
|
||||
expect(channel.selectedOptions[0]?.textContent).toBe("Telegram fallback");
|
||||
channel.dispatchEvent(new Event("change", { bubbles: true }));
|
||||
expect(onFormChange).toHaveBeenLastCalledWith({ failureAlertChannel: "telegram" });
|
||||
});
|
||||
|
||||
it("switches schedule inputs by segmented kind and wires kind changes", () => {
|
||||
|
||||
+373
-674
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user