"""Worldtree Conversation API session-lifecycle client.

Implements docs/contracts/issues/2.contract.md.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

import httpx


@dataclass(frozen=True)
class SessionInfo:
    """Worldtree session envelope; shared shape for create + list responses.

    INV-001 / INV-002: origin-conditional defaults — `create_session` sets fixed
    `name=None`, `archived=False`, `tags=[]`; `list_sessions` populates from the
    response item with the same absent/null defaults but `message_count=None`.
    """

    session_id: str
    agent_id: str
    created_at: str
    last_active: str
    metadata: dict[str, Any]
    message_count: int | None
    name: str | None
    archived: bool
    tags: list[str]


@dataclass(frozen=True)
class SessionPage:
    """One page of GET /sessions results. `next_cursor=None` on the last page."""

    items: list[SessionInfo]
    next_cursor: str | None


@dataclass(frozen=True)
class AgentInfo:
    """Worldtree agent envelope from GET /agents (issue #8).

    INV-005: required fields (`agent_id`, `name`, `description`) take the
    response value verbatim. Optional fields default to None / [] / {} when
    omitted by the server, mirroring SessionInfo's INV-001/INV-002
    origin-conditional defaulting.
    """

    agent_id: str
    name: str
    description: str
    version: str | None
    capabilities: list[str]
    supported_models: list[str]
    persona_traits: dict[str, Any]
    ui_hints: dict[str, Any]


@dataclass(frozen=True)
class BifrostBinding:
    """Session-create Bifrost binding (Worldtree BifrostBindingRequest, #160).

    Issue #17. `endpoint_url` is the WORLDTREE-VISIBLE base URL of ONE provider
    plane (memory :8391 / affect :8390). `scope` is an opaque pass-through copied
    into the handshake JWT unchanged (≤256 chars); ratatoskr does not interpret
    it and v1 sends None.
    """

    endpoint_url: str
    scope: str | None = None


class AgentNotFound(Exception):
    """Raised on HTTP 404 from POST /sessions — unknown agent_id."""

    def __init__(self, *, agent_id: str) -> None:
        super().__init__(f"unknown agent_id: {agent_id!r}")
        self.agent_id = agent_id


class InvalidCursor(Exception):
    """Raised on HTTP 422 cursor_invalid from GET /sessions."""

    def __init__(self, *, raw: str | None) -> None:
        super().__init__(f"server rejected cursor: {raw!r}")
        self.raw = raw


class SessionApiFailed(Exception):
    """Raised on unexpected response status from /sessions endpoints.

    `body` is truncated to ≤ 1024 bytes at construction (INV-004), consistent
    with the issue #1 precedent for `SseConnectFailed` / `CancelFailed`.
    """

    def __init__(self, *, status: int, body: bytes) -> None:
        body = body[:1024]
        super().__init__(f"sessions API failed: status={status}, body={body[:128]!r}")
        self.status = status
        self.body = body


# Issue #17 — Bifrost-bind failure modes on POST /sessions.
class BifrostConsumerKeyMissing(Exception):
    """A bifrost binding was requested without a consumer_key.

    Raised BEFORE any HTTP (INV-001): the bound create must never silently fall
    back to the client's default canary key — the consumer key IS the handshake
    identity Worldtree signs the Bifrost JWT with.
    """

    def __init__(self) -> None:
        super().__init__(
            "bifrost binding requires a non-empty consumer_key; refusing to "
            "fall back to the canary key (INV-001)"
        )


class BifrostHandshakeFailed(Exception):
    """Raised on HTTP 502 `bifrost_handshake_failed` from a bound POST /sessions.

    INV-002: the Bifrost handshake runs synchronously at session-create, so a
    handshake failure (bad URL / down provider / wrong key / HTTPS rejection)
    fails SESSION CREATION — surfaced on the create path, never deferred to the
    first turn. `bifrost_error` carries the spec-level code (e.g.
    `bifrost.auth_rejected`); `body` is truncated to ≤1024 bytes, consistent with
    `SessionApiFailed` (INV-004 precedent).
    """

    def __init__(self, *, bifrost_error: str | None, body: bytes) -> None:
        body = body[:1024]
        super().__init__(f"bifrost handshake failed: bifrost_error={bifrost_error!r}")
        self.bifrost_error = bifrost_error
        self.body = body


# Worldtree #204 / v0.28.0 — persona_state endpoint failure modes.
class PersonaNotConfigured(Exception):
    """Raised on HTTP 404 `persona_not_configured` from GET persona_state.

    Agent exists but has no persona surface: persona-disabled Tier 1/2
    agents (e.g. `domari`, `muninn`) and all Tier 3 consumer-defined
    agents (Phase 2.0). Distinct from `AgentNotAvailable` which means the
    agent_id is unknown entirely.
    """

    def __init__(self, *, agent_id: str) -> None:
        super().__init__(f"persona not configured for agent_id: {agent_id!r}")
        self.agent_id = agent_id


class AgentNotAvailable(Exception):
    """Raised on HTTP 404 `agent_not_available` from GET persona_state.

    The agent_id is unknown to the server. Distinct from
    `PersonaNotConfigured` (agent exists but has no persona).
    """

    def __init__(self, *, agent_id: str) -> None:
        super().__init__(f"agent not available: {agent_id!r}")
        self.agent_id = agent_id


class AuthScopeDenied(Exception):
    """Raised on HTTP 403 `auth_scope_denied` from a Heimdall-scoped endpoint.

    The API key lacks the required scope (e.g. `persona.read` for
    GET /agents/{id}/persona_state). User-tier keys carry `persona.read`
    by default; this surfaces when a narrower key is in use.
    """

    def __init__(self, *, scope: str) -> None:
        super().__init__(f"auth scope denied: required={scope!r}")
        self.scope = scope


class AuthoredHistoryUnavailable(Exception):
    """Raised on HTTP 404 from POST /sessions/{id}/history (#347 authored-history-write).

    Hide-existence (server INV-347-1): an ungranted caller, a non-owner, and an
    unknown session ALL receive a 404 byte-identical to a genuine
    `session_not_found` — the feature's existence is never revealed by status,
    body, or error_code. The consumer MUST treat this as feature-absent and fall
    back (a production consumer to a model-generated greeting), and MUST NOT
    capability-probe to distinguish the causes. Distinct from `SessionApiFailed`
    so callers branch feature-absent without inspecting a status code.
    """

    def __init__(self, *, session_id: str) -> None:
        super().__init__(
            f"authored-history write unavailable for session {session_id!r} "
            "(404 hide-existence: feature-absent / ungranted / session-absent)"
        )
        self.session_id = session_id


async def list_sessions(
    client: httpx.AsyncClient,
    *,
    include_archived: bool = False,
    limit: int = 50,
    cursor: str | None = None,
) -> SessionPage:
    """GET /sessions. See contract FN list_sessions."""
    assert client is not None
    assert 1 <= limit <= 200
    assert cursor is None or (isinstance(cursor, str) and cursor)

    params: dict[str, str] = {"limit": str(limit)}
    if include_archived:
        params["include_archived"] = "true"
    if cursor is not None:
        params["cursor"] = cursor

    resp = await client.get("/sessions", params=params)
    if resp.status_code == 422:
        try:
            err = resp.json()
        except ValueError:
            err = {}
        if err.get("error_code") == "cursor_invalid":
            raise InvalidCursor(raw=cursor)
        raise SessionApiFailed(status=422, body=resp.content)
    if resp.status_code != 200:
        raise SessionApiFailed(status=resp.status_code, body=resp.content)
    body = resp.json()
    items = [
        SessionInfo(
            session_id=item["session_id"],
            agent_id=item["agent_id"],
            created_at=item["created_at"],
            last_active=item["last_active"],
            metadata=item.get("metadata", {}),
            message_count=None,
            name=item.get("name"),
            archived=item.get("archived") or False,
            tags=item.get("tags") or [],
        )
        for item in body["items"]
    ]
    return SessionPage(items=items, next_cursor=body.get("next_cursor"))


def endpoint_for_plane(plane: str, base_host: str) -> str:
    """Map a provider plane name to its Worldtree-VISIBLE base URL.

    Issue #17 dev helper: `memory` → :8391, `affect` → :8390, `combined` → :8392
    (the #18 composite both-plane endpoint). Returns the Worldtree-visible base
    (e.g. `http://10.100.10.50:8391`), NOT the client's loopback — Worldtree must
    reach the provider over the network. `http://` is deliberate: the HTTPS
    relaxation is allowlist-side (Worldtree's BIFROST_CLIENT_ALLOWED_HOSTS), not a
    URL concern. A production HTTPS endpoint is supplied directly, bypassing this
    helper.
    """
    ports = {"memory": 8391, "affect": 8390, "combined": 8392}
    if plane not in ports:
        raise ValueError(
            f"unknown plane: {plane!r} "
            "(expected 'memory', 'affect', or 'combined')"
        )
    return f"http://{base_host}:{ports[plane]}"


def _bifrost_error_from(resp: httpx.Response) -> str | None:
    """Pull the spec-level `bifrost_error` from a 502 body.

    Tolerates both the FastAPI-nested `{"detail": {"bifrost_error": …}}` shape
    (the spec's documented form, §"Optional Bifrost binding") and a flat
    top-level `bifrost_error`, per the both-shape unwrap precedent established for
    persona_state errors (the real wire returns the detail-nested form).
    """
    try:
        err = resp.json()
    except ValueError:
        return None
    if not isinstance(err, dict):
        return None
    bifrost_error = err.get("bifrost_error")
    if bifrost_error is None and isinstance(err.get("detail"), dict):
        bifrost_error = err["detail"].get("bifrost_error")
    return bifrost_error


async def create_session(
    client: httpx.AsyncClient,
    agent_id: str,
    *,
    end_user_id: str | None = None,
    bifrost: BifrostBinding | None = None,
    consumer_key: str | None = None,
) -> SessionInfo:
    """POST /sessions to create a new session. See contract FN create_session.

    Per issue #5: pass `end_user_id` for per-end-user agents (lofn etc.).
    When None (default), the body shape matches the pre-#5 baseline
    `{"agent_id": agent_id}` so existing callers (mimir smoke) are unaffected.
    Empty-string `end_user_id` is rejected before HTTP (PRE-003).

    Per issue #17: when `bifrost` is set the request carries the binding and
    authenticates with `consumer_key` (NOT the client's default canary bearer);
    Worldtree handshakes synchronously to our provider before 201.
    """
    assert client is not None
    assert agent_id and isinstance(agent_id, str)
    assert end_user_id is None or (isinstance(end_user_id, str) and end_user_id)

    # PRE-001 (INV-001): a bifrost binding REQUIRES a non-empty consumer key,
    # enforced before any HTTP so a bound create never falls back to the canary.
    if bifrost is not None and not (isinstance(consumer_key, str) and consumer_key):
        raise BifrostConsumerKeyMissing()

    body: dict[str, Any] = {"agent_id": agent_id}
    if end_user_id is not None:
        body["end_user_id"] = end_user_id
    headers: dict[str, str] = {}
    if bifrost is not None:
        body["bifrost"] = {
            "endpoint_url": bifrost.endpoint_url,
            "scope": bifrost.scope,
        }
        # INV-001: the bound create authenticates with the consumer key,
        # overriding the httpx client's default canary bearer per-request.
        headers["Authorization"] = f"Bearer {consumer_key}"
    resp = await client.post("/sessions", json=body, headers=headers)
    if resp.status_code == 404:
        raise AgentNotFound(agent_id=agent_id)
    # POST-002 (INV-002): a 502 on a BOUND create is the synchronous Bifrost
    # handshake failing. Gated on `bifrost is not None` — an unbound create's
    # 502 is a generic upstream fault and stays SessionApiFailed.
    if bifrost is not None and resp.status_code == 502:
        raise BifrostHandshakeFailed(
            bifrost_error=_bifrost_error_from(resp), body=resp.content
        )
    if resp.status_code != 201:
        raise SessionApiFailed(status=resp.status_code, body=resp.content)
    body = resp.json()
    return SessionInfo(
        session_id=body["session_id"],
        agent_id=body["agent_id"],
        created_at=body["created_at"],
        last_active=body["last_active"],
        metadata=body.get("metadata", {}),
        message_count=body["message_count"],
        name=None,
        archived=False,
        tags=[],
    )


async def list_agents(client: httpx.AsyncClient) -> list[AgentInfo]:
    """GET /agents — list available agents. See contract FN list_agents (issue #8).

    No request params, no pagination. Returns server-ordered list. Optional
    fields are defaulted to None / [] / {} per INV-005.
    """
    assert client is not None

    resp = await client.get("/agents")
    if resp.status_code != 200:
        raise SessionApiFailed(status=resp.status_code, body=resp.content)
    body = resp.json()
    return [
        AgentInfo(
            agent_id=item["agent_id"],
            name=item["name"],
            description=item["description"],
            version=item.get("version"),
            capabilities=item.get("capabilities") or [],
            supported_models=item.get("supported_models") or [],
            persona_traits=item.get("persona_traits") or {},
            ui_hints=item.get("ui_hints") or {},
        )
        for item in body
    ]


async def get_persona_state(
    client: httpx.AsyncClient, agent_id: str
) -> dict[str, Any]:
    """GET /agents/{agent_id}/persona_state — fetch current persona snapshot.

    Worldtree #204 / v0.28.0. Returns the same `snapshot` dict shape as the
    `affect_update` SSE event's `status="current"` emission: pad,
    dominant_emotion, emotions_active, baseline_pad, mood_drift,
    last_updated_at. Bootstrap read for clients that want to populate a
    persona pane on session-open without waiting for turn-1's `affect_update`.

    Auth: requires Heimdall `persona.read` scope (user-tier default).

    Failure modes (mapped to typed exceptions per the spec error_codes):
      - 404 `persona_not_configured` → PersonaNotConfigured (persona-disabled
        agents: domari / muninn, and all Tier 3 in Phase 2.0)
      - 404 `agent_not_available` → AgentNotAvailable (unknown agent_id)
      - 403 `auth_scope_denied` → AuthScopeDenied (key lacks persona.read)
      - any other non-2xx → SessionApiFailed (preserves the broader-error
        precedent from list_agents / list_sessions / create_session)
    """
    assert client is not None
    assert agent_id and isinstance(agent_id, str)

    resp = await client.get(f"/agents/{agent_id}/persona_state")
    if resp.status_code == 200:
        return resp.json()
    # Discriminate the 4xx error_code sub-codes; everything else falls
    # through. Worldtree returns errors as either flat `{"error_code": …}`
    # OR FastAPI-default `{"detail": {"error_code": …}}` depending on
    # which handler raised — unwrap both shapes (real wire observed
    # 2026-05-28 returning the detail-nested form for auth_scope_denied
    # from /agents/{id}/persona_state).
    try:
        err = resp.json()
    except ValueError:
        err = None
    error_code: str | None = None
    if isinstance(err, dict):
        error_code = err.get("error_code")
        if error_code is None and isinstance(err.get("detail"), dict):
            error_code = err["detail"].get("error_code")
    if resp.status_code == 404 and error_code == "persona_not_configured":
        raise PersonaNotConfigured(agent_id=agent_id)
    if resp.status_code == 404 and error_code == "agent_not_available":
        raise AgentNotAvailable(agent_id=agent_id)
    if resp.status_code == 403 and error_code == "auth_scope_denied":
        raise AuthScopeDenied(scope="persona.read")
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def get_me(client: httpx.AsyncClient) -> dict[str, Any]:
    """GET /me — the authenticated principal's identity + key metadata (spec §GET /me).

    Boot-time whoami: verify the key without agent-config side effects. Returns
    the parsed dict verbatim (freeform per the frozen OpenAPI; the spec documents
    `{user_id, scopes, tier, display_name?, key_id?, key_label?, ...}`, optional
    fields omitted-not-null). 401 (bad/absent key when auth is enabled) — like
    every other non-200 — surfaces as SessionApiFailed (get_persona_state
    precedent). Read-only, rate-exempt, no audit emission.
    """
    assert client is not None
    resp = await client.get("/me")
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def list_character_models(client: httpx.AsyncClient) -> dict[str, Any]:
    """GET /models/available-for-characters — character-capable model profiles (#161).

    Requires `character.read`. Returns `{items: [{name, description, thinking}]}`.
    Parsed dict verbatim; any non-200 → SessionApiFailed.
    """
    assert client is not None
    resp = await client.get("/models/available-for-characters")
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def create_character(
    client: httpx.AsyncClient, character: dict[str, Any], *, state: dict[str, Any] | None = None
) -> dict[str, Any]:
    """POST /characters — create a transient character (#161). Requires `character.write`.

    Body is `{character, state}` (state optional — a CharacterStateSchema for
    mid-conversation rehydration). Returns 201 `{character_id, ttl_expires_at}`;
    any non-201 → SessionApiFailed.
    """
    assert client is not None
    assert isinstance(character, dict) and character
    resp = await client.post("/characters", json={"character": character, "state": state})
    if resp.status_code == 201:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def get_character_state(client: httpx.AsyncClient, character_id: str) -> dict[str, Any]:
    """GET /characters/{character_id}/state — live runtime state (#161). Requires `character.read`.

    Returns `{schema_version, pad, emotions_active, mood_drift, goal_signal_history}`;
    refreshes the character's TTL. Any non-200 → SessionApiFailed.
    """
    assert client is not None
    assert character_id and isinstance(character_id, str)
    resp = await client.get(f"/characters/{character_id}/state")
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def delete_character(client: httpx.AsyncClient, character_id: str) -> None:
    """DELETE /characters/{character_id} — remove a transient character (#161).

    Requires `character.write`. Bound sessions detach (next turn → 410
    character_not_found). 200/204 → None; any other status → SessionApiFailed.
    """
    assert client is not None
    assert character_id and isinstance(character_id, str)
    resp = await client.delete(f"/characters/{character_id}")
    if resp.status_code in (200, 204):
        return None
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def set_persona_state(
    client: httpx.AsyncClient, session_id: str, snapshot: dict[str, Any]
) -> None:
    """POST /sessions/{session_id}/persona_state — set a session's persona state (affect injection).

    The request body is FREEFORM on the wire (the OpenAPI declares no request
    schema), but worldtree-dev's prose now pins the canonical shape (#317):
    `{"pad": {"pleasure": p, "arousal": a, "dominance": d}}` — a named-key dict
    (each in [-1, 1]), NOT a bare list; PAD-only, session-scoped, pull-over-push
    (#289). The caller supplies the snapshot. 204 No Content → None; any other
    status → SessionApiFailed.
    """
    assert client is not None
    assert session_id and isinstance(session_id, str)
    assert isinstance(snapshot, dict)
    resp = await client.post(f"/sessions/{session_id}/persona_state", json=snapshot)
    if resp.status_code == 204:
        return None
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def get_session_bifrost(
    client: httpx.AsyncClient, session_id: str, *, admin_key: str
) -> dict[str, Any]:
    """GET /admin/sessions/{session_id}/bifrost — admin-scoped Bifrost dispatch state (#176).

    Returns the live Bifrost binding for a session: `{endpoint_url, consumer_id,
    connected, capabilities_granted, tools: [{name, description}]}`. Requires the
    `admin.sessions.read` scope (admin tier), so the request OVERRIDES the
    Authorization header with `admin_key` (distinct from the client's default
    consumer key). Read-only (audited server-side). Parsed dict verbatim; any
    non-200 → SessionApiFailed — notably 403 `auth_scope_denied` (key lacks the
    scope) and 404 `session_not_bifrost_bound` (session exists, no live client).
    """
    assert client is not None
    assert session_id and isinstance(session_id, str)
    assert admin_key and isinstance(admin_key, str)
    resp = await client.get(
        f"/admin/sessions/{session_id}/bifrost",
        headers={"Authorization": f"Bearer {admin_key}"},
    )
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def get_session_tools(client: httpx.AsyncClient, session_id: str) -> dict[str, Any]:
    """GET /sessions/{session_id}/tools — owner-scoped tool inventory (spec #183).

    Returns the merged tool list the LLM saw at turn-fire: `{agent_id,
    builtin_tools: [...], bifrost_tools: [{name, description, parameters}, ...]}`.
    Owner-scoped (`ctx.user_id == session.user_id`) — reachable with the consumer
    key, NO admin scope. Cross-owner access returns 404 `session_not_found`
    (existence-hiding); a revoked session returns 401 `auth_revoked`. Parsed dict
    verbatim; any non-200 → SessionApiFailed (mirrors get_persona_state).
    """
    assert client is not None
    assert session_id and isinstance(session_id, str)
    resp = await client.get(f"/sessions/{session_id}/tools")
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def get_capabilities(client: httpx.AsyncClient) -> dict[str, Any]:
    """GET /capabilities — server capability discovery (spec §Ephemeral Templates).

    Returns `{ephemeral_templates: {echo: {allowed_models, default_model,
    system_prompt_max_bytes}}}` — what the server offers before a client decides
    to instantiate. Any authenticated caller may read it (no scope). Parsed dict
    verbatim; any non-200 → SessionApiFailed.
    """
    assert client is not None
    resp = await client.get("/capabilities")
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def write_authored_history(
    client: httpx.AsyncClient,
    session_id: str,
    *,
    content: str,
    idempotency_key: str,
    author: str = "assistant",
    effects: str | None = None,
    claimed_original_at: str | None = None,
) -> dict[str, Any]:
    """POST /sessions/{session_id}/history — the #347 authored-history-write primitive.

    Write one model-visible turn into the session's ledger AS the bound agent,
    WITHOUT a generation and WITHOUT lived-turn side effects (the SillyTavern
    "first message"). v1: `author="assistant"`, `effects` omitted (== "none"),
    `idempotency_key` REQUIRED (per-session dedup). The server pins the body
    (`AuthoredWriteRequest`, `extra="forbid"`), so `effects` /
    `claimed_original_at` are sent only when non-None — never as null keys.

    Success is 201 (fresh) or 200 (idempotent replay, byte-identical body); both
    return the `AuthoredTurnResponse` dict verbatim (`{author, content_chars,
    injected_at, phase, seq, session_id, turn_id}` — provenance is audit-only,
    never on this body).

    404 → `AuthoredHistoryUnavailable` (hide-existence: feature-absent /
    ungranted / session-absent are indistinguishable by design; the caller falls
    back and NEVER capability-probes — server INV-347-1). Any other non-2xx →
    `SessionApiFailed` (notably 409 `generation_active`, 422 `content_too_long` /
    `validation_failed`).
    """
    assert client is not None
    assert session_id and isinstance(session_id, str)
    assert content and isinstance(content, str)
    assert idempotency_key and isinstance(idempotency_key, str)
    assert author and isinstance(author, str)
    body: dict[str, Any] = {
        "author": author,
        "content": content,
        "idempotency_key": idempotency_key,
    }
    if effects is not None:
        body["effects"] = effects
    if claimed_original_at is not None:
        body["claimed_original_at"] = claimed_original_at
    resp = await client.post(f"/sessions/{session_id}/history", json=body)
    if resp.status_code in (200, 201):
        return resp.json()
    if resp.status_code == 404:
        raise AuthoredHistoryUnavailable(session_id=session_id)
    raise SessionApiFailed(status=resp.status_code, body=resp.content)


async def get_session_messages(
    client: httpx.AsyncClient, session_id: str
) -> dict[str, Any]:
    """GET /sessions/{session_id}/messages — the session's message history.

    Un-deferred as the #347 seed read-back: a seeded turn renders as a normal
    `role=assistant` message (model-invisible provenance — indistinguishable
    from a lived turn on read). Returns `{session_id, items: [{seq, role,
    content, ...}], next_cursor}` verbatim; owner-scoped; any non-200 →
    `SessionApiFailed`. v1 reads the server default page (no pagination params —
    add limit/cursor when a caller needs scrollback).
    """
    assert client is not None
    assert session_id and isinstance(session_id, str)
    resp = await client.get(f"/sessions/{session_id}/messages")
    if resp.status_code == 200:
        return resp.json()
    raise SessionApiFailed(status=resp.status_code, body=resp.content)
