Jyotisha/frontend/src/lib/stream-smalltalk-response.ts

import { consultationAgentPublicEventSchema, type ConsultationAgentPublicEvent } from "./consultation-agent-events.ts";

/** Same NDJSON public envelope as consultation; no fake workflow or skill receipt. */
export function streamSmalltalkResponse(input: {
  requestId: string;
  reply: string;
  complete: () => Promise<void>;
  onError: () => Promise<void>;
}) {
  const encoder = new TextEncoder();
  let disconnected = false;
  const body = new ReadableStream<Uint8Array>({
    async start(controller) {
      const send = (event: ConsultationAgentPublicEvent) => {
        if (!disconnected) controller.enqueue(encoder.encode(`${JSON.stringify(consultationAgentPublicEventSchema.parse(event))}\n`));
      };
      try {
        // Persist/refund before emitting success or text; a rejected/failed RPC is not a reply.
        await input.complete();
        send({ type: "answer.delta", text: input.reply });
        send({ type: "run.completed", responseKind: "smalltalk" });
      } catch {
        await input.onError().catch(() => {});
        send({ type: "run.failed", code: "calculation_failed", message: "这次回复没能保存，请重试。" });
      } finally {
        if (!disconnected) controller.close();
      }
    },
    cancel() { disconnected = true; }, // Server settlement continues, as on the main stream.
  });
  return new Response(body, { headers: {
    "content-type": "application/x-ndjson; charset=utf-8",
    "cache-control": "no-cache, no-transform",
    "x-accel-buffering": "no",
    "x-ayanam-mode": "mastra-agentic",
    "x-ayanam-request-id": input.requestId,
    "x-jyotish-response-kind": "smalltalk",
  } });
}