#Step 09 — 스트리밍

import { ChatAnthropic } from "@langchain/anthropic";

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

// (A) invoke — 다 만들 때까지 아무것도 없다
const t0 = Date.now();
const res = await model.invoke("한국의 사계절을 각 두 문장씩 설명해 주세요.");
console.log(`invoke │ 첫 출력까지 ${Date.now() - t0}ms`);

// (B) stream — 첫 청크가 오는 순간을 찍는다
const t1 = Date.now();
let ttft = -1;
const stream = await model.stream("한국의 사계절을 각 두 문장씩 설명해 주세요.");
for await (const chunk of stream) {
  if (ttft === -1) ttft = Date.now() - t1;   // ← 첫 청크에서만
  process.stdout.write(chunk.text);
}
console.log(`\nstream │ 첫 출력까지 ${ttft}ms, 전체 ${Date.now() - t1}ms`);

const stream = await model.stream("스트리밍을 한 문장으로 설명해 주세요.");

for await (const chunk of stream) {
  console.log(chunk.constructor.name, JSON.stringify(chunk.text));
}

import { AIMessageChunk } from "@langchain/core/messages";

let full: AIMessageChunk | null = null;

const stream = await model.stream("스트리밍의 장점을 한 문장으로 말해 주세요.");
for await (const chunk of stream) {
  full = full === null ? chunk : full.concat(chunk);
}

console.log(full?.text);
console.log(JSON.stringify(full?.contentBlocks, null, 2));
console.log(JSON.stringify(full?.usage_metadata));

import { ChatOpenAI } from "@langchain/openai";

// (A) 모델 자체를 비스트리밍으로
const model = new ChatOpenAI({ model: "gpt-5.5", streaming: false });

// (B) 실행은 스트리밍으로 하되 스트림에서만 제외 — "nostream" 태그
const internalModel = new ChatAnthropic({
  model: "claude-haiku-4-5-20251001",
}).withConfig({ tags: ["nostream"] });

import { createAgent, tool } from "langchain";
import * as z from "zod";

const getWeather = tool(
  async ({ city }) => `${city}의 날씨는 맑음, 기온 21도입니다.`,
  {
    name: "get_weather",
    description: "도시 이름을 받아 현재 날씨를 반환합니다.",
    schema: z.object({ city: z.string() }),
  },
);

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  tools: [getWeather],
  systemPrompt: "당신은 간결하게 답하는 날씨 비서입니다.",
});

const question = { messages: [{ role: "user", content: "서울 날씨 알려줘" }] };

for await (const chunk of await agent.stream(question, { streamMode: "updates" })) {
  for (const [node, update] of Object.entries(chunk)) {
    console.log(`노드 ${node}:`);
    console.log(JSON.stringify(update, null, 2));
  }
}

for await (const chunk of await agent.stream(question, { streamMode: "values" })) {
  console.log(`messages 길이 = ${chunk.messages.length}`);
}

for await (const [token, metadata] of await agent.stream(question, {
  streamMode: "messages",
})) {
  console.log(`[${metadata.langgraph_node}] ${JSON.stringify(token.text)}`);
}

// 필터링 예시
if (metadata.langgraph_node === "model") { /* 모델 토큰만 */ }
if (metadata.tags?.includes("joke")) { /* 특정 태그가 붙은 호출만 */ }

for await (const chunk of await agent.stream(question, { streamMode: "debug" })) {
  console.log(JSON.stringify(chunk).slice(0, 200));
}

for await (const [mode, chunk] of await agent.stream(question, {
  streamMode: ["updates", "messages", "custom"],
})) {
  switch (mode) {
    case "messages": {
      const [token] = chunk;              // chunk 가 다시 [token, metadata] 튜플
      process.stdout.write(token.text);
      break;
    }
    case "updates":
      console.log(`\n[스텝 완료: ${Object.keys(chunk).join(", ")}]`);
      break;
    case "custom":
      console.log(`\n[진행] ${JSON.stringify(chunk)}`);
      break;
  }
}

const stream = await agent.streamEvents(question, { version: "v3" });

const stream = await agent.streamEvents(question, { version: "v3" });
for await (const event of stream) {
  console.log(event.method, JSON.stringify(event.params?.namespace));
}

const stream = await agent.streamEvents(question, { version: "v3" });

for await (const message of stream.messages) {
  for await (const delta of message.text) {
    process.stdout.write(delta);
  }
  console.log("\n--- 메시지 1건 종료 ---");
}

import { ChatAnthropic } from "@langchain/anthropic";

const agent = createAgent({
  model: new ChatAnthropic({
    model: "claude-sonnet-4-6",
    thinking: { type: "enabled", budget_tokens: 5000 },
  }),
  tools: [getWeather],
});

const stream = await agent.streamEvents(question, { version: "v3" });
for await (const message of stream.messages) {
  for await (const token of message.reasoning) {
    process.stdout.write(`[thinking] ${token}`);
  }
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const chunk of message.toolCalls) {
        console.log("인자 조각", chunk);        // 조각 단위 (부분 JSON)
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);      // 완성된 호출 단위
    }
  })(),
]);

for await (const subagent of stream.subgraphs) {
  if (subagent.name !== "weather_agent") continue;
  // 이 서브에이전트의 스트림만 처리
}

const stream = await agent.streamEvents(question, { version: "v3" });

for await (const message of stream.messages) {
  for await (const delta of message.text) process.stdout.write(delta);
}

const finalState = await stream.output;   // ← 스트림을 다 소비한 뒤에
console.log(finalState.messages.length);

const modelWithTools = model.bindTools([getWeather]);
const stream = await modelWithTools.stream("서울과 부산 날씨를 각각 알려줘");

for await (const chunk of stream) {
  for (const tc of chunk.tool_call_chunks ?? []) {
    console.log(`index=${tc.index} name=${tc.name ?? "(없음)"} args=${JSON.stringify(tc.args)}`);
  }
}

let full: AIMessageChunk | null = null;
for await (const chunk of stream) {
  full = full === null ? chunk : full.concat(chunk);
}

console.log(JSON.stringify(full?.tool_calls, null, 2));

const acc = new Map<number, { name: string; raw: string }>();

for await (const chunk of stream) {
  for (const tc of chunk.tool_call_chunks ?? []) {
    const index = tc.index ?? 0;
    const prev = acc.get(index) ?? { name: "", raw: "" };

    acc.set(index, {
      name: prev.name !== "" ? prev.name : (tc.name ?? ""),   // ← undefined 로 덮어쓰기 금지
      raw: prev.raw + (tc.args ?? ""),                        // ← index 별로 이어붙이기
    });

    const cur = acc.get(index)!;
    try {
      const args = JSON.parse(cur.raw);
      console.log(`index=${index} 완성!`, cur.name, args);
    } catch {
      // 아직 미완성 — 정상입니다. 다음 조각을 기다립니다.
    }
  }
}

import { type LangGraphRunnableConfig } from "@langchain/langgraph";

const queryDatabase = tool(
  async (input: { query: string }, config: LangGraphRunnableConfig) => {
    config.writer?.({ type: "progress", data: "0/100 건 조회" });
    // ... 실제 조회 ...
    config.writer?.({ type: "progress", data: "100/100 건 조회 완료" });
    return "조회 결과입니다.";
  },
  {
    name: "query_database",
    description: "데이터베이스를 조회합니다.",
    schema: z.object({ query: z.string() }),
  },
);

for await (const chunk of await agent.stream(question, { streamMode: "custom" })) {
  console.log(chunk);
}

import { createServer } from "node:http";

const server = createServer(async (req, res) => {
  if (req.url !== "/chat") { res.writeHead(404).end(); return; }

  // 1) SSE 헤더 3종 세트
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no",        // nginx 뒤에 있다면 필수
  });

  // 2) 클라이언트가 끊으면 에이전트도 끊는다
  const controller = new AbortController();
  req.on("close", () => controller.abort());

  const send = (event: string, data: unknown): void => {
    res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);   // ← 빈 줄로 프레임 종료
  };

  try {
    for await (const [mode, chunk] of await agent.stream(question, {
      streamMode: ["messages", "custom"],
      signal: controller.signal,
    })) {
      if (mode === "messages") {
        const [token, metadata] = chunk;
        if (metadata.langgraph_node === "model" && token.text !== "") {
          send("token", { text: token.text });
        }
      } else if (mode === "custom") {
        send("progress", chunk);
      }
    }
    send("done", { ok: true });
  } catch (err) {
    // 3) 이미 200 을 보냈다 → 500 으로 못 바꾼다. 에러도 "이벤트"로.
    if ((err as Error).name !== "AbortError") {
      send("error", { message: (err as Error).message });
    }
  } finally {
    res.end();
  }
});

server.listen(3000);

const es = new EventSource("/chat");
es.addEventListener("token", (e) => { output.textContent += JSON.parse(e.data).text; });
es.addEventListener("progress", (e) => { status.textContent = JSON.parse(e.data).data; });
es.addEventListener("error", (e) => { status.textContent = "오류가 발생했습니다"; });
es.addEventListener("done", () => es.close());

import { MemorySaver } from "@langchain/langgraph";

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: crypto.randomUUID() } };
const controller = new AbortController();
setTimeout(() => controller.abort(), 1500);   // 사용자가 "그만"을 누른 상황

let received = "";
let full: AIMessageChunk | null = null;

try {
  for await (const [mode, chunk] of await agent.stream(
    { messages: [{ role: "user", content: "서울 날씨를 알려주고 옷차림을 길게 조언해줘" }] },
    { ...config, streamMode: ["messages", "custom"], signal: controller.signal },
  )) {
    if (mode === "custom") {
      console.log(`\n[진행] ${JSON.stringify(chunk)}`);
      continue;
    }
    const [token, metadata] = chunk;
    if (metadata.langgraph_node !== "model") continue;     // 도구 결과 노출 방지

    process.stdout.write(token.text);                       // 표시는 .text
    received += token.text;
    full = full === null ? token : full.concat(token);      // 누적은 concat
  }
  console.log(`\n완료. 토큰: ${JSON.stringify(full?.usage_metadata)}`);
} catch (err) {
  if ((err as Error).name === "AbortError") {
    console.log(`\n[중단됨] 이미 보낸 ${received.length}자는 되돌릴 수 없습니다.`);
  } else {
    throw err;
  }
}

/**
 * Step 09 — 스트리밍
 * 실행: npx tsx docs/reference/langchain/step-09-streaming/practice.ts
 *
 * 본문 9-1 ~ 9-9 의 예제를 순서대로 담았습니다.
 * 블록 주석의 [9-x] 번호가 본문 소제목과 1:1 로 대응합니다.
 *
 * 이 파일은 실제 모델을 호출합니다. project/.env 에 ANTHROPIC_API_KEY 가 필요합니다.
 * (OpenAI 를 쓰려면 [9-2] 의 주석을 참고하세요.)
 */
import "dotenv/config";

import { createAgent, tool } from "langchain";
import { ChatAnthropic } from "@langchain/anthropic";
import { AIMessageChunk } from "@langchain/core/messages";
import { MemorySaver, type LangGraphRunnableConfig } from "@langchain/langgraph";
import * as z from "zod";

import { printSection } from "../project/src/lib/print.js";

/* ===== 공용 준비물 ===== */

// 이 스텝 전체에서 재사용할 도구.
// config 를 두 번째 인자로 받는 형태는 [9-7] 에서 writer 를 쓰기 위한 것입니다.
const getWeather = tool(
  async (input: { city: string }, config: LangGraphRunnableConfig) => {
    // writer 는 "custom" 스트림으로 임의의 데이터를 흘려보냅니다.
    // 구독자가 없어도(= streamMode 에 custom 이 없어도) 그냥 무시되므로 안전합니다.
    config.writer?.({ type: "progress", data: `${input.city} 관측소 조회 중...` });
    await new Promise((r) => setTimeout(r, 300)); // 느린 API 흉내
    config.writer?.({ type: "progress", data: `${input.city} 관측 데이터 수신 완료` });
    return `${input.city}의 날씨는 맑음, 기온 21도입니다.`;
  },
  {
    name: "get_weather",
    description: "도시 이름을 받아 현재 날씨를 반환합니다.",
    schema: z.object({
      city: z.string().describe("날씨를 조회할 도시 이름"),
    }),
  },
);

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

/* ===== [9-1] 왜 스트리밍인가 — TTFT 측정 ===== */

async function section_9_1(): Promise<void> {
  printSection("[9-1] invoke vs stream — 첫 글자까지 걸리는 시간(TTFT)");

  // (A) invoke: 모델이 문장을 다 만들 때까지 아무것도 못 본다.
  const t0 = Date.now();
  const res = await model.invoke("한국의 사계절을 각 두 문장씩 설명해 주세요.");
  const invokeTotal = Date.now() - t0;
  console.log(`invoke  │ 첫 출력까지 ${invokeTotal}ms  (= 전체 완료 시간과 같음)`);
  console.log(`invoke  │ 글자 수 ${res.text.length}`);

  // (B) stream: 첫 청크가 오는 순간 TTFT 가 찍힌다.
  const t1 = Date.now();
  let ttft = -1;
  let chars = 0;
  const stream = await model.stream("한국의 사계절을 각 두 문장씩 설명해 주세요.");
  for await (const chunk of stream) {
    if (ttft === -1) ttft = Date.now() - t1;
    chars += chunk.text.length;
  }
  const streamTotal = Date.now() - t1;
  console.log(`stream  │ 첫 출력까지 ${ttft}ms  (TTFT)`);
  console.log(`stream  │ 전체 완료 ${streamTotal}ms, 글자 수 ${chars}`);
  console.log(`\n→ 전체 시간은 비슷하지만, 사용자가 "기다린다"고 느끼는 시간은 ${ttft}ms 뿐입니다.`);
}

/* ===== [9-2] 모델 레벨 스트리밍 — AIMessageChunk 와 concat ===== */

async function section_9_2(): Promise<void> {
  printSection("[9-2] model.stream() — 청크를 눈으로 보기");

  // OpenAI 를 쓰려면 위쪽 model 정의를 이렇게 바꾸면 됩니다:
  //   import { ChatOpenAI } from "@langchain/openai";
  //   const model = new ChatOpenAI({ model: "gpt-5.5" });
  // stream() 의 사용법은 제공자와 무관하게 동일합니다.

  const stream = await model.stream("스트리밍을 한 문장으로 설명해 주세요.");

  let count = 0;
  for await (const chunk of stream) {
    count += 1;
    // chunk 는 AIMessage 가 아니라 AIMessageChunk 입니다. 이름이 다른 이유가 있습니다.
    if (count <= 3) {
      console.log(`청크 ${count} │ 클래스=${chunk.constructor.name} text=${JSON.stringify(chunk.text)}`);
    }
    process.stdout.write(chunk.text);
  }
  console.log(`\n\n총 ${count}개 청크`);

  printSection("[9-2] concat 으로 하나의 메시지로 합치기");

  const stream2 = await model.stream("스트리밍의 장점을 한 문장으로 말해 주세요.");

  // 누산기의 타입은 AIMessageChunk | null 입니다.
  // 첫 청크는 그대로 쓰고, 이후로는 .concat() 으로 병합합니다.
  let full: AIMessageChunk | null = null;
  for await (const chunk of stream2) {
    full = full === null ? chunk : full.concat(chunk);
  }

  if (full !== null) {
    console.log("합쳐진 text        :", full.text);
    console.log("합쳐진 contentBlocks:", JSON.stringify(full.contentBlocks, null, 2));
    // usage_metadata 는 보통 마지막 청크에만 실려 오는데,
    // concat 이 그것까지 병합해 줍니다. 문자열 이어붙이기로는 절대 얻을 수 없는 값입니다.
    console.log("usage_metadata     :", JSON.stringify(full.usage_metadata));
    console.log("response_metadata  :", JSON.stringify(full.response_metadata));
  }

  printSection("[9-2] 함정 — .content 를 문자열로 이어붙이면?");

  const stream3 = await model.stream("아무 인사말이나 짧게 해 주세요.");
  let naive = "";
  for await (const chunk of stream3) {
    // ⚠️ content 는 string 일 수도 있고 콘텐츠 블록 배열일 수도 있습니다.
    //    배열이면 "+= " 는 "[object Object]" 를 만들거나 조용히 어긋납니다.
    naive += typeof chunk.content === "string" ? chunk.content : JSON.stringify(chunk.content);
  }
  console.log("순진하게 이어붙인 결과:", naive.slice(0, 200));
  console.log("→ 텍스트 전용 모델에선 우연히 잘 보일 수 있습니다. 그게 이 함정이 위험한 이유입니다.");
}

/* ===== [9-3] 에이전트 스트리밍 — streamMode 해부 ===== */

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  tools: [getWeather],
  systemPrompt: "당신은 간결하게 답하는 날씨 비서입니다. 날씨 질문은 반드시 도구로 확인하세요.",
});

const question = { messages: [{ role: "user" as const, content: "서울 날씨 알려줘" }] };

async function section_9_3_updates(): Promise<void> {
  printSection('[9-3] streamMode: "updates" — 스텝이 끝날 때마다 "바뀐 것"만');

  for await (const chunk of await agent.stream(question, { streamMode: "updates" })) {
    // chunk 의 모양: { [노드이름]: { messages: [...] } }
    for (const [node, update] of Object.entries(chunk)) {
      console.log(`노드 ${node} 가 업데이트를 내놓음:`);
      console.log(JSON.stringify(update, null, 2).slice(0, 400));
    }
  }
}

async function section_9_3_values(): Promise<void> {
  printSection('[9-3] streamMode: "values" — 스텝마다 "전체 상태"');

  let step = 0;
  for await (const chunk of await agent.stream(question, { streamMode: "values" })) {
    step += 1;
    // values 는 매번 누적된 전체 messages 배열을 줍니다.
    // 그래서 뒤로 갈수록 payload 가 커집니다 — 네트워크로 그대로 흘리면 낭비입니다.
    const messages = (chunk as { messages?: unknown[] }).messages ?? [];
    console.log(`스텝 ${step} │ messages 길이 = ${messages.length}`);
  }
}

async function section_9_3_messages(): Promise<void> {
  printSection('[9-3] streamMode: "messages" — LLM 토큰 (token, metadata) 튜플');

  for await (const [token, metadata] of await agent.stream(question, { streamMode: "messages" })) {
    // token 은 AIMessageChunk (또는 ToolMessage), metadata 는 어느 노드에서 나왔는지 등.
    const node = (metadata as { langgraph_node?: string }).langgraph_node ?? "?";
    const text = (token as AIMessageChunk).text;
    if (text !== "") {
      console.log(`[${node}] ${JSON.stringify(text)}`);
    }
  }
}

async function section_9_3_debug(): Promise<void> {
  printSection('[9-3] streamMode: "debug" — 실행 중 모든 정보 (양이 많습니다)');

  let n = 0;
  for await (const chunk of await agent.stream(question, { streamMode: "debug" })) {
    n += 1;
    if (n <= 3) console.log(JSON.stringify(chunk).slice(0, 300));
  }
  console.log(`... 총 ${n}개 debug 이벤트 (프로덕션에서 사용자에게 흘리면 안 됩니다)`);
}

/* ===== [9-4] 여러 모드 동시 구독 ===== */

async function section_9_4(): Promise<void> {
  printSection('[9-4] streamMode: ["updates", "messages", "custom"] — 튜플이 [mode, chunk] 로 바뀐다');

  for await (const [mode, chunk] of await agent.stream(question, {
    streamMode: ["updates", "messages", "custom"],
  })) {
    // ⚠️ 배열로 주면 각 항목이 [모드이름, 페이로드] 튜플이 됩니다.
    //    단일 모드일 때의 코드를 그대로 두면 chunk 위치가 밀려서 조용히 깨집니다.
    switch (mode) {
      case "messages": {
        const [token] = chunk as [AIMessageChunk, unknown];
        if (token.text !== "") process.stdout.write(token.text);
        break;
      }
      case "updates": {
        console.log(`\n[updates] ${Object.keys(chunk as object).join(", ")} 노드 완료`);
        break;
      }
      case "custom": {
        console.log(`\n[custom] ${JSON.stringify(chunk)}`);
        break;
      }
      default:
        break;
    }
  }
  console.log("");
}

/* ===== [9-5] streamEvents — 세밀한 이벤트와 프로젝션 ===== */

async function section_9_5(): Promise<void> {
  printSection('[9-5] streamEvents(version: "v3") — 원본 프로토콜 이벤트');

  const raw = await agent.streamEvents(question, { version: "v3" });
  let count = 0;
  for await (const event of raw) {
    count += 1;
    if (count <= 5) {
      // 이벤트 봉투(envelope)의 모양: { method, params: { namespace, data } }
      console.log(`method=${event.method} namespace=${JSON.stringify(event.params?.namespace)}`);
    }
  }
  console.log(`... 총 ${count}개 원본 이벤트`);

  printSection("[9-5] 프로젝션 — stream.messages / message.text");

  const stream = await agent.streamEvents(question, { version: "v3" });
  // stream.messages 는 "LLM 호출 1건 = 메시지 스트림 1개" 로 묶어서 줍니다.
  // 그 안의 message.text 는 텍스트 델타만 뽑아 주는 프로젝션입니다.
  for await (const message of stream.messages) {
    for await (const delta of message.text) {
      process.stdout.write(delta);
    }
    console.log("\n--- 메시지 1건 종료 ---");
  }

  printSection("[9-5] 프로젝션 — stream.output (최종 상태)");

  const stream2 = await agent.streamEvents(question, { version: "v3" });
  for await (const message of stream2.messages) {
    for await (const _ of message.text) {
      // output 을 얻으려면 스트림을 끝까지 소비해야 합니다.
    }
  }
  const finalState = await stream2.output;
  console.log("최종 messages 개수:", (finalState as { messages?: unknown[] }).messages?.length);
}

/* ===== [9-6] 도구 호출 스트리밍 — tool_call_chunks 누적 ===== */

async function section_9_6(): Promise<void> {
  printSection("[9-6] tool_call_chunks — 인자가 부분 JSON 으로 쪼개져서 온다");

  const modelWithTools = model.bindTools([getWeather]);
  const stream = await modelWithTools.stream("서울과 부산 날씨를 각각 알려줘");

  let full: AIMessageChunk | null = null;
  for await (const chunk of stream) {
    for (const tc of chunk.tool_call_chunks ?? []) {
      // ⚠️ tc.args 는 완성된 JSON 이 아니라 "조각" 입니다.
      //    '{"ci' / 'ty": "서' / '울"}' 처럼 옵니다. JSON.parse 하면 터집니다.
      console.log(
        `index=${tc.index} name=${tc.name ?? "(없음)"} args조각=${JSON.stringify(tc.args)}`,
      );
    }
    full = full === null ? chunk : full.concat(chunk);
  }

  printSection("[9-6] concat 이 조각을 합쳐 준다 — full.tool_calls");

  if (full !== null) {
    // concat 은 index 를 기준으로 args 문자열을 이어붙이고,
    // 완성된 것만 tool_calls 에 파싱해서 올려 줍니다.
    console.log("tool_call_chunks 개수:", full.tool_call_chunks?.length);
    console.log("완성된 tool_calls   :", JSON.stringify(full.tool_calls, null, 2));
  }

  printSection("[9-6] 하면 안 되는 것 — 조각을 바로 파싱");

  const stream2 = await modelWithTools.stream("대구 날씨 알려줘");
  for await (const chunk of stream2) {
    for (const tc of chunk.tool_call_chunks ?? []) {
      try {
        JSON.parse(tc.args ?? "");
        console.log("파싱 성공(우연):", tc.args);
      } catch (e) {
        console.log(`파싱 실패 ← 정상입니다. args=${JSON.stringify(tc.args)} / ${(e as Error).name}`);
      }
    }
  }
}

/* ===== [9-7] 커스텀 데이터 스트리밍 ===== */

async function section_9_7(): Promise<void> {
  printSection('[9-7] streamMode: "custom" — 도구 안에서 진행상황 내보내기');

  // getWeather 도구는 이미 config.writer?.({ type, data }) 를 두 번 호출합니다.
  for await (const chunk of await agent.stream(question, { streamMode: "custom" })) {
    console.log("custom:", JSON.stringify(chunk));
  }

  printSection("[9-7] writer 는 구독자가 없어도 안전하다");

  // streamMode 에 custom 이 없으면 writer 로 보낸 것은 그냥 버려집니다. 에러가 아닙니다.
  for await (const chunk of await agent.stream(question, { streamMode: "updates" })) {
    console.log("updates 만 구독:", Object.keys(chunk).join(", "));
  }
}

/* ===== [9-8] 서버로 내보내기 — SSE ===== */

async function section_9_8(): Promise<void> {
  printSection("[9-8] SSE 서버 — 브라우저로 토큰 흘려보내기");

  const { createServer } = await import("node:http");

  const server = createServer(async (req, res) => {
    if (req.url !== "/chat") {
      res.writeHead(404).end();
      return;
    }

    // SSE 필수 헤더. 셋 중 하나라도 빠지면 프록시나 브라우저가 버퍼링합니다.
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no", // nginx 버퍼링 방지
    });

    // 클라이언트가 탭을 닫으면 close 가 옵니다. 이때 스트림을 끊지 않으면
    // 에이전트가 백그라운드에서 계속 돌며 토큰을 태웁니다.
    const controller = new AbortController();
    req.on("close", () => controller.abort());

    const send = (event: string, data: unknown): void => {
      res.write(`event: ${event}\n`);
      res.write(`data: ${JSON.stringify(data)}\n\n`);
    };

    try {
      for await (const [mode, chunk] of await agent.stream(question, {
        streamMode: ["messages", "custom"],
        signal: controller.signal,
      })) {
        if (mode === "messages") {
          const [token] = chunk as [AIMessageChunk, unknown];
          if (token.text !== "") send("token", { text: token.text });
        } else if (mode === "custom") {
          send("progress", chunk);
        }
      }
      send("done", { ok: true });
    } catch (err) {
      // ⚠️ 이미 200 을 보냈으므로 500 으로 바꿀 수 없습니다.
      //    에러도 "이벤트"로 흘려보내고, 클라이언트가 처리하게 해야 합니다.
      send("error", { message: (err as Error).message });
    } finally {
      res.end();
    }
  });

  await new Promise<void>((resolve) => server.listen(0, resolve));
  const address = server.address();
  const port = typeof address === "object" && address !== null ? address.port : 0;
  console.log(`SSE 서버 기동: http://127.0.0.1:${port}/chat`);

  // 같은 프로세스에서 fetch 로 직접 구독해 봅니다 (브라우저 EventSource 대신).
  const res = await fetch(`http://127.0.0.1:${port}/chat`);
  const reader = res.body?.getReader();
  const decoder = new TextDecoder();
  if (reader !== undefined) {
    for (;;) {
      const { done, value } = await reader.read();
      if (done) break;
      process.stdout.write(decoder.decode(value));
    }
  }

  server.close();
  console.log("\nSSE 서버 종료");
}

/* ===== [9-9] 종합 — 취소와 에러를 견디는 스트림 루프 ===== */

async function section_9_9(): Promise<void> {
  printSection("[9-9] 종합 — 취소 가능한 스트리밍 소비자");

  const agentWithMemory = createAgent({
    model: "anthropic:claude-sonnet-4-6",
    tools: [getWeather],
    systemPrompt: "당신은 간결하게 답하는 날씨 비서입니다.",
    checkpointer: new MemorySaver(),
  });

  const config = { configurable: { thread_id: crypto.randomUUID() } };
  const controller = new AbortController();

  // 1.5초 뒤 강제 취소 — 사용자가 "그만" 을 누른 상황.
  const timer = setTimeout(() => controller.abort(), 1500);

  let received = "";
  try {
    for await (const [token] of await agentWithMemory.stream(
      { messages: [{ role: "user", content: "서울 날씨를 알려주고, 옷차림을 길게 조언해줘" }] },
      { ...config, streamMode: "messages", signal: controller.signal },
    )) {
      const text = (token as AIMessageChunk).text;
      received += text;
      process.stdout.write(text);
    }
    console.log("\n→ 끝까지 완료");
  } catch (err) {
    // AbortError 는 "정상적인 취소" 입니다. 에러 로그로 올리면 안 됩니다.
    const name = (err as Error).name;
    console.log(`\n→ 중단됨 (${name}). 이미 사용자에게 보낸 ${received.length}자는 되돌릴 수 없습니다.`);
  } finally {
    clearTimeout(timer);
  }

  console.log("\n중요: break 로 for-await 를 빠져나가는 대신 signal 로 끊어야");
  console.log("      백그라운드 실행과 토큰 과금이 함께 멈춥니다.");
}

/* ===== 실행 ===== */

async function main(): Promise<void> {
  await section_9_1();
  await section_9_2();
  await section_9_3_updates();
  await section_9_3_values();
  await section_9_3_messages();
  await section_9_3_debug();
  await section_9_4();
  await section_9_5();
  await section_9_6();
  await section_9_7();
  await section_9_8();
  await section_9_9();
}

main().catch((err: unknown) => {
  console.error(err);
  process.exit(1);
});

/**
 * Step 09 — 스트리밍 · 연습문제
 * 실행: npx tsx docs/reference/langchain/step-09-streaming/exercise.ts
 *
 * 각 [문제 N] 블록의 TODO 를 채우세요.
 * 채우기 전에 실행하면 "미구현" 안내만 출력됩니다 — 정상입니다.
 * 정답과 해설은 solution.ts 에 있습니다.
 */
import "dotenv/config";

import { createAgent, tool } from "langchain";
import { ChatAnthropic } from "@langchain/anthropic";
import { AIMessageChunk } from "@langchain/core/messages";
import { type LangGraphRunnableConfig } from "@langchain/langgraph";
import * as z from "zod";

import { printSection } from "../project/src/lib/print.js";

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

const searchDocs = tool(
  async (input: { query: string }, _config: LangGraphRunnableConfig) => {
    await new Promise((r) => setTimeout(r, 200));
    return `"${input.query}" 검색 결과: 문서 3건을 찾았습니다.`;
  },
  {
    name: "search_docs",
    description: "사내 문서를 검색합니다.",
    schema: z.object({ query: z.string().describe("검색어") }),
  },
);

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  tools: [searchDocs],
  systemPrompt: "당신은 사내 문서 검색 비서입니다. 질문에는 반드시 도구로 근거를 찾아 답하세요.",
});

const ask = { messages: [{ role: "user" as const, content: "휴가 정책을 알려줘" }] };

/* =====================================================================
 * [문제 1] TTFT 측정
 *
 * model.stream() 을 호출해 "첫 청크가 도착한 시각 - 호출 시각"(TTFT)과
 * 전체 완료 시간을 각각 밀리초로 반환하세요.
 * 반환: { ttftMs, totalMs, text }
 *
 * 힌트: for await 안에서 "아직 TTFT 를 안 찍었으면 지금 찍는다" 로 처리합니다.
 * ===================================================================== */
async function problem1(
  prompt: string,
): Promise<{ ttftMs: number; totalMs: number; text: string }> {
  // TODO: 구현
  void prompt;
  return { ttftMs: -1, totalMs: -1, text: "" };
}

/* =====================================================================
 * [문제 2] concat 으로 청크 합치기
 *
 * model.stream() 의 청크를 concat 으로 합쳐 완성된 AIMessageChunk 하나를
 * 반환하세요. 문자열 이어붙이기(+=)를 쓰면 안 됩니다.
 *
 * 이 함수의 반환값에서 usage_metadata 가 살아 있어야 정답입니다.
 * (문자열로 이어붙이면 usage_metadata 는 절대 얻을 수 없습니다.)
 * ===================================================================== */
async function problem2(prompt: string): Promise<AIMessageChunk | null> {
  // TODO: 구현
  void prompt;
  return null;
}

/* =====================================================================
 * [문제 3] updates 모드로 "도구가 언제 불렸는지" 로그
 *
 * agent.stream(ask, { streamMode: "updates" }) 를 구독해서,
 * 모델이 도구를 호출하기로 결정한 순간과 도구 결과가 돌아온 순간을
 * 각각 한 줄씩 출력하세요.
 *
 * 힌트: chunk 는 { 노드이름: { messages: [...] } } 모양입니다.
 *       AIMessage 에 tool_calls 가 있으면 "호출 결정",
 *       ToolMessage(getType() === "tool") 가 오면 "결과 도착" 입니다.
 * ===================================================================== */
async function problem3(): Promise<void> {
  // TODO: 구현
}

/* =====================================================================
 * [문제 4] messages 모드에서 특정 노드의 토큰만 뽑기
 *
 * streamMode: "messages" 로 구독하되, metadata.langgraph_node 가
 * "model" 인 토큰만 출력하세요. (도구 결과는 걸러집니다.)
 *
 * 먼저 필터 없이 돌려서 어떤 노드 이름들이 나오는지 눈으로 확인한 뒤 거세요.
 * ===================================================================== */
async function problem4(): Promise<void> {
  // TODO: 구현
}

/* =====================================================================
 * [문제 5] 여러 모드 동시 구독
 *
 * streamMode: ["updates", "messages"] 로 구독해서
 * - messages 청크는 그대로 화면에 흘리고
 * - updates 청크가 오면 줄바꿈 후 "[스텝 완료: 노드이름]" 을 찍으세요.
 *
 * 함정: 배열을 주면 각 항목이 [mode, chunk] 튜플이 됩니다.
 *       단일 모드일 때의 구조분해를 그대로 쓰면 안 됩니다.
 * ===================================================================== */
async function problem5(): Promise<void> {
  // TODO: 구현
}

/* =====================================================================
 * [문제 6] tool_call_chunks 를 손으로 누적하기 ★어려움
 *
 * concat 을 쓰지 말고, tool_call_chunks 를 직접 index 별로 모아
 * "인자 JSON 이 완성되는 순간"을 감지하세요.
 *
 * 반환: 완성된 { index, name, args } 배열
 *
 * 규칙:
 * - tc.index 가 같은 조각끼리 args 문자열을 이어붙입니다.
 * - name 은 보통 첫 조각에만 실려 옵니다. 나중 조각의 undefined 로 덮어쓰면 안 됩니다.
 * - 매 조각마다 JSON.parse 를 시도하되, 실패는 "아직 미완성"이라는 뜻이므로
 *   에러를 던지지 말고 넘어가야 합니다.
 *
 * 힌트: model.bindTools([searchDocs]) 로 도구를 붙인 뒤 stream() 하세요.
 * ===================================================================== */
async function problem6(): Promise<Array<{ index: number; name: string; args: unknown }>> {
  // TODO: 구현
  return [];
}

/* =====================================================================
 * [문제 7] 커스텀 진행률 스트리밍
 *
 * (a) config.writer?.() 로 진행률을 3번 내보내는 도구를 새로 만드세요.
 *     예: { type: "progress", done: 1, total: 3 }
 * (b) 그 도구를 쓰는 에이전트를 만들고 streamMode: "custom" 으로 구독해
 *     "1/3", "2/3", "3/3" 을 출력하세요.
 *
 * 함정: writer 는 config 의 두 번째 인자로만 들어옵니다.
 *       도구 함수 시그니처에서 config 를 안 받으면 진행률이 조용히 사라집니다.
 * ===================================================================== */
async function problem7(): Promise<void> {
  // TODO: 구현
}

/* =====================================================================
 * [문제 8] SSE 핸들러 ★어려움
 *
 * node:http 서버를 띄워 /chat 에서 SSE 로 토큰을 흘리세요. 요구사항:
 * - Content-Type: text/event-stream, Cache-Control: no-cache
 * - 토큰은 `event: token` 으로
 * - 스트림 도중 예외가 나면 `event: error` 로 내보내고 res.end()
 *   (이미 200 을 보냈으니 500 으로 바꿀 수 없습니다)
 * - req 의 "close" 이벤트에서 AbortController 로 에이전트를 취소
 *
 * 검증: 같은 프로세스에서 fetch 로 구독해 출력해 보세요.
 * ===================================================================== */
async function problem8(): Promise<void> {
  // TODO: 구현
}

/* ===== 실행 ===== */

async function main(): Promise<void> {
  printSection("[문제 1] TTFT 측정");
  console.log(await problem1("스트리밍을 한 문장으로 설명해줘"));

  printSection("[문제 2] concat 으로 청크 합치기");
  console.log(await problem2("안녕이라고만 답해줘"));

  printSection("[문제 3] updates 로 도구 호출 추적");
  await problem3();

  printSection("[문제 4] messages 에서 model 노드만");
  await problem4();

  printSection("[문제 5] 여러 모드 동시 구독");
  await problem5();

  printSection("[문제 6] tool_call_chunks 수동 누적");
  console.log(await problem6());

  printSection("[문제 7] 커스텀 진행률");
  await problem7();

  printSection("[문제 8] SSE 핸들러");
  await problem8();
}

void ask;
void agent;

main().catch((err: unknown) => {
  console.error(err);
  process.exit(1);
});

/**
 * Step 09 — 스트리밍 · 정답과 해설
 * 실행: npx tsx docs/reference/langchain/step-09-streaming/solution.ts
 *
 * exercise.ts 를 스스로 풀어본 뒤에 열어보세요.
 * 각 정답 위 주석에 "왜 이렇게 해야 하는가"와 흔한 오답을 적어 두었습니다.
 */
import "dotenv/config";

import { createAgent, tool } from "langchain";
import { ChatAnthropic } from "@langchain/anthropic";
import { AIMessageChunk } from "@langchain/core/messages";
import { type LangGraphRunnableConfig } from "@langchain/langgraph";
import * as z from "zod";

import { printSection } from "../project/src/lib/print.js";

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

const searchDocs = tool(
  async (input: { query: string }, _config: LangGraphRunnableConfig) => {
    await new Promise((r) => setTimeout(r, 200));
    return `"${input.query}" 검색 결과: 문서 3건을 찾았습니다.`;
  },
  {
    name: "search_docs",
    description: "사내 문서를 검색합니다.",
    schema: z.object({ query: z.string().describe("검색어") }),
  },
);

const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  tools: [searchDocs],
  systemPrompt: "당신은 사내 문서 검색 비서입니다. 질문에는 반드시 도구로 근거를 찾아 답하세요.",
});

const ask = { messages: [{ role: "user" as const, content: "휴가 정책을 알려줘" }] };

/* =====================================================================
 * [정답 1] TTFT 측정
 *
 * 핵심은 "첫 청크가 도착한 시점"을 루프 안에서 단 한 번만 찍는 것입니다.
 * 흔한 오답: 루프가 끝난 뒤에 Date.now() 를 재는 것 — 그건 TTFT 가 아니라
 * 전체 완료 시간입니다. 사용자 체감은 전자가 좌우합니다.
 *
 * 실제로 돌려 보면 ttftMs 는 보통 totalMs 의 5~15% 수준입니다.
 * 이 격차가 스트리밍을 쓰는 유일한 이유입니다.
 * ===================================================================== */
async function problem1(
  prompt: string,
): Promise<{ ttftMs: number; totalMs: number; text: string }> {
  const started = Date.now();
  let ttftMs = -1;
  let text = "";

  const stream = await model.stream(prompt);
  for await (const chunk of stream) {
    if (ttftMs === -1) ttftMs = Date.now() - started; // 첫 청크에서만
    text += chunk.text; // .text 는 항상 문자열이라 안전합니다
  }

  return { ttftMs, totalMs: Date.now() - started, text };
}

/* =====================================================================
 * [정답 2] concat 으로 청크 합치기
 *
 * `full = full === null ? chunk : full.concat(chunk)` 가 이 스텝의 관용구입니다.
 *
 * 왜 문자열 += 가 안 되는가:
 * - text 만 필요하면 우연히 잘 동작합니다. 그래서 조용히 틀립니다.
 * - 하지만 usage_metadata, response_metadata, tool_call_chunks 는
 *   문자열로는 절대 복원되지 않습니다. usage_metadata 는 보통 마지막
 *   청크에만 실려 오는데, concat 만이 그걸 병합해서 올려 줍니다.
 * - reasoning 블록이 있는 모델은 content 가 블록 배열이라
 *   += 하면 "[object Object]" 가 섞입니다.
 * ===================================================================== */
async function problem2(prompt: string): Promise<AIMessageChunk | null> {
  let full: AIMessageChunk | null = null;

  const stream = await model.stream(prompt);
  for await (const chunk of stream) {
    full = full === null ? chunk : full.concat(chunk);
  }

  return full;
}

/* =====================================================================
 * [정답 3] updates 로 도구 호출 추적
 *
 * updates 청크는 { 노드이름: { messages: [...] } } 모양입니다.
 * "노드 하나가 방금 무엇을 상태에 더했나"만 담기므로 payload 가 작습니다.
 * 같은 걸 values 로 하면 매번 전체 messages 배열이 통째로 와서 낭비입니다.
 *
 * 흔한 오답: Object.entries 없이 chunk.messages 를 바로 읽는 것.
 * updates 는 노드 이름으로 한 겹 감싸여 있습니다.
 * ===================================================================== */
async function problem3(): Promise<void> {
  for await (const chunk of await agent.stream(ask, { streamMode: "updates" })) {
    for (const [node, update] of Object.entries(chunk)) {
      const messages = (update as { messages?: unknown[] }).messages ?? [];
      for (const m of messages) {
        const msg = m as {
          getType?: () => string;
          tool_calls?: Array<{ name: string; args: unknown }>;
          text?: string;
        };
        const type = msg.getType?.() ?? "?";

        if (type === "ai" && (msg.tool_calls?.length ?? 0) > 0) {
          for (const call of msg.tool_calls ?? []) {
            console.log(`[${node}] 도구 호출 결정 → ${call.name}(${JSON.stringify(call.args)})`);
          }
        } else if (type === "tool") {
          console.log(`[${node}] 도구 결과 도착 → ${String(msg.text).slice(0, 60)}`);
        } else if (type === "ai") {
          console.log(`[${node}] 최종 답변 ${String(msg.text).length}자`);
        }
      }
    }
  }
}

/* =====================================================================
 * [정답 4] messages 에서 특정 노드만
 *
 * metadata.langgraph_node 가 "어느 노드에서 나온 토큰인가"를 알려 줍니다.
 * createAgent 의 모델 노드 이름은 "model" 입니다.
 *
 * ⚠️ 이 필터가 필요한 이유: streamMode "messages" 는 LLM 토큰만이 아니라
 *    도구 노드가 만든 ToolMessage 도 함께 흘려보냅니다. 필터 없이
 *    화면에 다 찍으면 도구 결과 원문이 사용자에게 그대로 노출됩니다.
 *
 * 노드 이름은 프레임워크 내부 명칭이라 버전에 따라 바뀔 수 있습니다.
 * 하드코딩이 불안하면 metadata.tags 나 withConfig({ tags: [...] }) 로
 * 직접 붙인 태그로 거는 편이 안전합니다.
 * ===================================================================== */
async function problem4(): Promise<void> {
  for await (const [token, metadata] of await agent.stream(ask, { streamMode: "messages" })) {
    const node = (metadata as { langgraph_node?: string }).langgraph_node;
    if (node !== "model") continue;

    const text = (token as AIMessageChunk).text;
    if (text !== "") process.stdout.write(text);
  }
  console.log("");
}

/* =====================================================================
 * [정답 5] 여러 모드 동시 구독
 *
 * 배열을 주는 순간 반환 항목이 [mode, chunk] 튜플로 바뀝니다.
 * 이게 이 절의 함정입니다 — 단일 모드 코드를 그대로 두면
 * chunk 자리에 문자열 "messages" 가 들어와 .text 가 undefined 가 됩니다.
 *
 * 그리고 mode === "messages" 일 때 chunk 는 다시 [token, metadata] 튜플이라
 * 튜플이 두 겹입니다. 여기서 헷갈리는 사람이 많습니다.
 * ===================================================================== */
async function problem5(): Promise<void> {
  for await (const [mode, chunk] of await agent.stream(ask, {
    streamMode: ["updates", "messages"],
  })) {
    if (mode === "messages") {
      const [token] = chunk as [AIMessageChunk, unknown];
      if (token.text !== "") process.stdout.write(token.text);
    } else if (mode === "updates") {
      const nodes = Object.keys(chunk as object).join(", ");
      console.log(`\n[스텝 완료: ${nodes}]`);
    }
  }
  console.log("");
}

/* =====================================================================
 * [정답 6] tool_call_chunks 수동 누적 ★
 *
 * 이 문제의 정답은 실무에서 직접 쓸 일이 거의 없습니다 — concat 이 이걸
 * 대신 해 주니까요. 그럼에도 손으로 짜 보는 이유는 "부분 JSON" 이 무엇인지
 * 몸으로 알기 위해서입니다.
 *
 * 세 가지 포인트:
 * 1. index 로 묶는다. 병렬 도구 호출이면 index 0, 1 조각이 번갈아 옵니다.
 *    index 를 무시하고 args 를 한 문자열에 몰아 붙이면 두 호출의 JSON 이
 *    섞여서 영원히 파싱되지 않습니다.
 * 2. name 은 첫 조각에만 옵니다. `acc.name = tc.name` 으로 무조건 대입하면
 *    두 번째 조각의 undefined 가 이름을 지웁니다. `??` 로 보존해야 합니다.
 * 3. JSON.parse 실패는 에러가 아니라 "아직 안 왔다" 입니다.
 *    try/catch 로 삼키고 다음 조각을 기다리는 게 정상 동작입니다.
 * ===================================================================== */
async function problem6(): Promise<Array<{ index: number; name: string; args: unknown }>> {
  const modelWithTools = model.bindTools([searchDocs]);
  const stream = await modelWithTools.stream("휴가 정책 문서를 검색해줘");

  // index → 누적 상태
  const acc = new Map<number, { name: string; raw: string }>();
  const completed: Array<{ index: number; name: string; args: unknown }> = [];
  const doneIndexes = new Set<number>();

  for await (const chunk of stream) {
    for (const tc of chunk.tool_call_chunks ?? []) {
      const index = tc.index ?? 0;
      const prev = acc.get(index) ?? { name: "", raw: "" };

      acc.set(index, {
        // ⚠️ tc.name 이 undefined 인 조각이 대부분입니다. 덮어쓰면 안 됩니다.
        name: prev.name !== "" ? prev.name : (tc.name ?? ""),
        raw: prev.raw + (tc.args ?? ""),
      });

      // 매 조각마다 완성됐는지 시도해 봅니다.
      const cur = acc.get(index);
      if (cur === undefined || doneIndexes.has(index)) continue;
      try {
        const args: unknown = JSON.parse(cur.raw);
        doneIndexes.add(index);
        completed.push({ index, name: cur.name, args });
        console.log(`index=${index} 완성! name=${cur.name} args=${cur.raw}`);
      } catch {
        // 아직 미완성 — 정상. 다음 조각을 기다립니다.
        console.log(`index=${index} 미완성 raw=${JSON.stringify(cur.raw)}`);
      }
    }
  }

  return completed;
}

/* =====================================================================
 * [정답 7] 커스텀 진행률 스트리밍
 *
 * 포인트는 도구 함수의 "두 번째 인자" 입니다.
 *   async (input) => {...}                              ← writer 를 못 씀
 *   async (input, config: LangGraphRunnableConfig) => {} ← 정답
 *
 * config 를 안 받아도 타입 에러가 나지 않고 도구는 잘 동작합니다.
 * 진행률만 조용히 사라집니다. 그래서 이건 함정입니다.
 *
 * writer 를 `config.writer?.()` 로 옵셔널 호출하는 것도 중요합니다.
 * 스트리밍이 아닌 invoke() 경로에서는 writer 가 없을 수 있습니다.
 * ===================================================================== */
async function problem7(): Promise<void> {
  const bulkIndex = tool(
    async (input: { count: number }, config: LangGraphRunnableConfig) => {
      const total = 3;
      for (let done = 1; done <= total; done += 1) {
        await new Promise((r) => setTimeout(r, 150));
        config.writer?.({ type: "progress", done, total });
      }
      return `${input.count}건을 색인했습니다.`;
    },
    {
      name: "bulk_index",
      description: "문서를 대량 색인합니다. 진행률을 보고합니다.",
      schema: z.object({ count: z.number().describe("색인할 문서 수") }),
    },
  );

  const indexAgent = createAgent({
    model: "anthropic:claude-sonnet-4-6",
    tools: [bulkIndex],
    systemPrompt: "당신은 색인 작업자입니다. 요청받으면 도구를 사용하세요.",
  });

  for await (const chunk of await indexAgent.stream(
    { messages: [{ role: "user", content: "문서 100건을 색인해줘" }] },
    { streamMode: "custom" },
  )) {
    const p = chunk as { type?: string; done?: number; total?: number };
    if (p.type === "progress") {
      console.log(`진행률 ${p.done}/${p.total}`);
    }
  }
}

/* =====================================================================
 * [정답 8] SSE 핸들러 ★
 *
 * 실전에서 자주 놓치는 네 가지가 전부 여기 들어 있습니다.
 *
 * 1. 헤더 3종 세트. Cache-Control: no-cache 가 없으면 중간 프록시가
 *    응답을 통째로 모았다가 한 번에 보냅니다 — 스트리밍이 사라집니다.
 *    nginx 뒤에 있다면 X-Accel-Buffering: no 도 필요합니다.
 * 2. 에러를 "이벤트로" 보낸다. writeHead(200) 을 이미 호출했으므로
 *    나중에 500 으로 바꿀 수 없습니다. 상태 코드로 실패를 알리는
 *    평소 습관이 여기선 통하지 않습니다.
 * 3. req.on("close") → controller.abort(). 사용자가 탭을 닫아도
 *    에이전트는 계속 돌고 토큰은 계속 과금됩니다. 이걸 안 끊는 게
 *    스트리밍 서버의 대표적인 돈 새는 구멍입니다.
 * 4. SSE 프레임 규격: `data: ...\n\n` — 빈 줄 하나로 프레임이 끝납니다.
 *    \n 하나만 쓰면 브라우저가 이벤트를 영원히 기다립니다.
 * ===================================================================== */
async function problem8(): Promise<void> {
  const { createServer } = await import("node:http");

  const server = createServer(async (req, res) => {
    if (req.url !== "/chat") {
      res.writeHead(404).end();
      return;
    }

    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no",
    });

    const controller = new AbortController();
    req.on("close", () => controller.abort());

    const send = (event: string, data: unknown): void => {
      res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
    };

    try {
      for await (const [token, metadata] of await agent.stream(ask, {
        streamMode: "messages",
        signal: controller.signal,
      })) {
        if ((metadata as { langgraph_node?: string }).langgraph_node !== "model") continue;
        const text = (token as AIMessageChunk).text;
        if (text !== "") send("token", { text });
      }
      send("done", { ok: true });
    } catch (err) {
      const e = err as Error;
      // 취소(AbortError)는 사용자가 끊은 것이므로 에러 이벤트로 보낼 필요가 없습니다.
      if (e.name !== "AbortError") send("error", { message: e.message });
    } finally {
      res.end();
    }
  });

  await new Promise<void>((resolve) => server.listen(0, resolve));
  const address = server.address();
  const port = typeof address === "object" && address !== null ? address.port : 0;

  const res = await fetch(`http://127.0.0.1:${port}/chat`);
  const reader = res.body?.getReader();
  const decoder = new TextDecoder();
  if (reader !== undefined) {
    for (;;) {
      const { done, value } = await reader.read();
      if (done) break;
      process.stdout.write(decoder.decode(value));
    }
  }

  server.close();
}

/* ===== 실행 ===== */

async function main(): Promise<void> {
  printSection("[정답 1] TTFT 측정");
  const r1 = await problem1("스트리밍을 한 문장으로 설명해줘");
  console.log(`ttft=${r1.ttftMs}ms total=${r1.totalMs}ms (${r1.text.length}자)`);

  printSection("[정답 2] concat 으로 청크 합치기");
  const r2 = await problem2("안녕이라고만 답해줘");
  console.log("text          :", r2?.text);
  console.log("usage_metadata:", JSON.stringify(r2?.usage_metadata));

  printSection("[정답 3] updates 로 도구 호출 추적");
  await problem3();

  printSection("[정답 4] messages 에서 model 노드만");
  await problem4();

  printSection("[정답 5] 여러 모드 동시 구독");
  await problem5();

  printSection("[정답 6] tool_call_chunks 수동 누적");
  console.log(JSON.stringify(await problem6(), null, 2));

  printSection("[정답 7] 커스텀 진행률");
  await problem7();

  printSection("[정답 8] SSE 핸들러");
  await problem8();
}

main().catch((err: unknown) => {
  console.error(err);
  process.exit(1);
});