ctfy.sdk.resources.eval

  1"""``client.eval`` — the public leaderboard, plus your own agent shelf."""
  2
  3from __future__ import annotations
  4
  5from collections.abc import Sequence
  6from typing import Any
  7
  8from ctfy.core.state.models import EvalCampaignState, HarnessState, TaskState
  9from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 10from ctfy.sdk.base import BaseHttpClient
 11from ctfy.server.models import (
 12    EvalAllowanceRow,
 13    EvalLeaderboard,
 14    HarnessTags,
 15    InstanceTrafficPage,
 16    LlmCallBody,
 17    LlmCallPage,
 18    LlmModelOption,
 19    LlmProviderChoice,
 20    MyEvalRunPage,
 21    MyEvalRunRow,
 22    MyLlmKey,
 23    MyLlmKeyProbe,
 24    RegistryPushTarget,
 25    RunWriteup,
 26)
 27
 28
 29class EvalResource:
 30    """The public model leaderboard (no auth), and the signed-in caller's
 31    own harness registry (``/eval/harnesses``)."""
 32
 33    def __init__(self, http: BaseHttpClient) -> None:
 34        self._http = http
 35
 36    def leaderboard(self, *, competition_id: str = "", tier: str = "official") -> EvalLeaderboard:
 37        """Ranked model standings + per-dimension (difficulty / bucket / tag)
 38        breakdowns. ``competition_id`` scopes to one competition; empty = global.
 39
 40        ``tier`` selects the credibility band: ``official`` (default,
 41        platform-run), ``self_reported`` (vendor-run, unverified), or ``all``
 42        for both. Sandbox runs are never exposed here."""
 43        resp = self._http.request(
 44            "GET", "/eval/leaderboard", params={"competition_id": competition_id, "tier": tier}
 45        )
 46        _raise_for_status(resp)
 47        return EvalLeaderboard.model_validate(resp.json())
 48
 49    # -- your own agent images -------------------------------------------
 50
 51    def harnesses(self) -> list[HarnessState]:
 52        """The harnesses *you* registered, newest first.
 53
 54        Scoped to the caller — the platform's own images and other
 55        users' are not listed here (``/admin/eval-harnesses`` is the
 56        operator's whole-registry view).
 57        """
 58        resp = self._http.request("GET", "/eval/harnesses")
 59        _raise_for_status(resp)
 60        return [HarnessState.model_validate(row) for row in resp.json()]
 61
 62    def register_harness(
 63        self,
 64        harness_id: str,
 65        image: str,
 66        *,
 67        display_name: str = "",
 68        version: str = "",
 69        notes: str = "",
 70    ) -> HarnessState:
 71        """Register one of your own agent images.
 72
 73        ⚠️ The returned row is ``review_status="pending"`` and is **not
 74        dispatchable** until an operator approves it — registering an
 75        image and being trusted to run one are deliberately two things.
 76        ``owner_user_id`` and ``review_status`` are stamped by the
 77        server; anything you pass for them is ignored.
 78        """
 79        resp = self._http.request(
 80            "POST",
 81            "/eval/harnesses",
 82            json={
 83                "id": harness_id,
 84                "image": image,
 85                "display_name": display_name,
 86                "version": version,
 87                "notes": notes,
 88            },
 89        )
 90        _raise_for_status(resp)
 91        return HarnessState.model_validate(resp.json())
 92
 93    def update_harness(
 94        self,
 95        harness_id: str,
 96        *,
 97        image: str | None = None,
 98        display_name: str | None = None,
 99        version: str | None = None,
100        notes: str | None = None,
101        source_url: str | None = None,
102        active: bool | None = None,
103    ) -> HarnessState:
104        """Edit one of your own. Anything left ``None`` is untouched.
105
106        ⚠️ **Changing ``image`` puts the row back to
107        ``review_status="pending"``** and clears the approved digest —
108        the verdict was about specific content, and a dispatch pulls the
109        digest rather than the tag. Carrying the approval across would
110        leave the row naming a new reference while the node kept running
111        the old bytes. Re-sending the *same* image is a no-op and costs
112        nothing.
113
114        Anything that is not yours answers 404 rather than 403, like
115        :meth:`delete_harness`.
116        """
117        # ⚠️ Absence is what "leave it alone" means on the wire, so an
118        # unset argument must not appear in the body at all — the server
119        # reads ``model_fields_set``, and a key sent as ``null`` is a key
120        # that was sent.
121        body: dict[str, object] = {
122            name: value
123            for name, value in (
124                ("image", image),
125                ("display_name", display_name),
126                ("version", version),
127                ("notes", notes),
128                ("source_url", source_url),
129                ("active", active),
130            )
131            if value is not None
132        }
133        resp = self._http.request("PATCH", f"/eval/harnesses/{harness_id}", json=body)
134        _raise_for_status(resp)
135        return HarnessState.model_validate(resp.json())
136
137    def delete_harness(self, harness_id: str) -> bool:
138        """Remove one of your own. Anything that is not yours — another
139        user's, or a platform image — answers 404 rather than 403, so
140        this cannot be used to probe the registry."""
141        resp = self._http.request("DELETE", f"/eval/harnesses/{harness_id}")
142        _raise_for_status(resp)
143        return bool(resp.json().get("deleted", False))
144
145    def llm_providers(self) -> list[LlmProviderChoice]:
146        """The upstreams this deployment will relay to.
147
148        ⚠️ **Ask rather than assume.** The gateway makes the request to
149        whatever address a key names — from inside its own network, and
150        it returns the answer — so the set is closed and an operator may
151        replace it. A caller carrying its own list would offer addresses
152        this deployment refuses.
153
154        Each entry carries its dialect: a provider *is* a dialect plus an
155        address, and DeepSeek, Moonshot and OpenAI all speak ``openai``.
156        """
157        resp = self._http.request("GET", "/eval/llm-providers")
158        _raise_for_status(resp)
159        return [LlmProviderChoice.model_validate(r) for r in resp.json()]
160
161    def my_llm_keys(self) -> list[MyLlmKey]:
162        """Your own upstream credentials — what they are, never what they say.
163
164        ⚠️ There is no ``api_key`` on the way back, by design. A key is
165        pasted once; every read after that lets you tell two apart, not
166        copy one.
167        """
168        resp = self._http.request("GET", "/eval/llm-keys")
169        _raise_for_status(resp)
170        return [MyLlmKey.model_validate(r) for r in resp.json()]
171
172    def put_my_llm_key(
173        self, flavor: str, api_key: str, *, label: str = "", base_url: str = ""
174    ) -> MyLlmKey:
175        """Store or rotate the credential this account spends for ``flavor``.
176
177        A repeat dialect replaces: rotation is a write, because
178        delete-then-add leaves a window in which a run has no key.
179        Answers 503 on a deployment that cannot seal secrets at rest —
180        the key is never stored in the clear.
181        """
182        resp = self._http.request(
183            "PUT",
184            f"/eval/llm-keys/{flavor}",
185            json={"api_key": api_key, "label": label, "base_url": base_url},
186        )
187        _raise_for_status(resp)
188        return MyLlmKey.model_validate(resp.json())
189
190    def delete_my_llm_key(self, flavor: str, *, base_url: str = "") -> bool:
191        """Remove it. 404 for a provider you hold no key for — the same
192        answer somebody else's key gets, so this cannot report who holds
193        what.
194
195        ⚠️ ``base_url`` is part of the address, not a filter. Naming only
196        the dialect is ambiguous the moment you hold two ``openai``
197        providers, and resolving that by picking one is how somebody
198        loses the key they meant to keep. ``""`` names that dialect's
199        official endpoint, which is a value rather than a wildcard.
200        """
201        resp = self._http.request(
202            "DELETE", f"/eval/llm-keys/{flavor}", params={"base_url": base_url}
203        )
204        _raise_for_status(resp)
205        return bool(resp.json().get("deleted", False))
206
207    def probe_my_llm_key(self, flavor: str, *, base_url: str = "") -> MyLlmKeyProbe:
208        """Spend one token on your own credential and report what happened.
209
210        Storing a key says the write landed; it says nothing about
211        whether the credential **works**. The only other way to find
212        that out is to enter an agent, let a run be dispatched, and read
213        the failure out of the artifact tarball — two hours, on a full
214        range, to learn that a key is invalid.
215
216        ⚠️ **A code, never a boolean.** A rejected key, an address
217        nothing answers at, an account out of credit and a deployment
218        with no model registered at this endpoint are four different
219        fields to go and fix.
220
221        ⚠️ **200 with the outcome in the body** for every one of them —
222        so this does not raise on a failed credential. A 404 (and only a
223        404) means you hold no key for that provider.
224
225        ⚠️ ``base_url`` is part of the address, exactly as it is for the
226        removal beside it: naming the dialect alone picks whichever row
227        matched first the moment you hold two ``openai`` providers.
228        """
229        resp = self._http.request(
230            "POST", f"/eval/llm-keys/{flavor}/probe", params={"base_url": base_url}
231        )
232        _raise_for_status(resp)
233        return MyLlmKeyProbe.model_validate(resp.json())
234
235    def my_allowances(self) -> list[EvalAllowanceRow]:
236        """What an operator has granted this account, and how much is left.
237
238        ⚠️ Read ``remaining_usd`` rather than subtracting: the server
239        clamps an overshoot at zero, and a client doing its own
240        arithmetic would eventually disagree with the refusal that
241        actually decides.
242        """
243        resp = self._http.request("GET", "/eval/allowances")
244        _raise_for_status(resp)
245        return [EvalAllowanceRow.model_validate(r) for r in resp.json()]
246
247    def models(self) -> list[LlmModelOption]:
248        """The models an entry may name.
249
250        ⚠️ Approved and active only, decided by the server. The registry
251        holds more than this — the operator's own notes and the
252        addressing that says where their credential is spent — and none
253        of it travels.
254        """
255        resp = self._http.request("GET", "/eval/models")
256        _raise_for_status(resp)
257        return [LlmModelOption.model_validate(r) for r in resp.json()]
258
259    def enter(
260        self,
261        *,
262        harness_id: str,
263        model_id: str,
264        competition_id: str,
265        num_samples: int = 1,
266    ) -> EvalCampaignState:
267        """Put one of your own agents into one competition.
268
269        ⚠️ **The result is `self_reported`, and the server decides that.**
270        The default public leaderboard shows operator-run results; this
271        is your own claim about your own agent, so read it back with
272        ``tier="self_reported"`` or ``"all"``. There is no parameter for
273        it here because there is none on the wire.
274
275        ⚠️ **409 carries the reason as its ``code``** — whether to wait
276        for review, add a key, or pick another event. A caller that
277        reports only "entry refused" throws away the only actionable
278        part of the answer.
279        """
280        resp = self._http.request(
281            "POST",
282            "/eval/entries",
283            json={
284                "harness_id": harness_id,
285                "model_id": model_id,
286                "competition_id": competition_id,
287                "num_samples": num_samples,
288            },
289        )
290        _raise_for_status(resp)
291        return EvalCampaignState.model_validate(resp.json())
292
293    def my_entries(
294        self, *, competition_id: str = "", offset: int = 0, limit: int = 50
295    ) -> PagedList[EvalCampaignState]:
296        """The campaigns this account entered, newest first — one page,
297        with the total (``limit`` at most 200).
298
299        Only entries: an operator campaign the same account queued is
300        not one of them, because the two are different claims about the
301        same board. ``competition_id`` narrows to entries into that one
302        competition, on the server.
303
304        ⚠️ This returned a bare list, the server's newest 200, with
305        nothing to say the list was cut; ``total`` says how many exist.
306        """
307        params: dict[str, Any] = {"offset": offset, "limit": limit}
308        if competition_id:
309            params["competition_id"] = competition_id
310        resp = self._http.request("GET", "/eval/entries", params=params)
311        _raise_for_status(resp)
312        return _extract_items(resp.json(), EvalCampaignState)
313
314    def runs(self, *, competition_id: str = "", offset: int = 0, limit: int = 50) -> MyEvalRunPage:
315        """Evaluation runs you may read — your own plus your team-mates'.
316
317        ⚠️ The author set is the same membership join the per-run routes
318        enforce, so a run this lists can always be opened and a run it
319        omits always 404s. Two readings of "the same team" would drift
320        in the open direction.
321
322        ⚠️ ``mine`` on each row says whether *you* may open the
323        conversations. It is composed server-side because it gates a
324        control, and two clients deriving it would disagree.
325        Wraps ``GET /eval/runs``."""
326        resp = self._http.request(
327            "GET",
328            "/eval/runs",
329            params={"competition_id": competition_id, "offset": offset, "limit": limit},
330        )
331        _raise_for_status(resp)
332        return MyEvalRunPage.model_validate(resp.json())
333
334    def run(self, run_id: str) -> MyEvalRunRow:
335        """One of your runs, projected as the shelf projects it.
336
337        ⚠️ Addressable because :meth:`runs` is paginated: one campaign
338        over a large bucket is hundreds of runs, so finding a row by
339        scanning the first page is a bug with a threshold.
340
341        Metadata scope — a team-mate reads the row, and its ``traffic``
342        field says ``not_yours`` rather than the route refusing.
343        Wraps ``GET /eval/runs/{run_id}``."""
344        resp = self._http.request("GET", f"/eval/runs/{run_id}")
345        _raise_for_status(resp)
346        return MyEvalRunRow.model_validate(resp.json())
347
348    def run_llm_calls(self, run_id: str, *, offset: int = 0, limit: int = 50) -> LlmCallPage:
349        """Every model request one of your runs made, newest first.
350
351        Metadata only — token split, latency, queue wait, the estimated
352        cost. **Scoped to the run's team**, so a captain can see who is
353        burning the allowance without reading anybody's prompts, and to
354        ``principal == "agent"``, so a challenge container's own
355        conversation is never on this route.
356        Wraps ``GET /eval/runs/{run_id}/llm-calls``."""
357        resp = self._http.request(
358            "GET",
359            f"/eval/runs/{run_id}/llm-calls",
360            params={"offset": offset, "limit": limit},
361        )
362        _raise_for_status(resp)
363        return LlmCallPage.model_validate(resp.json())
364
365    def run_llm_call(self, run_id: str, call_id: str) -> LlmCallBody:
366        """The conversation itself — **the run author's only**.
367
368        ⚠️ Narrower than :meth:`run_llm_calls` on purpose: a team-mate
369        who can see the row is answered ``403
370        transcript_body_owner_only`` here, since an agent's prompt is its
371        author's work and ``HarnessStanding`` ranks the two of them
372        against each other.
373
374        ⚠️ ``available=False`` is an ordinary answer, not an error —
375        retention reclaims bodies and leaves the row behind.
376        Wraps ``GET /eval/runs/{run_id}/llm-calls/{call_id}``."""
377        resp = self._http.request("GET", f"/eval/runs/{run_id}/llm-calls/{call_id}")
378        _raise_for_status(resp)
379        return LlmCallBody.model_validate(resp.json())
380
381    def run_artifact(self, run_id: str) -> bytes:
382        """The files one of your runs left behind, as ``.tar.gz`` bytes.
383
384        Its working directory and, when the harness is Claude Code, its
385        session — collected from the container before it was removed.
386        The writing half, where :meth:`run_llm_call` is the deciding one
387        and :meth:`run_traffic` the doing one.
388
389        ⚠️ **The run author's only**, the narrowest of the three: a
390        team-mate who can see the row is answered ``403
391        run_artifact_owner_only``. An archive carries whatever the agent
392        wrote down, which for a solved challenge includes the flag.
393
394        ⚠️ **Bytes, not a model, and no URL.** The read goes through the
395        platform rather than handing out a bucket address, so there is
396        nothing to return but the archive itself. ⚠️ A run that kept no
397        files answers ``404 run_artifact_not_found`` rather than empty
398        bytes — which of the several reasons it is (the agent wrote
399        nothing, collection failed, retention took them) is on the run
400        row's ``artifact`` field, so the caller never has to guess from
401        an absence.
402
403        Wraps ``GET /eval/runs/{run_id}/artifact``."""
404        resp = self._http.request("GET", f"/eval/runs/{run_id}/artifact")
405        _raise_for_status(resp)
406        return bytes(resp.content)
407
408    def run_traffic(self, run_id: str, *, cursor: str = "", limit: int = 0) -> InstanceTrafficPage:
409        """Every request one of your runs sent at the challenge.
410
411        The doing half, where :meth:`run_llm_call` is the deciding half.
412        **The run author's only** — a team-mate who can see the row is
413        answered ``403 run_traffic_owner_only``, the same split the
414        conversation body makes, because the payloads an agent wrote are
415        its author's work.
416
417        ⚠️ Carry ``cursor`` forward from the previous page and you get
418        only what arrived since; pass an empty one and you get the whole
419        capture again. Every failure — an unreachable node, a run whose
420        estate the reaper has taken — answers 200 with your own cursor
421        and no flows, so a blip never re-delivers the capture. Which of
422        those it was, ``MyEvalRunRow.traffic`` says.
423
424        ⚠️ Redacted: this instance's minted answers are scrubbed out.
425        Wraps ``GET /eval/runs/{run_id}/traffic``."""
426        resp = self._http.request(
427            "GET",
428            f"/eval/runs/{run_id}/traffic",
429            params={"cursor": cursor, "limit": limit},
430        )
431        _raise_for_status(resp)
432        return InstanceTrafficPage.model_validate(resp.json())
433
434    def run_writeup(self, run_id: str) -> RunWriteup:
435        """What one of your runs said it did, in its own words.
436
437        ``CTFY-0001`` §3 invites an optional ``writeup`` in the result
438        file. The saying half, where :meth:`run_llm_call` is the
439        deciding one, :meth:`run_traffic` the doing one and
440        :meth:`run_artifact` the writing one.
441
442        ⚠️ **Telemetry, never a grading path.** Nothing in the result
443        file is scored — the run's own ``POST /submissions`` is — so a
444        write-up cannot earn a point and is not read as an answer.
445
446        ⚠️ **The run author's only**: a team-mate who can see the row is
447        answered ``403 run_writeup_owner_only``, the same split the
448        other three bodies make. ⚠️ A run whose image wrote none answers
449        ``404 run_writeup_not_found`` rather than empty prose, and the
450        run row's ``writeup`` field already said which — so the caller
451        never has to reach here to find out.
452
453        Wraps ``GET /eval/runs/{run_id}/writeup``."""
454        resp = self._http.request("GET", f"/eval/runs/{run_id}/writeup")
455        _raise_for_status(resp)
456        return RunWriteup.model_validate(resp.json())
457
458    def set_run_wall(self, run_id: str, *, published: bool) -> MyEvalRunRow:
459        """Put this run's agent on the event's projector wall, or take it off.
460
461        ``GET /competitions/{id}/agent-actions`` draws one scrubbed line
462        per agent saying what it is doing. The scrub answers what the
463        bytes make unsafe; this answers the part it cannot — the
464        conversation belongs to whoever ran it, so the wall shows an
465        entrant only once their author has said yes.
466
467        ⚠️ **The switch is on the *entrant*, not on this run.** The
468        transcript index groups by team and an agent's team is one per
469        ``(owner, harness, model, competition)``, so every run of the
470        same agent at the same event shares one answer — which is what
471        "this agent, at this event" means. The row that comes back is
472        the run's, with ``wall_published`` re-read from the entrant.
473
474        ⚠️ **The author's only**: a team-mate who can see the row is
475        answered ``403 run_wall_not_yours``, the same split
476        :meth:`run_writeup` makes, and for the same reason — the line
477        is read out of a conversation they already meet a 403 on.
478
479        ⚠️ A run with no agent entrant answers ``409`` naming which
480        kind of nothing it is; ``MyEvalRunRow.wall`` carries the same
481        code, so a caller never has to press to find out.
482
483        Wraps ``POST /eval/runs/{run_id}/wall``."""
484        resp = self._http.request(
485            "POST",
486            f"/eval/runs/{run_id}/wall",
487            json={"published": published},
488        )
489        _raise_for_status(resp)
490        return MyEvalRunRow.model_validate(resp.json())
491
492    def set_run_transcript(self, run_id: str, *, published: bool) -> MyEvalRunRow:
493        """Publish this run's agent's **full conversation**, or take it back.
494
495        A **second, stronger** consent than :meth:`set_run_wall`. The wall
496        shows one scrubbed line of what the agent is doing; this opens the
497        whole conversation — every turn the agent wrote — to spectators at
498        ``GET /competitions/{id}/awd/transcript/{call_id}``. The projection
499        still emits only the agent's own scrubbed turns (never the system
500        prompt, never a tool result), but the decision to publish a
501        conversation at all is the author's alone and separate from the
502        wall's.
503
504        ⚠️ **The switch is on the *entrant***, exactly as the wall's —
505        one answer per ``(owner, harness, model, competition)``.
506
507        ⚠️ **Refused for an anonymous entrant** (``403
508        run_transcript_anonymous``): a conversation de-anonymises whoever
509        wrote it, so the withheld-identity promise wins over the consent.
510        The author's-only and no-agent refusals mirror :meth:`set_run_wall`.
511
512        Wraps ``POST /eval/runs/{run_id}/transcript``."""
513        resp = self._http.request(
514            "POST",
515            f"/eval/runs/{run_id}/transcript",
516            json={"published": published},
517        )
518        _raise_for_status(resp)
519        return MyEvalRunRow.model_validate(resp.json())
520
521    def registry_push_target(self) -> RegistryPushTarget:
522        """Where this account pushes an agent image, if anywhere.
523
524        ⚠️ Read ``enabled``, not a non-empty ``repository_prefix``. The
525        server composes the verdict from four facts — a host, a token
526        service, and both halves of the signing pair — and a deployment
527        with a hostname but no signing pair refuses every login.
528        """
529        resp = self._http.request("GET", "/eval/registry")
530        _raise_for_status(resp)
531        return RegistryPushTarget.model_validate(resp.json())
532
533    def harness_tags(self, harness_id: str) -> HarnessTags:
534        """Which versions the registry publishes for one of your agents.
535
536        ⚠️ Addressed by **harness id**, not by image: the platform looks
537        the reference up in your own registry row, which is what keeps
538        this from being an outbound proxy. Answers 404 for anything that
539        is not yours, and 502 when the registry itself would not answer
540        — never an empty list, which would read as "no versions".
541        """
542        resp = self._http.request("GET", f"/eval/harnesses/{harness_id}/tags")
543        _raise_for_status(resp)
544        return HarnessTags.model_validate(resp.json())
545
546    def check_harness(self, harness_id: str) -> HarnessState:
547        """Ask the platform to look at one of your agent images.
548
549        Returns the row with the verdict on it: ``conformance_status``
550        (``""`` never checked / ``passed`` / ``failed`` / ``error``),
551        the digest the verdict describes, and a finding per thing worth
552        saying.
553
554        ⚠️ **Evidence, never a gate.** Nothing here changes
555        ``review_status``, and a ``failed`` verdict does not stop an
556        operator approving the image — the check exists so somebody can
557        see *why* a dispatch would go wrong, before it does.
558
559        ⚠️ **``error`` is not ``failed``, and it arrives as a 200.** A
560        registry that would not answer is not a fact about the image;
561        the message is on ``conformance_error``.
562
563        ⚠️ Addressed by harness id, like :meth:`harness_tags` and for
564        the same reason: the image is read off your own registry row, so
565        this cannot be pointed at an arbitrary host.
566        """
567        resp = self._http.request("POST", f"/eval/harnesses/{harness_id}/conformance")
568        _raise_for_status(resp)
569        return HarnessState.model_validate(resp.json())
570
571    def probe_harness(self, harness_id: str) -> TaskState:
572        """Actually run one of your agent images once, to check it.
573
574        Where :meth:`check_harness` reads a manifest, this starts the
575        container on a worker under the same isolation a real dispatch
576        uses, hands it a briefing marked ``dry_run``, and asks the one
577        question a manifest cannot answer: **did it write a result file
578        at the address it was given.**
579
580        ⚠️ **Returns a task, not a verdict.** The image has to be pulled
581        before the container starts, so this is submitted and polled —
582        read the harness row, or the task, for the outcome.
583
584        ⚠️ **No credential travels with the probe** and no platform URL
585        is named. See §3.1 of ``docs/agent-protocol.md``.
586        """
587        resp = self._http.request("POST", f"/eval/harnesses/{harness_id}/probe")
588        _raise_for_status(resp)
589        return TaskState.model_validate(resp.json())
590
591
592class RegistryResource:
593    """The container registry's token endpoint.
594
595    ⚠️ **Not a method anybody calls from Python.** ``docker`` follows
596    the registry's ``WWW-Authenticate`` realm here on its own, with HTTP
597    Basic; this exists so the SDK covers the surface and so a test can
598    drive the exchange the way docker does. The credential goes in the
599    *password* half — that is where ``docker login -p`` puts it.
600    """
601
602    def __init__(self, http: BaseHttpClient) -> None:
603        self._http = http
604
605    def token(self, *, service: str, scope: str | Sequence[str] = "") -> dict[str, Any]:
606        # ⚠️ A sequence is sent as a repeated ``scope`` parameter, because
607        # that is what docker does for a cross-repository blob mount —
608        # ``push`` on the destination and ``pull`` on the source, in one
609        # exchange. This method's whole stated purpose is that a test can
610        # drive the exchange the way docker does, and it could not.
611        params: list[tuple[str, str]] = [("service", service)]
612        if isinstance(scope, str):
613            if scope:
614                params.append(("scope", scope))
615        else:
616            params.extend(("scope", s) for s in scope if s)
617        resp = self._http.request("GET", "/registry/token", params=params)
618        _raise_for_status(resp)
619        body: dict[str, Any] = resp.json()
620        return body