"""Phase 07 task 03 — interpolation smoothness with a 60 ms latency.

The task's smoothness criterion, end to end: two browsers against a
server with `sim_latency_ms=60` (the injected server-side downlink
delay, `config.py`'s netcode seam), the host's tab drives the scripted
soup (the `test_us03_ready_and_start.py` choreography — the same
waypoints and holds), and the second tab's *rendered* state — the
single loop's `sample → lerpState` output published on
`window.__uc.lastView` — is sampled per animation frame while the
round runs.

The assertions:

* **the per-tick displacement is bounded** — a server tick moves a
  cook at most `MAX_SPEED * STEP_DT` (3.5 * 0.05 = 0.175); the bound
  the task asserts is `MAX_SPEED * (1/snapshot_hz) * 1.35`
  (0.23625). Checked on the buffer's *server frames* (consecutive
  ticks — the sampler records the frame's `t` and only pairs a
  displacement with its one-tick advance),
* **the rendered stream is smooth** — a sample's step is bounded by
  its gap's share of the 50 ms tick (a 100 ms gap moves at most two
  ticks) plus the 1.35 factor, and the velocity over a normal gap is
  at most `MAX_SPEED * 1.35` tiles/s,
* **both clients converge on `ROUND_END` within 1 s** — the round
  frame is broadcast once; with the 60 ms injected delay the second
  tab's result screen must follow the host's within the second.

The second tab's sampler is a browser-side `requestAnimationFrame`
promise (started while the host's choreography runs, so the sampling
is live), reading only — the single-loop rule is untouched (one loop
owns the buffer; the probe reads the loop's published `lastView` and
the buffer's `latest()`).
"""

from __future__ import annotations

import time
from typing import TYPE_CHECKING

import pytest

from conftest import ROUND_LEVELS_DIR, app_screen, load_app, open_two_pages

if TYPE_CHECKING:
    from playwright.sync_api import Browser

pytestmark = pytest.mark.e2e

#: The host's tab drives the soup — the `test_us03_ready_and_start.py`
#: choreography, the same waypoints and holds (the phase-06 two-client
#: machine, verbatim).
SOUP_CHOREOGRAPHY = """async () => {
  const app = window.__undercooked;
  const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
  const ownSeat = app.you.seat;
  const player = () => Object.values(app.view.players).find((p) => p.pi === ownSeat);

  const walkTo = async (target, band, settle, waitS) => {
    const deadline = performance.now() + waitS * 1000;
    for (;;) {
      const me = player();
      if (me === undefined) throw new Error('the scripted cook vanished from the view');
      const dx = target[0] - me.x;
      const dy = target[1] - me.y;
      const distance = Math.hypot(dx, dy);
      if (distance <= band) {
        if (settle) {
          app.sendInput(null, false);
          await sleep(150);  // the settle: the view catches up
          const again = player();
          if (again === undefined) throw new Error('the scripted cook vanished (settle)');
          const tail = Math.hypot(target[0] - again.x, target[1] - again.y);
          if (tail <= band + 0.2) return;  // the deceleration tail
          continue;  // the settle was not enough: walk again
        }
        return;  // pass the waypoint (a stopped cook never leaves the band)
      }
      if (performance.now() > deadline) throw new Error('timed out walking');
      const length = distance > 1e-6 ? distance : 1.0;
      app.sendInput({ x: dx / length, y: dy / length }, false);
      await sleep(50);  // ~20 Hz: one frame per tick — a tick with no
      // frame brakes the body, so the feed must ride every tick window
      // (walk and hold are sequential: the total is 20 Hz + the ping,
      // under the 40/s token bucket)
    }
  };

  const walk = async (waypoints, band, waitS) => {
    for (let i = 0; i < waypoints.length - 1; i += 1) {
      await walkTo(waypoints[i], 0.35, false, waitS);
    }
    await walkTo(waypoints[waypoints.length - 1], band, true, waitS);
  };

  const press = async () => {
    app.sendInput(null, true);  // the rising edge
    await sleep(90);
    app.sendInput(null, false);  // the release
  };

  // The pick after a hold, self-healing: an explicit release before
  // each attempt breaks a stuck hold; a missed attempt retries on a
  // fresh rising edge. The 250 ms spread keeps a tight 3-frame cluster
  // off the 40/s token bucket. An early pick (the last hold frame
  // landed late) is honoured: no press while the item is in hand.
  const pick = async (what, short) => {
    for (let i = 0; i < 3; i += 1) {
      if (carrying(short)) return;
      app.sendInput(null, false);
      await sleep(250);
      await press();
      const deadline = performance.now() + 6000;
      while (!carrying(short) && performance.now() < deadline) {
        await sleep(50);
      }
      if (carrying(short)) return;
    }
    throw new Error('timed out: ' + what);
  };

  const holdUntil = async (predicate, waitS) => {
    const deadline = performance.now() + waitS * 1000;
    while (!predicate()) {
      if (performance.now() > deadline) throw new Error('timed out holding');
      app.sendInput(null, true);
      await sleep(50);  // ~20 Hz: one frame per tick — a 1-tick gap resets
      // the gate's hold (the work freezes); the next frame re-presses
      // and re-claims it (20 Hz + the walk feed, under the 40/s bucket)
    }
  };

  const waitUntil = async (predicate, what, waitS) => {
    const deadline = performance.now() + waitS * 1000;
    while (!predicate()) {
      if (performance.now() > deadline) throw new Error('timed out: ' + what);
      await sleep(50);
    }
  };

  const stationReady = (kind) => {
    const station = Object.values(app.view.stations).find((s) => s.k === kind);
    return station !== undefined && station.s === 'ready';
  };
  const carrying = (short) => {
    const me = player();
    return me !== undefined && me.h === short;
  };

  // The phase-06 waypoints and staging points, verbatim.
  const SHELF_STAGE = [2.5, 1.5];  // facing LEFT at the shelf (1, 1)
  const BOARD_STAGE = [3.5, 1.5];  // facing RIGHT at the board (4, 1)
  const STOVE_STAGE = [7.5, 1.5];  // facing RIGHT at the stove (8, 1)
  const PREP_STAGE = [3.5, 3.5];  // facing LEFT at the prep (2, 3)
  const SERVE_STAGE = [8.8, 3.5];  // in front of the window (9, 3): a stop
  // west of its centre keeps the window ahead of the facing plane
  // (`not_facing`)

  // 1. The tomato, off the spawn shelf (the 3.5 corridor clears the
  //    blocking fixtures, the body's edge included). Self-healing: a
  //    dropped press frame is a silent no-op; the shelf is still full
  //    for a fresh rising edge.
  await walk([[3.5, 7.5], [3.5, 1.5], SHELF_STAGE], 0.4, 60);
  await pick('pick the tomato', 'tomato_raw');

  // 2. Chop it (0.9 s of work).
  await walk([BOARD_STAGE], 0.4, 45);
  await holdUntil(() => stationReady('chop_board'), 30);
  await pick('pick the chopped tomato', 'tomato_chopped');

  // 3. Cook it (6 s of work). The row-2 corridor clears the board and
  //    the stove; the last leg comes in west-to-east along row 1.
  await walk([[3.5, 2.5], [5.5, 2.5], [5.5, 1.5], STOVE_STAGE], 0.4, 60);
  await holdUntil(() => stationReady('stove'), 40);
  await pick('pick the cooked tomato', 'tomato_cooked');

  // 4. Plate it (0.6 s of work). The prep's staging point is its east
  //    neighbour, reached along row 3.
  await walk([[7.5, 3.5], PREP_STAGE], 0.4, 45);
  await holdUntil(() => stationReady('prep'), 30);
  await pick('pick the plated dish', 'tomato_plated');

  // 5. Serve it — retried: a dropped press frame (or a `not_facing`
  //    stop) is a silent no-op; the dish stays in the hand for a fresh
  //    rising edge. The round ends on the win (target 20), so an
  //    `input` frame after the verdict is a wrong_state (5 close 4404)
  //    — the screen check stops the feed the moment it does.
  await walk([SERVE_STAGE], 0.4, 45);
  for (let i = 0; i < 3; i += 1) {
    if (app.screen !== 'playing') break;
    app.sendInput(null, false);
    await sleep(250);
    if (app.screen !== 'playing') break;
    await press();
    await sleep(1500);
  }
  return true;
}"""

#: The sampler's contract: it records `{at, x, y, frameT, frameX,
#: frameY, frameStep}` per animation frame — `at` the
#: `performance.now()` of the sample, `x`/`y` the *rendered*
#: (interpolated) position of the remote cook, `frameT` the buffer's
#: latest server frame's tick, and `frameStep` the displacement from
#: the previous sample's server frame — set only when the frame
#: advanced by exactly one tick (a sample that straddles two ticks is
#: not a per-tick displacement). It resolves on the tab's result screen
#: (or after 120 s, whatever it has).
SAMPLER_JS = """
() => {
  window.__uc_sampling = new Promise((resolve) => {
    const samples = [];
    const startedAt = performance.now();
    const sample = () => {
      const uc = window.__uc;
      const app = window.__undercooked;
      if (uc !== undefined && app !== null && uc.lastView !== null) {
        const view = uc.lastView;
        const own = app.you === null ? -1 : app.you.seat;
        const remote = (view.p || []).find((p) => p.pi !== own);
        if (remote !== undefined) {
          const latest = uc.buffer !== null ? uc.buffer.latest() : null;
          let frameT = null;
          let frameX = null;
          let frameY = null;
          let frameStep = null;
          let frameJump = 1;
          if (latest !== null && latest.p.length > 0) {
            const latestRemote = latest.p.find((p) => p.pi !== own);
            if (latestRemote !== undefined) {
              frameT = latest.t;
              frameX = latestRemote.x;
              frameY = latestRemote.y;
              const last = samples[samples.length - 1];
              if (last !== undefined && last.frameT !== null) {
                const jump = frameT - last.frameT;
                if (jump > 1) frameJump = jump; // a delivery hiccup
                if (jump === 1) {
                  frameStep = Math.hypot(frameX - last.frameX, frameY - last.frameY);
                }
              }
            }
          }
          samples.push({
            at: performance.now(), x: remote.x, y: remote.y,
            frameT, frameX, frameY, frameStep, frameJump,
          });
        }
      }
      if (document.querySelector('#result-word') !== null) { resolve(samples); return; }
      if (performance.now() - startedAt > 120000) { resolve(samples); return; }
      requestAnimationFrame(sample);
    };
    sample();
  });
  return 'started';
}
"""

#: `MAX_SPEED * (1 / SNAPSHOT_HZ) * 1.35` — the task's per-frame bound.
STEP_BOUND = 3.5 * 0.05 * 1.35
#: `MAX_SPEED * 1.35` — the rendered velocity bound (tiles / second).
SPEED_BOUND = 3.5 * 1.35


def _assert_stream(samples: list[dict]) -> None:
    """The two smoothness assertions (the sample's step bound, the
    per-tick displacement bound)."""
    # Enough samples to talk about a stream (the round ran well past
    # the warm-up; 30 samples at ~60 fps is half a second).
    assert len(samples) >= 30, f"only {len(samples)} samples recorded"

    # The per-tick displacement (consecutive server ticks, from the
    # buffer's latest frames): the bound is
    # `MAX_SPEED * (1/snapshot_hz) * 1.35` (0.23625).
    frame_steps = [
        s["frameStep"] for s in samples if s.get("frameStep") is not None
    ]
    assert len(frame_steps) >= 30, (
        f"only {len(frame_steps)} one-tick displacements (the buffer was empty?)"
    )
    worst_tick = max(frame_steps)
    assert worst_tick <= STEP_BOUND, (
        f"a server tick moved the cook {worst_tick:.4f} tiles "
        f"(> {STEP_BOUND:.4f} = MAX_SPEED * 1/snapshot_hz * 1.35)"
    )

    # The rendered positions: a sample's step is bounded by its gap's
    # share of the 50 ms tick (a 100 ms gap moves at most two
    # ticks) plus the 1.35 factor; a delivery hiccup (the latest
    # frame's tick jumping ahead — a stall's resume) scales the
    # allowance by the jump, and the velocity over a normal gap is
    # at most `MAX_SPEED * 1.35` tiles/s (a hiccup's jump is not a
    # velocity).
    for i in range(1, len(samples)):
        a, b = samples[i - 1], samples[i]
        dt_ms = b["at"] - a["at"]
        if dt_ms <= 0:
            continue
        jump = max(1, b.get("frameJump", 1))
        step = ((b["x"] - a["x"]) ** 2 + (b["y"] - a["y"]) ** 2) ** 0.5
        allowance = STEP_BOUND * max(dt_ms / 50.0 + 0.5, jump)
        assert step <= allowance, (
            f"the rendered step {step:.4f} at a {dt_ms:.1f} ms gap (frame jump "
            f"{jump}) exceeds the allowed {allowance:.4f}"
        )
        if dt_ms >= 25.0 and jump <= dt_ms / 50.0 + 0.5:
            velocity = step / (dt_ms / 1000.0)
            assert velocity <= SPEED_BOUND, (
                f"the rendered velocity {velocity:.3f} tiles/s over a "
                f"{dt_ms:.1f} ms gap exceeds {SPEED_BOUND:.3f}"
            )


def _join_ui(page, code: str) -> None:
    """The keyboard-first join (the us03 machine)."""
    page.locator("#join-code").focus()
    page.keyboard.type(code)
    page.keyboard.press("Enter")


def test_the_rendered_stream_is_bounded_and_converges_within_a_second(
    browser: Browser, e2e_server
) -> None:
    """The 60 ms-latency round: the scripted soup on the host's tab,
    per-frame sampling of the second tab's rendered state, the per-tick
    displacement bounded by `MAX_SPEED * (1/snapshot_hz) * 1.35`, the
    rendered stream smooth, and both clients on the result screen
    within 1 s of each other."""
    url = e2e_server(ROUND_LEVELS_DIR, sim_latency_ms=60).url
    context_a, page_a, context_b, page_b = open_two_pages(browser, url)
    try:
        # The scripted tab drives the wire through the seam, so the
        # keyboard module is off for it (`?controls=off`, the documented
        # scripted-test mode): its 10 Hz zero-intent keep-alive would
        # race the script's frames on the same seat (an `act=false`
        # frame breaks the holds, a zero vector brakes the walks).
        load_app(page_a, url + "?controls=off")
        load_app(page_b, url)
        page_a.evaluate("() => window.__undercooked.createLobby({ name: 'Ada' })")
        page_a.locator("#room-code").wait_for(state="visible", timeout=20_000)
        code = page_a.locator("#room-code").inner_text()
        _join_ui(page_b, code)
        page_b.locator("#room-code").wait_for(state="visible", timeout=20_000)

        page_b.get_by_role("button", name="Ready up").focus()
        page_b.keyboard.press("Enter")
        # The host's Start is enabled once the ready-up's re-broadcast
        # lands (a visible-but-disabled Start swallows the Enter).
        start_enabled = (
            "() => { const b = document.querySelector('#start-btn'); "
            "return b !== null && !b.disabled; }"
        )
        page_a.wait_for_function(start_enabled, timeout=20_000)
        page_a.get_by_role("button", name="Start").focus()
        page_a.keyboard.press("Enter")

        for page in (page_a, page_b):
            page.locator("#kcanvas").wait_for(state="attached", timeout=30_000)
        assert app_screen(page_a) == "playing"
        assert app_screen(page_b) == "playing"

        # The second tab's sampler starts now (a browser-side rAF
        # promise — live while the choreography's evaluate blocks this
        # one); the host's tab drives the soup (the ~60 s run, the
        # us03 margins).
        assert page_b.evaluate(SAMPLER_JS) == "started"
        page_a.set_default_timeout(300_000)
        try:
            page_a.evaluate(SOUP_CHOREOGRAPHY)
        finally:
            page_a.set_default_timeout(30_000)

        # The host is on the result screen (the round frame landed).
        # The window covers the 60 s buzzer: with the injected 60 ms
        # lag the host's *own* view trails the server truth, so the
        # soup's final serve press can miss the window's reach zone
        # and the round runs to the buzzer (a LOST is as good as a
        # WON for the smoothness and convergence assertions).
        page_a.set_default_timeout(90_000)
        try:
            page_a.locator("#result-word").wait_for(state="visible", timeout=65_000)
        finally:
            page_a.set_default_timeout(30_000)
        assert app_screen(page_a) == "round_end"
        convergence_start = time.monotonic()

        # The second tab's sampler resolves on *its* result screen.
        page_b.set_default_timeout(120_000)
        try:
            samples = page_b.evaluate("() => window.__uc_sampling")
        finally:
            page_b.set_default_timeout(30_000)
        convergence_s = time.monotonic() - convergence_start

        page_b.locator("#result-word").wait_for(state="visible", timeout=10_000)
        assert app_screen(page_b) == "round_end"
        assert convergence_s <= 1.0, (
            f"the second tab's result screen came {convergence_s:.2f} s after the host's "
            "(the 60 ms latency round must converge within the second)"
        )
        _assert_stream(samples)
    finally:
        context_a.close()
        context_b.close()