Hankan/app/src/domain/tutor-client.ts

/* The real tutor, over SSE.

   Implements the same Sample contract as the stub, so swapping one for the
   other touches nothing in the UI.

   Not EventSource: it can only GET, and a turn is a POST carrying the
   system prompt and the transcript. fetch + a reader gives us the body we
   need and an AbortController for free — and the abort is load-bearing,
   because it is what stops a cancelled turn from being billed. */

import {
  SampleError,
  type Sample,
  type SampleOptions,
  type SampleRequest,
  type SampleResult,
} from "./stub-tutor.js";

export interface TutorEndpoint {
  baseUrl: string;
  token: string;
}

interface WireEvent {
  type: "delta" | "done" | "error";
  text?: string;
  code?: string;
  message?: string;
}

/**
 * Split an SSE byte stream into events.
 *
 * Events are separated by a blank line and may span reads, so a partial
 * event has to be carried over rather than parsed early. Lines beginning
 * with ":" are heartbeat comments and carry no data.
 */
async function* readEvents(body: ReadableStream<Uint8Array>): AsyncGenerator<WireEvent> {
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  try {
    for (;;) {
      const { done, value } = await reader.read();
      if (done) break;
      buffer += decoder.decode(value, { stream: true });

      let split: number;
      while ((split = buffer.indexOf("\n\n")) !== -1) {
        const frame = buffer.slice(0, split);
        buffer = buffer.slice(split + 2);

        const data = frame
          .split("\n")
          .filter((l) => l.startsWith("data:"))
          .map((l) => l.slice(5).trim())
          .join("");
        if (!data) continue; // heartbeat, or an event with an empty payload

        try {
          yield JSON.parse(data) as WireEvent;
        } catch {
          // A frame we cannot parse is not worth killing the turn over.
        }
      }
    }
  } finally {
    // Release the stream whatever happened. No `return` here: a return
    // inside `finally` would swallow an in-flight exception.
    reader.cancel().catch(() => undefined);
  }
}

/** A Sample backed by the Pi. */
export function makeRemoteTutor(endpoint: TutorEndpoint): Sample {
  return async (req: SampleRequest, opts: SampleOptions = {}): Promise<SampleResult> => {
    const { signal, onText } = opts;
    const history = req.messages.slice(0, -1);
    const last = req.messages[req.messages.length - 1];

    let res: Response;
    try {
      res = await fetch(`${endpoint.baseUrl.replace(/\/$/, "")}/api/tutor`, {
        method: "POST",
        signal,
        headers: {
          "content-type": "application/json",
          authorization: `Bearer ${endpoint.token}`,
        },
        body: JSON.stringify({
          system: req.system,
          history,
          message: last?.content ?? "",
        }),
      });
    } catch (err) {
      if (signal?.aborted) throw new SampleError("cancelled");
      throw new SampleError("offline", err instanceof Error ? err.message : "Could not reach 선생님.");
    }

    if (res.status === 401) throw new SampleError("unauthorized", "The tutor token was rejected.");
    if (!res.ok) throw new SampleError("upstream", `Tutor endpoint returned HTTP ${res.status}.`);
    if (!res.body) throw new SampleError("upstream", "Tutor endpoint returned no body.");

    // The UI wants cumulative text; the wire carries deltas.
    let text = "";

    for await (const event of readEvents(res.body)) {
      if (event.type === "delta" && event.text) {
        text += event.text;
        onText?.({ text });
      } else if (event.type === "error") {
        // Keep whatever streamed before the failure — a partial lesson is
        // worth more than an empty bubble.
        throw new SampleError(event.code ?? "upstream", event.message, text);
      } else if (event.type === "done") {
        break;
      }
    }

    if (signal?.aborted) throw new SampleError("cancelled", "stopped", text);
    return { text };
  };
}