memind/message-stream.mjs

/**
 * Merge streaming assistant chunks from SSE Message events.
 * Upstream may send deltas, cumulative snapshots, or multi-part content arrays.
 */
export function mergeStreamingChunk(previous, incoming) {
  const prev = String(previous ?? '');
  const next = String(incoming ?? '');
  if (!next) return prev;
  if (!prev) return next;
  if (next === prev) return prev;
  if (next.startsWith(prev)) return next;
  if (prev.startsWith(next)) return prev;
  if (prev.endsWith(next)) return prev;

  const maxOverlap = Math.min(prev.length, next.length);
  for (let size = maxOverlap; size > 0; size -= 1) {
    if (prev.endsWith(next.slice(0, size))) {
      return prev + next.slice(size);
    }
  }

  return prev + next;
}

function findLastContentIndex(content, type) {
  for (let index = content.length - 1; index >= 0; index -= 1) {
    if (content[index]?.type === type) return index;
  }
  return -1;
}

function mergeTextLikeBlock(block, incomingValue, field) {
  block[field] = mergeStreamingChunk(block[field], incomingValue);
}

export function mergeMessageContent(existingContent, incomingContent) {
  const merged = [...existingContent];

  for (const item of incomingContent) {
    if (item.type === 'text') {
      const index = findLastContentIndex(merged, 'text');
      if (index >= 0) {
        mergeTextLikeBlock(merged[index], item.text, 'text');
        continue;
      }
      merged.push({ ...item });
      continue;
    }

    if (item.type === 'thinking') {
      const index = findLastContentIndex(merged, 'thinking');
      if (index >= 0) {
        mergeTextLikeBlock(merged[index], item.thinking, 'thinking');
        continue;
      }
      merged.push({ ...item });
      continue;
    }

    const tail = merged[merged.length - 1];
    if (tail && JSON.stringify(tail) === JSON.stringify(item)) continue;
    merged.push({ ...item });
  }

  return merged;
}