ctfy.sdk.resources.competitions

  1"""``client.competitions`` — discover competitions (list / get).
  2
  3Acting *within* a competition — register, your team, invites, challenges,
  4instances, submissions, standings — lives on the scoped handle
  5``client.competition(id)`` (:class:`ctfy.sdk.competition.Competition`).
  6"""
  7
  8from __future__ import annotations
  9
 10import builtins
 11from typing import Any
 12
 13from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 14from ctfy.sdk.base import BaseHttpClient
 15from ctfy.server.models import (
 16    AgentActionsPage,
 17    AgentPulsePage,
 18    CapacityWatchResult,
 19    CompetitionAttachmentsResponse,
 20    CompetitionDetail,
 21    CompetitionEligibility,
 22    CompetitionExclusion,
 23    CompetitionInfo,
 24    CompetitionTopicInfo,
 25    CompetitionTopicsResponse,
 26    PublicRecap,
 27    ScoringChangeInfo,
 28    ScoringChangesResponse,
 29    VirtualGhostBoard,
 30)
 31
 32
 33class CompetitionsResource:
 34    """Browse competitions. To act within one, use ``client.competition(id)``."""
 35
 36    def __init__(self, http: BaseHttpClient) -> None:
 37        self._http = http
 38
 39    def list(
 40        self, phase: str = "", offset: int = 0, limit: int = 50, kind: str = ""
 41    ) -> PagedList[CompetitionInfo]:
 42        """Public competition list. ``phase`` ∈ ``"" | "upcoming" |
 43        "running" | "past"`` ("" = all). ``kind`` ∈ ``"" | "event" | "suite"``
 44        ("" = both): an *event* is a hosted contest with a window and
 45        registration, a *suite* a public benchmark set open at any time --
 46        to evaluate an agent, ask for ``kind="suite"`` (or
 47        ``client.suites.list()``). Start here, then scope into one with
 48        ``client.competition(comp.id)``."""
 49        params: dict[str, Any] = {"offset": offset, "limit": limit}
 50        if phase:
 51            params["phase"] = phase
 52        if kind:
 53            params["kind"] = kind
 54        resp = self._http.request("GET", "/competitions", params=params)
 55        _raise_for_status(resp)
 56        return _extract_items(resp.json(), CompetitionInfo)
 57
 58    def get(self, competition_id: str) -> CompetitionDetail:
 59        """Full competition detail including resolved challenge summaries
 60        (``.challenges``). Same as ``client.competition(id).detail()``."""
 61        resp = self._http.request("GET", f"/competitions/{competition_id}")
 62        _raise_for_status(resp)
 63        return CompetitionDetail.model_validate(resp.json())
 64
 65    def attachments(self, competition_id: str) -> CompetitionAttachmentsResponse:
 66        """The competition's rules / handbook / notice files, in display
 67        order and every language's (``locale`` ``""`` is for everyone).
 68        Read these before playing — they are the rules as the organiser
 69        published them. Fetch one with :meth:`download_attachment`."""
 70        resp = self._http.request("GET", f"/competitions/{competition_id}/attachments")
 71        _raise_for_status(resp)
 72        return CompetitionAttachmentsResponse.model_validate(resp.json())
 73
 74    def download_attachment(self, competition_id: str, attachment_id: str) -> bytes:
 75        """One competition file's raw bytes (id from :meth:`attachments`)."""
 76        resp = self._http.request(
 77            "GET", f"/competitions/{competition_id}/attachments/{attachment_id}"
 78        )
 79        _raise_for_status(resp)
 80        return resp.content
 81
 82    def topics(self, competition_id: str) -> CompetitionTopicsResponse:
 83        """The competition's released topics, in release order (#1965).
 84
 85        An organiser schedules each with a ``release_at`` and the clock
 86        releases it — nothing is pushed, so **poll this** (around the
 87        start of each authoring window) to learn the day's topic. For
 88        registered entrants: anybody else is ``403 not_registered``, and
 89        a competition you cannot see is ``404``."""
 90        resp = self._http.request("GET", f"/competitions/{competition_id}/topics")
 91        _raise_for_status(resp)
 92        return CompetitionTopicsResponse.model_validate(resp.json())
 93
 94    def topic(self, competition_id: str, topic_id: str) -> CompetitionTopicInfo:
 95        """One released topic (id from :meth:`topics`); an unreleased one
 96        is ``404 competition_topic_not_found``, like an unknown id."""
 97        resp = self._http.request("GET", f"/competitions/{competition_id}/topics/{topic_id}")
 98        _raise_for_status(resp)
 99        return CompetitionTopicInfo.model_validate(resp.json())
100
101    def scoring_changes(self, competition_id: str) -> builtins.list[ScoringChangeInfo]:
102        """Every change to the competition's scoring rule made while it was
103        live, oldest first (#1965 H) — when, why, and the rule and params
104        before and after. The board is recomputed under the new rule as a
105        whole, so **poll this** beside the scoreboard: standings that moved
106        with no new solve moved because the rule did. Empty for an event
107        whose scoring never changed while live."""
108        resp = self._http.request("GET", f"/competitions/{competition_id}/scoring-changes")
109        _raise_for_status(resp)
110        return ScoringChangesResponse.model_validate(resp.json()).changes
111
112    def recap(self, competition_id: str) -> PublicRecap:
113        """The competition's published recap (#3023): reach and the contest,
114        as the public board shows them (frozen, anonymised). 404
115        ``recap_not_published`` until an organiser publishes it."""
116        resp = self._http.request("GET", f"/competitions/{competition_id}/recap")
117        _raise_for_status(resp)
118        return PublicRecap.model_validate(resp.json())
119
120    def eligibility(self, competition_id: str) -> CompetitionEligibility:
121        """Whether you may register, and what to do if not (#2184).
122
123        ``restricted`` is false for an ordinary competition. For one limited to
124        certain institutions, ``state`` is ``eligible`` (you hold a proved
125        address at an accepted domain), ``needs_verification`` (you hold one
126        but never proved it: ``pending_email``) or ``needs_email`` (you hold
127        none). ``exempt`` is an organiser's invitation. Gate on ``may_register``.
128        """
129        resp = self._http.request("GET", f"/competitions/{competition_id}/eligibility")
130        _raise_for_status(resp)
131        return CompetitionEligibility.model_validate(resp.json())
132
133    def watch_capacity(self, competition_id: str, *, on: bool = True) -> CapacityWatchResult:
134        """Ask to be told, once, when a place frees in a full competition (#2767).
135
136        ``on=False`` withdraws the request. Refused with ``409
137        registration_not_full`` when the competition has no cap or a free
138        place: register instead. The request is consumed when it is reported,
139        so asking again after a notice is a new request.
140        """
141        if on:
142            resp = self._http.request(
143                "PUT", f"/competitions/{competition_id}/registration-capacity/watch"
144            )
145        else:
146            resp = self._http.request(
147                "DELETE", f"/competitions/{competition_id}/registration-capacity/watch"
148            )
149        _raise_for_status(resp)
150        return CapacityWatchResult.model_validate(resp.json())
151
152    def exclusion(self, competition_id: str) -> CompetitionExclusion:
153        """Which other competitions this one is mutually exclusive with (#2301).
154
155        When ``exclusive`` is true, a person may hold a team in only one
156        competition of the group. ``peers`` are the ones you may open; the
157        rest are counted in ``hidden`` and never named. ``blocked`` is
158        ``holding`` (you already hold a team in ``blocked_by``) or
159        ``switch_closed`` (you left one after the group started) when
160        registering here would be refused right now, else ``None``.
161        """
162        resp = self._http.request("GET", f"/competitions/{competition_id}/exclusion")
163        _raise_for_status(resp)
164        return CompetitionExclusion.model_validate(resp.json())
165
166    def start_virtual(self, competition_id: str) -> CompetitionInfo:
167        """Replay a finished contest on your own clock.
168
169        Returns the *replay*, which is an ordinary competition you are
170        already registered in — so every other call takes its
171        ``.id`` unchanged: launch instances, submit answers, submit
172        patches, read its board.
173
174        **Idempotent.** Asking twice returns the replay you already
175        started rather than minting a second sitting, so a retry cannot
176        split one attempt across two competitions.
177
178        Refused (400) when a replay is not defined: the source has not
179        finished, has no window, is multi-session, never ran, or is
180        itself a replay. The error code says which.
181        """
182        resp = self._http.request("POST", f"/competitions/{competition_id}/virtual")
183        _raise_for_status(resp)
184        return CompetitionInfo.model_validate(resp.json())
185
186    def ghosts(self, competition_id: str) -> VirtualGhostBoard:
187        """The source contest's standings at the instant this replay has
188        reached, with you merged in and ranked among them.
189
190        Takes the **replay's** id, not the source's. At T+40 minutes
191        into your run you see the contest as it stood at T+40 minutes on
192        the day — which is what makes a replay a race rather than an
193        exercise, since a board holding only your own solves answers
194        nothing.
195
196        Refused (400 ``not_a_virtual_sitting``) on an ordinary
197        competition. A source that has since been deleted costs the
198        ghosts and nothing else: the reply comes back with no items and
199        the run itself keeps working.
200        """
201        resp = self._http.request("GET", f"/competitions/{competition_id}/ghosts")
202        _raise_for_status(resp)
203        return VirtualGhostBoard.model_validate(resp.json())
204
205    def agent_actions(self, competition_id: str) -> AgentActionsPage:
206        """One scrubbed line per agent saying what it is doing now.
207
208        :meth:`agent_pulse`'s sibling, and the whole difference is that
209        this one carries content. What makes that publishable is two
210        gates, neither of which stands in for the other:
211
212        * the line is a single tool call picked out of the newest
213          finished conversation, run through the platform's secret-shape
214          table and cut — so the round's minted flag and the untrusted
215          markup do not travel;
216        * a row exists only for an entrant whose **author** opted in
217          (``POST /eval/runs/{run_id}/wall``), which is the answer to
218          the one rule no scrubbing dissolves: the conversation belongs
219          to the person who ran it.
220
221        ⚠️ **An empty ``rows`` is four different things and the three
222        numbers beside it say which**: ``indexing: false`` is a
223        deployment with no transcript index, ``entrants: 0`` is nobody
224        having entered an agent, ``published: 0`` is nobody's author
225        having agreed, and all three set with no rows is everybody
226        having agreed and no call having finished yet.
227
228        ⚠️ **At most ``max_rows`` entrants are read**, because each one
229        costs an archived-transcript read; the reply publishes that
230        ceiling beside ``published`` so a truncated answer cannot read
231        as a complete one.
232
233        Refused (403 ``silent_standings``) on a silent competition, for
234        :meth:`agent_pulse`'s reason.
235
236        Wraps ``GET /competitions/{competition_id}/agent-actions``."""
237        resp = self._http.request("GET", f"/competitions/{competition_id}/agent-actions")
238        _raise_for_status(resp)
239        return AgentActionsPage.model_validate(resp.json())
240
241    def agent_pulse(self, competition_id: str, *, limit: int = 200) -> AgentPulsePage:
242        """What this event's agents have been asking a model, without it.
243
244        One lane per entrant, one beat per **finished** model call. The
245        gateway composes a record once, at completion — streamed
246        completions included — so no caller of this can see a
247        conversation in flight, whatever cadence they poll at.
248
249        ⚠️ **Carries no prompt and no completion, and cannot be made
250        to.** A relayed conversation holds the challenge's own minted
251        answer, which on an AWD gamebox is the round's flag, and its
252        prompt half is written by whoever is attacking the challenge —
253        so what travels is the shape of the traffic: how much each call
254        wrote, how long it took, and what went wrong. The bodies live
255        behind ``/admin/llm-calls/{id}`` and the run's own author-scoped
256        route, both of which rule on who may read them.
257
258        Empty means one of two opposite things and the reply says
259        which: ``indexing: false`` is a deployment that keeps no
260        transcript index at all.
261
262        Refused (403 ``silent_standings``) on a silent competition, like
263        the board — a lane per entrant saying who is still working is
264        the field's progress.
265        """
266        resp = self._http.request(
267            "GET",
268            f"/competitions/{competition_id}/agent-pulse",
269            params={"limit": limit},
270        )
271        _raise_for_status(resp)
272        return AgentPulsePage.model_validate(resp.json())