teng-lin--notebooklm-py/tests/unit/test_runtime_lifecycle.py

"""Unit tests for :mod:`notebooklm._runtime.lifecycle`.

Covers the load-bearing behaviors of :class:`ClientLifecycle` directly, in
addition to the existing ``Session``-shaped tests in
``test_session_close.py`` / ``test_client_keepalive.py`` / ``test_vcr_config.py``
which exercise the same helper through the compat facade.

Specifically pinned here:

* :meth:`ClientLifecycle.open` is **idempotent** — a second call while the
  client is already open is a no-op (the first ``httpx.AsyncClient`` instance
  is preserved).
* :meth:`ClientLifecycle.close` **cancels and awaits the keepalive task
  cleanly** — the task exits and is set to ``None``; the call doesn't leak a
  ``CancelledError``.
* ``_bound_loop`` **mismatch raises ``RuntimeError``** — the cross-loop guard
  in :meth:`RuntimeTransport.perform_authed_post` reads ``_bound_loop`` through
  the lifecycle and raises actionably when the loops differ.
* :meth:`ClientLifecycle.save_cookies` **invokes** the
  :class:`CookiePersistence` collaborator's ``save`` method with the right
  ``jar`` and ``path`` arguments AND with the ``save_cookies_to_storage``
  value resolved from ``notebooklm._auth.storage`` at call time.
* The httpx ``AsyncClient`` **always uses httpx's default transport** —
  Tier-12 PR 12.6 lifted synthetic-error injection into the chain
  (:class:`notebooklm._middleware.error_injection.ErrorInjectionMiddleware`)
  and PR 12.9 deleted the legacy ``_SyntheticErrorTransport`` class.
  The lifecycle constructs a plain transport regardless of
  ``NOTEBOOKLM_VCR_RECORD_ERRORS``.
* :meth:`ClientLifecycle._keepalive_loop` **respects the min-interval
  clamp** — ``_resolve_keepalive_interval`` floors the configured interval
  at ``keepalive_min_interval`` so a sub-floor user value gets bumped up.

Tests are intentionally helper-shaped (instantiate :class:`ClientLifecycle`
directly with a stub collaborator bundle) so they cover the lifecycle
without taking on a ``Session`` dependency. Wave 2 of plan
``host-protocol-removal`` narrowed the lifecycle method signatures from
the legacy Session-shaped ``host`` Protocol to explicit keyword-only
collaborators; the :class:`_StubHost` fixture now serves purely as a
convenience bundle, paired with module-level :func:`_open` / :func:`_close`
adapters that unpack the bundle into the new kwarg shape.
"""

from __future__ import annotations

import asyncio
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock

import httpx
import pytest

from notebooklm._runtime.helpers import _resolve_keepalive_interval
from notebooklm._runtime.lifecycle import (
    ClientLifecycle,
    _default_cookie_rotator,
    _default_cookie_saver,
)
from notebooklm._transport_drain import TransportDrainTracker
from notebooklm.auth import AuthTokens
from notebooklm.types import ConnectionLimits
from tests._helpers.client_factory import build_client_shell_for_tests


class _StubHost:
    """Test-side collaborator bundle for :class:`ClientLifecycle` unit tests.

    Wave 2 of plan ``host-protocol-removal`` narrowed the four
    :class:`ClientLifecycle` methods to take explicit keyword-only
    collaborators rather than a Session-shaped ``host`` Protocol. This
    fixture survives as a convenience bundle — it holds the same set of
    stub collaborators every lifecycle test needs (auth, drain tracker,
    auth coordinator, reqid counter, cookie persistence) in one place,
    so each test can do ``lifecycle.open(auth=host.auth,
    drain_tracker=host._drain_tracker, ...)`` without re-building five
    mocks at every call site.

    Mirrors the live ``Session`` attribute names for grep continuity:

    * ``auth`` — a real :class:`AuthTokens` so :meth:`ClientLifecycle.open`
      can read ``cookies`` / ``cookie_jar`` / ``storage_path``.
    * ``_drain_tracker`` / ``_auth_coord`` / ``_reqid`` — ``MagicMock``s;
      the lifecycle calls ``_drain_tracker.reset_after_open()`` (Wave 1
      of host-protocol-removal replaced the legacy direct
      ``_drain_tracker._draining = False`` write) and ``set_bound_loop``
      on each of the three helpers (drain / reqid / auth_coord) from the
      open() path so cross-loop misuse can be caught.
    * ``cookie_persistence`` — a ``MagicMock`` with an async ``save``
      coroutine; assertions check it was called with the right args.
    * ``_drain_tracker.run_drain_hooks`` — called by close(); set to an
      ``AsyncMock`` so tests can assert it ran and inspect call order.

    Stage B1 PR 2 of the post-refactoring plan dropped the close-time
    ``host._rpc_executor = None`` line from
    :meth:`ClientLifecycle.close` — the executor now persists across
    ``close()`` → ``open()`` cycles. The corresponding sentinel and the
    ``test_close_nulls_rpc_executor`` regression test were removed in
    that PR; see :mod:`tests.unit.test_lifecycle_executor_reuse` for
    the replacement contract.
    """

    def __init__(self) -> None:
        self.auth = AuthTokens(
            csrf_token="CSRF",
            session_id="SID",
            cookies={"SID": "v1"},
            storage_path=None,
        )
        self._drain_tracker = MagicMock()
        # ``open()`` calls ``drain_tracker.reset_after_open()`` (Wave 1 of
        # host-protocol-removal — the encapsulated form of the legacy
        # ``_drain_tracker._draining = False`` write). The ``MagicMock``
        # default lets the call land without configuring a side effect; the
        # invocation is asserted by ``test_open_captures_bound_loop_and_resets_drain``.
        # Seed ``_draining = True`` so a future regression that re-introduces
        # a direct field read in the lifecycle would still see "drained".
        self._drain_tracker._draining = True
        # Wave 2 of session-decoupling: drain hooks live on the tracker.
        # ``close()`` calls ``drain_tracker.run_drain_hooks()`` so the mock
        # needs an async implementation.
        self._drain_tracker.run_drain_hooks = AsyncMock()
        self._auth_coord = MagicMock()
        # Wave 1 of host-protocol-removal: ``close()`` no longer reads the
        # private ``_refresh_task`` slot directly — it calls the awaitable
        # ``cancel_inflight_refresh`` method on the coordinator. Default
        # ``MagicMock()`` would return a non-awaitable, so the stub needs an
        # ``AsyncMock`` for that method. The real coordinator handles the
        # no-op / already-done / in-flight branches internally (covered by
        # the focused unit tests in ``tests/unit/test_runtime_auth.py``).
        # ``_refresh_task`` is kept as ``None`` on the stub for
        # forward-compatibility with any future test that probes the slot
        # directly (the lifecycle itself no longer reads it).
        self._auth_coord._refresh_task = None
        self._auth_coord.cancel_inflight_refresh = AsyncMock()
        # ``_reqid`` is targeted by ``set_bound_loop`` from open() (P0-2).
        self._reqid = MagicMock()
        # ``open()`` also propagates the bound loop into the composition
        # holder and resets the lazy RPC semaphore (issue #1169): it calls
        # ``composed.set_bound_loop(loop)`` and ``composed.reset_after_open()``
        # so a client reopened on a different loop rebuilds the semaphore on
        # the new loop. The ``MagicMock`` default lets both calls land
        # without configuring side effects; the invocations are asserted by
        # ``test_open_captures_bound_loop_and_resets_drain``.
        self._composed = MagicMock()
        # ``open()`` also propagates the bound loop into the Sources upload
        # pipeline and resets its lazy upload semaphore (issue #1196 upload
        # variant): it calls ``uploader.set_bound_loop(loop)`` and
        # ``uploader.reset_after_open()`` so a client reopened on a different
        # loop rebuilds the upload semaphore on the new loop. The ``MagicMock``
        # default lets both calls land without configuring side effects; the
        # invocations are asserted by
        # ``test_open_captures_bound_loop_and_resets_drain``.
        self._uploader = MagicMock()
        # ``open()`` also propagates the bound loop into the ChatAPI and resets
        # its lazy per-conversation / per-notebook lock maps (#1225): it calls
        # ``chat.set_bound_loop(loop)`` and ``chat.reset_after_open()`` so a
        # client reopened on a different loop rebuilds the conversation locks on
        # the new loop. The ``MagicMock`` default lets both calls land without
        # configuring side effects; the invocations are asserted by
        # ``test_open_captures_bound_loop_and_resets_drain``.
        self._chat = MagicMock()
        self.cookie_persistence = MagicMock()
        self.cookie_persistence.save = AsyncMock()
        self.cookie_persistence.capture_open_snapshot = MagicMock()
        # Stage B1 PR 2 dropped the close-time null on ``_rpc_executor``;
        # the slot is left as-set by the composition root. Set a stable
        # sentinel here in case future regression tests want to assert
        # the value is untouched across an open/close cycle. The lifecycle
        # itself no longer reads this slot.
        self._rpc_executor: Any = "RPC_EXECUTOR_SENTINEL"


def _make_lifecycle(
    *,
    keepalive_interval: float | None = None,
    keepalive_storage_path: Path | None = None,
) -> ClientLifecycle:
    """Construct a :class:`ClientLifecycle` with defaults safe for unit tests.

    Default ``keepalive_interval=None`` means no background keepalive task is
    spawned on :meth:`open` — tests that want the task pass an interval
    explicitly.
    """
    return ClientLifecycle(
        timeout=30.0,
        connect_timeout=10.0,
        limits=ConnectionLimits(),
        keepalive_interval=keepalive_interval,
        keepalive_storage_path=keepalive_storage_path,
    )


async def _open(lifecycle: ClientLifecycle, host: _StubHost) -> None:
    """Adapter that forwards a :class:`_StubHost` bundle into the new
    explicit-kwargs :meth:`ClientLifecycle.open` signature.

    Wave 2 of plan ``host-protocol-removal`` narrowed the lifecycle to
    take collaborators by keyword instead of a Session-shaped host. The
    test fixtures still bundle the collaborators into a stub for
    convenience; this helper bridges the two shapes so each test stays
    a single readable line.
    """
    await lifecycle.open(
        auth=host.auth,
        drain_tracker=host._drain_tracker,
        auth_coord=host._auth_coord,
        reqid=host._reqid,
        cookie_persistence=host.cookie_persistence,
        composed=host._composed,
        uploader=host._uploader,
        chat=host._chat,
    )


async def _close(lifecycle: ClientLifecycle, host: _StubHost) -> None:
    """Adapter for :meth:`ClientLifecycle.close` — see :func:`_open`."""
    await lifecycle.close(
        auth_coord=host._auth_coord,
        drain_tracker=host._drain_tracker,
        cookie_persistence=host.cookie_persistence,
    )


# ---------------------------------------------------------------------------
# open() — idempotency, bound-loop capture, AsyncClient construction
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_open_idempotent_preserves_existing_client() -> None:
    """Second ``open()`` while already open is a no-op — same ``httpx.AsyncClient``."""
    lifecycle = _make_lifecycle()
    host = _StubHost()

    await _open(lifecycle, host)
    first_client = lifecycle._http_client
    assert first_client is not None
    assert lifecycle.is_open()

    await _open(lifecycle, host)
    second_client = lifecycle._http_client

    assert second_client is first_client, (
        "open() must be idempotent — re-opening on an already-open lifecycle "
        "should preserve the existing AsyncClient instance, not build a fresh one."
    )

    await _close(lifecycle, host)


@pytest.mark.asyncio
async def test_open_captures_bound_loop_and_resets_drain() -> None:
    """``open()`` binds the running loop and calls ``reset_after_open`` on the tracker.

    Wave 1 of plan ``host-protocol-removal`` encapsulated the legacy
    direct write ``host._drain_tracker._draining = False`` behind
    :meth:`TransportDrainTracker.reset_after_open`. The lifecycle's
    obligation is now to CALL that method on every ``open()``; the
    method's own behavior (clearing ``_draining`` while leaving
    in-flight counters intact) is pinned by the focused unit tests
    further down in this file.

    Stubs ``host._drain_tracker`` as a ``MagicMock`` so this test
    captures the call without depending on a real
    :class:`TransportDrainTracker`. The companion full-stack
    open-then-close test that exercises a real tracker lives in
    ``tests/integration/`` (and the AST-guarded lint forbids any
    lifecycle code from writing to ``_draining`` directly outside the
    tracker itself, see the acceptance-criteria ``rg`` check in the
    plan).
    """
    lifecycle = _make_lifecycle()
    host = _StubHost()
    assert lifecycle._bound_loop is None

    await _open(lifecycle, host)

    assert lifecycle._bound_loop is asyncio.get_running_loop()
    assert lifecycle.get_bound_loop() is asyncio.get_running_loop()
    host._drain_tracker.reset_after_open.assert_called_once_with()
    # Issue #1169: the composition holder is the fourth loop-bound primitive
    # and must receive the same set_bound_loop / reset_after_open treatment as
    # the drain tracker so the lazy RPC semaphore rebinds on close→reopen.
    host._composed.set_bound_loop.assert_called_once_with(asyncio.get_running_loop())
    host._composed.reset_after_open.assert_called_once_with()
    # Issue #1196 upload variant: the Sources upload pipeline is the second
    # lazily-built loop-bound semaphore and must receive the same
    # set_bound_loop / reset_after_open treatment so the upload semaphore
    # rebinds on close→reopen.
    host._uploader.set_bound_loop.assert_called_once_with(asyncio.get_running_loop())
    host._uploader.reset_after_open.assert_called_once_with()
    # Issue #1225: the ChatAPI conversation locks are the last lazily-built
    # loop-bound primitives and must receive the same set_bound_loop /
    # reset_after_open treatment so the per-conversation / per-notebook locks
    # rebind on close→reopen.
    host._chat.set_bound_loop.assert_called_once_with(asyncio.get_running_loop())
    host._chat.reset_after_open.assert_called_once_with()

    await _close(lifecycle, host)


@pytest.mark.asyncio
async def test_open_close_open_rebinds_loop() -> None:
    """``close()`` does not unbind, but a subsequent ``open()`` re-captures
    the current loop (used by clients that close + re-open within one loop)."""
    lifecycle = _make_lifecycle()
    host = _StubHost()

    await _open(lifecycle, host)
    bound_after_first_open = lifecycle._bound_loop
    await _close(lifecycle, host)

    # close() does NOT clear _bound_loop — the cross-loop guard fires on the
    # next call against a different loop if the user mistakenly hands the
    # client off after close.
    assert lifecycle._bound_loop is bound_after_first_open
    assert lifecycle.is_open() is False

    # Re-open on the same loop. New AsyncClient instance; same bound loop.
    await _open(lifecycle, host)
    assert lifecycle._bound_loop is asyncio.get_running_loop()
    assert lifecycle.is_open() is True
    await _close(lifecycle, host)


@pytest.mark.asyncio
async def test_open_captures_cookie_snapshot() -> None:
    """``open()`` calls ``cookie_persistence.capture_open_snapshot`` with the
    live ``httpx.Cookies`` jar AFTER the AsyncClient is built — preserving
    the contract that the open-time baseline reflects httpx-normalized
    domains.
    """
    lifecycle = _make_lifecycle()
    host = _StubHost()

    await _open(lifecycle, host)
    try:
        host.cookie_persistence.capture_open_snapshot.assert_called_once()
        passed_jar = host.cookie_persistence.capture_open_snapshot.call_args.args[0]
        # The jar passed to capture is the AsyncClient's live jar.
        assert passed_jar is lifecycle._http_client.cookies  # type: ignore[union-attr]
    finally:
        await _close(lifecycle, host)


# ---------------------------------------------------------------------------
# Synthetic-error injection — lifted to the chain in PR 12.6
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_open_uses_default_httpx_transport_by_default(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Default path: httpx's default ``AsyncHTTPTransport`` is in place
    (no custom transport wrapping). Post-Tier-12 the synthetic-error
    substitution lives in ``ErrorInjectionMiddleware``; the lifecycle
    constructs a plain transport regardless of any env var, so the test
    asserts the lifecycle's transport construction directly without
    monkeypatching the now-middleware-only error-injection seam.
    """
    from notebooklm import _error_injection

    monkeypatch.setattr(_error_injection, "_get_error_injection_mode", lambda: None)
    lifecycle = _make_lifecycle()
    host = _StubHost()

    await _open(lifecycle, host)
    try:
        client = lifecycle._http_client
        assert client is not None
        assert isinstance(client._transport, httpx.AsyncHTTPTransport)
    finally:
        await _close(lifecycle, host)


@pytest.mark.asyncio
async def test_open_uses_default_httpx_transport_when_env_var_set(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``AsyncClient`` uses httpx's default transport even with env var set.

    Pre-Tier-12 the lifecycle wrapped the inner transport in a synthetic
    httpx transport (deleted in PR 12.9). After Tier-12 the substitution
    lives in the chain (``ErrorInjectionMiddleware``); the lifecycle
    constructs a plain transport regardless of the env var.
    """
    from notebooklm import _error_injection

    monkeypatch.setattr(_error_injection, "_get_error_injection_mode", lambda: "429")
    lifecycle = _make_lifecycle()
    host = _StubHost()

    await _open(lifecycle, host)
    try:
        client = lifecycle._http_client
        assert client is not None
        assert isinstance(client._transport, httpx.AsyncHTTPTransport)
    finally:
        await _close(lifecycle, host)


# ---------------------------------------------------------------------------
# close() — keepalive cancellation, sentinel null-out, idempotency
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_close_cancels_keepalive_cleanly() -> None:
    """``close()`` cancels and awaits the keepalive task; no leaked exception.

    Uses a very short interval (the lifecycle does not re-clamp; the caller
    is expected to have passed the pre-clamped value) so the task has had a
    chance to park on its ``asyncio.sleep`` before close() cancels it.
    """
    lifecycle = _make_lifecycle(keepalive_interval=0.01)
    host = _StubHost()

    await _open(lifecycle, host)
    task = lifecycle._keepalive_task
    assert task is not None
    assert not task.done()

    # Yield once so the keepalive task actually parks on its sleep.
    await asyncio.sleep(0)

    await _close(lifecycle, host)
    assert lifecycle._keepalive_task is None, (
        "close() must null out _keepalive_task after the cancel+gather."
    )
    assert task.cancelled() or task.done(), (
        "keepalive task should be finished (cancelled) after close()."
    )


@pytest.mark.asyncio
async def test_close_when_never_opened_is_noop() -> None:
    """Closing a never-opened lifecycle is safe and does nothing harmful."""
    lifecycle = _make_lifecycle()
    host = _StubHost()

    # No exception, no state churn beyond what's already None/sentinel.
    await _close(lifecycle, host)
    assert lifecycle._http_client is None
    assert lifecycle._keepalive_task is None


@pytest.mark.asyncio
async def test_close_runs_drain_hooks_before_transport_teardown() -> None:
    """``close()`` invokes ``run_drain_hooks`` on the tracker before tearing down the HTTP client.

    Wave 2 of session-decoupling: drain hooks live on ``TransportDrainTracker``;
    the lifecycle just calls ``host._drain_tracker.run_drain_hooks()`` and the
    tracker handles the firing + exception suppression.
    """
    lifecycle = _make_lifecycle()
    host = _StubHost()

    # Record ordering: drain hooks must run *before* the HTTP client teardown
    # (so a hook that needs the live client — e.g. an in-flight cookie save —
    # can still see it).
    events: list[str] = []

    async def fake_run_drain_hooks() -> None:
        assert lifecycle._http_client is not None, (
            "drain hooks must run while the HTTP client is still open"
        )
        events.append("run_drain_hooks")

    host._drain_tracker.run_drain_hooks = fake_run_drain_hooks

    original_aclose = lifecycle._kernel.aclose

    async def recording_aclose() -> None:
        events.append("kernel_aclose")
        await original_aclose()

    lifecycle._kernel.aclose = recording_aclose  # type: ignore[method-assign]

    await _open(lifecycle, host)
    await _close(lifecycle, host)

    assert events == ["run_drain_hooks", "kernel_aclose"], (
        f"close() must run drain hooks before kernel.aclose(); got {events}"
    )
    assert lifecycle._http_client is None


# ---------------------------------------------------------------------------
# save_cookies — invokes cookie_persistence with right args
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_save_cookies_invokes_cookie_persistence(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """``save_cookies(host, jar, path)`` delegates to
    ``host.cookie_persistence.save(...)``, forwarding the lifecycle's
    ``_cookie_saver`` wrapper as the storage writer.

    Phase 2 PR 3 introduced an injectable ``cookie_saver`` seam; the
    default ``_default_cookie_saver`` wrapper still late-binds at call
    time so a swap of the canonical ``_auth.storage.save_cookies_to_storage``
    attribute fires through. (Phase 4 retargeted the wrapper's late-bind
    from ``notebooklm._core`` to ``notebooklm._auth.storage`` when the
    ``_core`` compatibility shim was deleted.) This assertion is BEHAVIORAL
    (invoke the wrapper, observe the sentinel was called) rather than
    identity-based, because the wrapper indirection is the whole point of
    the seam.
    """
    from notebooklm._auth import storage as storage_module

    sentinel = MagicMock()
    monkeypatch.setattr(storage_module, "save_cookies_to_storage", sentinel)

    lifecycle = _make_lifecycle()
    host = _StubHost()
    jar = httpx.Cookies()
    jar.set("SID", "v2", domain=".google.com")
    target_path = tmp_path / "storage_state.json"

    await lifecycle.save_cookies(host.cookie_persistence, jar, target_path)

    host.cookie_persistence.save.assert_awaited_once()
    call = host.cookie_persistence.save.call_args
    assert call.args[0] is jar
    assert call.args[1] == target_path
    # The kwarg is the lifecycle's wrapper (not the raw sentinel), so the
    # ``CookiePersistence._save`` worker-thread invocation goes through
    # ``_default_cookie_saver``'s late-bound ``_auth.storage`` lookup.
    forwarded_saver = call.kwargs["save_cookies_to_storage"]
    assert forwarded_saver is lifecycle._cookie_saver, (
        "lifecycle.save_cookies must forward self._cookie_saver as the "
        "storage writer (the wrapper indirection is what preserves the "
        "canonical monkeypatch surface)."
    )
    # Behavioral check: invoking the captured wrapper hits the monkeypatched
    # sentinel via late-bound canonical-module resolution.
    forwarded_saver(jar, target_path)
    sentinel.assert_called_once_with(jar, target_path)
    assert call.kwargs["to_thread"] is asyncio.to_thread


# ---------------------------------------------------------------------------
# _bound_loop accessor + cross-loop guard
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_bound_loop_get_returns_running_loop_after_open() -> None:
    """``get_bound_loop()`` returns the captured loop after open().

    The cross-loop affinity ``RuntimeError`` is raised by
    ``RuntimeTransport.perform_authed_post`` on actual cross-loop reuse —
    see ``tests/integration/concurrency/test_cross_loop_affinity.py`` for
    the end-to-end exercise. Here we only assert the lifecycle exposes the
    captured loop via :meth:`get_bound_loop`.
    """
    lifecycle = _make_lifecycle()
    host = _StubHost()

    assert lifecycle.get_bound_loop() is None
    await _open(lifecycle, host)
    try:
        assert lifecycle.get_bound_loop() is asyncio.get_running_loop()
    finally:
        await _close(lifecycle, host)


def test_bound_loop_mismatch_via_session_raises_runtime_error() -> None:
    """Cross-loop reuse of a single NotebookLMClient raises cleanly.

    The ``RuntimeError`` appears on the second loop's first authed POST. The
    test runs two separate ``asyncio.run`` invocations to materialise two
    distinct loops.
    """

    auth = AuthTokens(csrf_token="CSRF", session_id="SID", cookies={"SID": "v1"})
    core = build_client_shell_for_tests(auth=auth)

    async def _open_on_loop_a() -> None:
        await core.__aenter__()
        # We deliberately do NOT call core.close() because close() resets
        # _http_client (which would let loop B's open() re-bind the loop
        # and skip the guard). The whole point is that the guard fires when
        # _bound_loop is set from a different loop and a request is attempted
        # without an intervening close().

    def _build_request_stub(snapshot: Any) -> tuple[httpx.Request, Any]:
        return (
            httpx.Request(
                "POST",
                "https://notebooklm.google.com/_/LabsTailwindUi/data/batchexecute",
            ),
            None,
        )

    async def _attempt_post_on_loop_b() -> Exception | None:
        # ``open()`` is idempotent — since loop A left ``_http_client``
        # populated, this is a no-op and ``_bound_loop`` stays bound to loop A.
        await core.__aenter__()
        try:
            await core._composed.transport.perform_authed_post(
                build_request=_build_request_stub,
                log_label="test.cross_loop",
            )
        except RuntimeError as exc:
            return exc
        return None

    asyncio.run(_open_on_loop_a())
    exc = asyncio.run(_attempt_post_on_loop_b())
    assert isinstance(exc, RuntimeError), (
        f"Cross-loop authed POST must raise RuntimeError; got {exc!r}"
    )
    # The guard's message mentions the loop affinity invariant — match a
    # stable substring rather than the exact phrasing.
    assert "loop" in str(exc).lower(), f"Unexpected RuntimeError text: {exc!r}"


# ---------------------------------------------------------------------------
# _resolve_keepalive_interval clamping
# ---------------------------------------------------------------------------


def test_resolve_keepalive_interval_clamps_to_min_floor() -> None:
    """``_resolve_keepalive_interval`` floors a too-small user value at
    ``min_interval`` — preserving the "accidentally rate-limiting Google's
    identity surface" guard the lifecycle inherits from the resolver.

    The resolver now lives in ``notebooklm._runtime.helpers``; this test belongs
    alongside the lifecycle suite because the
    clamped value is what the lifecycle stores in ``_keepalive_interval``.
    """
    # User asks for 1s — much lower than the 60s default floor.
    resolved = _resolve_keepalive_interval(keepalive=1.0, min_interval=60.0)
    assert resolved == 60.0


def test_resolve_keepalive_interval_passes_through_above_floor() -> None:
    """A user value above the floor passes through unchanged."""
    resolved = _resolve_keepalive_interval(keepalive=120.0, min_interval=60.0)
    assert resolved == 120.0


def test_resolve_keepalive_interval_none_disables() -> None:
    """``None`` disables the keepalive (no background task spawned)."""
    resolved = _resolve_keepalive_interval(keepalive=None, min_interval=60.0)
    assert resolved is None


def test_resolve_keepalive_interval_rejects_non_positive() -> None:
    """Zero / negative / NaN values raise ``ValueError`` instead of silently
    disabling — surface misconfiguration loudly at construction time."""
    with pytest.raises(ValueError):
        _resolve_keepalive_interval(keepalive=0, min_interval=60.0)
    with pytest.raises(ValueError):
        _resolve_keepalive_interval(keepalive=-1.0, min_interval=60.0)
    with pytest.raises(ValueError):
        _resolve_keepalive_interval(keepalive=1.0, min_interval=0)


# ---------------------------------------------------------------------------
# Construction-time invariants
# ---------------------------------------------------------------------------


def test_init_is_event_loop_agnostic() -> None:
    """Constructing a ``ClientLifecycle`` outside a running loop must not
    raise. The helper stores only plain values and ``None`` placeholders;
    the ``httpx.AsyncClient`` and keepalive task are deferred to ``open()``.
    """
    # Outside ``asyncio.run`` — no running loop available.
    lifecycle = ClientLifecycle(
        timeout=30.0,
        connect_timeout=10.0,
        limits=ConnectionLimits(),
        keepalive_interval=60.0,
        keepalive_storage_path=Path("/tmp/storage.json"),
    )
    assert lifecycle._http_client is None
    assert lifecycle._bound_loop is None
    assert lifecycle._keepalive_task is None
    assert lifecycle._keepalive_interval == 60.0
    assert lifecycle._keepalive_storage_path == Path("/tmp/storage.json")
    assert lifecycle._timeout == 30.0
    assert lifecycle._connect_timeout == 10.0
    assert lifecycle.is_open() is False
    assert lifecycle.get_bound_loop() is None


# ---------------------------------------------------------------------------
# Injectable seams
#
# Three load-bearing properties pinned here:
#
# 1. ``_default_cookie_saver`` performs a late-bound
#    ``_auth.storage.save_cookies_to_storage`` lookup inside its function body.
#    Monkeypatching the canonical storage seam AFTER the wrapper exists must
#    still affect the wrapper's behavior.
#
# 2. ``_default_cookie_rotator`` performs the same late-bound lookup for
#    ``_auth.keepalive._rotate_cookies``. The keepalive-loop equivalent of (1).
#
# 3. ``ClientLifecycle.__init__`` wires the defaults when ``cookie_saver`` /
#    ``cookie_rotator`` are ``None`` (or omitted), and accepts custom
#    callables when supplied. The ``or _default_*`` resolution pattern is what
#    lets the production assembly path omit custom seam callables when no
#    override is supplied.
# ---------------------------------------------------------------------------


def test_default_cookie_saver_late_binds_to_canonical_seam(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``_default_cookie_saver`` resolves
    ``_auth.storage.save_cookies_to_storage`` at CALL time, not at
    module-import time.

    Establish a sentinel AFTER ``_default_cookie_saver`` already exists,
    then invoke the wrapper and prove the sentinel was called. A non-late-
    bound wrapper would have captured the original ``save_cookies_to_storage``
    reference at module load and silently ignored the monkeypatch.
    (Phase 4 retargeted the late-bind from ``notebooklm._core`` to
    ``notebooklm._auth.storage`` when the ``_core`` compatibility shim
    was deleted.)
    """
    from notebooklm._auth import storage as storage_module

    sentinel = MagicMock(return_value=True)
    monkeypatch.setattr(storage_module, "save_cookies_to_storage", sentinel)

    jar = httpx.Cookies()
    path = Path("/tmp/storage.json")
    result = _default_cookie_saver(jar, path)

    sentinel.assert_called_once_with(jar, path)
    assert result is True


@pytest.mark.asyncio
async def test_default_cookie_rotator_late_binds_to_canonical_seam(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``_default_cookie_rotator`` resolves
    ``_auth.keepalive._rotate_cookies`` at CALL time and awaits it.
    Async-shape counterpart to the saver test. (Phase 4 retargeted the
    late-bind from ``notebooklm._core`` to ``notebooklm._auth.keepalive``
    when the ``_core`` compatibility shim was deleted.)
    """
    from notebooklm._auth import keepalive as keepalive_module

    sentinel = AsyncMock(return_value=None)
    monkeypatch.setattr(keepalive_module, "_rotate_cookies", sentinel)

    client = MagicMock(spec=httpx.AsyncClient)
    path = Path("/tmp/storage.json")
    await _default_cookie_rotator(client, path)

    sentinel.assert_awaited_once_with(client, path)


def test_init_wires_default_seams_when_none_supplied() -> None:
    """When ``cookie_saver`` / ``cookie_rotator`` are omitted (or ``None``),
    ``ClientLifecycle.__init__`` wires the module-level late-binding
    defaults; supplying custom callables overrides them.

    This is what lets the production assembly path omit custom seam callables
    when no override is supplied: it constructs ``ClientLifecycle(...)``
    without the new kwargs, and the ``or _default_*`` resolution preserves the
    canonical late-bound seams.
    """
    # Defaults: omit the kwargs entirely.
    default_lifecycle = ClientLifecycle(
        timeout=30.0,
        connect_timeout=10.0,
        limits=ConnectionLimits(),
        keepalive_interval=None,
        keepalive_storage_path=None,
    )
    assert default_lifecycle._cookie_saver is _default_cookie_saver
    assert default_lifecycle._cookie_rotator is _default_cookie_rotator

    # Explicit ``None`` resolves the same way as omission.
    explicit_none_lifecycle = ClientLifecycle(
        timeout=30.0,
        connect_timeout=10.0,
        limits=ConnectionLimits(),
        keepalive_interval=None,
        keepalive_storage_path=None,
        cookie_saver=None,
        cookie_rotator=None,
    )
    assert explicit_none_lifecycle._cookie_saver is _default_cookie_saver
    assert explicit_none_lifecycle._cookie_rotator is _default_cookie_rotator

    # Custom callables override the defaults — pure pass-through, no
    # ``_core`` indirection.
    custom_saver = MagicMock(return_value=True)
    custom_rotator = AsyncMock(return_value=None)
    custom_lifecycle = ClientLifecycle(
        timeout=30.0,
        connect_timeout=10.0,
        limits=ConnectionLimits(),
        keepalive_interval=None,
        keepalive_storage_path=None,
        cookie_saver=custom_saver,
        cookie_rotator=custom_rotator,
    )
    assert custom_lifecycle._cookie_saver is custom_saver
    assert custom_lifecycle._cookie_rotator is custom_rotator


# ---------------------------------------------------------------------------
# TransportDrainTracker.reset_after_open — Wave 1 of host-protocol-removal
# encapsulated ``host._drain_tracker._draining = False`` (previously written
# directly by ``ClientLifecycle.open``) behind a method on the tracker. The
# method is intentionally narrow: it clears ONLY the ``_draining`` flag and
# leaves in-flight counters / depth maps untouched. These two tests pin
# both halves of that contract on the collaborator directly so a regression
# in the tracker is caught without driving the full lifecycle open() path.
# ---------------------------------------------------------------------------


def test_drain_tracker_reset_after_open_clears_draining_flag() -> None:
    """``reset_after_open`` flips ``_draining`` from ``True`` back to ``False``.

    Pins the encapsulation of the legacy
    ``host._drain_tracker._draining = False`` write performed inside
    ``ClientLifecycle.open``. The behavior is the same — a tracker that was
    previously drained, then re-opened, admits new top-level work again —
    just routed through a named method instead of a direct field write.
    """
    tracker = TransportDrainTracker()
    tracker._draining = True

    tracker.reset_after_open()

    assert tracker._draining is False


@pytest.mark.asyncio
async def test_drain_tracker_reset_after_open_does_not_touch_inflight_counts() -> None:
    """``reset_after_open`` ONLY clears ``_draining`` — every other piece
    of bookkeeping state is left intact.

    Pins the "intentionally narrow" half of the encapsulation. If a
    well-meaning maintainer "helpfully" expanded this method to also zero
    ``_in_flight_posts`` or reset / clear ``_operation_depths``, the
    load-bearing in-flight invariants asserted by
    ``tests/unit/test_observability.py::test_drain_allows_nested_work_inside_accepted_operation``
    and ``tests/unit/concurrency/test_close_cancellation_leak.py`` would
    break. This regression test catches that expansion at the tracker
    level so the failure points at the right code rather than surfacing
    as a confusing in-flight count mismatch elsewhere.

    Seeds an actual operation-depth entry (not just an identity-stable
    empty map) so a regression that calls
    ``self._operation_depths.clear()`` — which would preserve map identity
    but wipe the contents — also fails this test. Async-marked so the
    seeded task ``asyncio.current_task()`` returns a real task to key the
    WeakKeyDictionary on.
    """
    tracker = TransportDrainTracker()
    tracker._draining = True
    # Seed non-default values so the assertions below would fail loudly
    # if ``reset_after_open`` overwrote them.
    tracker._in_flight_posts = 3
    seed_task = asyncio.current_task()
    assert seed_task is not None, "test runs under @pytest.mark.asyncio"
    tracker._operation_depths[seed_task] = 2
    pre_depths_id = id(tracker._operation_depths)

    async def _drain_hook() -> None:
        return None

    tracker.register_drain_hook("seed_hook", _drain_hook)

    tracker.reset_after_open()

    assert tracker._draining is False, "the one flag this method *does* clear"
    assert tracker._in_flight_posts == 3, (
        "reset_after_open must not touch _in_flight_posts — clearing it "
        "would lose track of in-flight operations and let drain() return "
        "prematurely on the next close()."
    )
    assert id(tracker._operation_depths) == pre_depths_id, (
        "reset_after_open must preserve the _operation_depths "
        "WeakKeyDictionary identity — replacing it would orphan per-task "
        "depth bookkeeping for already-admitted operations."
    )
    assert tracker._operation_depths.get(seed_task) == 2, (
        "reset_after_open must not clear() _operation_depths contents — "
        "a regression that wiped per-task depths would reject already-"
        "admitted nested operations after the next open()."
    )
    assert "seed_hook" in tracker._drain_hooks, (
        "reset_after_open must not touch registered drain hooks; feature "
        "code registers them at construction-time on the tracker."
    )