ctfy.sdk.admin_resources.challenges

  1"""``client.admin.challenges`` — catalog rescan, pre-build, stats, feedback audit."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from ctfy.sdk._helpers import _extract_items, _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    AdminChallengeLatencyBucket,
 11    AdminChallengeStatsRow,
 12    AdminChallengeTimeseries,
 13    AdminFeedbackRow,
 14    AdminTaskInfo,
 15    ChallengeBuildStateResponse,
 16    ChallengePullStateResponse,
 17    LlmBudgetPage,
 18    LlmBudgetResetInfo,
 19    QuestionAttemptResetInfo,
 20)
 21
 22
 23class AdminChallengesResource:
 24    """Operator challenge tooling: reload the catalog, pre-build on nodes,
 25    inspect attempt/solve stats, audit feedback, reset attempt caps."""
 26
 27    def __init__(self, http: BaseHttpClient) -> None:
 28        self._http = http
 29
 30    def rescan(self) -> AdminTaskInfo:
 31        """Submit a catalog-rescan background task (reload the platform's
 32        spec cache + fan the rescan out to every online node). Returns the
 33        task record; poll ``admin.tasks.get(id)`` (or watch the Tasks page)
 34        for the per-node outcome in ``result``. Wraps
 35        ``POST /admin/challenges/rescan``."""
 36        resp = self._http.request("POST", "/admin/challenges/rescan")
 37        _raise_for_status(resp)
 38        return AdminTaskInfo.model_validate(resp.json())
 39
 40    def build(self, challenge_id: str) -> AdminTaskInfo:
 41        """Submit a single-challenge pre-build background task.
 42
 43        Fans the build out to every online node and polls it to
 44        completion; poll :meth:`build_state` (or the Tasks surface) to
 45        watch ``building`` → ``built`` / ``failed``. Returns the task
 46        record. Wraps ``POST /admin/challenges/{id}/build``."""
 47        resp = self._http.request("POST", f"/admin/challenges/{challenge_id}/build")
 48        _raise_for_status(resp)
 49        return AdminTaskInfo.model_validate(resp.json())
 50
 51    def build_all(self, *, competition_id: str = "") -> AdminTaskInfo:
 52        """Submit a bulk pre-build background task. With ``competition_id``
 53        the build is scoped to that competition's effective challenge set;
 54        otherwise the whole catalog. The task kicks each online node off
 55        and polls them to completion; the per-node outcome + final
 56        per-challenge status land in ``result``. Returns the task record.
 57        Wraps ``POST /admin/challenges/build-all``."""
 58        params: dict[str, Any] = {}
 59        if competition_id:
 60            params["competition_id"] = competition_id
 61        resp = self._http.request("POST", "/admin/challenges/build-all", params=params)
 62        _raise_for_status(resp)
 63        return AdminTaskInfo.model_validate(resp.json())
 64
 65    def build_state(self) -> ChallengeBuildStateResponse:
 66        """Aggregate per-challenge build state across every online node.
 67
 68        Returns one row per known challenge with an ``aggregated``
 69        worst-case status plus the per-node breakdown for the modal
 70        drill-down. Wraps ``GET /admin/challenges/build-state``."""
 71        resp = self._http.request("GET", "/admin/challenges/build-state")
 72        _raise_for_status(resp)
 73        return ChallengeBuildStateResponse.model_validate(resp.json())
 74
 75    def pull(self, challenge_id: str) -> AdminTaskInfo:
 76        """Submit a single-challenge pre-pull background task (pull-side
 77        twin of :meth:`build`: warms registry-only ``image:`` services).
 78        Returns the task record. Wraps ``POST /admin/challenges/{id}/pull``."""
 79        resp = self._http.request("POST", f"/admin/challenges/{challenge_id}/pull")
 80        _raise_for_status(resp)
 81        return AdminTaskInfo.model_validate(resp.json())
 82
 83    def pull_all(self, *, competition_id: str = "") -> AdminTaskInfo:
 84        """Submit a bulk pre-pull background task (pull-side twin of
 85        :meth:`build_all`). Returns the task record. Wraps
 86        ``POST /admin/challenges/pull-all``."""
 87        params: dict[str, Any] = {}
 88        if competition_id:
 89            params["competition_id"] = competition_id
 90        resp = self._http.request("POST", "/admin/challenges/pull-all", params=params)
 91        _raise_for_status(resp)
 92        return AdminTaskInfo.model_validate(resp.json())
 93
 94    def pull_state(self) -> ChallengePullStateResponse:
 95        """Aggregate per-challenge pull state across every online node.
 96
 97        Pull-side twin of :meth:`build_state`. Wraps
 98        ``GET /admin/challenges/pull-state``."""
 99        resp = self._http.request("GET", "/admin/challenges/pull-state")
100        _raise_for_status(resp)
101        return ChallengePullStateResponse.model_validate(resp.json())
102
103    def stats(self, offset: int = 0, limit: int = 50) -> list[AdminChallengeStatsRow]:
104        """Per-challenge attempt / solve / first-blood counts."""
105        resp = self._http.request(
106            "GET",
107            "/admin/challenges/stats",
108            params={"offset": offset, "limit": limit},
109        )
110        _raise_for_status(resp)
111        return [AdminChallengeStatsRow.model_validate(r) for r in resp.json()["items"]]
112
113    def timeseries(
114        self,
115        challenge_id: str,
116        *,
117        window: int = 86400,
118        bucket: int = 3600,
119    ) -> AdminChallengeTimeseries:
120        """Per-challenge timeseries for the row-level drawer on
121        ``/admin/challenges``."""
122        resp = self._http.request(
123            "GET",
124            f"/admin/challenges/{challenge_id}/timeseries",
125            params={"window": window, "bucket": bucket},
126        )
127        _raise_for_status(resp)
128        return AdminChallengeTimeseries.model_validate(resp.json())
129
130    def latency_histogram(self) -> list[AdminChallengeLatencyBucket]:
131        """Instance-start latency distribution across challenges (the
132        admin challenges latency chart)."""
133        resp = self._http.request("GET", "/admin/challenges/latency-histogram")
134        _raise_for_status(resp)
135        return [AdminChallengeLatencyBucket.model_validate(b) for b in resp.json()]
136
137    def feedback(self, challenge_id: str, *, competition_id: str = "") -> list[AdminFeedbackRow]:
138        """Admin audit list: every reaction row on a challenge with
139        denormalised user identity. Requires admin role."""
140        params: dict[str, Any] = {}
141        if competition_id:
142            params["competition_id"] = competition_id
143        resp = self._http.request(
144            "GET", f"/admin/challenges/{challenge_id}/feedback", params=params
145        )
146        _raise_for_status(resp)
147        return _extract_items(resp.json(), AdminFeedbackRow)
148
149    def reset_question_attempts(
150        self,
151        team_id: str,
152        challenge_id: str,
153        question_id: str,
154        *,
155        reason: str = "",
156    ) -> QuestionAttemptResetInfo:
157        """Reset the per-question wrong-attempt counter for one team.
158
159        Writes a fresh baseline timestamp; the cap counter ignores all
160        wrong submissions before the new baseline, so the team gets a
161        fresh batch of attempts on this question without any audit
162        rows being deleted. Returns the new baseline row.
163        Wraps ``POST /admin/teams/{team_id}/questions/{challenge_id}/{question_id}/reset-attempts``."""
164        resp = self._http.request(
165            "POST",
166            f"/admin/teams/{team_id}/questions/{challenge_id}/{question_id}/reset-attempts",
167            json={"reason": reason},
168        )
169        _raise_for_status(resp)
170        return QuestionAttemptResetInfo.model_validate(resp.json())
171
172    def list_llm_budgets(self, *, competition_id: str = "") -> LlmBudgetPage:
173        """Who has spent what against the challenge-LLM gateway.
174
175        The reader for :meth:`reset_llm_budget` — without it an organiser
176        can only hand back an allowance to a team somebody already named.
177        ⚠️ ``gateway_configured`` is on the envelope because an empty
178        list otherwise reads as "nobody spent anything" when it means
179        "this deployment enforces no budgets", and those call for
180        opposite actions.
181        Wraps ``GET /admin/llm-budgets``."""
182        params = {"competition_id": competition_id} if competition_id else None
183        resp = self._http.request("GET", "/admin/llm-budgets", params=params)
184        _raise_for_status(resp)
185        return LlmBudgetPage.model_validate(resp.json())
186
187    def reset_llm_budget(
188        self,
189        team_id: str,
190        competition_id: str,
191        challenge_id: str,
192        *,
193        reason: str = "",
194    ) -> LlmBudgetResetInfo:
195        """Give one team its LLM allowance back on one challenge.
196
197        Clears the gateway's live tally **and** the platform's durable
198        total — either alone un-does itself, so the route does both or
199        refuses. ⚠️ Always answers 200; read ``.outcome`` rather than the
200        status, because ``unreachable`` means nothing changed and is
201        worth retrying.
202        Wraps ``POST /admin/teams/{team_id}/llm-budget/{competition_id}/{challenge_id}/reset``."""
203        resp = self._http.request(
204            "POST",
205            f"/admin/teams/{team_id}/llm-budget/{competition_id}/{challenge_id}/reset",
206            json={"reason": reason},
207        )
208        _raise_for_status(resp)
209        return LlmBudgetResetInfo.model_validate(resp.json())