67 lines
1.9 KiB
JavaScript
67 lines
1.9 KiB
JavaScript
|
export const RUN_STREAM_STATUS_EVENTS = new Set([
|
||
|
|
'queued',
|
||
|
|
'running',
|
||
|
|
'retryable',
|
||
|
|
'succeeded',
|
||
|
|
'failed',
|
||
|
|
]);
|
||
|
|
|
||
|
|
export const RUN_STREAM_TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
|
||
|
|
|
||
|
|
function envFlag(value, fallback = false) {
|
||
|
|
const raw = String(value ?? '').trim().toLowerCase();
|
||
|
|
if (!raw) return fallback;
|
||
|
|
return ['1', 'true', 'yes', 'on'].includes(raw);
|
||
|
|
}
|
||
|
|
|
||
|
|
export function isRunStreamReplayEnabled(env = process.env) {
|
||
|
|
return envFlag(env.MEMIND_RUN_STREAM_REPLAY, false);
|
||
|
|
}
|
||
|
|
|
||
|
|
export function parseRunStreamLastEventId(value) {
|
||
|
|
const id = String(value ?? '').trim();
|
||
|
|
return id || null;
|
||
|
|
}
|
||
|
|
|
||
|
|
export function extractRunFromStreamEvent(event) {
|
||
|
|
if (!event) return null;
|
||
|
|
if (event.eventType === 'run_snapshot' && event.data?.run) {
|
||
|
|
return event.data.run;
|
||
|
|
}
|
||
|
|
return null;
|
||
|
|
}
|
||
|
|
|
||
|
|
export function shouldEmitRunUpdateForStreamEvent(eventType) {
|
||
|
|
return RUN_STREAM_STATUS_EVENTS.has(String(eventType ?? '').trim())
|
||
|
|
|| String(eventType ?? '').trim() === 'run_snapshot';
|
||
|
|
}
|
||
|
|
|
||
|
|
export function formatRunStreamSseChunk({ id = null, event, data }) {
|
||
|
|
const lines = [];
|
||
|
|
if (id) lines.push(`id: ${id}`);
|
||
|
|
lines.push(`event: ${event}`);
|
||
|
|
lines.push(`data: ${JSON.stringify(data)}`);
|
||
|
|
return `${lines.join('\n')}\n\n`;
|
||
|
|
}
|
||
|
|
|
||
|
|
export function projectRunStreamEvents(events, { fallbackRun = null } = {}) {
|
||
|
|
let latestRun = fallbackRun;
|
||
|
|
const emittedRuns = [];
|
||
|
|
for (const event of Array.isArray(events) ? events : []) {
|
||
|
|
const snapshotRun = extractRunFromStreamEvent(event);
|
||
|
|
if (snapshotRun) {
|
||
|
|
latestRun = snapshotRun;
|
||
|
|
emittedRuns.push({ eventId: event.id, run: snapshotRun });
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
if (shouldEmitRunUpdateForStreamEvent(event.eventType) && latestRun) {
|
||
|
|
emittedRuns.push({ eventId: event.id, run: latestRun });
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return { latestRun, emittedRuns };
|
||
|
|
}
|
||
|
|
|
||
|
|
export function isTerminalRunStatus(status) {
|
||
|
|
return RUN_STREAM_TERMINAL_STATUSES.has(String(status ?? '').trim());
|
||
|
|
}
|