From 0602c5f537d9681896247068e1ce6376d76ef3f9 Mon Sep 17 00:00:00 2001
From: John | Elite Encoder <john@eliteencoder.net>
Date: Mon, 5 Oct 2026 15:54:04 -0400
Subject: [PATCH 1/6] Carry the paid manifest_id on call_runner streams and
 errors

LiveRunnerCallStream.session_id carries the answered payment challenge's
manifest_id, as LiveRunnerCallResult.session_id already does for unary calls;
it is "" when no challenge was answered.

Errors raised from a call_runner attempt carry payment_sent, True when that
attempt's request already carried Livepeer-Payment headers, and manifest_id,
the paid challenge's manifest_id ("" when unpaid). A gateway may fail over to
another runner only while payment_sent is False, and can attribute the
network cost of a paid call that failed. When the challenge retries run out,
payment_sent covers every attempt and manifest_id is the last paid one.

manifest_id is the key the signer's create_signed_ticket Kafka event carries
as manifest_id; it is not the event's session_id (the signer state id) or
pm_session_id (the orchestrator ticket params' recipientRandHash).

Replayed onto livepeer/livepeer-python-gateway main without the multipart
commit this change was previously stacked on.
---
 src/livepeer_gateway/errors.py      |  15 ++-
 src/livepeer_gateway/live_runner.py |  36 ++++-
 tests/test_live_runner.py           | 197 ++++++++++++++++++++++++++++
 3 files changed, 243 insertions(+), 5 deletions(-)

diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py
index 9b44713..61c4897 100644
--- a/src/livepeer_gateway/errors.py
+++ b/src/livepeer_gateway/errors.py
@@ -4,7 +4,20 @@
 
 
 class LivepeerGatewayError(RuntimeError):
-    """Base error for the library."""
+    """Base error for the library.
+
+    ``payment_sent`` is True when the error comes from a ``call_runner`` attempt
+    whose request carried ``Livepeer-Payment`` headers, that is, after a payment
+    challenge had already been answered. A gateway may retry another runner only
+    while it is False; once tickets are sent the request is bound to that runner.
+
+    ``manifest_id`` is the answered payment challenge's ``manifest_id`` when
+    ``payment_sent`` is True, and ``""`` otherwise, so a gateway can attribute
+    the cost of a paid call that failed.
+    """
+
+    payment_sent: bool = False
+    manifest_id: str = ""
 
 
 class LivepeerHTTPError(LivepeerGatewayError):
diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 75faa28..3a0b773 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -196,6 +196,9 @@ class LiveRunnerCallStream:
     _response: aiohttp.ClientResponse = field(repr=False, compare=False)
     # True once the orchestrator reported the backing session gone.
     released: bool = False
+    # The payment challenge's manifest_id, as LiveRunnerCallResult.session_id
+    # carries it for a unary call; "" when no challenge was answered.
+    session_id: str = ""
     _payment_task: asyncio.Task[None] | None = field(
         default=None, repr=False, compare=False
     )
@@ -807,7 +810,10 @@ async def call_runner(
 
     With ``signer_url`` set, payment is automatic and **per call**: a 402 challenge is
     paid via the signer and retried (up to ``max_payment_challenge_retries``), one job,
-    one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors.
+    one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. Every error
+    raised from a runner request carries ``payment_sent``: True when the failed
+    attempt already carried ``Livepeer-Payment`` headers, so a gateway knows whether
+    it may still try another runner.
 
     ``application/json`` and ``+json`` types parse into ``result.data``; anything else
     (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``.
@@ -827,6 +833,8 @@ async def call_runner(
         signer = await get_signer_info(signer_url, _freeze_headers(signer_headers))
         payer_address = cast(str, signer.address)
     challenge: LivePaymentChallenge | None = None
+    any_payment_sent = False
+    paid_manifest_id = ""
     attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2
     for attempt in range(attempts):
         payment_session: LivePaymentSession | None = None
@@ -865,6 +873,11 @@ async def call_runner(
             # Metered pricing bills for as long as the work runs.
             needs_ongoing_funding = payment_type in _METERED_PAYMENT_TYPES
 
+        # Once tickets go out with the request, the call is bound to this runner.
+        payment_sent = "Livepeer-Payment" in request_headers
+        any_payment_sent = any_payment_sent or payment_sent
+        if payment_sent:
+            paid_manifest_id = session_id
         try:
             request_kwargs: dict[str, Any] = {"timeout": timeout}
             if request_headers:
@@ -887,6 +900,7 @@ async def call_runner(
                     None if payment_type == "fixed" else payment_session,
                     session,
                     resp,
+                    session_id=session_id,
                 )
                 # The stream outlives this call, so it owns the funding.
                 if needs_ongoing_funding:
@@ -937,13 +951,27 @@ async def call_runner(
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
+                e.payment_sent = payment_sent
+                e.manifest_id = session_id
                 raise
             if not signer_url:
                 raise LivepeerGatewayError("Live runner paid call requires signer_url") from e
-            challenge = _parse_runner_payment_challenge(e)
+            try:
+                challenge = _parse_runner_payment_challenge(e)
+            except LivepeerGatewayError as parse_error:
+                parse_error.payment_sent = payment_sent
+                parse_error.manifest_id = session_id
+                raise
             continue
-
-    raise LivepeerGatewayError("Live runner call exhausted payment challenge retries")
+        except LivepeerGatewayError as e:
+            e.payment_sent = payment_sent
+            e.manifest_id = session_id
+            raise
+
+    exhausted = LivepeerGatewayError("Live runner call exhausted payment challenge retries")
+    exhausted.payment_sent = any_payment_sent
+    exhausted.manifest_id = paid_manifest_id
+    raise exhausted
 
 
 def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> LivePaymentChallenge:
diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py
index 3cd2033..fbf8586 100644
--- a/tests/test_live_runner.py
+++ b/tests/test_live_runner.py
@@ -1,13 +1,17 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import os
+import socket
 import sys
+from collections.abc import AsyncIterator, Iterator
 from types import SimpleNamespace
 from unittest import mock
 
 import pytest
+from aiohttp import web
 
 from livepeer_gateway import live_runner
 from livepeer_gateway.errors import (
@@ -1833,3 +1837,196 @@ def test_torch_detects_current_device(self) -> None:
             gpu = live_runner._detect_gpu_torch()
 
         assert gpu == LiveRunnerGPU(id="1", name="Torch GPU", vram_mb=32)
+
+
+# ---------------------------------------------------------------------------
+# call_runner against a real runner. The signer is faked; the runner is a real
+# aiohttp application on an ephemeral port.
+# ---------------------------------------------------------------------------
+
+
+@contextlib.asynccontextmanager
+async def _serve(app: web.Application) -> AsyncIterator[str]:
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    port = site._server.sockets[0].getsockname()[1]  # type: ignore[union-attr]
+    try:
+        yield f"http://127.0.0.1:{port}"
+    finally:
+        await runner.cleanup()
+
+
+class _RecordedRequest:
+    def __init__(self, request: web.Request, body: bytes) -> None:
+        self.path = request.path
+        self.method = request.method
+        self.headers = dict(request.headers)
+        self.body = body
+
+
+class _FakePaymentSession:
+    """Stands in for LivePaymentSession: returns fixed tickets, funds forever."""
+
+    instances: list[_FakePaymentSession] = []
+
+    def __init__(self, signer_url: str, **kwargs: object) -> None:
+        self.signer_url = signer_url
+        self.kwargs = kwargs
+        type(self).instances.append(self)
+
+    async def get_payment(self) -> object:
+        return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64")
+
+    async def run_payments(self) -> bool:
+        await asyncio.Event().wait()
+        return False
+
+
+@contextlib.contextmanager
+def _fake_signer() -> Iterator[None]:
+    _FakePaymentSession.instances = []
+    with (
+        mock.patch.object(live_runner, "LivePaymentSession", _FakePaymentSession),
+        mock.patch.object(
+            live_runner,
+            "get_signer_info",
+            new_callable=mock.AsyncMock,
+            return_value=SimpleNamespace(address="opaque-payer", sig="opaque-signature"),
+        ),
+    ):
+        yield
+
+
+def _runner_app(
+    seen: list[_RecordedRequest],
+    *,
+    statuses: list[int] | None = None,
+    stream: bool = False,
+) -> web.Application:
+    """A runner that answers each request with the next status in ``statuses``.
+
+    402 carries a payment challenge; 200 answers JSON, or two SSE frames when
+    ``stream`` is set.
+    """
+    pending = list(statuses or [200])
+
+    async def handler(request: web.Request) -> web.StreamResponse:
+        body = await request.read()
+        seen.append(_RecordedRequest(request, body))
+        status = pending.pop(0) if len(pending) > 1 else pending[0]
+        if status == 402:
+            return web.Response(
+                status=402,
+                text=_payment_challenge_body("manifest-1"),
+                content_type="application/json",
+            )
+        if status != 200:
+            return web.Response(status=status, text="no")
+        if stream:
+            resp = web.StreamResponse(headers={"Content-Type": "text/event-stream"})
+            await resp.prepare(request)
+            await resp.write(b"data: one\n\n")
+            await resp.write(b"data: two\n\n")
+            await resp.write_eof()
+            return resp
+        return web.json_response({"text": "hello", "n": len(seen)})
+
+    app = web.Application()
+    app.router.add_route("*", "/call", handler)
+    return app
+
+
+def _free_port() -> int:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def _paid_call(base: str, **kwargs: object):
+    return call_runner(
+        f"{base}/call",
+        payload={"prompt": "hi"},
+        signer_url="https://signer.example.com",
+        payment_unit="seconds",
+        **kwargs,
+    )
+
+
+class TestCallRunnerStreamSessionId:
+    async def test_stream_carries_manifest_id(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402, 200], stream=True)) as base:
+            with _fake_signer():
+                async with await _paid_call(base, stream=True) as stream:
+                    assert stream.session_id == "manifest-1"
+                    assert stream.payment_session is _FakePaymentSession.instances[0]
+                    lines = [line async for line in stream.aiter_lines() if line]
+
+        assert lines == ["data: one", "data: two"]
+        assert len(seen) == 2
+
+    async def test_stream_without_signer_has_empty_session_id(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, stream=True)) as base:
+            async with await call_runner(f"{base}/call", payload={}, stream=True) as stream:
+                assert stream.session_id == ""
+
+
+class TestCallRunnerPaymentSent:
+    @pytest.mark.parametrize("status", [500, 400])
+    async def test_error_after_payment_carries_the_manifest(self, status: int) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402, status])) as base:
+            with _fake_signer():
+                with pytest.raises(LivepeerHTTPError) as info:
+                    await _paid_call(base)
+        assert info.value.status_code == status
+        assert info.value.payment_sent is True
+        assert info.value.manifest_id == "manifest-1"
+        assert len(seen) == 2
+
+    @pytest.mark.parametrize("status", [400, 503])
+    async def test_error_before_payment_has_no_manifest(self, status: int) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[status])) as base:
+            with _fake_signer():
+                with pytest.raises(LivepeerHTTPError) as info:
+                    await _paid_call(base)
+        assert info.value.status_code == status
+        assert info.value.payment_sent is False
+        assert info.value.manifest_id == ""
+        assert len(seen) == 1
+
+    async def test_connection_refused_before_challenge(self) -> None:
+        with pytest.raises(LivepeerGatewayError, match="connection refused") as info:
+            await call_runner(f"http://127.0.0.1:{_free_port()}/call", payload={"prompt": "hi"})
+        assert info.value.payment_sent is False
+
+    async def test_stream_error_after_payment_says_so(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402, 500])) as base:
+            with _fake_signer():
+                with pytest.raises(LivepeerHTTPError) as info:
+                    await _paid_call(base, stream=True)
+        assert info.value.status_code == 500
+        assert info.value.payment_sent is True
+        assert info.value.manifest_id == "manifest-1"
+
+    async def test_exhausted_retries_after_payment_says_so(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402])) as base:
+            with _fake_signer():
+                with pytest.raises(LivepeerGatewayError, match="exhausted") as info:
+                    await _paid_call(base, max_payment_challenge_retries=1)
+        assert info.value.payment_sent is True
+        assert info.value.manifest_id == "manifest-1"
+
+    async def test_paid_call_without_signer_is_unpaid(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402])) as base:
+            with pytest.raises(LivepeerGatewayError, match="requires signer_url") as info:
+                await call_runner(f"{base}/call", payload={"prompt": "hi"})
+        assert info.value.payment_sent is False
+        assert info.value.manifest_id == ""

From 5868d81d5bfcad305af96c57c679acee79e07edb Mon Sep 17 00:00:00 2001
From: John | Elite Encoder <john@eliteencoder.net>
Date: Tue, 6 Oct 2026 16:39:02 -0400
Subject: [PATCH 2/6] Classify runner rejections and record whether they paid

Selection keeps each failed attempt as capacity, unreachable, or other, and copies payment_sent and the paid manifest id onto the rejection so a gateway can stop after a paid attempt.
---
 src/livepeer_gateway/errors.py    | 49 +++++++++++++++++-
 src/livepeer_gateway/scope.py     |  7 ++-
 src/livepeer_gateway/selection.py |  6 +--
 tests/test_errors.py              | 85 +++++++++++++++++++++++++++++++
 tests/test_selection.py           | 52 ++++++++++++++++++-
 5 files changed, 190 insertions(+), 9 deletions(-)

diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py
index 61c4897..1ba1d03 100644
--- a/src/livepeer_gateway/errors.py
+++ b/src/livepeer_gateway/errors.py
@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import Literal
+
+import aiohttp
 
 
 class LivepeerGatewayError(RuntimeError):
@@ -37,11 +40,52 @@ class OrchestratorRejection:
     reason: str
 
 
+RunnerFailureKind = Literal["capacity", "unreachable", "other"]
+
+
 @dataclass
 class RunnerRejection:
-    """Records a single runner that was tried and rejected."""
+    """Records a single runner that was tried and rejected.
+
+    ``kind`` is ``capacity`` for an HTTP 503, ``unreachable`` when the runner
+    never answered, and ``other`` for every remaining failure. ``payment_sent``
+    is copied from the ``call_runner`` error: True only after that attempt's
+    request carried ``Livepeer-Payment`` headers. ``manifest_id`` is that
+    attempt's paid challenge id, or ``""`` when it did not pay.
+    """
+
     url: str
     reason: str
+    kind: RunnerFailureKind = "other"
+    payment_sent: bool = False
+    manifest_id: str = ""
+
+
+def runner_rejection(url: str, error: BaseException) -> RunnerRejection:
+    """Record one failed runner attempt without dropping payment or its kind."""
+    payment_sent = False
+    manifest_id = ""
+    if isinstance(error, LivepeerGatewayError) and error.payment_sent:
+        payment_sent = True
+        manifest_id = error.manifest_id
+    return RunnerRejection(
+        url=url,
+        reason=str(error),
+        kind=_runner_failure_kind(error),
+        payment_sent=payment_sent,
+        manifest_id=manifest_id,
+    )
+
+
+def _runner_failure_kind(error: BaseException) -> RunnerFailureKind:
+    if isinstance(error, LivepeerHTTPError) and error.status_code == 503:
+        return "capacity"
+    current: BaseException | None = error
+    while current is not None:
+        if isinstance(current, (TimeoutError, ConnectionError, OSError, aiohttp.ClientConnectionError)):
+            return "unreachable"
+        current = current.__cause__
+    return "other"
 
 
 class NoOrchestratorAvailableError(LivepeerGatewayError):
@@ -65,6 +109,9 @@ class NoRunnerAvailableError(LivepeerGatewayError):
     def __init__(self, message: str, rejections: list[RunnerRejection] | None = None) -> None:
         super().__init__(message)
         self.rejections: list[RunnerRejection] = rejections or []
+        paid = [rejection for rejection in self.rejections if rejection.payment_sent]
+        self.payment_sent = bool(paid)
+        self.manifest_id = paid[-1].manifest_id if paid else ""
 
     def __str__(self) -> str:
         message = super().__str__()
diff --git a/src/livepeer_gateway/scope.py b/src/livepeer_gateway/scope.py
index 76cd979..999a364 100644
--- a/src/livepeer_gateway/scope.py
+++ b/src/livepeer_gateway/scope.py
@@ -3,7 +3,7 @@
 import logging
 from typing import Any, Optional, Sequence
 
-from .errors import LivepeerGatewayError, NoRunnerAvailableError, RunnerRejection
+from .errors import LivepeerGatewayError, NoRunnerAvailableError, runner_rejection
 from .http import post_json
 from .lv2v import LiveVideoToVideo, StartJobRequest
 from .selection import runner_selector
@@ -129,12 +129,11 @@ async def _start_scope_with_runner(
                 raise LivepeerGatewayError("Scope response missing manifest_id")
             return job
         except Exception as e:
-            reason = str(e)
             runner_url = result.runner_url.strip()
             if not runner_url and result.runner is not None:
                 runner_url = result.runner.url
-            _LOG.debug("scope runner startup failed: %s (%s)", runner_url, reason)
-            cursor.rejections.append(RunnerRejection(url=runner_url, reason=reason))
+            _LOG.debug("scope runner startup failed: %s (%s)", runner_url, e)
+            cursor.rejections.append(runner_rejection(runner_url, e))
 
 
 def _is_serverless_runner(runner: object) -> bool:
diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py
index 9a5d47e..a01ae03 100644
--- a/src/livepeer_gateway/selection.py
+++ b/src/livepeer_gateway/selection.py
@@ -18,6 +18,7 @@
     NoRunnerAvailableError,
     OrchestratorRejection,
     RunnerRejection,
+    runner_rejection,
 )
 from .live_runner import (
     LiveRunnerCallResult,
@@ -205,13 +206,12 @@ async def next(self) -> LiveRunnerCallResult:
                     kwargs["signer_headers"] = self._signer_headers
                 result = await call_runner(**kwargs)
             except Exception as e:
-                reason = str(e)
                 _LOG.debug(
                     "select_runner candidate failed: %s (%s)",
                     runner.url,
-                    reason,
+                    e,
                 )
-                self.rejections.append(RunnerRejection(url=runner.url, reason=reason))
+                self.rejections.append(runner_rejection(runner.url, e))
                 continue
 
             _LOG.debug("select_runner selected: %s", runner.url)
diff --git a/tests/test_errors.py b/tests/test_errors.py
index da34da7..75a6ed8 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -1,6 +1,13 @@
+import aiohttp
+
 from livepeer_gateway.errors import (
+    LivepeerGatewayError,
+    LivepeerHTTPError,
     NoOrchestratorAvailableError,
+    NoRunnerAvailableError,
     OrchestratorRejection,
+    RunnerRejection,
+    runner_rejection,
 )
 
 
@@ -28,3 +35,81 @@ def test_no_orchestrator_available_error_includes_rejections() -> None:
         "https://orch-a.example.com: connection refused; "
         "https://orch-b.example.com: request timed out"
     )
+
+
+def test_paid_503_is_a_capacity_rejection() -> None:
+    error = LivepeerHTTPError(503, "https://runner.example.com", "busy")
+    error.payment_sent = True
+    error.manifest_id = "manifest-1"
+
+    rejection = runner_rejection("https://runner.example.com", error)
+
+    assert rejection.kind == "capacity"
+    assert rejection.payment_sent is True
+    assert rejection.manifest_id == "manifest-1"
+    assert rejection.reason == str(error)
+
+
+def test_http_400_is_another_kind_of_rejection() -> None:
+    error = LivepeerHTTPError(400, "https://runner.example.com", "bad")
+
+    rejection = runner_rejection("https://runner.example.com", error)
+
+    assert rejection.kind == "other"
+    assert rejection.payment_sent is False
+    assert rejection.manifest_id == ""
+
+
+def test_connection_refused_is_unreachable() -> None:
+    try:
+        raise LivepeerGatewayError("connection refused") from ConnectionRefusedError()
+    except LivepeerGatewayError as error:
+        rejection = runner_rejection("https://runner.example.com", error)
+
+    assert rejection.kind == "unreachable"
+    assert rejection.payment_sent is False
+
+
+def test_server_disconnect_is_unreachable() -> None:
+    try:
+        raise LivepeerGatewayError("failed to reach endpoint") from aiohttp.ServerDisconnectedError()
+    except LivepeerGatewayError as error:
+        rejection = runner_rejection("https://runner.example.com", error)
+
+    assert rejection.kind == "unreachable"
+
+
+def test_payload_error_is_not_unreachable() -> None:
+    try:
+        raise LivepeerGatewayError("short read") from aiohttp.ClientPayloadError("short")
+    except LivepeerGatewayError as error:
+        rejection = runner_rejection("https://runner.example.com", error)
+
+    assert rejection.kind == "other"
+
+
+def test_text_mentioning_capacity_is_not_a_capacity_refusal() -> None:
+    error = LivepeerGatewayError("capacity exhausted")
+
+    rejection = runner_rejection("https://runner.example.com", error)
+
+    assert rejection.kind == "other"
+
+
+def test_no_runner_available_keeps_the_paid_rejection() -> None:
+    error = NoRunnerAvailableError(
+        "All runners failed (1 tried)",
+        rejections=[
+            RunnerRejection(
+                url="https://runner.example.com",
+                reason="busy",
+                kind="capacity",
+                payment_sent=True,
+                manifest_id="manifest-1",
+            )
+        ],
+    )
+
+    assert error.payment_sent is True
+    assert error.manifest_id == "manifest-1"
+    assert str(error) == "All runners failed (1 tried): https://runner.example.com: busy"
diff --git a/tests/test_selection.py b/tests/test_selection.py
index e56d371..5f8a5c9 100644
--- a/tests/test_selection.py
+++ b/tests/test_selection.py
@@ -5,7 +5,7 @@
 import pytest
 
 from livepeer_gateway import selection
-from livepeer_gateway.errors import LivepeerGatewayError, NoRunnerAvailableError
+from livepeer_gateway.errors import LivepeerGatewayError, LivepeerHTTPError, NoRunnerAvailableError
 from livepeer_gateway.live_runner import LiveRunnerCallResult, LiveRunnerInstance
 
 
@@ -178,6 +178,56 @@ async def _call_runner(
         assert len(cursor.rejections) == 1
         assert cursor.rejections[0].url == "https://orch-a/apps/a/session"
         assert cursor.rejections[0].reason == "capacity exhausted"
+        assert cursor.rejections[0].kind == "other"
+        assert cursor.rejections[0].payment_sent is False
+
+    async def test_runner_selector_keeps_payment_and_failure_kind(self) -> None:
+        async def _call_runner(
+            *,
+            runner: LiveRunnerInstance,
+            payload: dict[str, object],
+            method: str,
+            timeout: float,
+        ) -> LiveRunnerCallResult:
+            del payload, method, timeout
+            if runner.url.endswith("/a/session"):
+                refused = LivepeerHTTPError(503, runner.url, "busy")
+                refused.payment_sent = True
+                refused.manifest_id = "manifest-1"
+                raise refused
+            try:
+                raise LivepeerGatewayError("connection refused") from ConnectionRefusedError()
+            except LivepeerGatewayError as unreachable:
+                raise unreachable
+
+        with (
+            mock.patch.object(
+                selection,
+                "discover_runners",
+                return_value=[
+                    _entry(
+                        [
+                            {"url": "https://orch-a/apps/a/session", "app": "app-a"},
+                            {"url": "https://orch-a/apps/b/session", "app": "app-b"},
+                        ]
+                    )
+                ],
+            ),
+            mock.patch.object(selection, "call_runner", side_effect=_call_runner),
+        ):
+            cursor = await selection.runner_selector(
+                discovery_url="https://example.com/discovery"
+            )
+            with pytest.raises(NoRunnerAvailableError) as raised:
+                await cursor.next()
+
+        assert cursor.rejections[0].kind == "capacity"
+        assert cursor.rejections[0].payment_sent is True
+        assert cursor.rejections[0].manifest_id == "manifest-1"
+        assert cursor.rejections[1].kind == "unreachable"
+        assert cursor.rejections[1].payment_sent is False
+        assert raised.value.payment_sent is True
+        assert raised.value.manifest_id == "manifest-1"
 
     async def test_runner_selector_empty_discovery_raises_no_runner_available(
         self,

From 3250c08cb6424d339614f1e9095064fbe7e6da6e Mon Sep 17 00:00:00 2001
From: John | Elite Encoder <john@eliteencoder.net>
Date: Wed, 7 Oct 2026 14:48:43 -0400
Subject: [PATCH 3/6] Update error handling to track signer auth IDs

Refactor the error handling in LivepeerGatewayError and related classes to replace the manifest_id with auth_ids, which now captures the signer auth IDs for each payment session. This change allows for better tracking of payment sessions across errors and runner rejections, ensuring that the system can accurately attribute costs and manage retries based on the associated auth IDs. Update tests to reflect these changes and ensure proper functionality.
---
 src/livepeer_gateway/errors.py        | 31 +++++++------
 src/livepeer_gateway/live_runner.py   | 67 +++++++++++++++++++++------
 src/livepeer_gateway/remote_signer.py | 33 +++++++++++++
 tests/test_errors.py                  | 27 +++++++----
 tests/test_live_payment_session.py    | 41 ++++++++++++++++
 tests/test_live_runner.py             | 26 ++++++++---
 tests/test_selection.py               |  6 +--
 7 files changed, 185 insertions(+), 46 deletions(-)

diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py
index 1ba1d03..9e48559 100644
--- a/src/livepeer_gateway/errors.py
+++ b/src/livepeer_gateway/errors.py
@@ -14,13 +14,13 @@ class LivepeerGatewayError(RuntimeError):
     challenge had already been answered. A gateway may retry another runner only
     while it is False; once tickets are sent the request is bound to that runner.
 
-    ``manifest_id`` is the answered payment challenge's ``manifest_id`` when
-    ``payment_sent`` is True, and ``""`` otherwise, so a gateway can attribute
-    the cost of a paid call that failed.
+    ``auth_ids`` are the signer auth ids of every payment session this call
+    opened. The signer charges an allocation when it signs, which can be before
+    ``payment_sent`` becomes True. Empty when no session was signed.
     """
 
     payment_sent: bool = False
-    manifest_id: str = ""
+    auth_ids: tuple[str, ...] = ()
 
 
 class LivepeerHTTPError(LivepeerGatewayError):
@@ -50,30 +50,30 @@ class RunnerRejection:
     ``kind`` is ``capacity`` for an HTTP 503, ``unreachable`` when the runner
     never answered, and ``other`` for every remaining failure. ``payment_sent``
     is copied from the ``call_runner`` error: True only after that attempt's
-    request carried ``Livepeer-Payment`` headers. ``manifest_id`` is that
-    attempt's paid challenge id, or ``""`` when it did not pay.
+    request carried ``Livepeer-Payment`` headers. ``auth_ids`` are the signer
+    auth ids that attempt opened, which can be set before ``payment_sent``.
     """
 
     url: str
     reason: str
     kind: RunnerFailureKind = "other"
     payment_sent: bool = False
-    manifest_id: str = ""
+    auth_ids: tuple[str, ...] = ()
 
 
 def runner_rejection(url: str, error: BaseException) -> RunnerRejection:
     """Record one failed runner attempt without dropping payment or its kind."""
     payment_sent = False
-    manifest_id = ""
-    if isinstance(error, LivepeerGatewayError) and error.payment_sent:
-        payment_sent = True
-        manifest_id = error.manifest_id
+    auth_ids: tuple[str, ...] = ()
+    if isinstance(error, LivepeerGatewayError):
+        payment_sent = error.payment_sent
+        auth_ids = error.auth_ids
     return RunnerRejection(
         url=url,
         reason=str(error),
         kind=_runner_failure_kind(error),
         payment_sent=payment_sent,
-        manifest_id=manifest_id,
+        auth_ids=auth_ids,
     )
 
 
@@ -111,7 +111,12 @@ def __init__(self, message: str, rejections: list[RunnerRejection] | None = None
         self.rejections: list[RunnerRejection] = rejections or []
         paid = [rejection for rejection in self.rejections if rejection.payment_sent]
         self.payment_sent = bool(paid)
-        self.manifest_id = paid[-1].manifest_id if paid else ""
+        auth_ids: list[str] = []
+        for rejection in self.rejections:
+            for auth_id in rejection.auth_ids:
+                if auth_id not in auth_ids:
+                    auth_ids.append(auth_id)
+        self.auth_ids = tuple(auth_ids)
 
     def __str__(self) -> str:
         message = super().__str__()
diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 3a0b773..5466f9e 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -176,6 +176,9 @@ class LiveRunnerCallResult:
     # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty.
     content: bytes | None = field(default=None, repr=False)
     content_type: str = ""
+    # Signer auth ids of every payment session this call opened. Fixed-price
+    # calls still list them here, because their payment_session is not returned.
+    auth_ids: tuple[str, ...] = ()
 
 
 @dataclass
@@ -198,7 +201,10 @@ class LiveRunnerCallStream:
     released: bool = False
     # The payment challenge's manifest_id, as LiveRunnerCallResult.session_id
     # carries it for a unary call; "" when no challenge was answered.
+    # Correlation with the orchestrator, not a billing key.
     session_id: str = ""
+    # Signer auth ids of every payment session this call opened.
+    auth_ids: tuple[str, ...] = ()
     _payment_task: asyncio.Task[None] | None = field(
         default=None, repr=False, compare=False
     )
@@ -813,7 +819,8 @@ async def call_runner(
     one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. Every error
     raised from a runner request carries ``payment_sent``: True when the failed
     attempt already carried ``Livepeer-Payment`` headers, so a gateway knows whether
-    it may still try another runner.
+    it may still try another runner. Results, streams, and those errors also carry
+    ``auth_ids``, the signer auth id of every payment session the call opened.
 
     ``application/json`` and ``+json`` types parse into ``result.data``; anything else
     (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``.
@@ -834,8 +841,14 @@ async def call_runner(
         payer_address = cast(str, signer.address)
     challenge: LivePaymentChallenge | None = None
     any_payment_sent = False
-    paid_manifest_id = ""
+    auth_ids: list[str] = []
     attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2
+
+    def remember_auth_id(session: LivePaymentSession | None) -> None:
+        if session is None or not session.auth_id or session.auth_id in auth_ids:
+            return
+        auth_ids.append(session.auth_id)
+
     for attempt in range(attempts):
         payment_session: LivePaymentSession | None = None
         session_id = ""
@@ -858,6 +871,7 @@ async def call_runner(
                 )
             except SignerRefreshRequired as e:
                 if attempt + 1 >= attempts:
+                    e.auth_ids = tuple(auth_ids)
                     raise
                 # Could happen if embedded payment params expire; just retry in this case.
                 _LOG.info(
@@ -866,6 +880,13 @@ async def call_runner(
                 )
                 challenge = None
                 continue
+            except LivepeerGatewayError as e:
+                for auth_id in e.auth_ids:
+                    if auth_id not in auth_ids:
+                        auth_ids.append(auth_id)
+                e.auth_ids = tuple(auth_ids)
+                raise
+            remember_auth_id(payment_session)
             request_headers["Livepeer-Payment"] = payment.payment
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
@@ -876,8 +897,11 @@ async def call_runner(
         # Once tickets go out with the request, the call is bound to this runner.
         payment_sent = "Livepeer-Payment" in request_headers
         any_payment_sent = any_payment_sent or payment_sent
-        if payment_sent:
-            paid_manifest_id = session_id
+
+        def stamp(error: LivepeerGatewayError) -> None:
+            error.payment_sent = payment_sent
+            error.auth_ids = tuple(auth_ids)
+
         try:
             request_kwargs: dict[str, Any] = {"timeout": timeout}
             if request_headers:
@@ -901,6 +925,7 @@ async def call_runner(
                     session,
                     resp,
                     session_id=session_id,
+                    auth_ids=tuple(auth_ids),
                 )
                 # The stream outlives this call, so it owns the funding.
                 if needs_ongoing_funding:
@@ -948,29 +973,29 @@ async def call_runner(
                 payment_session=None if payment_type == "fixed" else payment_session,
                 content=None if is_json else body,
                 content_type=content_type,
+                auth_ids=tuple(auth_ids),
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
-                e.payment_sent = payment_sent
-                e.manifest_id = session_id
+                stamp(e)
                 raise
             if not signer_url:
-                raise LivepeerGatewayError("Live runner paid call requires signer_url") from e
+                unpaid = LivepeerGatewayError("Live runner paid call requires signer_url")
+                stamp(unpaid)
+                raise unpaid from e
             try:
                 challenge = _parse_runner_payment_challenge(e)
             except LivepeerGatewayError as parse_error:
-                parse_error.payment_sent = payment_sent
-                parse_error.manifest_id = session_id
+                stamp(parse_error)
                 raise
             continue
         except LivepeerGatewayError as e:
-            e.payment_sent = payment_sent
-            e.manifest_id = session_id
+            stamp(e)
             raise
 
     exhausted = LivepeerGatewayError("Live runner call exhausted payment challenge retries")
     exhausted.payment_sent = any_payment_sent
-    exhausted.manifest_id = paid_manifest_id
+    exhausted.auth_ids = tuple(auth_ids)
     raise exhausted
 
 
@@ -1039,14 +1064,26 @@ async def _get_runner_payment(
         app=app,
         max_price=max_price.to_json() if max_price is not None else None,
     )
-    payment = await session.get_payment()
+    try:
+        payment = await session.get_payment()
+    except LivepeerGatewayError as error:
+        error.auth_ids = _signed_auth_ids(session)
+        raise
     if not payment.payment:
-        raise LivepeerGatewayError("Live runner payment response missing payment")
+        missing = LivepeerGatewayError("Live runner payment response missing payment")
+        missing.auth_ids = _signed_auth_ids(session)
+        raise missing
     if not payment.seg_creds:
-        raise LivepeerGatewayError("Live runner payment response missing segCreds")
+        missing = LivepeerGatewayError("Live runner payment response missing segCreds")
+        missing.auth_ids = _signed_auth_ids(session)
+        raise missing
     return session, payment
 
 
+def _signed_auth_ids(session: LivePaymentSession) -> tuple[str, ...]:
+    return (session.auth_id,) if session.auth_id else ()
+
+
 def _runner_payment_type(
     runner: LiveRunnerInstance | None,
     payment_unit: str | None = None,
diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index dd4554f..637036d 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -6,6 +6,7 @@
 import logging
 import re
 import ssl
+from collections.abc import Mapping
 from dataclasses import dataclass, replace
 from functools import lru_cache
 from typing import Any, Optional
@@ -26,6 +27,28 @@
 # Must stay under the signer's opening payment: 10s per-second, 60s pixel.
 PAYMENT_INTERVAL_S = 3.0
 
+
+def auth_id_from_signer_state(state: Mapping[str, Any] | None) -> str:
+    """Read the auth id sealed in a remote-signer payment state.
+
+    go-livepeer returns that state as base64-encoded JSON under ``State``.
+    ``AuthID`` is the id its authorization webhook assigned. An unreadable
+    state, or one with no auth id yet, reads as empty.
+    """
+    if not state:
+        return ""
+    raw = state.get("State")
+    if not isinstance(raw, str) or not raw:
+        return ""
+    try:
+        payload = json.loads(base64.standard_b64decode(raw))
+    except (ValueError, UnicodeDecodeError):
+        return ""
+    if not isinstance(payload, dict):
+        return ""
+    auth_id = payload.get("AuthID")
+    return auth_id if isinstance(auth_id, str) else ""
+
 @dataclass(frozen=True)
 class GetPaymentResponse:
     payment: str
@@ -233,6 +256,11 @@ def __init__(
         self._max_refresh_retries = max(0, int(max_refresh_retries))
         self._state: dict[str, Any] | None = None
 
+    @property
+    def auth_id(self) -> str:
+        """The signer's sealed auth id, empty until the first payment response."""
+        return auth_id_from_signer_state(self._state)
+
     async def get_payment(self) -> GetPaymentResponse:
         if not self._signer_url:
             return GetPaymentResponse(payment="", seg_creds=None)
@@ -388,6 +416,11 @@ def __init__(
         self._max_refresh_retries = max(0, int(max_refresh_retries))
         self._state: Optional[dict[str, str]] = None
 
+    @property
+    def auth_id(self) -> str:
+        """The signer's sealed auth id, empty until the first payment response."""
+        return auth_id_from_signer_state(self._state)
+
     def set_manifest_id(self, manifest_id: str) -> None:
         if not isinstance(manifest_id, str) or not manifest_id.strip():
             raise PaymentError("manifest_id must be a non-empty string")
diff --git a/tests/test_errors.py b/tests/test_errors.py
index 75a6ed8..8397436 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -40,13 +40,13 @@ def test_no_orchestrator_available_error_includes_rejections() -> None:
 def test_paid_503_is_a_capacity_rejection() -> None:
     error = LivepeerHTTPError(503, "https://runner.example.com", "busy")
     error.payment_sent = True
-    error.manifest_id = "manifest-1"
+    error.auth_ids = ("auth-1",)
 
     rejection = runner_rejection("https://runner.example.com", error)
 
     assert rejection.kind == "capacity"
     assert rejection.payment_sent is True
-    assert rejection.manifest_id == "manifest-1"
+    assert rejection.auth_ids == ("auth-1",)
     assert rejection.reason == str(error)
 
 
@@ -57,7 +57,7 @@ def test_http_400_is_another_kind_of_rejection() -> None:
 
     assert rejection.kind == "other"
     assert rejection.payment_sent is False
-    assert rejection.manifest_id == ""
+    assert rejection.auth_ids == ()
 
 
 def test_connection_refused_is_unreachable() -> None:
@@ -98,18 +98,29 @@ def test_text_mentioning_capacity_is_not_a_capacity_refusal() -> None:
 
 def test_no_runner_available_keeps_the_paid_rejection() -> None:
     error = NoRunnerAvailableError(
-        "All runners failed (1 tried)",
+        "All runners failed (2 tried)",
         rejections=[
             RunnerRejection(
                 url="https://runner.example.com",
                 reason="busy",
                 kind="capacity",
                 payment_sent=True,
-                manifest_id="manifest-1",
-            )
+                auth_ids=("auth-1",),
+            ),
+            RunnerRejection(
+                url="https://runner-b.example.com",
+                reason="busy",
+                kind="capacity",
+                payment_sent=True,
+                auth_ids=("auth-2",),
+            ),
         ],
     )
 
     assert error.payment_sent is True
-    assert error.manifest_id == "manifest-1"
-    assert str(error) == "All runners failed (1 tried): https://runner.example.com: busy"
+    assert error.auth_ids == ("auth-1", "auth-2")
+    assert str(error) == (
+        "All runners failed (2 tried): "
+        "https://runner.example.com: busy; "
+        "https://runner-b.example.com: busy"
+    )
diff --git a/tests/test_live_payment_session.py b/tests/test_live_payment_session.py
index 35afbb5..4e50b4b 100644
--- a/tests/test_live_payment_session.py
+++ b/tests/test_live_payment_session.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import base64
+import json
 import types
 from unittest import mock
 
@@ -14,6 +16,7 @@
     LivePaymentChallenge,
     LivePaymentSession,
     PaymentSession,
+    auth_id_from_signer_state,
     get_signer_info,
 )
 
@@ -360,3 +363,41 @@ async def _post_json(
         assert first.address == "opaque-sender"
         assert first.sig == "opaque-signature"
         assert len(calls) == 1
+
+    async def test_auth_id_is_read_from_the_signed_state(self) -> None:
+        signed = base64.standard_b64encode(
+            json.dumps({"StateID": "state-1", "AuthID": "auth-1"}).encode()
+        ).decode()
+
+        async def _post_json(
+            url: str,
+            payload: dict[str, object],
+            *,
+            headers: dict[str, str] | None = None,
+            timeout: float = 5.0,
+        ) -> dict[str, object]:
+            del url, payload, headers, timeout
+            return {
+                "payment": "payment",
+                "segCreds": "segment",
+                "state": {"State": signed, "Sig": "sig"},
+            }
+
+        with mock.patch("livepeer_gateway.http.post_json", side_effect=_post_json):
+            session = LivePaymentSession(
+                "https://signer.example.com",
+                type="fixed",
+                challenge=_challenge(),
+            )
+            assert session.auth_id == ""
+            await session.get_payment()
+
+        assert session.auth_id == "auth-1"
+
+
+def test_auth_id_is_empty_when_the_signer_state_has_none() -> None:
+    assert auth_id_from_signer_state(None) == ""
+    assert auth_id_from_signer_state({"state": "lowercase"}) == ""
+    assert auth_id_from_signer_state({"State": "not-json"}) == ""
+    encoded = base64.standard_b64encode(json.dumps({"StateID": "state-1"}).encode()).decode()
+    assert auth_id_from_signer_state({"State": encoded}) == ""
diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py
index fbf8586..767c0a5 100644
--- a/tests/test_live_runner.py
+++ b/tests/test_live_runner.py
@@ -301,6 +301,8 @@ def __init__(self, signer_url: str, **kwargs: object) -> None:
                 payment_sessions.append(self)
                 sessions.append({"signer_url": signer_url, **kwargs})
 
+            auth_id = ""
+
             async def get_payment(self) -> object:
                 return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64")
 
@@ -396,6 +398,8 @@ async def test_paid_scope_runner_uses_lv2v_payment_type(self) -> None:
         )
 
         class _PaymentSession:
+            auth_id = ""
+
             def __init__(self, signer_url: str, **kwargs: object) -> None:
                 sessions.append({"signer_url": signer_url, **kwargs})
 
@@ -469,6 +473,7 @@ async def test_paid_fixed_runner_retries_with_fresh_payment_and_no_renewal_sessi
         class _PaymentSession:
             def __init__(self, signer_url: str, **kwargs: object) -> None:
                 sessions.append({"signer_url": signer_url, **kwargs})
+                self.auth_id = f"auth-{len(sessions)}"
 
             async def get_payment(self) -> object:
                 payment_number = len(sessions)
@@ -532,6 +537,7 @@ def _request_body(
         assert sessions[0]["challenge"] == _payment_challenge("fixed-manifest")
         assert sessions[1]["challenge"] == _payment_challenge("fixed-manifest")
         assert result.payment_session is None
+        assert result.auth_ids == ("auth-1", "auth-2")
 
     async def test_paid_call_restarts_challenge_when_signer_requests_refresh(
         self,
@@ -546,6 +552,8 @@ async def test_paid_call_restarts_challenge_when_signer_requests_refresh(
         runner_url = "https://service.example.com/apps/runner-1/session"
 
         class _PaymentSession:
+            auth_id = ""
+
             def __init__(self, signer_url: str, **kwargs: object) -> None:
                 payment_sessions.append(self)
                 sessions.append({"signer_url": signer_url, **kwargs})
@@ -645,6 +653,8 @@ async def test_paid_call_repeated_refresh_requests_are_bounded(self) -> None:
         runner_url = "https://service.example.com/apps/runner-1/session"
 
         class _PaymentSession:
+            auth_id = ""
+
             def __init__(self, signer_url: str, **kwargs: object) -> None:
                 del signer_url, kwargs
 
@@ -1874,6 +1884,7 @@ class _FakePaymentSession:
     def __init__(self, signer_url: str, **kwargs: object) -> None:
         self.signer_url = signer_url
         self.kwargs = kwargs
+        self.auth_id = f"auth-{len(type(self).instances) + 1}"
         type(self).instances.append(self)
 
     async def get_payment(self) -> object:
@@ -1961,6 +1972,7 @@ async def test_stream_carries_manifest_id(self) -> None:
             with _fake_signer():
                 async with await _paid_call(base, stream=True) as stream:
                     assert stream.session_id == "manifest-1"
+                    assert stream.auth_ids == ("auth-1",)
                     assert stream.payment_session is _FakePaymentSession.instances[0]
                     lines = [line async for line in stream.aiter_lines() if line]
 
@@ -1976,7 +1988,7 @@ async def test_stream_without_signer_has_empty_session_id(self) -> None:
 
 class TestCallRunnerPaymentSent:
     @pytest.mark.parametrize("status", [500, 400])
-    async def test_error_after_payment_carries_the_manifest(self, status: int) -> None:
+    async def test_error_after_payment_carries_the_auth_id(self, status: int) -> None:
         seen: list[_RecordedRequest] = []
         async with _serve(_runner_app(seen, statuses=[402, status])) as base:
             with _fake_signer():
@@ -1984,11 +1996,11 @@ async def test_error_after_payment_carries_the_manifest(self, status: int) -> No
                     await _paid_call(base)
         assert info.value.status_code == status
         assert info.value.payment_sent is True
-        assert info.value.manifest_id == "manifest-1"
+        assert info.value.auth_ids == ("auth-1",)
         assert len(seen) == 2
 
     @pytest.mark.parametrize("status", [400, 503])
-    async def test_error_before_payment_has_no_manifest(self, status: int) -> None:
+    async def test_error_before_payment_has_no_auth_id(self, status: int) -> None:
         seen: list[_RecordedRequest] = []
         async with _serve(_runner_app(seen, statuses=[status])) as base:
             with _fake_signer():
@@ -1996,7 +2008,7 @@ async def test_error_before_payment_has_no_manifest(self, status: int) -> None:
                     await _paid_call(base)
         assert info.value.status_code == status
         assert info.value.payment_sent is False
-        assert info.value.manifest_id == ""
+        assert info.value.auth_ids == ()
         assert len(seen) == 1
 
     async def test_connection_refused_before_challenge(self) -> None:
@@ -2012,7 +2024,7 @@ async def test_stream_error_after_payment_says_so(self) -> None:
                     await _paid_call(base, stream=True)
         assert info.value.status_code == 500
         assert info.value.payment_sent is True
-        assert info.value.manifest_id == "manifest-1"
+        assert info.value.auth_ids == ("auth-1",)
 
     async def test_exhausted_retries_after_payment_says_so(self) -> None:
         seen: list[_RecordedRequest] = []
@@ -2021,7 +2033,7 @@ async def test_exhausted_retries_after_payment_says_so(self) -> None:
                 with pytest.raises(LivepeerGatewayError, match="exhausted") as info:
                     await _paid_call(base, max_payment_challenge_retries=1)
         assert info.value.payment_sent is True
-        assert info.value.manifest_id == "manifest-1"
+        assert info.value.auth_ids == ("auth-1", "auth-2", "auth-3")
 
     async def test_paid_call_without_signer_is_unpaid(self) -> None:
         seen: list[_RecordedRequest] = []
@@ -2029,4 +2041,4 @@ async def test_paid_call_without_signer_is_unpaid(self) -> None:
             with pytest.raises(LivepeerGatewayError, match="requires signer_url") as info:
                 await call_runner(f"{base}/call", payload={"prompt": "hi"})
         assert info.value.payment_sent is False
-        assert info.value.manifest_id == ""
+        assert info.value.auth_ids == ()
diff --git a/tests/test_selection.py b/tests/test_selection.py
index 5f8a5c9..3ab3ece 100644
--- a/tests/test_selection.py
+++ b/tests/test_selection.py
@@ -193,7 +193,7 @@ async def _call_runner(
             if runner.url.endswith("/a/session"):
                 refused = LivepeerHTTPError(503, runner.url, "busy")
                 refused.payment_sent = True
-                refused.manifest_id = "manifest-1"
+                refused.auth_ids = ("auth-1",)
                 raise refused
             try:
                 raise LivepeerGatewayError("connection refused") from ConnectionRefusedError()
@@ -223,11 +223,11 @@ async def _call_runner(
 
         assert cursor.rejections[0].kind == "capacity"
         assert cursor.rejections[0].payment_sent is True
-        assert cursor.rejections[0].manifest_id == "manifest-1"
+        assert cursor.rejections[0].auth_ids == ("auth-1",)
         assert cursor.rejections[1].kind == "unreachable"
         assert cursor.rejections[1].payment_sent is False
         assert raised.value.payment_sent is True
-        assert raised.value.manifest_id == "manifest-1"
+        assert raised.value.auth_ids == ("auth-1",)
 
     async def test_runner_selector_empty_discovery_raises_no_runner_available(
         self,

From e389f49a878d326c3bc435c4dc8f8857b3d4cdd2 Mon Sep 17 00:00:00 2001
From: John | Elite Encoder <john@eliteencoder.net>
Date: Mon, 12 Oct 2026 12:20:32 -0400
Subject: [PATCH 4/6] Deduplicate auth id bookkeeping in call_runner

Merge auth ids through one unique_auth_ids helper, attach the signed
session's auth id in a single place in _get_runner_payment, decode the
signer state's auth id once per payment response, drop the unused
PaymentSession.auth_id, and simplify the unreachable exception check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/livepeer_gateway/errors.py        | 17 +++++----
 src/livepeer_gateway/live_runner.py   | 50 +++++++++++----------------
 src/livepeer_gateway/remote_signer.py |  9 ++---
 3 files changed, 33 insertions(+), 43 deletions(-)

diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py
index 9e48559..9f928ea 100644
--- a/src/livepeer_gateway/errors.py
+++ b/src/livepeer_gateway/errors.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import Literal
 
@@ -23,6 +24,11 @@ class LivepeerGatewayError(RuntimeError):
     auth_ids: tuple[str, ...] = ()
 
 
+def unique_auth_ids(auth_ids: Iterable[str]) -> tuple[str, ...]:
+    """Non-empty auth ids, first occurrence order, without repeats."""
+    return tuple(dict.fromkeys(auth_id for auth_id in auth_ids if auth_id))
+
+
 class LivepeerHTTPError(LivepeerGatewayError):
     """Raised when an HTTP endpoint returns a non-success status."""
 
@@ -82,7 +88,7 @@ def _runner_failure_kind(error: BaseException) -> RunnerFailureKind:
         return "capacity"
     current: BaseException | None = error
     while current is not None:
-        if isinstance(current, (TimeoutError, ConnectionError, OSError, aiohttp.ClientConnectionError)):
+        if isinstance(current, (OSError, aiohttp.ClientConnectionError)):
             return "unreachable"
         current = current.__cause__
     return "other"
@@ -111,12 +117,9 @@ def __init__(self, message: str, rejections: list[RunnerRejection] | None = None
         self.rejections: list[RunnerRejection] = rejections or []
         paid = [rejection for rejection in self.rejections if rejection.payment_sent]
         self.payment_sent = bool(paid)
-        auth_ids: list[str] = []
-        for rejection in self.rejections:
-            for auth_id in rejection.auth_ids:
-                if auth_id not in auth_ids:
-                    auth_ids.append(auth_id)
-        self.auth_ids = tuple(auth_ids)
+        self.auth_ids = unique_auth_ids(
+            auth_id for rejection in self.rejections for auth_id in rejection.auth_ids
+        )
 
     def __str__(self) -> str:
         message = super().__str__()
diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 5466f9e..5077818 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -27,7 +27,12 @@
 from aiohttp.helpers import parse_mimetype
 
 from .channel_reader import ChannelReader
-from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired
+from .errors import (
+    LivepeerGatewayError,
+    LivepeerHTTPError,
+    SignerRefreshRequired,
+    unique_auth_ids,
+)
 from .http import _post_empty, _request_body, open_stream, post_json, request_json
 from .remote_signer import (
     GetPaymentResponse,
@@ -841,14 +846,9 @@ async def call_runner(
         payer_address = cast(str, signer.address)
     challenge: LivePaymentChallenge | None = None
     any_payment_sent = False
-    auth_ids: list[str] = []
+    auth_ids: tuple[str, ...] = ()
     attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2
 
-    def remember_auth_id(session: LivePaymentSession | None) -> None:
-        if session is None or not session.auth_id or session.auth_id in auth_ids:
-            return
-        auth_ids.append(session.auth_id)
-
     for attempt in range(attempts):
         payment_session: LivePaymentSession | None = None
         session_id = ""
@@ -871,7 +871,7 @@ def remember_auth_id(session: LivePaymentSession | None) -> None:
                 )
             except SignerRefreshRequired as e:
                 if attempt + 1 >= attempts:
-                    e.auth_ids = tuple(auth_ids)
+                    e.auth_ids = unique_auth_ids((*auth_ids, *e.auth_ids))
                     raise
                 # Could happen if embedded payment params expire; just retry in this case.
                 _LOG.info(
@@ -881,12 +881,9 @@ def remember_auth_id(session: LivePaymentSession | None) -> None:
                 challenge = None
                 continue
             except LivepeerGatewayError as e:
-                for auth_id in e.auth_ids:
-                    if auth_id not in auth_ids:
-                        auth_ids.append(auth_id)
-                e.auth_ids = tuple(auth_ids)
+                e.auth_ids = unique_auth_ids((*auth_ids, *e.auth_ids))
                 raise
-            remember_auth_id(payment_session)
+            auth_ids = unique_auth_ids((*auth_ids, payment_session.auth_id))
             request_headers["Livepeer-Payment"] = payment.payment
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
@@ -900,7 +897,7 @@ def remember_auth_id(session: LivePaymentSession | None) -> None:
 
         def stamp(error: LivepeerGatewayError) -> None:
             error.payment_sent = payment_sent
-            error.auth_ids = tuple(auth_ids)
+            error.auth_ids = auth_ids
 
         try:
             request_kwargs: dict[str, Any] = {"timeout": timeout}
@@ -925,7 +922,7 @@ def stamp(error: LivepeerGatewayError) -> None:
                     session,
                     resp,
                     session_id=session_id,
-                    auth_ids=tuple(auth_ids),
+                    auth_ids=auth_ids,
                 )
                 # The stream outlives this call, so it owns the funding.
                 if needs_ongoing_funding:
@@ -973,7 +970,7 @@ def stamp(error: LivepeerGatewayError) -> None:
                 payment_session=None if payment_type == "fixed" else payment_session,
                 content=None if is_json else body,
                 content_type=content_type,
-                auth_ids=tuple(auth_ids),
+                auth_ids=auth_ids,
             )
         except LivepeerHTTPError as e:
             if e.status_code != 402:
@@ -995,7 +992,7 @@ def stamp(error: LivepeerGatewayError) -> None:
 
     exhausted = LivepeerGatewayError("Live runner call exhausted payment challenge retries")
     exhausted.payment_sent = any_payment_sent
-    exhausted.auth_ids = tuple(auth_ids)
+    exhausted.auth_ids = auth_ids
     raise exhausted
 
 
@@ -1066,24 +1063,17 @@ async def _get_runner_payment(
     )
     try:
         payment = await session.get_payment()
+        if not payment.payment:
+            raise LivepeerGatewayError("Live runner payment response missing payment")
+        if not payment.seg_creds:
+            raise LivepeerGatewayError("Live runner payment response missing segCreds")
     except LivepeerGatewayError as error:
-        error.auth_ids = _signed_auth_ids(session)
+        # The signer charges when it signs, so a signed session counts even on failure.
+        error.auth_ids = unique_auth_ids((session.auth_id,))
         raise
-    if not payment.payment:
-        missing = LivepeerGatewayError("Live runner payment response missing payment")
-        missing.auth_ids = _signed_auth_ids(session)
-        raise missing
-    if not payment.seg_creds:
-        missing = LivepeerGatewayError("Live runner payment response missing segCreds")
-        missing.auth_ids = _signed_auth_ids(session)
-        raise missing
     return session, payment
 
 
-def _signed_auth_ids(session: LivePaymentSession) -> tuple[str, ...]:
-    return (session.auth_id,) if session.auth_id else ()
-
-
 def _runner_payment_type(
     runner: LiveRunnerInstance | None,
     payment_unit: str | None = None,
diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 637036d..4ff9337 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -255,11 +255,12 @@ def __init__(
         self._max_price = dict(max_price) if max_price is not None else None
         self._max_refresh_retries = max(0, int(max_refresh_retries))
         self._state: dict[str, Any] | None = None
+        self._auth_id = ""
 
     @property
     def auth_id(self) -> str:
         """The signer's sealed auth id, empty until the first payment response."""
-        return auth_id_from_signer_state(self._state)
+        return self._auth_id
 
     async def get_payment(self) -> GetPaymentResponse:
         if not self._signer_url:
@@ -362,6 +363,7 @@ async def _payment_request(self) -> GetPaymentResponse:
             )
 
         self._state = state
+        self._auth_id = auth_id_from_signer_state(state)
         return GetPaymentResponse(payment=payment, seg_creds=seg_creds)
 
     async def _refresh_payment_params(self) -> None:
@@ -416,11 +418,6 @@ def __init__(
         self._max_refresh_retries = max(0, int(max_refresh_retries))
         self._state: Optional[dict[str, str]] = None
 
-    @property
-    def auth_id(self) -> str:
-        """The signer's sealed auth id, empty until the first payment response."""
-        return auth_id_from_signer_state(self._state)
-
     def set_manifest_id(self, manifest_id: str) -> None:
         if not isinstance(manifest_id, str) or not manifest_id.strip():
             raise PaymentError("manifest_id must be a non-empty string")

From d8e2b80048feae7f831be1f10206f63fa73b963e Mon Sep 17 00:00:00 2001
From: John | Elite Encoder <john@eliteencoder.net>
Date: Mon, 12 Oct 2026 12:24:07 -0400
Subject: [PATCH 5/6] Record each call_runner attempt in a shared
 RunnerCallRecord

call_runner now writes every attempt to a RunnerCallRecord as it happens:
the challenge answered, the signer auth id, whether the request left with
Livepeer-Payment, the runner's status, and how it ended. The result,
stream, and any error hold that same record, and a caller can pass its
own to read it after cancellation. payment_sent and auth_ids on errors,
results, and streams are derived from it, which replaces per-raise-site
stamping that missed signer failures, refresh retries, and non-gateway
exceptions.

Attempt outcomes report "unreachable" only when the request provably
never left; a bare timeout reads as "timeout". Unexpected exceptions are
wrapped in LivepeerGatewayError so they carry the record, and open_stream
now closes its session when reading an error body fails.

Runner failover (selection.py, scope.py, RunnerRejection) is reverted to
main and moves to a follow-up PR.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/livepeer_gateway/errors.py      |  87 +++------
 src/livepeer_gateway/http.py        |   8 +-
 src/livepeer_gateway/live_runner.py | 272 ++++++++++++++++++++++------
 src/livepeer_gateway/scope.py       |   7 +-
 src/livepeer_gateway/selection.py   |   6 +-
 tests/test_errors.py                |  96 ----------
 tests/test_live_runner.py           | 138 +++++++++++++-
 tests/test_selection.py             |  52 +-----
 8 files changed, 386 insertions(+), 280 deletions(-)

diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py
index 9f928ea..8c50eb8 100644
--- a/src/livepeer_gateway/errors.py
+++ b/src/livepeer_gateway/errors.py
@@ -1,32 +1,39 @@
 from __future__ import annotations
 
-from collections.abc import Iterable
 from dataclasses import dataclass
-from typing import Literal
+from typing import TYPE_CHECKING
 
-import aiohttp
+if TYPE_CHECKING:
+    from .live_runner import RunnerCallRecord
 
 
 class LivepeerGatewayError(RuntimeError):
     """Base error for the library.
 
-    ``payment_sent`` is True when the error comes from a ``call_runner`` attempt
-    whose request carried ``Livepeer-Payment`` headers, that is, after a payment
-    challenge had already been answered. A gateway may retry another runner only
-    while it is False; once tickets are sent the request is bound to that runner.
-
-    ``auth_ids`` are the signer auth ids of every payment session this call
-    opened. The signer charges an allocation when it signs, which can be before
-    ``payment_sent`` becomes True. Empty when no session was signed.
+    ``record`` is set on every error ``call_runner`` raises: the attempts it
+    made against the runner, with what was signed, sent, and how each ended.
+    ``payment_sent`` and ``auth_ids`` read from it, and are empty without one.
     """
 
-    payment_sent: bool = False
-    auth_ids: tuple[str, ...] = ()
+    record: RunnerCallRecord | None = None
+
+    @property
+    def payment_sent(self) -> bool:
+        """True once any attempt's request carried ``Livepeer-Payment`` headers.
 
+        The call is then bound to this runner; another runner is safe only
+        while this is False.
+        """
+        return self.record is not None and self.record.payment_sent
 
-def unique_auth_ids(auth_ids: Iterable[str]) -> tuple[str, ...]:
-    """Non-empty auth ids, first occurrence order, without repeats."""
-    return tuple(dict.fromkeys(auth_id for auth_id in auth_ids if auth_id))
+    @property
+    def auth_ids(self) -> tuple[str, ...]:
+        """Signer auth ids of every payment session the call opened.
+
+        The signer charges when it signs, so these can be set while
+        ``payment_sent`` is still False.
+        """
+        return self.record.auth_ids if self.record is not None else ()
 
 
 class LivepeerHTTPError(LivepeerGatewayError):
@@ -46,52 +53,11 @@ class OrchestratorRejection:
     reason: str
 
 
-RunnerFailureKind = Literal["capacity", "unreachable", "other"]
-
-
 @dataclass
 class RunnerRejection:
-    """Records a single runner that was tried and rejected.
-
-    ``kind`` is ``capacity`` for an HTTP 503, ``unreachable`` when the runner
-    never answered, and ``other`` for every remaining failure. ``payment_sent``
-    is copied from the ``call_runner`` error: True only after that attempt's
-    request carried ``Livepeer-Payment`` headers. ``auth_ids`` are the signer
-    auth ids that attempt opened, which can be set before ``payment_sent``.
-    """
-
+    """Records a single runner that was tried and rejected."""
     url: str
     reason: str
-    kind: RunnerFailureKind = "other"
-    payment_sent: bool = False
-    auth_ids: tuple[str, ...] = ()
-
-
-def runner_rejection(url: str, error: BaseException) -> RunnerRejection:
-    """Record one failed runner attempt without dropping payment or its kind."""
-    payment_sent = False
-    auth_ids: tuple[str, ...] = ()
-    if isinstance(error, LivepeerGatewayError):
-        payment_sent = error.payment_sent
-        auth_ids = error.auth_ids
-    return RunnerRejection(
-        url=url,
-        reason=str(error),
-        kind=_runner_failure_kind(error),
-        payment_sent=payment_sent,
-        auth_ids=auth_ids,
-    )
-
-
-def _runner_failure_kind(error: BaseException) -> RunnerFailureKind:
-    if isinstance(error, LivepeerHTTPError) and error.status_code == 503:
-        return "capacity"
-    current: BaseException | None = error
-    while current is not None:
-        if isinstance(current, (OSError, aiohttp.ClientConnectionError)):
-            return "unreachable"
-        current = current.__cause__
-    return "other"
 
 
 class NoOrchestratorAvailableError(LivepeerGatewayError):
@@ -115,11 +81,6 @@ class NoRunnerAvailableError(LivepeerGatewayError):
     def __init__(self, message: str, rejections: list[RunnerRejection] | None = None) -> None:
         super().__init__(message)
         self.rejections: list[RunnerRejection] = rejections or []
-        paid = [rejection for rejection in self.rejections if rejection.payment_sent]
-        self.payment_sent = bool(paid)
-        self.auth_ids = unique_auth_ids(
-            auth_id for rejection in self.rejections for auth_id in rejection.auth_ids
-        )
 
     def __str__(self) -> str:
         message = super().__str__()
diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py
index 699d042..c4bef5d 100644
--- a/src/livepeer_gateway/http.py
+++ b/src/livepeer_gateway/http.py
@@ -365,9 +365,11 @@ async def open_stream(
             f"HTTP stream error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})"
         ) from e
     if resp.status >= 400:
-        raw = await resp.text()
-        resp.release()
-        await session.close()
+        try:
+            raw = await resp.text()
+        finally:
+            resp.release()
+            await session.close()
         _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items()))
     return session, resp
 
diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 5077818..7d0bdf6 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -27,12 +27,7 @@
 from aiohttp.helpers import parse_mimetype
 
 from .channel_reader import ChannelReader
-from .errors import (
-    LivepeerGatewayError,
-    LivepeerHTTPError,
-    SignerRefreshRequired,
-    unique_auth_ids,
-)
+from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired
 from .http import _post_empty, _request_body, open_stream, post_json, request_json
 from .remote_signer import (
     GetPaymentResponse,
@@ -167,6 +162,69 @@ class LiveRunnerProxy:
     url: str
 
 
+RunnerAttemptOutcome = Literal[
+    "pending",  # still in flight
+    "succeeded",
+    "payment_required",  # runner answered 402; the next attempt pays
+    "refused",  # runner answered 503: no capacity
+    "http",  # runner answered with any other error status
+    "payment",  # the signer failed, so the request never left
+    "unreachable",  # the request never reached the runner
+    "timeout",  # no answer in time; the runner may have the request
+    "disconnected",  # the connection dropped after the request left
+    "invalid_response",
+    "cancelled",
+    "error",
+]
+
+
+@dataclass
+class RunnerAttempt:
+    """One request ``call_runner`` made, or set out to make, to the runner."""
+
+    # The payment challenge this attempt answered; "" when it went out unpaid.
+    manifest_id: str = ""
+    # Signer auth id of the payment session this attempt opened. The signer
+    # charges when it signs, so this is set even if the request never left.
+    auth_id: str = ""
+    # True once the request left with Livepeer-Payment headers.
+    payment_sent: bool = False
+    # The runner's HTTP status, when known.
+    status_code: int | None = None
+    outcome: RunnerAttemptOutcome = "pending"
+
+
+@dataclass
+class RunnerCallRecord:
+    """Every attempt of one ``call_runner`` call, written as each step happens.
+
+    The result, stream, or error of the call holds this same record. Pass one
+    to ``call_runner`` to read it even when the call is cancelled.
+    """
+
+    runner_url: str = ""
+    attempts: list[RunnerAttempt] = field(default_factory=list)
+
+    @property
+    def payment_sent(self) -> bool:
+        """True once any attempt's request left with ``Livepeer-Payment`` headers.
+
+        The call is then bound to this runner; another runner is safe only
+        while this is False.
+        """
+        return any(attempt.payment_sent for attempt in self.attempts)
+
+    @property
+    def auth_ids(self) -> tuple[str, ...]:
+        """Signer auth ids of every payment session the call opened, in order."""
+        return tuple(dict.fromkeys(a.auth_id for a in self.attempts if a.auth_id))
+
+    @property
+    def outcome(self) -> RunnerAttemptOutcome:
+        """How the last attempt ended; ``pending`` before the first one."""
+        return self.attempts[-1].outcome if self.attempts else "pending"
+
+
 @dataclass(frozen=True)
 class LiveRunnerCallResult:
     data: dict[str, Any]
@@ -181,9 +239,23 @@ class LiveRunnerCallResult:
     # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty.
     content: bytes | None = field(default=None, repr=False)
     content_type: str = ""
-    # Signer auth ids of every payment session this call opened. Fixed-price
-    # calls still list them here, because their payment_session is not returned.
-    auth_ids: tuple[str, ...] = ()
+    record: RunnerCallRecord = field(
+        default_factory=RunnerCallRecord,
+        repr=False,
+        compare=False,
+    )
+
+    @property
+    def payment_sent(self) -> bool:
+        return self.record.payment_sent
+
+    @property
+    def auth_ids(self) -> tuple[str, ...]:
+        """Signer auth ids of every payment session the call opened.
+
+        Fixed-price calls still list them, though their payment_session is not returned.
+        """
+        return self.record.auth_ids
 
 
 @dataclass
@@ -208,8 +280,11 @@ class LiveRunnerCallStream:
     # carries it for a unary call; "" when no challenge was answered.
     # Correlation with the orchestrator, not a billing key.
     session_id: str = ""
-    # Signer auth ids of every payment session this call opened.
-    auth_ids: tuple[str, ...] = ()
+    record: RunnerCallRecord = field(
+        default_factory=RunnerCallRecord,
+        repr=False,
+        compare=False,
+    )
     _payment_task: asyncio.Task[None] | None = field(
         default=None, repr=False, compare=False
     )
@@ -218,6 +293,14 @@ class LiveRunnerCallStream:
     def content_type(self) -> str:
         return self._response.content_type
 
+    @property
+    def payment_sent(self) -> bool:
+        return self.record.payment_sent
+
+    @property
+    def auth_ids(self) -> tuple[str, ...]:
+        return self.record.auth_ids
+
     async def aiter_bytes(self) -> AsyncIterator[bytes]:
         async for chunk in self._response.content.iter_any():
             yield chunk
@@ -782,6 +865,7 @@ async def call_runner(
     payment_unit: str | None = ...,
     timeout: float = ...,
     max_payment_challenge_retries: int = ...,
+    record: RunnerCallRecord | None = ...,
     stream: Literal[False] = False,
 ) -> LiveRunnerCallResult:
     pass
@@ -799,6 +883,7 @@ async def call_runner(
     payment_unit: str | None = ...,
     timeout: float = ...,
     max_payment_challenge_retries: int = ...,
+    record: RunnerCallRecord | None = ...,
     stream: Literal[True],
 ) -> LiveRunnerCallStream:
     pass
@@ -815,26 +900,76 @@ async def call_runner(
     payment_unit: str | None = None,
     timeout: float = 5.0,
     max_payment_challenge_retries: int = 3,
+    record: RunnerCallRecord | None = None,
     stream: bool = False,
 ) -> LiveRunnerCallResult | LiveRunnerCallStream:
     """Call a runner once and return its result (or a live stream if ``stream=True``).
 
     With ``signer_url`` set, payment is automatic and **per call**: a 402 challenge is
     paid via the signer and retried (up to ``max_payment_challenge_retries``), one job,
-    one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. Every error
-    raised from a runner request carries ``payment_sent``: True when the failed
-    attempt already carried ``Livepeer-Payment`` headers, so a gateway knows whether
-    it may still try another runner. Results, streams, and those errors also carry
-    ``auth_ids``, the signer auth id of every payment session the call opened.
+    one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors, and
+    ``LivepeerGatewayError`` on any other failure.
+
+    Every attempt is written to ``record`` as it happens: what was signed, whether
+    the request left with payment, and how it ended. The result, stream, or error
+    holds that record, and its ``payment_sent`` says whether a gateway may still try
+    another runner. Pass ``record`` in to keep it readable if the call is cancelled.
 
     ``application/json`` and ``+json`` types parse into ``result.data``; anything else
     (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``.
 
     The request asks for no particular format, so the app picks what it returns.
     """
+    if record is None:
+        record = RunnerCallRecord()
+    try:
+        return await _call_runner_attempts(
+            record,
+            runner_url,
+            runner=runner,
+            payload=payload,
+            method=method,
+            signer_url=signer_url,
+            signer_headers=signer_headers,
+            payment_unit=payment_unit,
+            timeout=timeout,
+            max_payment_challenge_retries=max_payment_challenge_retries,
+            stream=stream,
+        )
+    except asyncio.CancelledError:
+        _end_attempt(record, "cancelled")
+        raise
+    except LivepeerGatewayError as e:
+        _end_attempt(record, _attempt_outcome(e))
+        e.record = record
+        raise
+    except Exception as e:
+        _end_attempt(record, _attempt_outcome(e))
+        error = LivepeerGatewayError(
+            f"Live runner call failed: {e.__class__.__name__}: {e} (url={record.runner_url})"
+        )
+        error.record = record
+        raise error from e
+
+
+async def _call_runner_attempts(
+    record: RunnerCallRecord,
+    runner_url: str,
+    *,
+    runner: LiveRunnerInstance | None,
+    payload: dict[str, Any] | None,
+    method: str,
+    signer_url: str | None,
+    signer_headers: dict[str, str] | None,
+    payment_unit: str | None,
+    timeout: float,
+    max_payment_challenge_retries: int,
+    stream: bool,
+) -> LiveRunnerCallResult | LiveRunnerCallStream:
     runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "")
     if not runner_url:
         raise LivepeerGatewayError("Live runner call requires runner_url")
+    record.runner_url = runner_url
     request_payload = payload or {}
     payment_type = _runner_payment_type(runner, payment_unit) if signer_url else ""
     max_price: LiveRunnerPriceInfo | None = None
@@ -845,11 +980,10 @@ async def call_runner(
         signer = await get_signer_info(signer_url, _freeze_headers(signer_headers))
         payer_address = cast(str, signer.address)
     challenge: LivePaymentChallenge | None = None
-    any_payment_sent = False
-    auth_ids: tuple[str, ...] = ()
-    attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2
-
-    for attempt in range(attempts):
+    max_attempts = (max(0, int(max_payment_challenge_retries)) + 1) * 2
+    for attempt_index in range(max_attempts):
+        attempt = RunnerAttempt()
+        record.attempts.append(attempt)
         payment_session: LivePaymentSession | None = None
         session_id = ""
         needs_ongoing_funding = False
@@ -860,18 +994,21 @@ async def call_runner(
             request_headers[_LIVE_RUNNER_PAYER_ADDRESS_HEADER] = payer_address
         # Pending challenge means payment is needed.
         if challenge is not None:
+            attempt.manifest_id = challenge.manifest_id
             try:
                 payment_session, payment = await _get_runner_payment(
                     challenge,
+                    attempt,
                     payment_type=payment_type,
                     app=runner.app if runner is not None else None,
                     signer_url=signer_url or "",
                     signer_headers=signer_headers,
                     max_price=max_price,
                 )
-            except SignerRefreshRequired as e:
-                if attempt + 1 >= attempts:
-                    e.auth_ids = unique_auth_ids((*auth_ids, *e.auth_ids))
+            except Exception as e:
+                # Any failure here is the signer's, never the runner's.
+                attempt.outcome = "payment"
+                if not isinstance(e, SignerRefreshRequired) or attempt_index + 1 >= max_attempts:
                     raise
                 # Could happen if embedded payment params expire; just retry in this case.
                 _LOG.info(
@@ -880,10 +1017,6 @@ async def call_runner(
                 )
                 challenge = None
                 continue
-            except LivepeerGatewayError as e:
-                e.auth_ids = unique_auth_ids((*auth_ids, *e.auth_ids))
-                raise
-            auth_ids = unique_auth_ids((*auth_ids, payment_session.auth_id))
             request_headers["Livepeer-Payment"] = payment.payment
             request_headers["Livepeer-Segment"] = payment.seg_creds or ""
             session_id = challenge.manifest_id
@@ -892,13 +1025,7 @@ async def call_runner(
             needs_ongoing_funding = payment_type in _METERED_PAYMENT_TYPES
 
         # Once tickets go out with the request, the call is bound to this runner.
-        payment_sent = "Livepeer-Payment" in request_headers
-        any_payment_sent = any_payment_sent or payment_sent
-
-        def stamp(error: LivepeerGatewayError) -> None:
-            error.payment_sent = payment_sent
-            error.auth_ids = auth_ids
-
+        attempt.payment_sent = "Livepeer-Payment" in request_headers
         try:
             request_kwargs: dict[str, Any] = {"timeout": timeout}
             if request_headers:
@@ -913,6 +1040,8 @@ def stamp(error: LivepeerGatewayError) -> None:
                     payload=request_payload,
                     headers=request_headers or None,
                 )
+                attempt.status_code = resp.status
+                attempt.outcome = "succeeded"
                 call_stream = LiveRunnerCallStream(
                     resp.status,
                     resp.headers,
@@ -922,7 +1051,7 @@ def stamp(error: LivepeerGatewayError) -> None:
                     session,
                     resp,
                     session_id=session_id,
-                    auth_ids=auth_ids,
+                    record=record,
                 )
                 # The stream outlives this call, so it owns the funding.
                 if needs_ongoing_funding:
@@ -951,14 +1080,17 @@ def stamp(error: LivepeerGatewayError) -> None:
                 try:
                     data = json.loads(body)
                 except (UnicodeDecodeError, json.JSONDecodeError) as e:
+                    attempt.outcome = "invalid_response"
                     raise LivepeerGatewayError(
                         f"HTTP JSON error: endpoint did not return valid JSON: {e} "
                         f"(url={runner_url}, content_type={content_type})"
                     ) from e
                 if not isinstance(data, dict):
+                    attempt.outcome = "invalid_response"
                     raise LivepeerGatewayError(
                         f"Live runner call expected JSON object, got {type(data).__name__}"
                     )
+            attempt.outcome = "succeeded"
             return LiveRunnerCallResult(
                 data,
                 runner_url=runner_url,
@@ -970,30 +1102,56 @@ def stamp(error: LivepeerGatewayError) -> None:
                 payment_session=None if payment_type == "fixed" else payment_session,
                 content=None if is_json else body,
                 content_type=content_type,
-                auth_ids=auth_ids,
+                record=record,
             )
         except LivepeerHTTPError as e:
+            attempt.status_code = e.status_code
             if e.status_code != 402:
-                stamp(e)
+                attempt.outcome = "refused" if e.status_code == 503 else "http"
                 raise
             if not signer_url:
-                unpaid = LivepeerGatewayError("Live runner paid call requires signer_url")
-                stamp(unpaid)
-                raise unpaid from e
+                attempt.outcome = "payment_required"
+                raise LivepeerGatewayError("Live runner paid call requires signer_url") from e
             try:
                 challenge = _parse_runner_payment_challenge(e)
-            except LivepeerGatewayError as parse_error:
-                stamp(parse_error)
+            except LivepeerGatewayError:
+                attempt.outcome = "invalid_response"
                 raise
+            attempt.outcome = "payment_required"
             continue
-        except LivepeerGatewayError as e:
-            stamp(e)
-            raise
 
-    exhausted = LivepeerGatewayError("Live runner call exhausted payment challenge retries")
-    exhausted.payment_sent = any_payment_sent
-    exhausted.auth_ids = auth_ids
-    raise exhausted
+    raise LivepeerGatewayError("Live runner call exhausted payment challenge retries")
+
+
+def _end_attempt(record: RunnerCallRecord, outcome: RunnerAttemptOutcome) -> None:
+    """Close the attempt still in flight, if any, with ``outcome``."""
+    if record.attempts and record.attempts[-1].outcome == "pending":
+        record.attempts[-1].outcome = outcome
+
+
+def _attempt_outcome(error: BaseException) -> RunnerAttemptOutcome:
+    """Classify a failed attempt by the network error behind it.
+
+    Only errors that prove the request never left read as ``unreachable``.
+    aiohttp's total timeout raises a bare TimeoutError whether or not it had
+    connected, so that reads as ``timeout``: the runner may have the request.
+    """
+    current: BaseException | None = error
+    while current is not None:
+        if isinstance(
+            current,
+            (aiohttp.ClientConnectorError, aiohttp.ConnectionTimeoutError, ConnectionRefusedError),
+        ):
+            return "unreachable"
+        if isinstance(current, TimeoutError):
+            return "timeout"
+        if isinstance(
+            current,
+            (aiohttp.ServerDisconnectedError, aiohttp.ClientPayloadError, aiohttp.ClientOSError, ConnectionError),
+        ):
+            return "disconnected"
+        current = current.__cause__
+    return "error"
 
 
 def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> LivePaymentChallenge:
@@ -1046,6 +1204,7 @@ async def _stop_funding(task: asyncio.Task[None] | None) -> None:
 
 async def _get_runner_payment(
     challenge: LivePaymentChallenge,
+    attempt: RunnerAttempt,
     *,
     payment_type: str,
     signer_url: str,
@@ -1063,14 +1222,13 @@ async def _get_runner_payment(
     )
     try:
         payment = await session.get_payment()
-        if not payment.payment:
-            raise LivepeerGatewayError("Live runner payment response missing payment")
-        if not payment.seg_creds:
-            raise LivepeerGatewayError("Live runner payment response missing segCreds")
-    except LivepeerGatewayError as error:
+    finally:
         # The signer charges when it signs, so a signed session counts even on failure.
-        error.auth_ids = unique_auth_ids((session.auth_id,))
-        raise
+        attempt.auth_id = session.auth_id
+    if not payment.payment:
+        raise LivepeerGatewayError("Live runner payment response missing payment")
+    if not payment.seg_creds:
+        raise LivepeerGatewayError("Live runner payment response missing segCreds")
     return session, payment
 
 
diff --git a/src/livepeer_gateway/scope.py b/src/livepeer_gateway/scope.py
index 999a364..76cd979 100644
--- a/src/livepeer_gateway/scope.py
+++ b/src/livepeer_gateway/scope.py
@@ -3,7 +3,7 @@
 import logging
 from typing import Any, Optional, Sequence
 
-from .errors import LivepeerGatewayError, NoRunnerAvailableError, runner_rejection
+from .errors import LivepeerGatewayError, NoRunnerAvailableError, RunnerRejection
 from .http import post_json
 from .lv2v import LiveVideoToVideo, StartJobRequest
 from .selection import runner_selector
@@ -129,11 +129,12 @@ async def _start_scope_with_runner(
                 raise LivepeerGatewayError("Scope response missing manifest_id")
             return job
         except Exception as e:
+            reason = str(e)
             runner_url = result.runner_url.strip()
             if not runner_url and result.runner is not None:
                 runner_url = result.runner.url
-            _LOG.debug("scope runner startup failed: %s (%s)", runner_url, e)
-            cursor.rejections.append(runner_rejection(runner_url, e))
+            _LOG.debug("scope runner startup failed: %s (%s)", runner_url, reason)
+            cursor.rejections.append(RunnerRejection(url=runner_url, reason=reason))
 
 
 def _is_serverless_runner(runner: object) -> bool:
diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py
index a01ae03..9a5d47e 100644
--- a/src/livepeer_gateway/selection.py
+++ b/src/livepeer_gateway/selection.py
@@ -18,7 +18,6 @@
     NoRunnerAvailableError,
     OrchestratorRejection,
     RunnerRejection,
-    runner_rejection,
 )
 from .live_runner import (
     LiveRunnerCallResult,
@@ -206,12 +205,13 @@ async def next(self) -> LiveRunnerCallResult:
                     kwargs["signer_headers"] = self._signer_headers
                 result = await call_runner(**kwargs)
             except Exception as e:
+                reason = str(e)
                 _LOG.debug(
                     "select_runner candidate failed: %s (%s)",
                     runner.url,
-                    e,
+                    reason,
                 )
-                self.rejections.append(runner_rejection(runner.url, e))
+                self.rejections.append(RunnerRejection(url=runner.url, reason=reason))
                 continue
 
             _LOG.debug("select_runner selected: %s", runner.url)
diff --git a/tests/test_errors.py b/tests/test_errors.py
index 8397436..da34da7 100644
--- a/tests/test_errors.py
+++ b/tests/test_errors.py
@@ -1,13 +1,6 @@
-import aiohttp
-
 from livepeer_gateway.errors import (
-    LivepeerGatewayError,
-    LivepeerHTTPError,
     NoOrchestratorAvailableError,
-    NoRunnerAvailableError,
     OrchestratorRejection,
-    RunnerRejection,
-    runner_rejection,
 )
 
 
@@ -35,92 +28,3 @@ def test_no_orchestrator_available_error_includes_rejections() -> None:
         "https://orch-a.example.com: connection refused; "
         "https://orch-b.example.com: request timed out"
     )
-
-
-def test_paid_503_is_a_capacity_rejection() -> None:
-    error = LivepeerHTTPError(503, "https://runner.example.com", "busy")
-    error.payment_sent = True
-    error.auth_ids = ("auth-1",)
-
-    rejection = runner_rejection("https://runner.example.com", error)
-
-    assert rejection.kind == "capacity"
-    assert rejection.payment_sent is True
-    assert rejection.auth_ids == ("auth-1",)
-    assert rejection.reason == str(error)
-
-
-def test_http_400_is_another_kind_of_rejection() -> None:
-    error = LivepeerHTTPError(400, "https://runner.example.com", "bad")
-
-    rejection = runner_rejection("https://runner.example.com", error)
-
-    assert rejection.kind == "other"
-    assert rejection.payment_sent is False
-    assert rejection.auth_ids == ()
-
-
-def test_connection_refused_is_unreachable() -> None:
-    try:
-        raise LivepeerGatewayError("connection refused") from ConnectionRefusedError()
-    except LivepeerGatewayError as error:
-        rejection = runner_rejection("https://runner.example.com", error)
-
-    assert rejection.kind == "unreachable"
-    assert rejection.payment_sent is False
-
-
-def test_server_disconnect_is_unreachable() -> None:
-    try:
-        raise LivepeerGatewayError("failed to reach endpoint") from aiohttp.ServerDisconnectedError()
-    except LivepeerGatewayError as error:
-        rejection = runner_rejection("https://runner.example.com", error)
-
-    assert rejection.kind == "unreachable"
-
-
-def test_payload_error_is_not_unreachable() -> None:
-    try:
-        raise LivepeerGatewayError("short read") from aiohttp.ClientPayloadError("short")
-    except LivepeerGatewayError as error:
-        rejection = runner_rejection("https://runner.example.com", error)
-
-    assert rejection.kind == "other"
-
-
-def test_text_mentioning_capacity_is_not_a_capacity_refusal() -> None:
-    error = LivepeerGatewayError("capacity exhausted")
-
-    rejection = runner_rejection("https://runner.example.com", error)
-
-    assert rejection.kind == "other"
-
-
-def test_no_runner_available_keeps_the_paid_rejection() -> None:
-    error = NoRunnerAvailableError(
-        "All runners failed (2 tried)",
-        rejections=[
-            RunnerRejection(
-                url="https://runner.example.com",
-                reason="busy",
-                kind="capacity",
-                payment_sent=True,
-                auth_ids=("auth-1",),
-            ),
-            RunnerRejection(
-                url="https://runner-b.example.com",
-                reason="busy",
-                kind="capacity",
-                payment_sent=True,
-                auth_ids=("auth-2",),
-            ),
-        ],
-    )
-
-    assert error.payment_sent is True
-    assert error.auth_ids == ("auth-1", "auth-2")
-    assert str(error) == (
-        "All runners failed (2 tried): "
-        "https://runner.example.com: busy; "
-        "https://runner-b.example.com: busy"
-    )
diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py
index 767c0a5..1ff7f3a 100644
--- a/tests/test_live_runner.py
+++ b/tests/test_live_runner.py
@@ -17,6 +17,7 @@
 from livepeer_gateway.errors import (
     LivepeerGatewayError,
     LivepeerHTTPError,
+    PaymentError,
     SignerRefreshRequired,
 )
 from livepeer_gateway.live_runner import (
@@ -26,6 +27,7 @@
     LiveRunnerPriceInfo,
     LiveRunnerRegistration,
     LiveRunnerSessionEvent,
+    RunnerCallRecord,
     register_runner,
     stop_runner_session,
     create_proxy,
@@ -1880,6 +1882,8 @@ class _FakePaymentSession:
     """Stands in for LivePaymentSession: returns fixed tickets, funds forever."""
 
     instances: list[_FakePaymentSession] = []
+    # What each session's get_payment raises, in order; None signs normally.
+    errors: list[Exception | None] = []
 
     def __init__(self, signer_url: str, **kwargs: object) -> None:
         self.signer_url = signer_url
@@ -1888,6 +1892,9 @@ def __init__(self, signer_url: str, **kwargs: object) -> None:
         type(self).instances.append(self)
 
     async def get_payment(self) -> object:
+        error = type(self).errors.pop(0) if type(self).errors else None
+        if error is not None:
+            raise error
         return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64")
 
     async def run_payments(self) -> bool:
@@ -1896,8 +1903,9 @@ async def run_payments(self) -> bool:
 
 
 @contextlib.contextmanager
-def _fake_signer() -> Iterator[None]:
+def _fake_signer(errors: list[Exception | None] | None = None) -> Iterator[None]:
     _FakePaymentSession.instances = []
+    _FakePaymentSession.errors = list(errors or [])
     with (
         mock.patch.object(live_runner, "LivePaymentSession", _FakePaymentSession),
         mock.patch.object(
@@ -1913,13 +1921,14 @@ def _fake_signer() -> Iterator[None]:
 def _runner_app(
     seen: list[_RecordedRequest],
     *,
-    statuses: list[int] | None = None,
+    statuses: list[int | str] | None = None,
     stream: bool = False,
 ) -> web.Application:
     """A runner that answers each request with the next status in ``statuses``.
 
     402 carries a payment challenge; 200 answers JSON, or two SSE frames when
-    ``stream`` is set.
+    ``stream`` is set. ``"hang"`` never answers in time, and ``"truncated"``
+    starts a 500 and drops the connection mid-body.
     """
     pending = list(statuses or [200])
 
@@ -1927,6 +1936,17 @@ async def handler(request: web.Request) -> web.StreamResponse:
         body = await request.read()
         seen.append(_RecordedRequest(request, body))
         status = pending.pop(0) if len(pending) > 1 else pending[0]
+        if status == "hang":
+            await asyncio.sleep(1)
+            return web.Response(status=500)
+        if status == "truncated":
+            resp = web.StreamResponse(status=500)
+            resp.content_length = 1000
+            await resp.prepare(request)
+            await resp.write(b"partial")
+            assert request.transport is not None
+            request.transport.close()
+            return resp
         if status == 402:
             return web.Response(
                 status=402,
@@ -1934,7 +1954,7 @@ async def handler(request: web.Request) -> web.StreamResponse:
                 content_type="application/json",
             )
         if status != 200:
-            return web.Response(status=status, text="no")
+            return web.Response(status=int(status), text="no")
         if stream:
             resp = web.StreamResponse(headers={"Content-Type": "text/event-stream"})
             await resp.prepare(request)
@@ -2042,3 +2062,113 @@ async def test_paid_call_without_signer_is_unpaid(self) -> None:
                 await call_runner(f"{base}/call", payload={"prompt": "hi"})
         assert info.value.payment_sent is False
         assert info.value.auth_ids == ()
+
+
+class TestCallRunnerRecord:
+    async def test_attempts_record_the_challenge_and_success(self) -> None:
+        seen: list[_RecordedRequest] = []
+        record = RunnerCallRecord()
+        async with _serve(_runner_app(seen, statuses=[402, 200])) as base:
+            with _fake_signer():
+                result = await _paid_call(base, record=record)
+        assert result.record is record
+        assert result.payment_sent is True
+        assert record.runner_url == f"{base}/call"
+        assert [a.outcome for a in record.attempts] == ["payment_required", "succeeded"]
+        assert [a.status_code for a in record.attempts] == [402, None]
+        assert [a.payment_sent for a in record.attempts] == [False, True]
+        assert [a.manifest_id for a in record.attempts] == ["", "manifest-1"]
+        assert [a.auth_id for a in record.attempts] == ["", "auth-1"]
+
+    async def test_signer_failure_after_a_paid_attempt_still_says_paid(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402])) as base:
+            with _fake_signer([None, PaymentError("signer down")]):
+                with pytest.raises(PaymentError) as info:
+                    await _paid_call(base)
+        record = info.value.record
+        assert record is not None
+        assert [a.outcome for a in record.attempts] == ["payment_required", "payment_required", "payment"]
+        assert info.value.payment_sent is True
+        assert info.value.auth_ids == ("auth-1", "auth-2")
+
+    async def test_unpaid_retry_after_refresh_still_says_paid(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402, 402, 500])) as base:
+            with _fake_signer([None, SignerRefreshRequired("refresh")]):
+                with pytest.raises(LivepeerHTTPError) as info:
+                    await _paid_call(base)
+        assert info.value.status_code == 500
+        assert info.value.payment_sent is True
+        assert info.value.auth_ids == ("auth-1", "auth-2")
+        assert info.value.record is not None
+        assert info.value.record.attempts[-1].payment_sent is False
+        assert info.value.record.outcome == "http"
+
+    async def test_refresh_on_the_last_attempt_keeps_earlier_auth_ids(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402])) as base:
+            with _fake_signer([None, None, SignerRefreshRequired("refresh")]):
+                with pytest.raises(SignerRefreshRequired) as info:
+                    await _paid_call(base, max_payment_challenge_retries=1)
+        assert info.value.payment_sent is True
+        assert info.value.auth_ids == ("auth-1", "auth-2", "auth-3")
+        assert info.value.record is not None
+        assert info.value.record.outcome == "payment"
+
+    async def test_truncated_stream_error_after_payment_is_wrapped(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[402, "truncated"], stream=True)) as base:
+            with _fake_signer():
+                with pytest.raises(LivepeerGatewayError, match="Live runner call failed") as info:
+                    await _paid_call(base, stream=True)
+        assert not isinstance(info.value, LivepeerHTTPError)
+        assert info.value.payment_sent is True
+        assert info.value.auth_ids == ("auth-1",)
+        assert info.value.record is not None
+        assert info.value.record.outcome == "disconnected"
+
+    async def test_cancelled_call_leaves_the_record_readable(self) -> None:
+        seen: list[_RecordedRequest] = []
+        record = RunnerCallRecord()
+        async with _serve(_runner_app(seen, statuses=[402, "hang"])) as base:
+            with _fake_signer():
+                with pytest.raises(TimeoutError):
+                    await asyncio.wait_for(_paid_call(base, record=record), 0.5)
+        assert record.payment_sent is True
+        assert record.auth_ids == ("auth-1",)
+        assert record.outcome == "cancelled"
+
+    async def test_refused_connection_is_unreachable(self) -> None:
+        with pytest.raises(LivepeerGatewayError) as info:
+            await call_runner(f"http://127.0.0.1:{_free_port()}/call", payload={})
+        assert info.value.record is not None
+        assert info.value.record.outcome == "unreachable"
+        assert info.value.payment_sent is False
+
+    async def test_slow_runner_is_a_timeout_not_unreachable(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=["hang"])) as base:
+            with pytest.raises(LivepeerGatewayError) as info:
+                await call_runner(f"{base}/call", payload={}, timeout=0.2)
+        assert len(seen) == 1
+        assert info.value.record is not None
+        assert info.value.record.outcome == "timeout"
+
+    @pytest.mark.parametrize(("status", "outcome"), [(503, "refused"), (400, "http")])
+    async def test_runner_error_status_outcomes(self, status: int, outcome: str) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, statuses=[status])) as base:
+            with pytest.raises(LivepeerHTTPError) as info:
+                await call_runner(f"{base}/call", payload={})
+        assert info.value.record is not None
+        assert info.value.record.outcome == outcome
+        assert info.value.record.attempts[0].status_code == status
+
+    async def test_stream_records_its_status(self) -> None:
+        seen: list[_RecordedRequest] = []
+        async with _serve(_runner_app(seen, stream=True)) as base:
+            async with await call_runner(f"{base}/call", payload={}, stream=True) as stream:
+                assert stream.record.outcome == "succeeded"
+                assert stream.record.attempts[0].status_code == 200
+                assert stream.payment_sent is False
diff --git a/tests/test_selection.py b/tests/test_selection.py
index 3ab3ece..e56d371 100644
--- a/tests/test_selection.py
+++ b/tests/test_selection.py
@@ -5,7 +5,7 @@
 import pytest
 
 from livepeer_gateway import selection
-from livepeer_gateway.errors import LivepeerGatewayError, LivepeerHTTPError, NoRunnerAvailableError
+from livepeer_gateway.errors import LivepeerGatewayError, NoRunnerAvailableError
 from livepeer_gateway.live_runner import LiveRunnerCallResult, LiveRunnerInstance
 
 
@@ -178,56 +178,6 @@ async def _call_runner(
         assert len(cursor.rejections) == 1
         assert cursor.rejections[0].url == "https://orch-a/apps/a/session"
         assert cursor.rejections[0].reason == "capacity exhausted"
-        assert cursor.rejections[0].kind == "other"
-        assert cursor.rejections[0].payment_sent is False
-
-    async def test_runner_selector_keeps_payment_and_failure_kind(self) -> None:
-        async def _call_runner(
-            *,
-            runner: LiveRunnerInstance,
-            payload: dict[str, object],
-            method: str,
-            timeout: float,
-        ) -> LiveRunnerCallResult:
-            del payload, method, timeout
-            if runner.url.endswith("/a/session"):
-                refused = LivepeerHTTPError(503, runner.url, "busy")
-                refused.payment_sent = True
-                refused.auth_ids = ("auth-1",)
-                raise refused
-            try:
-                raise LivepeerGatewayError("connection refused") from ConnectionRefusedError()
-            except LivepeerGatewayError as unreachable:
-                raise unreachable
-
-        with (
-            mock.patch.object(
-                selection,
-                "discover_runners",
-                return_value=[
-                    _entry(
-                        [
-                            {"url": "https://orch-a/apps/a/session", "app": "app-a"},
-                            {"url": "https://orch-a/apps/b/session", "app": "app-b"},
-                        ]
-                    )
-                ],
-            ),
-            mock.patch.object(selection, "call_runner", side_effect=_call_runner),
-        ):
-            cursor = await selection.runner_selector(
-                discovery_url="https://example.com/discovery"
-            )
-            with pytest.raises(NoRunnerAvailableError) as raised:
-                await cursor.next()
-
-        assert cursor.rejections[0].kind == "capacity"
-        assert cursor.rejections[0].payment_sent is True
-        assert cursor.rejections[0].auth_ids == ("auth-1",)
-        assert cursor.rejections[1].kind == "unreachable"
-        assert cursor.rejections[1].payment_sent is False
-        assert raised.value.payment_sent is True
-        assert raised.value.auth_ids == ("auth-1",)
 
     async def test_runner_selector_empty_discovery_raises_no_runner_available(
         self,

From 1214d19c6a3a88e6c7617b6e10b55b2680f404ce Mon Sep 17 00:00:00 2001
From: John | Elite Encoder <john@eliteencoder.net>
Date: Mon, 12 Oct 2026 14:53:22 -0400
Subject: [PATCH 6/6] Move the call record types into errors and read auth_id
 from signer state

RunnerAttempt, RunnerAttemptOutcome and RunnerCallRecord now live in errors.py,
so LivepeerGatewayError can reference RunnerCallRecord without a TYPE_CHECKING
import. LivePaymentSession.auth_id is parsed from the stored signer state
instead of being cached separately.
---
 src/livepeer_gateway/errors.py        | 68 +++++++++++++++++++++++--
 src/livepeer_gateway/live_runner.py   | 72 +++------------------------
 src/livepeer_gateway/remote_signer.py |  4 +-
 3 files changed, 73 insertions(+), 71 deletions(-)

diff --git a/src/livepeer_gateway/errors.py b/src/livepeer_gateway/errors.py
index 8c50eb8..144fe56 100644
--- a/src/livepeer_gateway/errors.py
+++ b/src/livepeer_gateway/errors.py
@@ -1,10 +1,70 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from dataclasses import dataclass, field
+from typing import Literal
+
+
+RunnerAttemptOutcome = Literal[
+    "pending",  # still in flight
+    "succeeded",
+    "payment_required",  # runner answered 402; the next attempt pays
+    "refused",  # runner answered 503: no capacity
+    "http",  # runner answered with any other error status
+    "payment",  # the signer failed, so the request never left
+    "unreachable",  # the request never reached the runner
+    "timeout",  # no answer in time; the runner may have the request
+    "disconnected",  # the connection dropped after the request left
+    "invalid_response",
+    "cancelled",
+    "error",
+]
 
-if TYPE_CHECKING:
-    from .live_runner import RunnerCallRecord
+
+@dataclass
+class RunnerAttempt:
+    """One request ``call_runner`` made, or set out to make, to the runner."""
+
+    # The payment challenge this attempt answered; "" when it went out unpaid.
+    manifest_id: str = ""
+    # Signer auth id of the payment session this attempt opened. The signer
+    # charges when it signs, so this is set even if the request never left.
+    auth_id: str = ""
+    # True once the request left with Livepeer-Payment headers.
+    payment_sent: bool = False
+    # The runner's HTTP status, when known.
+    status_code: int | None = None
+    outcome: RunnerAttemptOutcome = "pending"
+
+
+@dataclass
+class RunnerCallRecord:
+    """Every attempt of one ``call_runner`` call, written as each step happens.
+
+    The result, stream, or error of the call holds this same record. Pass one
+    to ``call_runner`` to read it even when the call is cancelled.
+    """
+
+    runner_url: str = ""
+    attempts: list[RunnerAttempt] = field(default_factory=list)
+
+    @property
+    def payment_sent(self) -> bool:
+        """True once any attempt's request left with ``Livepeer-Payment`` headers.
+
+        The call is then bound to this runner; another runner is safe only
+        while this is False.
+        """
+        return any(attempt.payment_sent for attempt in self.attempts)
+
+    @property
+    def auth_ids(self) -> tuple[str, ...]:
+        """Signer auth ids of every payment session the call opened, in order."""
+        return tuple(dict.fromkeys(a.auth_id for a in self.attempts if a.auth_id))
+
+    @property
+    def outcome(self) -> RunnerAttemptOutcome:
+        """How the last attempt ended; ``pending`` before the first one."""
+        return self.attempts[-1].outcome if self.attempts else "pending"
 
 
 class LivepeerGatewayError(RuntimeError):
diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py
index 7d0bdf6..5e16357 100644
--- a/src/livepeer_gateway/live_runner.py
+++ b/src/livepeer_gateway/live_runner.py
@@ -27,7 +27,14 @@
 from aiohttp.helpers import parse_mimetype
 
 from .channel_reader import ChannelReader
-from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired
+from .errors import (
+    LivepeerGatewayError,
+    LivepeerHTTPError,
+    RunnerAttempt,
+    RunnerAttemptOutcome,
+    RunnerCallRecord,
+    SignerRefreshRequired,
+)
 from .http import _post_empty, _request_body, open_stream, post_json, request_json
 from .remote_signer import (
     GetPaymentResponse,
@@ -162,69 +169,6 @@ class LiveRunnerProxy:
     url: str
 
 
-RunnerAttemptOutcome = Literal[
-    "pending",  # still in flight
-    "succeeded",
-    "payment_required",  # runner answered 402; the next attempt pays
-    "refused",  # runner answered 503: no capacity
-    "http",  # runner answered with any other error status
-    "payment",  # the signer failed, so the request never left
-    "unreachable",  # the request never reached the runner
-    "timeout",  # no answer in time; the runner may have the request
-    "disconnected",  # the connection dropped after the request left
-    "invalid_response",
-    "cancelled",
-    "error",
-]
-
-
-@dataclass
-class RunnerAttempt:
-    """One request ``call_runner`` made, or set out to make, to the runner."""
-
-    # The payment challenge this attempt answered; "" when it went out unpaid.
-    manifest_id: str = ""
-    # Signer auth id of the payment session this attempt opened. The signer
-    # charges when it signs, so this is set even if the request never left.
-    auth_id: str = ""
-    # True once the request left with Livepeer-Payment headers.
-    payment_sent: bool = False
-    # The runner's HTTP status, when known.
-    status_code: int | None = None
-    outcome: RunnerAttemptOutcome = "pending"
-
-
-@dataclass
-class RunnerCallRecord:
-    """Every attempt of one ``call_runner`` call, written as each step happens.
-
-    The result, stream, or error of the call holds this same record. Pass one
-    to ``call_runner`` to read it even when the call is cancelled.
-    """
-
-    runner_url: str = ""
-    attempts: list[RunnerAttempt] = field(default_factory=list)
-
-    @property
-    def payment_sent(self) -> bool:
-        """True once any attempt's request left with ``Livepeer-Payment`` headers.
-
-        The call is then bound to this runner; another runner is safe only
-        while this is False.
-        """
-        return any(attempt.payment_sent for attempt in self.attempts)
-
-    @property
-    def auth_ids(self) -> tuple[str, ...]:
-        """Signer auth ids of every payment session the call opened, in order."""
-        return tuple(dict.fromkeys(a.auth_id for a in self.attempts if a.auth_id))
-
-    @property
-    def outcome(self) -> RunnerAttemptOutcome:
-        """How the last attempt ended; ``pending`` before the first one."""
-        return self.attempts[-1].outcome if self.attempts else "pending"
-
-
 @dataclass(frozen=True)
 class LiveRunnerCallResult:
     data: dict[str, Any]
diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py
index 4ff9337..fed782f 100644
--- a/src/livepeer_gateway/remote_signer.py
+++ b/src/livepeer_gateway/remote_signer.py
@@ -255,12 +255,11 @@ def __init__(
         self._max_price = dict(max_price) if max_price is not None else None
         self._max_refresh_retries = max(0, int(max_refresh_retries))
         self._state: dict[str, Any] | None = None
-        self._auth_id = ""
 
     @property
     def auth_id(self) -> str:
         """The signer's sealed auth id, empty until the first payment response."""
-        return self._auth_id
+        return auth_id_from_signer_state(self._state)
 
     async def get_payment(self) -> GetPaymentResponse:
         if not self._signer_url:
@@ -363,7 +362,6 @@ async def _payment_request(self) -> GetPaymentResponse:
             )
 
         self._state = state
-        self._auth_id = auth_id_from_signer_state(state)
         return GetPaymentResponse(payment=payment, seg_creds=seg_creds)
 
     async def _refresh_payment_params(self) -> None: