john
632 lines
23 KiB
JavaScript
632 lines
23 KiB
JavaScript
import { deriveAssistantFacingText, deriveUserFacingText } from './conversation-display.mjs';
|
||
|
||
const TABLE = 'h5_episodic_memory_items';
|
||
const DEFAULT_LIMIT = 3;
|
||
const MAX_LIMIT = 8;
|
||
const MAX_SEARCH_CHARS = 120_000;
|
||
const MAX_SEGMENT_CHARS = 360;
|
||
const MAX_EXCERPT_CHARS = 900;
|
||
|
||
const EPISODIC_RECALL_PATTERNS = [
|
||
/(?:我们|我和你|咱们).{0,12}(?:之前|以前|上次|曾经|过去)?.{0,12}(?:聊过|聊了|讨论过|讨论了|谈过|谈了|提到过|提了|说过|说了)/u,
|
||
/(?:之前|以前|上次|曾经|过去).{0,20}(?:聊过|聊了|讨论过|讨论了|谈过|谈了|提到过|提了|说过|说了|交流过)/u,
|
||
/你(?:还)?记得.{0,24}(?:聊过|聊了|讨论过|讨论了|谈过|谈了|提到过|说过)/u,
|
||
/(?:聊过|聊了|讨论过|讨论了|谈过|谈了|提到过|说过).{0,24}你(?:还)?记得/u,
|
||
];
|
||
|
||
function readFlag(value, fallback = false) {
|
||
if (value == null || value === '') return fallback;
|
||
const normalized = String(value).trim().toLowerCase();
|
||
if (['1', 'true', 'yes', 'on'].includes(normalized)) return true;
|
||
if (['0', 'false', 'no', 'off'].includes(normalized)) return false;
|
||
return fallback;
|
||
}
|
||
|
||
function normalizeRolloutMode(value) {
|
||
const normalized = String(value ?? 'off').trim().toLowerCase();
|
||
return ['off', 'canary', 'active'].includes(normalized) ? normalized : 'off';
|
||
}
|
||
|
||
function normalizeUserIds(value) {
|
||
return [...new Set(String(value ?? '')
|
||
.split(/[\s,]+/u)
|
||
.map((item) => item.trim())
|
||
.filter(Boolean))].slice(0, 1_000);
|
||
}
|
||
|
||
function boundedNumber(value, fallback, { min, max }) {
|
||
const parsed = Number(value);
|
||
if (!Number.isFinite(parsed)) return fallback;
|
||
return Math.min(max, Math.max(min, parsed));
|
||
}
|
||
|
||
function collapseText(value) {
|
||
return String(value ?? '').replace(/\s+/gu, ' ').trim();
|
||
}
|
||
|
||
function clipText(value, maxChars) {
|
||
const text = collapseText(value);
|
||
if (text.length <= maxChars) return text;
|
||
return `${text.slice(0, Math.max(0, maxChars - 1))}…`;
|
||
}
|
||
|
||
function messageText(message) {
|
||
const content = Array.isArray(message?.content)
|
||
? message.content
|
||
.filter((item) => typeof item === 'string' || item?.type === 'text')
|
||
.map((item) => (typeof item === 'string' ? item : item.text ?? ''))
|
||
.join('\n')
|
||
: (message?.content ?? message?.text ?? message?.value ?? '');
|
||
const displayText = message?.metadata?.displayText;
|
||
const raw = typeof displayText === 'string' && displayText.trim() ? displayText : content;
|
||
if (message?.role === 'user') return deriveUserFacingText(raw);
|
||
if (message?.role === 'assistant') return deriveAssistantFacingText(raw);
|
||
return '';
|
||
}
|
||
|
||
function parseTimestamp(value, fallback = 0) {
|
||
if (value == null || value === '') return fallback;
|
||
const numeric = Number(value);
|
||
if (Number.isFinite(numeric) && numeric > 0) return numeric;
|
||
const parsed = Date.parse(String(value));
|
||
return Number.isFinite(parsed) ? parsed : fallback;
|
||
}
|
||
|
||
function escapeLike(value) {
|
||
return String(value).replace(/=/g, '==').replace(/%/g, '=%').replace(/_/g, '=_');
|
||
}
|
||
|
||
function parseJson(value, fallback) {
|
||
if (value == null || value === '') return fallback;
|
||
if (typeof value === 'object') return value;
|
||
try {
|
||
return JSON.parse(value);
|
||
} catch {
|
||
return fallback;
|
||
}
|
||
}
|
||
|
||
export function isEpisodicRecallQuery(query) {
|
||
const normalized = collapseText(query);
|
||
return Boolean(normalized && EPISODIC_RECALL_PATTERNS.some((pattern) => pattern.test(normalized)));
|
||
}
|
||
|
||
export function extractEpisodicRecallTopic(query) {
|
||
const normalized = collapseText(query)
|
||
.replace(/[？?！!。]+$/u, '')
|
||
.trim();
|
||
if (!normalized) return '';
|
||
|
||
const afterConversationVerb = normalized.match(
|
||
/(?:聊过|聊了|讨论过|讨论了|谈过|谈了|提到过|提了|说过|说了|交流过)(?:的|关于)?(.+)$/u,
|
||
)?.[1];
|
||
let topic = afterConversationVerb ?? normalized;
|
||
topic = topic
|
||
.replace(/^(?:关于|有关|一下|一些|那个|这个)\s*/u, '')
|
||
.replace(/[，,；;：:]?(?:你)?(?:还)?记得(?:吗|么|呢|不|没有|吧)?$/u, '')
|
||
.replace(/(?:这件事|这个话题)(?:吗|么|呢)?$/u, '')
|
||
.replace(/^(?:你(?:还)?记得|我们|我和你|咱们|我|之前|以前|上次|曾经|过去)+/u, '')
|
||
.replace(/(?:吗|呢|吧)$/u, '')
|
||
.trim();
|
||
if (topic.endsWith('么') && !/(?:什么|怎么)$/u.test(topic)) {
|
||
topic = topic.slice(0, -1).trim();
|
||
}
|
||
if (/^(?:什么|啥|哪些|什么内容|什么话题|哪些内容|哪些话题)$/u.test(topic)) return '';
|
||
if (!afterConversationVerb || topic.length < 2) {
|
||
const beforeConversationVerb = normalized.match(
|
||
/(?:之前|以前|上次|曾经|过去)(?:有|也)?(?:关于)?(.{2,80}?)(?:聊过|聊了|讨论过|讨论了|谈过|谈了|提到过|提了|说过|说了|交流过)/u,
|
||
)?.[1]?.trim();
|
||
if (beforeConversationVerb) topic = beforeConversationVerb;
|
||
}
|
||
return topic.length >= 2 ? topic.slice(0, 120) : '';
|
||
}
|
||
|
||
function topicTerms(query) {
|
||
const topic = extractEpisodicRecallTopic(query);
|
||
if (!topic) return [];
|
||
const terms = [topic];
|
||
for (const part of topic.split(/[\s，,。；;：:、/|()（）《》“”"'？?！!]+|(?:关于|有关|以及|还有|是什么|有什么|如何|怎么|为何|为什么|的|在|和|与)/u)) {
|
||
const normalized = part.trim();
|
||
if (normalized.length >= 2 && normalized !== topic) terms.push(normalized);
|
||
}
|
||
return [...new Set(terms)].slice(0, 4);
|
||
}
|
||
|
||
function visibleSegments(messages) {
|
||
const segments = [];
|
||
let usedChars = 0;
|
||
for (const message of Array.isArray(messages) ? messages : []) {
|
||
if (!['user', 'assistant'].includes(message?.role)) continue;
|
||
if (message?.metadata?.userVisible === false) continue;
|
||
const text = clipText(messageText(message), MAX_SEGMENT_CHARS);
|
||
if (!text) continue;
|
||
const prefix = message.role === 'user' ? '用户' : '助手';
|
||
const segment = `${prefix}：${text}`;
|
||
if (usedChars + segment.length > MAX_SEARCH_CHARS) break;
|
||
segments.push(segment);
|
||
usedChars += segment.length + 1;
|
||
}
|
||
return segments;
|
||
}
|
||
|
||
export function buildEpisodicSnapshotDocument({ sessionId, userId, session = {}, messages = [] } = {}) {
|
||
const segments = visibleSegments(messages);
|
||
const title = clipText(
|
||
session.display_title ?? session.displayTitle ?? session.name ?? '',
|
||
512,
|
||
);
|
||
const summarySegments = segments.length <= 6
|
||
? segments
|
||
: [...segments.slice(0, 2), ...segments.slice(-4)];
|
||
const summaryText = clipText(summarySegments.join('\n'), 2_400);
|
||
const searchText = [title ? `标题：${title}` : '', ...segments].filter(Boolean).join('\n');
|
||
const fallbackTimestamp = Date.now();
|
||
return {
|
||
sessionId: String(sessionId ?? '').trim(),
|
||
userId: String(userId ?? '').trim(),
|
||
title,
|
||
summaryText,
|
||
searchText,
|
||
messageCount: segments.length,
|
||
sourceUpdatedAt: String(session.updated_at ?? session.source_updated_at ?? '').trim(),
|
||
sessionUpdatedAt: parseTimestamp(
|
||
session.updated_at ?? session.source_updated_at,
|
||
fallbackTimestamp,
|
||
),
|
||
};
|
||
}
|
||
|
||
export function buildEpisodicMemorySchemaSql({ table = TABLE } = {}) {
|
||
if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new Error('Invalid episodic memory table name');
|
||
return `CREATE TABLE IF NOT EXISTS \`${table}\` (
|
||
agent_session_id VARCHAR(128) NOT NULL PRIMARY KEY,
|
||
user_id CHAR(36) NOT NULL,
|
||
display_title VARCHAR(512) NOT NULL DEFAULT '',
|
||
summary_text TEXT NOT NULL,
|
||
search_text MEDIUMTEXT NOT NULL,
|
||
message_count INT NOT NULL DEFAULT 0,
|
||
source_updated_at VARCHAR(64) NOT NULL DEFAULT '',
|
||
session_updated_at BIGINT NOT NULL,
|
||
indexed_at BIGINT NOT NULL,
|
||
KEY idx_h5_episodic_user_updated (user_id, session_updated_at),
|
||
CONSTRAINT fk_h5_episodic_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`;
|
||
}
|
||
|
||
function indexedRowToDocument(row) {
|
||
return {
|
||
sessionId: String(row.agent_session_id),
|
||
userId: String(row.user_id),
|
||
title: String(row.display_title ?? ''),
|
||
summaryText: String(row.summary_text ?? ''),
|
||
searchText: String(row.search_text ?? ''),
|
||
messageCount: Number(row.message_count ?? 0),
|
||
sourceUpdatedAt: String(row.source_updated_at ?? ''),
|
||
sessionUpdatedAt: Number(row.session_updated_at ?? 0),
|
||
source: 'episodic-index',
|
||
};
|
||
}
|
||
|
||
function snapshotRowToDocument(row) {
|
||
const messages = parseJson(row.messages_json, []);
|
||
return {
|
||
...buildEpisodicSnapshotDocument({
|
||
sessionId: row.agent_session_id,
|
||
userId: row.user_id,
|
||
session: {
|
||
display_title: row.display_title,
|
||
name: row.name,
|
||
updated_at: row.source_updated_at || row.updated_at_str || row.synced_at,
|
||
},
|
||
messages,
|
||
}),
|
||
sessionUpdatedAt: parseTimestamp(
|
||
row.source_updated_at || row.updated_at_str,
|
||
Number(row.synced_at ?? 0),
|
||
),
|
||
source: 'session-snapshot-fallback',
|
||
messages,
|
||
};
|
||
}
|
||
|
||
function matchingExcerpt(document, terms) {
|
||
const segments = String(document.searchText ?? '').split('\n').filter(Boolean);
|
||
const loweredTerms = terms.map((term) => term.toLocaleLowerCase());
|
||
const matchIndex = segments.findIndex((segment) => {
|
||
const lowered = segment.toLocaleLowerCase();
|
||
return loweredTerms.some((term) => lowered.includes(term));
|
||
});
|
||
if (matchIndex < 0) return clipText(document.summaryText, MAX_EXCERPT_CHARS);
|
||
return clipText(
|
||
segments.slice(Math.max(0, matchIndex - 1), matchIndex + 2).join('；'),
|
||
MAX_EXCERPT_CHARS,
|
||
);
|
||
}
|
||
|
||
function documentScore(document, terms) {
|
||
const title = String(document.title ?? '').toLocaleLowerCase();
|
||
const haystack = String(document.searchText ?? '').toLocaleLowerCase();
|
||
let score = 0;
|
||
terms.forEach((term, index) => {
|
||
const lowered = term.toLocaleLowerCase();
|
||
if (haystack.includes(lowered)) score += 20 - Math.min(index * 3, 9);
|
||
if (title.includes(lowered)) score += 8;
|
||
});
|
||
score += Math.min(5, Math.max(0, Number(document.sessionUpdatedAt ?? 0)) / 10 ** 15);
|
||
return score;
|
||
}
|
||
|
||
function formatDate(timestamp, sourceUpdatedAt) {
|
||
const parsed = parseTimestamp(sourceUpdatedAt, Number(timestamp ?? 0));
|
||
if (!parsed) return '';
|
||
try {
|
||
return new Date(parsed).toISOString().slice(0, 10);
|
||
} catch {
|
||
return '';
|
||
}
|
||
}
|
||
|
||
function toMemory(document, terms) {
|
||
const excerpt = matchingExcerpt(document, terms);
|
||
if (!excerpt) return null;
|
||
const date = formatDate(document.sessionUpdatedAt, document.sourceUpdatedAt);
|
||
const title = clipText(document.title, 80);
|
||
const descriptor = [date, title ? `会话“${title}”` : '历史会话'].filter(Boolean).join('，');
|
||
return {
|
||
id: `episodic:${document.sessionId}`,
|
||
type: 'episodic',
|
||
label: '历史会话',
|
||
text: `${descriptor || '历史会话'}：${excerpt}`,
|
||
sessionId: document.sessionId,
|
||
source: document.source,
|
||
evidence: {
|
||
sessionId: document.sessionId,
|
||
sourceUpdatedAt: document.sourceUpdatedAt || null,
|
||
matchedTerms: terms.filter((term) => String(document.searchText).toLocaleLowerCase().includes(term.toLocaleLowerCase())),
|
||
},
|
||
};
|
||
}
|
||
|
||
async function raceWithTimeout(promise, timeoutMs) {
|
||
if (!timeoutMs) return promise;
|
||
let timer;
|
||
try {
|
||
return await Promise.race([
|
||
promise,
|
||
new Promise((_, reject) => {
|
||
timer = setTimeout(() => {
|
||
const error = new Error('Episodic memory resolve timed out');
|
||
error.code = 'EPISODIC_MEMORY_TIMEOUT';
|
||
reject(error);
|
||
}, timeoutMs);
|
||
}),
|
||
]);
|
||
} finally {
|
||
if (timer) clearTimeout(timer);
|
||
}
|
||
}
|
||
|
||
export function createEpisodicMemoryService(pool, {
|
||
table = TABLE,
|
||
env = process.env,
|
||
getEffectiveEnv = null,
|
||
logger = console,
|
||
now = () => Date.now(),
|
||
} = {}) {
|
||
if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new Error('Invalid episodic memory table name');
|
||
const quotedTable = `\`${table}\``;
|
||
let effectiveEnvCache = null;
|
||
const metrics = {
|
||
resolveStarted: 0,
|
||
resolved: 0,
|
||
skipped: 0,
|
||
degraded: 0,
|
||
matched: 0,
|
||
indexed: 0,
|
||
indexSkipped: 0,
|
||
indexFailed: 0,
|
||
lastReason: null,
|
||
};
|
||
|
||
async function resolveEnv() {
|
||
if (typeof getEffectiveEnv !== 'function') return env;
|
||
const currentTime = now();
|
||
if (effectiveEnvCache && currentTime - effectiveEnvCache.loadedAt < 1_000) {
|
||
return effectiveEnvCache.value;
|
||
}
|
||
try {
|
||
const value = { ...env, ...(await getEffectiveEnv()) };
|
||
effectiveEnvCache = { loadedAt: currentTime, value };
|
||
return value;
|
||
} catch {
|
||
return env;
|
||
}
|
||
}
|
||
|
||
async function resolvePolicy(userId = null) {
|
||
const effectiveEnv = await resolveEnv();
|
||
const retrieverEnabled = readFlag(effectiveEnv?.MEMORY_RETRIEVER_ENABLED, false);
|
||
const episodicEnabled = readFlag(effectiveEnv?.MEMORY_RETRIEVER_EPISODIC_ENABLED, false);
|
||
const configuredEnabled = retrieverEnabled && episodicEnabled;
|
||
const mode = normalizeRolloutMode(effectiveEnv?.MEMORY_RETRIEVER_EPISODIC_MODE);
|
||
const canaryUserIds = normalizeUserIds(effectiveEnv?.MEMORY_RETRIEVER_EPISODIC_CANARY_USER_IDS);
|
||
const normalizedUserId = String(userId ?? '').trim();
|
||
const inRollout = mode === 'active'
|
||
|| (mode === 'canary' && normalizedUserId && canaryUserIds.includes(normalizedUserId));
|
||
return {
|
||
configuredEnabled,
|
||
retrieverEnabled,
|
||
episodicEnabled,
|
||
enabled: configuredEnabled && inRollout,
|
||
mode,
|
||
canaryUserIds,
|
||
inRollout: Boolean(inRollout),
|
||
limit: Math.round(boundedNumber(effectiveEnv?.MEMORY_RETRIEVER_LIMIT, DEFAULT_LIMIT, { min: 1, max: MAX_LIMIT })),
|
||
charBudget: Math.round(boundedNumber(effectiveEnv?.MEMORY_RETRIEVER_TOKEN_BUDGET, 1_800, { min: 80, max: 4_000 })) * 3,
|
||
timeoutMs: Math.round(boundedNumber(effectiveEnv?.MEMORY_RETRIEVER_TIMEOUT_MS, 1_200, { min: 0, max: 10_000 })),
|
||
};
|
||
}
|
||
|
||
async function ensureSchema() {
|
||
if (!pool?.query) return { ok: false, reason: 'database_unavailable' };
|
||
await pool.query(buildEpisodicMemorySchemaSql({ table }));
|
||
return { ok: true };
|
||
}
|
||
|
||
async function upsertSnapshot({ sessionId, userId, session, messages } = {}) {
|
||
if (!pool?.query || !sessionId || !userId) {
|
||
metrics.indexSkipped += 1;
|
||
return { ok: false, skipped: true, reason: 'invalid_input' };
|
||
}
|
||
const policy = await resolvePolicy(userId);
|
||
if (!policy.enabled) {
|
||
const reason = !policy.configuredEnabled
|
||
? 'disabled'
|
||
: policy.mode === 'off' ? 'rollout_off' : 'outside_canary';
|
||
metrics.indexSkipped += 1;
|
||
return { ok: true, skipped: true, reason };
|
||
}
|
||
const document = buildEpisodicSnapshotDocument({ sessionId, userId, session, messages });
|
||
if (!document.searchText || document.messageCount <= 0) {
|
||
metrics.indexSkipped += 1;
|
||
return { ok: true, skipped: true, reason: 'empty_snapshot' };
|
||
}
|
||
try {
|
||
await pool.query(
|
||
`INSERT INTO ${quotedTable}
|
||
(agent_session_id, user_id, display_title, summary_text, search_text,
|
||
message_count, source_updated_at, session_updated_at, indexed_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
user_id = VALUES(user_id),
|
||
display_title = VALUES(display_title),
|
||
summary_text = VALUES(summary_text),
|
||
search_text = VALUES(search_text),
|
||
message_count = VALUES(message_count),
|
||
source_updated_at = VALUES(source_updated_at),
|
||
session_updated_at = VALUES(session_updated_at),
|
||
indexed_at = VALUES(indexed_at)`,
|
||
[
|
||
document.sessionId,
|
||
document.userId,
|
||
document.title,
|
||
document.summaryText,
|
||
document.searchText,
|
||
document.messageCount,
|
||
document.sourceUpdatedAt,
|
||
document.sessionUpdatedAt,
|
||
now(),
|
||
],
|
||
);
|
||
} catch (err) {
|
||
metrics.indexFailed += 1;
|
||
throw err;
|
||
}
|
||
metrics.indexed += 1;
|
||
return { ok: true, skipped: false, document };
|
||
}
|
||
|
||
async function remove(sessionId) {
|
||
if (!pool?.query || !sessionId) return { ok: false, skipped: true };
|
||
await pool.query(`DELETE FROM ${quotedTable} WHERE agent_session_id = ?`, [String(sessionId)]);
|
||
return { ok: true };
|
||
}
|
||
|
||
async function queryIndexed({ userId, sessionId, terms, candidateLimit }) {
|
||
const where = ['user_id = ?'];
|
||
const params = [String(userId)];
|
||
if (sessionId) {
|
||
where.push('agent_session_id <> ?');
|
||
params.push(String(sessionId));
|
||
}
|
||
if (terms.length) {
|
||
const clauses = [];
|
||
for (const term of terms) {
|
||
clauses.push("(display_title LIKE ? ESCAPE '=' OR search_text LIKE ? ESCAPE '=')");
|
||
const pattern = `%${escapeLike(term)}%`;
|
||
params.push(pattern, pattern);
|
||
}
|
||
where.push(`(${clauses.join(' OR ')})`);
|
||
}
|
||
params.push(candidateLimit);
|
||
const [rows] = await pool.query(
|
||
`SELECT agent_session_id, user_id, display_title, summary_text, search_text,
|
||
message_count, source_updated_at, session_updated_at
|
||
FROM ${quotedTable}
|
||
WHERE ${where.join(' AND ')}
|
||
ORDER BY session_updated_at DESC
|
||
LIMIT ?`,
|
||
params,
|
||
);
|
||
return rows.map(indexedRowToDocument);
|
||
}
|
||
|
||
async function querySnapshots({ userId, sessionId, terms, candidateLimit }) {
|
||
const where = ['user_id = ?'];
|
||
const params = [String(userId)];
|
||
if (sessionId) {
|
||
where.push('agent_session_id <> ?');
|
||
params.push(String(sessionId));
|
||
}
|
||
if (terms.length) {
|
||
const clauses = [];
|
||
for (const term of terms) {
|
||
clauses.push("(display_title LIKE ? ESCAPE '=' OR messages_json LIKE ? ESCAPE '=')");
|
||
const pattern = `%${escapeLike(term)}%`;
|
||
params.push(pattern, pattern);
|
||
}
|
||
where.push(`(${clauses.join(' OR ')})`);
|
||
}
|
||
params.push(candidateLimit);
|
||
const [rows] = await pool.query(
|
||
`SELECT agent_session_id, user_id, name, display_title, updated_at_str,
|
||
source_updated_at, messages_json, synced_at
|
||
FROM h5_session_snapshots
|
||
WHERE ${where.join(' AND ')}
|
||
ORDER BY synced_at DESC
|
||
LIMIT ?`,
|
||
params,
|
||
);
|
||
return rows.map(snapshotRowToDocument).filter((item) => item.searchText);
|
||
}
|
||
|
||
async function resolveCore({ userId, sessionId, query, limit, policy }) {
|
||
if (!pool?.query || !userId) {
|
||
return { memories: [], skipped: true, reason: 'database_unavailable', source: 'episodic' };
|
||
}
|
||
if (!policy.enabled) {
|
||
const reason = !policy.configuredEnabled
|
||
? 'disabled'
|
||
: policy.mode === 'off' ? 'rollout_off' : 'outside_canary';
|
||
return { memories: [], skipped: true, reason, source: 'episodic', mode: policy.mode };
|
||
}
|
||
if (!isEpisodicRecallQuery(query)) {
|
||
return { memories: [], skipped: true, reason: 'not_historical_recall', source: 'episodic' };
|
||
}
|
||
|
||
const terms = topicTerms(query);
|
||
const resolvedLimit = Math.min(policy.limit, Math.max(1, Number(limit) || policy.limit));
|
||
const candidateLimit = Math.min(32, Math.max(8, resolvedLimit * 4));
|
||
const candidates = new Map();
|
||
let indexFailed = false;
|
||
try {
|
||
for (const document of await queryIndexed({ userId, sessionId, terms, candidateLimit })) {
|
||
candidates.set(document.sessionId, document);
|
||
}
|
||
} catch (err) {
|
||
indexFailed = true;
|
||
logger?.warn?.(`[episodic-memory] index query degraded: ${err instanceof Error ? err.message : err}`);
|
||
}
|
||
|
||
// Existing users are covered by the snapshot fallback until their first
|
||
// matching lookup lazily creates an index row. Once the index already has
|
||
// a match, avoid scanning the larger snapshot JSON on every recall.
|
||
if (indexFailed || candidates.size === 0) {
|
||
try {
|
||
const fallback = await querySnapshots({ userId, sessionId, terms, candidateLimit });
|
||
for (const document of fallback) {
|
||
if (!candidates.has(document.sessionId)) candidates.set(document.sessionId, document);
|
||
}
|
||
for (const document of fallback) {
|
||
if (document.source !== 'session-snapshot-fallback') continue;
|
||
void upsertSnapshot({
|
||
sessionId: document.sessionId,
|
||
userId: document.userId,
|
||
session: {
|
||
display_title: document.title,
|
||
updated_at: document.sourceUpdatedAt || document.sessionUpdatedAt,
|
||
},
|
||
messages: document.messages,
|
||
}).catch(() => {});
|
||
}
|
||
} catch (err) {
|
||
logger?.warn?.(`[episodic-memory] snapshot fallback degraded: ${err instanceof Error ? err.message : err}`);
|
||
if (candidates.size === 0) {
|
||
return {
|
||
memories: [],
|
||
skipped: false,
|
||
degraded: true,
|
||
reason: indexFailed ? 'index_and_snapshot_query_failed' : 'snapshot_query_failed',
|
||
source: 'episodic',
|
||
};
|
||
}
|
||
}
|
||
}
|
||
|
||
const ranked = [...candidates.values()]
|
||
.map((document) => ({ document, score: documentScore(document, terms) }))
|
||
.filter(({ score }) => terms.length === 0 || score > 0)
|
||
.sort((a, b) => b.score - a.score || b.document.sessionUpdatedAt - a.document.sessionUpdatedAt)
|
||
.slice(0, resolvedLimit);
|
||
const memories = [];
|
||
let remainingChars = policy.charBudget;
|
||
for (const { document } of ranked) {
|
||
const memory = toMemory(document, terms);
|
||
if (!memory || remainingChars < 80) break;
|
||
memory.text = clipText(memory.text, remainingChars);
|
||
remainingChars -= memory.text.length;
|
||
memories.push(memory);
|
||
}
|
||
return {
|
||
memories,
|
||
skipped: false,
|
||
degraded: indexFailed,
|
||
reason: memories.length ? null : 'no_historical_match',
|
||
source: memories.some((item) => item.source === 'session-snapshot-fallback')
|
||
? 'episodic-snapshot-fallback'
|
||
: 'episodic-index',
|
||
queryTopic: extractEpisodicRecallTopic(query) || null,
|
||
mode: policy.mode,
|
||
};
|
||
}
|
||
|
||
async function resolve(input = {}) {
|
||
const policy = await resolvePolicy(input?.userId);
|
||
metrics.resolveStarted += 1;
|
||
let result;
|
||
try {
|
||
result = await raceWithTimeout(resolveCore({ ...input, policy }), policy.timeoutMs);
|
||
} catch (err) {
|
||
logger?.warn?.(`[episodic-memory] resolve failed open: ${err instanceof Error ? err.message : err}`);
|
||
result = {
|
||
memories: [],
|
||
skipped: false,
|
||
degraded: true,
|
||
reason: err?.code === 'EPISODIC_MEMORY_TIMEOUT' ? 'timeout' : 'resolve_failed',
|
||
source: 'episodic',
|
||
};
|
||
}
|
||
if (result.skipped) metrics.skipped += 1;
|
||
else metrics.resolved += 1;
|
||
if (result.degraded) metrics.degraded += 1;
|
||
metrics.matched += Array.isArray(result.memories) ? result.memories.length : 0;
|
||
metrics.lastReason = result.reason ?? null;
|
||
return result;
|
||
}
|
||
|
||
async function getStatus() {
|
||
const policy = await resolvePolicy(null);
|
||
return {
|
||
configuredEnabled: policy.configuredEnabled,
|
||
retrieverEnabled: policy.retrieverEnabled,
|
||
episodicEnabled: policy.episodicEnabled,
|
||
mode: policy.mode,
|
||
activeForAll: policy.configuredEnabled && policy.mode === 'active',
|
||
canaryUserCount: policy.canaryUserIds.length,
|
||
limit: policy.limit,
|
||
charBudget: policy.charBudget,
|
||
timeoutMs: policy.timeoutMs,
|
||
metrics: { ...metrics },
|
||
};
|
||
}
|
||
|
||
return {
|
||
ensureSchema,
|
||
resolvePolicy,
|
||
upsertSnapshot,
|
||
remove,
|
||
resolve,
|
||
getStatus,
|
||
};
|
||
}
|