john
77 lines
2.6 KiB
JavaScript
77 lines
2.6 KiB
JavaScript
export const SESSION_STREAM_TERMINAL_TYPES = new Set(['Finish', 'Error']);
|
|
|
|
function envFlag(value, fallback = false) {
|
|
const raw = String(value ?? '').trim().toLowerCase();
|
|
if (!raw) return fallback;
|
|
return ['1', 'true', 'yes', 'on'].includes(raw);
|
|
}
|
|
|
|
export function isSessionStreamReplayEnabled(env = process.env) {
|
|
return envFlag(env.MEMIND_SESSION_STREAM_REPLAY, false);
|
|
}
|
|
|
|
export function parseSessionStreamLastEventId(value) {
|
|
const id = String(value ?? '').trim();
|
|
return id || null;
|
|
}
|
|
|
|
/**
|
|
* Portal replay ids are local database cursors and are not necessarily valid
|
|
* Goose SSE ids. Resume Goose from the newest persisted upstream cursor we
|
|
* actually know about. If none exists, return null so the caller omits
|
|
* Last-Event-ID and lets Goose replay the authoritative stream from the start.
|
|
*
|
|
* @param {Array<{ upstreamEventId?: string | null }>} events
|
|
* @param {{ upstreamEventId?: string | null } | null} cursorEvent
|
|
* @returns {string | null}
|
|
*/
|
|
export function resolveUpstreamResumeEventId(events, cursorEvent = null) {
|
|
const candidates = Array.isArray(events) ? [...events].reverse() : [];
|
|
if (cursorEvent) candidates.push(cursorEvent);
|
|
for (const event of candidates) {
|
|
const id = String(event?.upstreamEventId ?? '').trim();
|
|
if (id) return id;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
export function isTerminalSessionEvent(event) {
|
|
const type = String(event?.type ?? '').trim();
|
|
return SESSION_STREAM_TERMINAL_TYPES.has(type);
|
|
}
|
|
|
|
export function shouldPersistSessionStreamEvent(event) {
|
|
const type = String(event?.type ?? '').trim();
|
|
if (!type) return false;
|
|
return type !== 'Ping';
|
|
}
|
|
|
|
export function parseSessionSseBlock(block) {
|
|
const lines = String(block ?? '').split('\n');
|
|
let id = null;
|
|
let eventName = null;
|
|
let data = null;
|
|
for (const line of lines) {
|
|
if (line.startsWith('id:')) id = line.slice(3).trim() || null;
|
|
if (line.startsWith('event:')) eventName = line.slice(6).trim() || null;
|
|
if (line.startsWith('data:')) data = line.slice(5).trim();
|
|
}
|
|
return { id, eventName, data };
|
|
}
|
|
|
|
export function formatSessionStreamSseChunk({ id = null, event = null, data }) {
|
|
const lines = [];
|
|
if (id) lines.push(`id: ${id}`);
|
|
if (event) lines.push(`event: ${event}`);
|
|
lines.push(`data: ${JSON.stringify(data)}`);
|
|
return `${lines.join('\n')}\n\n`;
|
|
}
|
|
|
|
export function shouldSkipUpstreamAfterSessionReplay(events, { cursorMiss = false } = {}) {
|
|
if (!Array.isArray(events) || events.length === 0) {
|
|
return cursorMiss ? false : false;
|
|
}
|
|
const last = events[events.length - 1];
|
|
return isTerminalSessionEvent(last?.payload ?? last);
|
|
}
|