// scenarios/prompt-routing.mjs — web-side subscribe → live event fan-out
// (including unpersisted message_update), REST prompt routed to the agent WS,
// 409 when the agent is offline.

import { agentClient, agentEvent, agentHello, rest, sessionSnapshot, sid, waitUntil, webClient } from "../lib.mjs";

export async function run(ctx) {
  const { r, base, token, agentUrl, webUrl, state } = ctx;
  const id = sid("e2e-prompt");
  const snapshot = sessionSnapshot(id, { name: "prompt-routing", repo: "e2e/harness" });

  const browser = await webClient(webUrl, token);
  const initList = await browser.waitForFrame((f) => f.type === "session_list");
  r.check("web WS sends initial session_list on connect", initList !== null);
  browser.send({ type: "subscribe", sessionId: id });

  const agent = await agentClient(agentUrl, token);
  const welcome = await agentHello(agent, snapshot);
  r.check("agent handshake under active web subscription", welcome?.type === "welcome");

  // session change (hello registers the session) must reach the browser.
  const listed = await browser.waitForFrame(
    (f) => f.type === "session_list" && f.sessions?.some((s) => s.id === id),
  );
  r.check("session_list frame carries the new session", listed !== null);
  const listedRow = listed?.sessions?.find((s) => s.id === id);
  r.check(
    "session_list row matches hello snapshot",
    listedRow?.name === "prompt-routing" && listedRow?.repo === "e2e/harness" && listedRow?.online === true,
    JSON.stringify(listedRow ?? null),
  );

  // Live fan-out: persisted + unpersisted kinds both batched to the subscriber.
  agentEvent(agent, id, 1, "message_start", { message: { role: "assistant", id: "m1" } });
  agentEvent(agent, id, 2, "message_update", { delta: "live delta one " });
  agentEvent(agent, id, 3, "message_update", { delta: "live delta two" });
  agentEvent(agent, id, 4, "message_end", {
    message: { role: "assistant", id: "m1", text: "live delta one live delta two", thinking: null, toolCalls: [], toolCallId: null },
  });

  const frame = await browser.waitForFrame((f) => f.type === "events" && f.sessionId === id);
  r.check("events frame received for subscribed session", frame !== null);
  // Deltas flush in ≤40ms batches; wait until the seq-4 message_end fan-out lands.
  const fanoutDone = await browser.waitForFrame(
    (f) => f.type === "events" && f.sessionId === id && (f.events ?? []).some((e) => e.seq === 4),
  );
  r.check("browser received events through message_end (seq 4)", fanoutDone !== null);
  const all = browser.frames
    .filter((f) => f.type === "events" && f.sessionId === id)
    .flatMap((f) => f.events ?? []);
  r.check(
    "browser saw live message_update deltas (seq 2,3)",
    all.some((e) => e.seq === 2 && e.type === "message_update" && e.delta === "live delta one ") &&
      all.some((e) => e.seq === 3 && e.delta === "live delta two"),
    JSON.stringify(all.map((e) => `${e.seq}:${e.type}`)),
  );
  const evFrames = browser.frames.filter((f) => f.type === "events" && f.sessionId === id);
  r.check(
    "events frame after = first seq - 1",
    evFrames.every((f) => f.after === (f.events?.[0]?.seq ?? f.after + 1) - 1),
    JSON.stringify(evFrames.map((f) => f.after)),
  );

  // Prompt routing: REST → agent WS.
  const prompted = await rest(base, token, `/api/sessions/${id}/prompt`, {
    method: "POST",
    body: { message: "e2e routed prompt" },
  });
  r.check("POST prompt while online → 200 {ok:true}", prompted.status === 200 && prompted.json?.ok === true, JSON.stringify(prompted.json));

  const promptFrame = await agent.waitForFrame((f) => f.type === "prompt");
  r.check("agent received prompt frame", promptFrame !== null);
  r.check(
    "prompt frame envelope + payload",
    promptFrame?.v === 1 && promptFrame?.sessionId === id && promptFrame?.message === "e2e routed prompt" && typeof promptFrame?.promptId === "string",
    JSON.stringify(promptFrame ?? null),
  );

  agent.close();
  const goneOffline = await waitUntil(async () => {
    const res = await rest(base, token, "/api/sessions");
    return res.json?.find((s) => s.id === id)?.online === false;
  });
  r.check("agent offline after close", goneOffline);

  const rejected = await rest(base, token, `/api/sessions/${id}/prompt`, {
    method: "POST",
    body: { message: "nobody home" },
  });
  r.check("POST prompt while offline → 409", rejected.status === 409, `got ${rejected.status}`);

  browser.close();
  state.prompt = { id };
}