ChatWindow.tsx

const sendPayload = server$(async function*(input: unknown, config: unknown) {
  const self = this as unknown as RequestEventBase;
  for await (const s of fetchEventSource(`${self.env.get('LANGSERVE_BASE_URL') ?? ''}/stream_log`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ input, config }),
    signal: self.signal
  })) {
    yield s;
  }
});

// ...

const initiateStream = $(async (input: unknown, config: unknown) => {
    const response = await sendPayload(input, config);
    for await (const r of response) {
      switch (r.event) {
        case EventSourceEvent.OPEN:
          messages.value = [
            ...messages.value,
            { type: "ai", content: "" },
          ];
          break;
        case EventSourceEvent.MESSAGE:
          const finalOutput = r.aggregatedState?.final_output;
          // By default the lookback index is -1, hwoever if there is a retry in the stream library then 
          // the OPEN event will create an empty ai content above which will need to be replaced, therefore
          // below we need to look at the previous message and if empty, make sure we replace accordingly
          const lookbackIdx = [null, ""].includes(messages.value[messages.value.length - 2]?.content) ? -2 : -1;
          if (typeof finalOutput === "string") {
            messages.value = [
              ...messages.value.slice(0, lookbackIdx),
              { type: "ai", content: finalOutput, runId: r.aggregatedState?.id }
            ];
          } else if (isAIMessage(finalOutput)) {
            messages.value = [
              ...messages.value.slice(0, lookbackIdx),
              { type: "ai", content: finalOutput.content, runId: r.aggregatedState?.id }
            ];
          }
          break;
        case EventSourceEvent.CLOSE:
          isLoading.value = false;
          break;
        case EventSourceEvent.ERROR:
          isLoading.value = false;
          toast.error(`${r.error?.message}\nCheck your backend logs for errors.`);
          console.error(messages.value[messages.value.length - 2]?.content, r.error);
          currentInputValue.value = messages.value[messages.value.length - 2]?.content;
          messages.value = [
            ...messages.value.slice(0, -2),
          ];
      }
    }
  });