ctfy.sdk

108class AdminClient:
109    """Admin / operator surface, grouped into resource namespaces.
110
111    Namespaces: :attr:`users`, :attr:`competitions`, :attr:`suites`, :attr:`announcements`,
112    :attr:`achievements`, :attr:`challenges`, :attr:`gauntlet`,
113    :attr:`instances`,
114    :attr:`records`, :attr:`nodes`, :attr:`observability`, :attr:`settings`,
115    :attr:`competition_admins`.
116    """
117
118    def __init__(self, http: BaseHttpClient) -> None:
119        #: Shared transport (the parent PlatformClient when reached via
120        #: ``client.admin``, or an owned client when built via ``connect``).
121        self._http = http
122
123    @classmethod
124    def connect(
125        cls,
126        server_url: str,
127        token: str = "",
128        *,
129        max_retries: int = 2,
130        timeout: int = DEFAULT_CLIENT_TIMEOUT,
131    ) -> AdminClient:
132        """Build an ``AdminClient`` that owns its own transport.
133
134        For operator tooling that doesn't already hold a
135        :class:`PlatformClient`. The returned client is a context manager
136        that closes its transport on exit.
137        """
138        base = server_url.rstrip("/")
139        http = BaseHttpClient(f"{base}/api/v1", token, timeout=timeout, max_retries=max_retries)
140        return cls(http)
141
142    @cached_property
143    def users(self) -> AdminUsersResource:
144        return AdminUsersResource(self._http)
145
146    @cached_property
147    def competitions(self) -> AdminCompetitionsResource:
148        return AdminCompetitionsResource(self._http)
149
150    @cached_property
151    def suites(self) -> AdminSuitesResource:
152        """A suite through its own door: its own fields only (#2896)."""
153        return AdminSuitesResource(self._http)
154
155    @cached_property
156    def announcements(self) -> AdminAnnouncementsResource:
157        return AdminAnnouncementsResource(self._http)
158
159    @cached_property
160    def changelog(self) -> AdminChangelogResource:
161        return AdminChangelogResource(self._http)
162
163    @cached_property
164    def contact_messages(self) -> AdminContactMessagesResource:
165        """What visitors wrote on the Contact page (#3414)."""
166        return AdminContactMessagesResource(self._http)
167
168    @cached_property
169    def institutions(self) -> AdminInstitutionsResource:
170        return AdminInstitutionsResource(self._http)
171
172    @cached_property
173    def exclusion_groups(self) -> AdminExclusionGroupsResource:
174        return AdminExclusionGroupsResource(self._http)
175
176    @cached_property
177    def achievements(self) -> AdminAchievementsResource:
178        return AdminAchievementsResource(self._http)
179
180    @cached_property
181    def challenges(self) -> AdminChallengesResource:
182        return AdminChallengesResource(self._http)
183
184    @cached_property
185    def gauntlet(self) -> AdminGauntletResource:
186        return AdminGauntletResource(self._http)
187
188    @cached_property
189    def email(self) -> AdminEmailResource:
190        return AdminEmailResource(self._http)
191
192    @cached_property
193    def ip_location(self) -> AdminIpLocationResource:
194        return AdminIpLocationResource(self._http)
195
196    @cached_property
197    def llm_calls(self) -> AdminLlmCallsResource:
198        return AdminLlmCallsResource(self._http)
199
200    @cached_property
201    def llm_providers(self) -> AdminLlmProvidersResource:
202        return AdminLlmProvidersResource(self._http)
203
204    @cached_property
205    def patches(self) -> AdminPatchesResource:
206        return AdminPatchesResource(self._http)
207
208    @cached_property
209    def reports(self) -> AdminReportsResource:
210        return AdminReportsResource(self._http)
211
212    @cached_property
213    def awd(self) -> AdminAwdResource:
214        return AdminAwdResource(self._http)
215
216    @cached_property
217    def series(self) -> AdminSeriesResource:
218        return AdminSeriesResource(self._http)
219
220    @cached_property
221    def instances(self) -> AdminInstancesResource:
222        return AdminInstancesResource(self._http)
223
224    @cached_property
225    def records(self) -> AdminRecordsResource:
226        return AdminRecordsResource(self._http)
227
228    @cached_property
229    def nodes(self) -> AdminNodesResource:
230        return AdminNodesResource(self._http)
231
232    @cached_property
233    def observability(self) -> AdminObservabilityResource:
234        return AdminObservabilityResource(self._http)
235
236    @cached_property
237    def settings(self) -> AdminSettingsResource:
238        return AdminSettingsResource(self._http)
239
240    @cached_property
241    def competition_admins(self) -> AdminCompetitionAdminsResource:
242        return AdminCompetitionAdminsResource(self._http)
243
244    @cached_property
245    def admin_invites(self) -> AdminAdminInvitesResource:
246        """Invite an address with no account yet to be an administrator."""
247        return AdminAdminInvitesResource(self._http)
248
249    @cached_property
250    def competition_invites(self) -> AdminCompetitionInvitesResource:
251        return AdminCompetitionInvitesResource(self._http)
252
253    @cached_property
254    def competition_applications(self) -> AdminCompetitionApplicationsResource:
255        """Applications to a private competition, and the verdict on each."""
256        return AdminCompetitionApplicationsResource(self._http)
257
258    @cached_property
259    def competition_attachments(self) -> AdminCompetitionAttachmentsResource:
260        """A competition's rules / handbook / notice files (#1808)."""
261        return AdminCompetitionAttachmentsResource(self._http)
262
263    @cached_property
264    def competition_boards(self) -> AdminCompetitionBoardsResource:
265        """A competition's extra boards — published and shadow (#1965 G)."""
266        return AdminCompetitionBoardsResource(self._http)
267
268    @cached_property
269    def recap(self) -> AdminRecapResource:
270        """A competition's recap, its publication and its report email (#3023)."""
271        return AdminRecapResource(self._http)
272
273    @cached_property
274    def challenge_schedule(self) -> AdminChallengeScheduleResource:
275        """When each challenge of a competition appears and shuts (#2093)."""
276        return AdminChallengeScheduleResource(self._http)
277
278    @cached_property
279    def release_plan(self) -> AdminReleasePlanResource:
280        """Waves, per-challenge rules and hints: drafts, publish, release (#2093, #2101)."""
281        return AdminReleasePlanResource(self._http)
282
283    @cached_property
284    def env_sharing(self) -> AdminEnvSharingResource:
285        """How many teams stand on one environment, the groups behind it and
286        the health of the shared environments (#2129 B)."""
287        return AdminEnvSharingResource(self._http)
288
289    @cached_property
290    def machines(self) -> AdminMachinesResource:
291        """Machine logins handed to teams, captain-only (#2132)."""
292        return AdminMachinesResource(self._http)
293
294    @cached_property
295    def messages(self) -> AdminMessagesResource:
296        """Site messages from an event's organisers or the platform (#2073)."""
297        return AdminMessagesResource(self._http)
298
299    @cached_property
300    def prewarm(self) -> AdminPrewarmResource:
301        """Whether a competition's images are warm on its nodes (#2129)."""
302        return AdminPrewarmResource(self._http)
303
304    @cached_property
305    def competition_topics(self) -> AdminCompetitionTopicsResource:
306        """A competition's scheduled topics, released by the clock (#1965)."""
307        return AdminCompetitionTopicsResource(self._http)
308
309    @cached_property
310    def registrations(self) -> AdminRegistrationsResource:
311        """Entrant roster, eligibility review, and CSV export."""
312        return AdminRegistrationsResource(self._http)
313
314    @cached_property
315    def teams(self) -> AdminTeamsResource:
316        """Enforcement against a squad — disqualify and reinstate.
317
318        Separate from ``registrations`` on purpose: that one rules on
319        eligibility (which decides prizes), this one decides whether the
320        team is in the event at all.
321        """
322        return AdminTeamsResource(self._http)
323
324    @cached_property
325    def challenge_stats(self) -> AdminChallengeStatsResource:
326        """How players did against an event's challenges: the organiser's
327        table and the single-challenge reading (#2146)."""
328        return AdminChallengeStatsResource(self._http)
329
330    @cached_property
331    def risk(self) -> AdminRiskResource:
332        """Risk control: the case queue (organisers) and an account's sealed
333        sightings (platform administrators). Detection reports; people decide."""
334        return AdminRiskResource(self._http)
335
336    @cached_property
337    def tasks(self) -> AdminTasksResource:
338        return AdminTasksResource(self._http)
339
340    @cached_property
341    def scheduled_jobs(self) -> AdminScheduledJobsResource:
342        return AdminScheduledJobsResource(self._http)
343
344    @cached_property
345    def authored_challenges(self) -> AdminAuthoredChallengesResource:
346        """The review queue for player-authored challenges."""
347        return AdminAuthoredChallengesResource(self._http)
348
349    @cached_property
350    def authoring_reviewers(self) -> AdminAuthoringReviewersResource:
351        """Who may rule on one competition's uploads (super-admin)."""
352        return AdminAuthoringReviewersResource(self._http)
353
354    @cached_property
355    def competition_authored_challenges(self) -> AdminCompetitionAuthoredChallengesResource:
356        """One competition's uploads, for its delegated reviewer: read,
357        audit and rule — never publish."""
358        return AdminCompetitionAuthoredChallengesResource(self._http)
359
360    @cached_property
361    def llm_models(self) -> AdminLlmModelsResource:
362        return AdminLlmModelsResource(self._http)
363
364    @cached_property
365    def eval_vendors(self) -> AdminEvalVendorsResource:
366        return AdminEvalVendorsResource(self._http)
367
368    @cached_property
369    def eval_harnesses(self) -> AdminEvalHarnessesResource:
370        return AdminEvalHarnessesResource(self._http)
371
372    @property
373    def eval_allowances(self) -> AdminEvalAllowancesResource:
374        return AdminEvalAllowancesResource(self._http)
375
376    @property
377    def registry(self) -> AdminRegistryResource:
378        """Registry storage: the usage report and the manual reclaim."""
379        return AdminRegistryResource(self._http)
380
381    @cached_property
382    def eval_runs(self) -> AdminEvalRunsResource:
383        return AdminEvalRunsResource(self._http)
384
385    @cached_property
386    def eval_campaigns(self) -> AdminEvalCampaignsResource:
387        return AdminEvalCampaignsResource(self._http)
388
389    def close(self) -> None:
390        """Close the underlying transport.
391
392        Only call this on a standalone client built via :meth:`connect`;
393        when reached via ``PlatformClient.admin`` the transport is shared
394        with — and closed by — the parent player client.
395        """
396        self._http.close()
397
398    def __enter__(self) -> Self:
399        return self
400
401    def __exit__(
402        self,
403        exc_type: type[BaseException] | None,
404        exc: BaseException | None,
405        tb: TracebackType | None,
406    ) -> None:
407        self.close()

 248class NodeClient(BaseHttpClient):
 249    """Thin sync HTTP client; one instance per node URL."""
 250
 251    def __init__(
 252        self,
 253        node_url: str,
 254        token: str,
 255        *,
 256        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 257        max_retries: int = 2,
 258        transport: httpx.BaseTransport | None = None,
 259        disabled_features: Iterable[str] = (),
 260    ) -> None:
 261        """*disabled_features* are the features an admin switched off on this
 262        node (#3037, `ctfy.core.node_features`): :meth:`capabilities` answers
 263        ``False`` for each, whatever the node says, so a caller reads the
 264        effective value and cannot forget the switch. ``()`` is the raw answer
 265        — every platform construction passes the node row's set
 266        (`test_every_node_client_carries_the_switches`).
 267
 268        *transport* is `BaseHttpClient`'s own seam, passed through so a
 269        test can drive a real node app over the real client — the hop the
 270        fakes on either side of it cannot see.
 271
 272        *max_retries* is passed through too: a best-effort read or a stop
 273        is tried once (#2812), because a retry on a timeout multiplies the
 274        one number that has to stay small.
 275
 276        ⚠️ **A call that starts long work retries only a request that never
 277        left** (#3162, :data:`~ctfy.sdk.base.RETRY_UNSENT`): every POST here
 278        that names its own HTTP timeout says so per call, because a
 279        ``ReadTimeout`` is a request the node has and is still running.
 280        ``start_instance`` keeps the default — the node replays it from the
 281        instance id. Guarded by `ctfy/tests/test_long_node_work_is_sent_once.py`."""
 282        super().__init__(
 283            f"{node_url.rstrip('/')}/api/v1",
 284            token,
 285            timeout=timeout,
 286            max_retries=max_retries,
 287            transport=transport,
 288        )
 289        self.disabled_features = frozenset(disabled_features)
 290
 291    # -- lifecycle ----------------------------------------------------------
 292
 293    def start_instance(
 294        self,
 295        *,
 296        challenge_id: str,
 297        instance_id: str,
 298        ttl: int,
 299        answers: dict[str, str],
 300        proxy_output_dir: str | None = None,
 301        env: dict[str, str] | None = None,
 302        publish_gamebox: bool = False,
 303        match_network: MatchNetwork | None = None,
 304        open_shell_password: str = "",
 305        authored_revision: str = "",
 306        internet: InternetAccess = InternetAccess.ALLOW,
 307        tcp_access: TcpAccess = TcpAccess.TLS_SNI,
 308        egress_allow_hosts: list[str] | None = None,
 309        resource_budget: ResourceBudget | None = None,
 310        compose_up_timeout_s: int = 0,
 311    ) -> InstanceStarted:
 312        resp = self.request(
 313            "POST",
 314            "/instances",
 315            json={
 316                # #1835: the authored tree's revision this start must run;
 317                # a node holding another (or none) fetches it first.
 318                "authored_revision": authored_revision,
 319                "challenge_id": challenge_id,
 320                "instance_id": instance_id,
 321                "ttl": ttl,
 322                "answers": answers,
 323                "proxy_output_dir": proxy_output_dir,
 324                "env": env or {},
 325                "publish_gamebox": publish_gamebox,
 326                # ``None`` rather than an omitted key: the node's model
 327                # defaults to None either way, and sending it explicitly
 328                # keeps the body's shape stable across both branches.
 329                "match_network": (dataclasses.asdict(match_network) if match_network else None),
 330                # Classic-AWD open shell (#1725): the box's starting root
 331                # password, empty for every managed / non-AWD start.
 332                "open_shell_password": open_shell_password,
 333                # #1969: ``deny`` holds the box off the internet; the hosts
 334                # are the platform-internal ones it still reaches.
 335                "internet": str(internet),
 336                # #2082: how a player reaches the box's TCP services.
 337                "tcp_access": str(tcp_access),
 338                "egress_allow_hosts": list(egress_allow_hosts or []),
 339                # The CPU / memory a player-authored box is held to; ``None``
 340                # for the corpus and from an older platform.
 341                "resource_budget": (
 342                    resource_budget.model_dump() if resource_budget is not None else None
 343                ),
 344                # #3165: the platform's ``compose_up_timeout_s``, so the node
 345                # spends the budget the platform waits on; ``0`` = the node's own.
 346                "compose_up_timeout_s": compose_up_timeout_s,
 347            },
 348        )
 349        _raise_for_node_status(resp)
 350        return _parsed(InstanceStarted, resp)
 351
 352    def stop_instance(self, instance_id: str) -> None:
 353        resp = self.request("DELETE", f"/instances/{instance_id}")
 354        resp.raise_for_status()
 355
 356    def apply_fabric(
 357        self,
 358        *,
 359        index: int,
 360        interface: dict[str, Any],
 361        match_network: str = "",
 362    ) -> FabricStatus:
 363        """Put this node on one match's WireGuard fabric.
 364
 365        ⚠️ **Returns the model, never ``resp.json()``.** The platform
 366        reads the verdict field by field, and on a dict a renamed field
 367        is ``None`` — falsy — which reads as "the arena did not open",
 368        so a wire change would quietly report every node as broken.
 369
 370        Returns the node's verdict rather than raising on a refusal: the
 371        route answers **200 with a body** precisely so that "this node is
 372        unreachable" and "this node rejected the plan" stay different
 373        facts, and a caller fanning out over an arena needs to tell them
 374        apart — one is waited on, the other is looked at.
 375
 376        *match_network* lets the node check the tunnel landed in the
 377        namespace its boxes are in, and open the bridge's one hole. It is
 378        the platform's own fact (`awd_network.network_name`), so the node
 379        is told rather than guessing.
 380        """
 381        resp = self.request(
 382            "POST",
 383            "/fabric",
 384            json={
 385                "index": index,
 386                "interface": interface,
 387                "match_network": match_network,
 388            },
 389        )
 390        resp.raise_for_status()
 391        return _parsed(FabricStatus, resp)
 392
 393    def probe_fabric(
 394        self,
 395        *,
 396        index: int,
 397        peers: list[dict[str, str]],
 398        mtu: int = 0,
 399    ) -> FabricProbeReport:
 400        """Ask named peers the three link questions. Applies nothing.
 401
 402        ⚠️ **Returns the model, never ``resp.json()``** — same reason
 403        `apply_fabric` gives: on a dict a renamed field reads as falsy,
 404        which here means "your tunnel is down" about a wire change.
 405        """
 406        resp = self.request(
 407            "POST",
 408            "/fabric/probe",
 409            json={"index": index, "peers": peers, "mtu": mtu},
 410        )
 411        resp.raise_for_status()
 412        return _parsed(FabricProbeReport, resp)
 413
 414    def remove_fabric(self, *, index: int, match_network: str = "") -> FabricStatus:
 415        """Take this node off one match's fabric and close its hole.
 416
 417        ⚠️ *match_network* is what lets the node resolve the bridge and
 418        delete the ``DOCKER-USER`` rules by name. Without it the tunnel
 419        goes and the hole stays — open on whatever docker gives that
 420        bridge name next.
 421        """
 422        resp = self.request(
 423            "DELETE",
 424            f"/fabric/{index}",
 425            params={"match_network": match_network},
 426        )
 427        resp.raise_for_status()
 428        return _parsed(FabricStatus, resp)
 429
 430    def stop_instance_service(self, instance_id: str, service: str) -> int:
 431        """Stop one compose service, leaving the instance running.
 432
 433        The exercise format's session close (§4b.4). Returns how many
 434        containers the node stopped — 0 means the challenge declares no
 435        such service, which is a real answer rather than a failure, so
 436        the caller can distinguish it from a cut tunnel.
 437        """
 438        resp = self.request(
 439            "POST",
 440            f"/instances/{instance_id}/stop-service",
 441            json={"service": service},
 442        )
 443        resp.raise_for_status()
 444        body = resp.json()
 445        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
 446
 447    def capabilities(
 448        self, *, harness_admission: bool = False, agent_ingress: bool = False
 449    ) -> NodeCapabilities:
 450        """What this node can do beyond the original verbs.
 451
 452        ⚠️ **A 404 is an answer, not an error**: a node from before hold /
 453        resume has no such route, and "cannot hold" is exactly what the
 454        platform needs to hear -- the caller then falls back to deleting.
 455        Any other failure raises, because an unreachable node is not a node
 456        that cannot hold.
 457        """
 458        # ⚠️ #2356: ``harness_admission`` makes the node prove it can hold
 459        # an admitted run, which writes its host firewall — asked only by
 460        # the admission paths, never by hold / internet.
 461        # #3020: ``agent_ingress`` makes the node converge its listener,
 462        # asked only before handing a published target over WireGuard.
 463        params = {
 464            **({"harness_admission": "true"} if harness_admission else {}),
 465            **({"agent_ingress": "true"} if agent_ingress else {}),
 466        }
 467        resp = self.request("GET", "/capabilities", params=params or None)
 468        if resp.status_code == 404:
 469            return NodeCapabilities()
 470        resp.raise_for_status()
 471        return masked_capabilities(_parsed(NodeCapabilities, resp), self.disabled_features)
 472
 473    def create_agent_ingress_peer(
 474        self, *, run_id: str, public_key: str, ports: list[int], ttl_s: int
 475    ) -> AgentIngressPeer:
 476        """Add *run_id*'s peer to the node's agent-ingress listener (#3020).
 477
 478        Only the public key crosses; the caller keeps the private half.
 479        Raises on any non-2xx (the node's refusal is in the detail).
 480        """
 481        body = AgentIngressPeerRequest(
 482            run_id=run_id, public_key=public_key, ports=ports, ttl_s=ttl_s
 483        )
 484        resp = self.request("POST", "/agent-ingress/peers", json=body.model_dump(mode="json"))
 485        resp.raise_for_status()
 486        return _parsed(AgentIngressPeer, resp)
 487
 488    def delete_agent_ingress_peer(self, run_id: str) -> bool:
 489        """Revoke *run_id*'s peer; ``False`` when the node held none."""
 490        resp = self.request("DELETE", f"/agent-ingress/peers/{run_id}")
 491        resp.raise_for_status()
 492        return _parsed(AgentIngressPeerRemoved, resp).removed
 493
 494    def hold_instance(self, instance_id: str, mode: str) -> int:
 495        """Pause or stop an instance, removing nothing. Returns containers held."""
 496        resp = self.request("POST", f"/instances/{instance_id}/hold", json={"mode": mode})
 497        resp.raise_for_status()
 498        return _parsed(InstanceHeld, resp).count
 499
 500    def resume_instance(self, instance_id: str) -> int:
 501        """Bring a held instance back. Returns containers brought back."""
 502        resp = self.request("POST", f"/instances/{instance_id}/resume")
 503        resp.raise_for_status()
 504        return _parsed(InstanceHeld, resp).count
 505
 506    def change_instance_internet(
 507        self, instance_id: str, change: InstanceInternetChange
 508    ) -> InstanceInternetChanged:
 509        """Change a running box's internet verdict (#2067).
 510
 511        Raises :class:`NodeRequestError` on any refusal, carrying the node's
 512        ``code`` (:data:`~ctfy.core.exceptions.NODE_INTERNET_CHANGE_FAILED`
 513        when its firewall could not converge — the node has then put its
 514        record back, so nothing changed).
 515        """
 516        resp = self.request(
 517            "POST", f"/instances/{instance_id}/internet", json=change.model_dump(mode="json")
 518        )
 519        _raise_for_node_status(resp)
 520        return _parsed(InstanceInternetChanged, resp)
 521
 522    def stop_all(self) -> None:
 523        resp = self.request("POST", "/admin/stop-all")
 524        resp.raise_for_status()
 525
 526    def put_authored_challenge(self, push: NodeAuthoredPush) -> NodeAuthoredResult:
 527        """Ask the node to write an approved player upload into its catalog.
 528
 529        ⚠️ **The bytes travel here, unlike `put_generated_range`.** That
 530        one sends a scenario because the node can compile it; a
 531        player-authored tree has no compiler and no corpus entry, so the
 532        archive is what there is to send.
 533
 534        The node answers **200 with ``ok: false``** for a refusal, so a
 535        node that declined the archive and a node that could not be
 536        reached stay different facts; only the second raises here.
 537        """
 538        resp = self.request("POST", "/admin/authored", json=push.model_dump(mode="json"))
 539        resp.raise_for_status()
 540        return _parsed(NodeAuthoredResult, resp)
 541
 542    def smoke_authored_challenge(
 543        self, push: NodeAuthoredSmokePush, *, timeout: int | None = None
 544    ) -> NodeAuthoredSmokeResult:
 545        """Ask the node to run an authored challenge's own reference exploit.
 546
 547        ⚠️ **Takes its own timeout.** A smoke run brings a whole estate
 548        up; the client default is sized for a status call and would give
 549        up while the node is still doing what it was asked.
 550        """
 551        kwargs: dict[str, Any] = {"json": push.model_dump(mode="json")}
 552        if timeout is not None:
 553            kwargs["timeout"] = timeout
 554        resp = self.request("POST", "/admin/authored/smoke", retry_on=RETRY_UNSENT, **kwargs)
 555        resp.raise_for_status()
 556        return _parsed(NodeAuthoredSmokeResult, resp)
 557
 558    def smoke_progress(self, run_id: str, since: int = 0) -> NodeSmokeProgress | None:
 559        """The steps and lines a running smoke has produced since *since* (#2145).
 560
 561        ``None`` when the node holds no feed for *run_id* — a node older than
 562        the feed, or a run that never started — which a watcher reads as
 563        *unwatched*. Any other failure raises: the caller decides that a
 564        watcher's trouble must never fail the run it watches.
 565        """
 566        # ⚠️ A short timeout of its own: this is read once a second beside a
 567        # run that may be minutes long, and a read that hangs on a busy worker
 568        # must not hold the watcher past the run it watches.
 569        resp = self.request(
 570            "GET",
 571            f"/admin/authored/smoke/{run_id}/progress",
 572            params={"since": since},
 573            timeout=10,
 574        )
 575        if resp.status_code == 404:
 576            return None
 577        resp.raise_for_status()
 578        return _parsed(NodeSmokeProgress, resp)
 579
 580    def smoke_entrance(self, run_id: str) -> NodeSmokeEntrance:
 581        """The player's entrance to the smoke box held for *run_id* (#2819 W2).
 582
 583        ``pending`` / ``ready`` / ``gone``; a 404 (a node older than the
 584        route) raises, which the caller reads as *this node cannot hold*.
 585        """
 586        resp = self.request("GET", f"/admin/authored/smoke/{run_id}/entrance", timeout=15)
 587        resp.raise_for_status()
 588        return _parsed(NodeSmokeEntrance, resp)
 589
 590    def post_smoke_exploit(self, run_id: str, outcome: NodeSmokeExploitOutcome) -> bool:
 591        """Hand the held smoke box its exploit outcome (#2819 W2).
 592
 593        ``True`` when the node took it; ``False`` when it holds no box for
 594        *run_id* any more (it tore down on its own timeout) or had one.
 595        """
 596        resp = self.request(
 597            "POST",
 598            f"/admin/authored/smoke/{run_id}/exploit",
 599            json=outcome.model_dump(mode="json"),
 600            timeout=30,
 601            # First outcome wins on the node, so a resend is harmless; but a
 602            # read timeout is a post that may have landed (#3162).
 603            retry_on=RETRY_UNSENT,
 604        )
 605        if resp.status_code == 404:
 606            return False
 607        resp.raise_for_status()
 608        return bool(resp.json().get("taken"))
 609
 610    def harness_log(self, run_id: str, after: int = 0) -> NodeAgentLog | None:
 611        """An agent run's console lines numbered above *after* (#2401).
 612
 613        ``None`` when the node holds no console for *run_id* — a node older
 614        than the feed, or a run that never started there — which a watcher
 615        reads as *unwatched*. Any other failure raises: the caller decides
 616        that a watcher's trouble must never fail the run it watches.
 617        """
 618        # ⚠️ A short timeout of its own, as ``smoke_progress``: read once a
 619        # second beside a run that may be hours long.
 620        resp = self.request(
 621            "GET", f"/harness/runs/{run_id}/log", params={"after": after}, timeout=10
 622        )
 623        if resp.status_code == 404:
 624            return None
 625        resp.raise_for_status()
 626        return _parsed(NodeAgentLog, resp)
 627
 628    def harness_result(self, run_id: str) -> HarnessRunResult | None:
 629        """What the node knows of a run's harness call (#3021).
 630
 631        ``None`` when the node has no such route — a node older than the
 632        result store, which can never return a run's outcome after its call
 633        is gone. ``state="unknown"`` is the node answering that it holds
 634        nothing for *run_id*. Any other failure raises: an unreachable node
 635        is neither of those, and the caller bounds how long it waits on one.
 636        """
 637        resp = self.request("GET", f"/harness/runs/{run_id}/result", timeout=30)
 638        if resp.status_code == 404:
 639            return None
 640        resp.raise_for_status()
 641        return _parsed(HarnessRunResult, resp)
 642
 643    def put_generated_range(self, push: NodeRangePush) -> NodeRangeResult:
 644        """Ask the node to build a generated range from its scenario.
 645
 646        ⚠️ **The scenario travels, not the estate.** The corpus is
 647        mounted on the node and the compiler ships inside it, so 70-odd
 648        rendered files never cross this API — and the corpus commit that
 649        rides along is what stops one scenario becoming two different
 650        estates under one id across a fleet.
 651
 652        The node answers **200 with ``ok: false``** for a refusal, so a
 653        node that declined and a node that could not be reached stay
 654        different facts; only the second raises here.
 655        """
 656        resp = self.request("POST", "/admin/ranges", json=push.model_dump(mode="json"))
 657        resp.raise_for_status()
 658        return _parsed(NodeRangeResult, resp)
 659
 660    def verify_generated_range(
 661        self, push: NodeRangeVerifyPush, *, timeout: int | None = None
 662    ) -> NodeRangeVerifyResult:
 663        """Ask the node to build a scenario, solve it, and take it down.
 664
 665        ⚠️ **The longest call in this API**, so the caller passes its own
 666        timeout: the node's budget is ``compose_timeout_s +
 667        solve_timeout_s`` and a client that used the default would give
 668        up while the node was still building — then retry, and start a
 669        second estate beside the first.
 670
 671        ⚠️ A client timeout does **not** leave a range running: the node
 672        tears down in its own ``finally``. What is lost is the answer,
 673        not the host.
 674
 675        The node answers **200 with an ``error``** for a refusal, the
 676        same split ``put_generated_range`` uses; only an unreachable node
 677        raises.
 678        """
 679        kwargs: dict[str, Any] = {"json": push.model_dump(mode="json")}
 680        if timeout is not None:
 681            # httpx takes a per-request override; the client-wide value
 682            # stays whatever the caller constructed this with.
 683            kwargs["timeout"] = timeout
 684        resp = self.request("POST", "/admin/ranges/verify", retry_on=RETRY_UNSENT, **kwargs)
 685        resp.raise_for_status()
 686        return _parsed(NodeRangeVerifyResult, resp)
 687
 688    def cancel_generated_range_verify(self, run_id: str) -> bool:
 689        """Tell the node to stop a verification it is running.
 690
 691        ⚠️ **A miss is success, not a 404 to report.** The cancel races
 692        the run's own finish, so "no such run" means the run already
 693        ended — which is what the operator asked for. The node answers
 694        ``{"cancelled": false}`` for that case rather than raising, and
 695        this returns it so a caller that wants to say "it had already
 696        finished" can.
 697        """
 698        resp = self.request("POST", f"/admin/ranges/verify/{run_id}/cancel")
 699        resp.raise_for_status()
 700        body = resp.json()
 701        return bool(body.get("cancelled"))
 702
 703    def rescan_challenges(self) -> ChallengeReloadResult:
 704        """Tell the node to drop its spec cache and re-scan challenges_dir.
 705
 706        Returns ``{total, added, removed}`` so the platform can report
 707        the per-node outcome of a cluster-wide rescan."""
 708        resp = self.request("POST", "/admin/rescan-challenges")
 709        resp.raise_for_status()
 710        return _parsed(ChallengeReloadResult, resp)
 711
 712    # -- admin pre-build (image cache warming) ------------------------------
 713
 714    def build_challenge(
 715        self,
 716        challenge_id: str,
 717        *,
 718        authored_revision: str = "",
 719        resource_budget: ResourceBudget | None = None,
 720    ) -> ChallengeBuildState:
 721        """Ask the node to pre-build images for *challenge_id*.
 722
 723        Fires-and-returns: the node persists ``status="building"`` and
 724        spawns a daemon thread; the body returned here is that initial
 725        state row. Polling :meth:`get_build_state` is how the platform
 726        learns when it lands on ``built`` / ``failed``. *authored_revision*
 727        is #1835's: the tree the build must be of, fetched if not held.
 728        """
 729        params = {"authored_revision": authored_revision} if authored_revision else None
 730        # #2504: an authored tree pre-builds under this budget, or not at all.
 731        kwargs: dict[str, Any] = {}
 732        if resource_budget is not None:
 733            kwargs["json"] = {"resource_budget": resource_budget.model_dump()}
 734        resp = self.request(
 735            "POST", f"/admin/challenges/{challenge_id}/build", params=params, **kwargs
 736        )
 737        _raise_for_node_status(resp)
 738        return _parsed(ChallengeBuildState, resp)
 739
 740    def build_all_challenges(
 741        self,
 742        challenge_ids: list[str] | None = None,
 743        *,
 744        resource_budget: ResourceBudget | None = None,
 745    ) -> BuildAllQueued:
 746        """Queue background pre-build on the node.
 747
 748        With ``challenge_ids=None`` the node builds every spec it knows.
 749        With an explicit list (the platform scoping a bulk build to a
 750        competition's challenge set) only those are built. Returns
 751        ``{queued, skipped_built, skipped_in_progress}`` so the platform
 752        can report per-node what got picked up. Sequential on the node
 753        side — no fan-out across the corpus.
 754        """
 755        kwargs: dict[str, Any] = {}
 756        body: dict[str, Any] = {}
 757        if challenge_ids is not None:
 758            body["challenge_ids"] = challenge_ids
 759        if resource_budget is not None:
 760            # #2504: what an authored id in the scope pre-builds under.
 761            body["resource_budget"] = resource_budget.model_dump()
 762        if body:
 763            kwargs["json"] = body
 764        resp = self.request("POST", "/admin/challenges/build-all", **kwargs)
 765        _raise_for_node_status(resp)
 766        return _parsed(BuildAllQueued, resp)
 767
 768    def build_push_challenge(
 769        self,
 770        challenge_id: str,
 771        *,
 772        push_ticket: str = "",
 773        deadline_s: int = 0,
 774        cleanup: bool = False,
 775        cache_max_bytes: int = 0,
 776        confine_network: bool = False,
 777        authored_revision: str = "",
 778        run_id: str = "",
 779        resource_budget: ResourceBudget | None = None,
 780    ) -> ChallengeFreezeResult:
 781        """Build one challenge's ``build:`` services and freeze them to the registry.
 782
 783        Layer 2 producer (P0). **Blocks** until the challenge is built +
 784        pushed (or skipped as already frozen), unlike the fire-and-return
 785        :meth:`build_challenge`. ⚠️ *deadline_s* bounds the node's whole
 786        freeze (#3167, `ctfy.core.freeze_budget`; ``0`` sends the node's own
 787        default), and the HTTP wait is derived from it —
 788        `freeze_client_wait_s`, past the node's own answer — so the client
 789        never abandons a freeze the node is still running. Sent once: a
 790        ``ReadTimeout`` is a request the node has (#3162). Returns the per-service freeze result;
 791        the platform's ``challenge_build_push`` task calls this one
 792        challenge at a time. *push_ticket* is the platform's short-lived
 793        registry credential the node's ``docker push`` authenticates with
 794        — the node runs it in a container with no ambient login.
 795        *cleanup* asks the node to remove what the build left and prune its
 796        build cache to *cache_max_bytes* afterwards, and *confine_network*
 797        keeps every ``RUN`` off the host's network (both the freeze pool, #1827).
 798        *run_id* names this call for :meth:`cancel_build_push` and
 799        :meth:`build_push_progress` while it is in flight (#1949).
 800        """
 801        resp = self.request(
 802            "POST",
 803            f"/admin/challenges/{challenge_id}/build-push",
 804            json={
 805                "push_ticket": push_ticket,
 806                "cleanup": cleanup,
 807                "cache_max_bytes": cache_max_bytes,
 808                "confine_network": confine_network,
 809                "authored_revision": authored_revision,
 810                "run_id": run_id,
 811                "deadline_s": deadline_s,
 812                # #2504: the build limits an authored tree is frozen under;
 813                # a node refuses one that arrives without them.
 814                "resource_budget": (
 815                    resource_budget.model_dump() if resource_budget is not None else None
 816                ),
 817            },
 818            timeout=freeze_client_wait_s(deadline_s or freeze_deadline_s()),
 819            retry_on=RETRY_UNSENT,
 820        )
 821        _raise_for_node_status(resp)
 822        return _parsed(ChallengeFreezeResult, resp)
 823
 824    def cancel_build_push(self, run_id: str) -> BuildPushCancelled:
 825        """Cancel the freeze :meth:`build_push_challenge` started under *run_id* (#1949).
 826
 827        Idempotent; ``found`` false means the node holds no such run.
 828        """
 829        resp = self.request("POST", "/admin/build-push/cancel", json={"run_id": run_id})
 830        _raise_for_node_status(resp)
 831        return _parsed(BuildPushCancelled, resp)
 832
 833    def build_push_progress(self, run_id: str) -> BuildPushProgress:
 834        """What the freeze running under *run_id* is doing now (#1949)."""
 835        resp = self.request("GET", f"/admin/build-push/{run_id}/progress")
 836        _raise_for_node_status(resp)
 837        return _parsed(BuildPushProgress, resp)
 838
 839    def get_disk_free(self) -> NodeDiskFree:
 840        """Free space on the node's data filesystem, read now (#1827).
 841
 842        Raises on any failure — including a node predating the route, which
 843        answers 404 — and the freeze reads every failure as *cannot tell*,
 844        which stops it.
 845        """
 846        resp = self.request("GET", "/admin/disk")
 847        _raise_for_node_status(resp)
 848        return _parsed(NodeDiskFree, resp)
 849
 850    def get_build_state(self) -> BuildStateRows:
 851        """Fetch every per-challenge build-state row on this node.
 852
 853        Returns ``{"rows": [{challenge_id, status, built_at, error}, …]}``.
 854        ``status`` is one of ``unbuilt`` / ``building`` / ``built`` /
 855        ``failed``; ``unbuilt`` placeholders are synthesised for specs
 856        the node has seen but never been asked to build.
 857        """
 858        resp = self.request("GET", "/admin/challenges/build-state")
 859        _raise_for_node_status(resp)
 860        return _parsed(BuildStateRows, resp)
 861
 862    def pull_challenge(self, challenge_id: str) -> ChallengePullState:
 863        """Ask the node to pre-pull registry images for *challenge_id*.
 864
 865        Pull-side twin of :meth:`build_challenge`: the node persists
 866        ``status="pulling"`` and spawns a daemon thread; poll
 867        :meth:`get_pull_state` for the ``pulled`` / ``failed`` landing.
 868        """
 869        resp = self.request("POST", f"/admin/challenges/{challenge_id}/pull")
 870        _raise_for_node_status(resp)
 871        return _parsed(ChallengePullState, resp)
 872
 873    def pull_all_challenges(self, challenge_ids: list[str] | None = None) -> PullAllQueued:
 874        """Queue background pre-pull on the node.
 875
 876        With ``challenge_ids=None`` the node pulls every spec it knows;
 877        with an explicit list only those (the platform scoping to a
 878        competition). Returns ``{queued, skipped_pulled,
 879        skipped_in_progress}``.
 880        """
 881        kwargs: dict[str, Any] = {}
 882        if challenge_ids is not None:
 883            kwargs["json"] = {"challenge_ids": challenge_ids}
 884        resp = self.request("POST", "/admin/challenges/pull-all", **kwargs)
 885        _raise_for_node_status(resp)
 886        return _parsed(PullAllQueued, resp)
 887
 888    def get_pull_state(self) -> PullStateRows:
 889        """Fetch every per-challenge pull-state row on this node.
 890
 891        Returns ``{"rows": [{challenge_id, status, pulled_at, error}, …]}``.
 892        ``status`` is one of ``unpulled`` / ``pulling`` / ``pulled`` /
 893        ``failed``; ``unpulled`` placeholders are synthesised for specs
 894        the node has seen but never been asked to pull.
 895        """
 896        resp = self.request("GET", "/admin/challenges/pull-state")
 897        _raise_for_node_status(resp)
 898        return _parsed(PullStateRows, resp)
 899
 900    # -- status / health ----------------------------------------------------
 901
 902    def get_status(self, instance_id: str) -> InstanceStatusReport:
 903        resp = self.request("GET", f"/instances/{instance_id}/status")
 904        resp.raise_for_status()
 905        return _parsed(InstanceStatusReport, resp)
 906
 907    def list_instances(self) -> list[NodeInstanceSummary]:
 908        """Every instance the node still has bookkeeping for.
 909
 910        Used by the platform's boot re-adoption pass to decide which
 911        durable claims still have containers behind them. ``get_status``
 912        answers the same question one id at a time; asking it per claim
 913        would cost one round trip per instance to a node that may be
 914        slow or down, on the path that gates startup.
 915        """
 916        resp = self.request("GET", "/instances")
 917        resp.raise_for_status()
 918        return _parsed(NodeInstanceList, resp).items
 919
 920    def check_health(self, instance_id: str) -> bool:
 921        return self.check_health_verbose(instance_id).healthy
 922
 923    def check_health_verbose(self, instance_id: str) -> HealthReading:
 924        """Health plus whatever agent-deception telemetry the node read.
 925
 926        One call, because the platform already polls this endpoint for
 927        every live instance every tick and a second round trip per
 928        instance would be the whole cost of the feature. ``check_health``
 929        stays a bool for the callers that only ever wanted one.
 930
 931        Never an error, whatever arrives: this rides the health poll,
 932        and telemetry must not be able to mark a healthy instance
 933        unhealthy.
 934
 935        ⚠️ **The two ways this can come back empty are different facts
 936        and used to be one.** An older node sends no ``deception`` key
 937        at all — it *did not say*, which is ``channel=None`` and
 938        authorises nothing. A body this platform cannot parse is
 939        something else: the node did say, and we could not read it. That
 940        is ``unreadable``, the same value the reader gives for a missing
 941        file, because it is the same fact — there is no usable channel
 942        here — arriving through a different door.
 943
 944        The parse failure that matters is version skew in the other
 945        direction: a **newer** node reporting an ``ObservationKind`` this
 946        platform does not have. ``Observation.kind`` is typed, so one
 947        unknown member fails the whole model rather than dropping one
 948        row — which is the right refusal (the node's word is not ours to
 949        edit) and exactly why it must not land as "reported nothing".
 950        """
 951        resp = self.request("GET", f"/instances/{instance_id}/health")
 952        resp.raise_for_status()
 953        body = resp.json()
 954        raw = body.get("deception")
 955        if not raw:
 956            summary = DeceptionSummary()
 957        else:
 958            try:
 959                summary = DeceptionSummary.model_validate(raw)
 960            except ValidationError as exc:
 961                log.info(
 962                    "unreadable deception telemetry from node",
 963                    instance_id=instance_id,
 964                    error=tail(str(exc), limit=200),
 965                )
 966                summary = DeceptionSummary(channel="unreadable")
 967        raw_contact = body.get("contact")
 968        if not raw_contact:
 969            contact = ContactReading()
 970        else:
 971            try:
 972                contact = ContactReading.model_validate(raw_contact)
 973            except ValidationError as exc:
 974                log.info(
 975                    "unreadable contact telemetry from node",
 976                    instance_id=instance_id,
 977                    error=tail(str(exc), limit=200),
 978                )
 979                contact = ContactReading(channel="unreadable")
 980        raw_procs = body.get("new_processes")
 981        # ⚠️ Shape-checked rather than trusted: an older node sends
 982        # nothing and a newer one could send a shape this platform does
 983        # not model, and neither may reach a reader as *a container full
 984        # of intruders*. Anything unreadable is silence, which for a
 985        # signal built to under-report is the honest reading.
 986        procs: dict[str, list[str]] = {}
 987        if isinstance(raw_procs, dict):
 988            for container, names in raw_procs.items():
 989                if isinstance(container, str) and isinstance(names, list):
 990                    procs[container] = [str(n) for n in names]
 991        return HealthReading(
 992            healthy=bool(body.get("is_healthy")),
 993            deception=summary,
 994            contact=contact,
 995            new_processes=procs,
 996        )
 997
 998    def node_health(self) -> dict[str, Any]:
 999        """Liveness + ``{running, capacity}`` for heartbeat."""
1000        resp = self.request("GET", "/health")
1001        resp.raise_for_status()
1002        body: dict[str, Any] = resp.json()
1003        return body
1004
1005    # -- traffic / runtime --------------------------------------------------
1006
1007    def get_traffic_page(
1008        self, instance_id: str, *, cursor: str = "", limit: int = 0
1009    ) -> InstanceTrafficPage:
1010        """One page of mitmproxy flows; the capture lives on the node's FS.
1011
1012        ⚠️ **The one method that must not use ``_parsed``.** A node from
1013        before paging answers with a bare *list* — it sent the whole
1014        capture and understands no cursor — and a strict
1015        ``model_validate`` turns that supported skew into a raised
1016        request. So the body goes through ``coerce_page``, which already
1017        owns every decision about an unexpected shape, and the caller
1018        gets the model.
1019
1020        This is the same move ``coerce_page`` was written for, one layer
1021        down. Its docstring used to say the parsing belonged
1022        "platform-side … so a route and this client cannot come to
1023        disagree" — and there were *three* platform-side calls doing it.
1024        There is one call now, at the seam that knows the request
1025        cursor, which is what the fallback needs; the routes hold the
1026        type instead of re-deciding its shape.
1027
1028        ⚠️ **An empty ``cursor`` means "from the beginning".** Passing
1029        one on every poll re-reads the whole capture, which is the cost
1030        the paged read exists to remove — arriving through the caller
1031        rather than through the node.
1032        """
1033        params: dict[str, Any] = {}
1034        if cursor:
1035            params["cursor"] = cursor
1036        if limit > 0:
1037            params["limit"] = limit
1038        resp = self.request("GET", f"/instances/{instance_id}/traffic", params=params)
1039        resp.raise_for_status()
1040        page = coerce_page(resp.json(), cursor)
1041        return InstanceTrafficPage(
1042            flows=page.flows,
1043            cursor=page.cursor,
1044            truncated=page.truncated,
1045            capture=page.capture,
1046        )
1047
1048    def get_traffic_witnesses(
1049        self, instance_id: str, *, cursor: str = "", limit: int = 0
1050    ) -> InstanceTrafficWitnesses:
1051        """Who spoke to this box since *cursor*, with no bodies moved.
1052
1053        §7.1's correlation reads one gamebox per team per round. Through
1054        :meth:`get_traffic_page` that is every flow, both bodies
1055        included (~100 kB apiece by ``CLAUDE.md``'s measurement),
1056        pulled across the wire to extract a handful of addresses; the
1057        node reduces instead and this carries the result.
1058
1059        ⚠️ **Strict parsing here, unlike its neighbour, and that is the
1060        difference between an old field and an old route.**
1061        ``get_traffic_page`` is lenient because a node from before
1062        paging answers the *same* path with a bare list, which is
1063        supported skew. This route did not exist on such a node at all,
1064        so the honest answer from one is a 404 — and a lenient parse of
1065        an error body would hand the round loop an empty witness set,
1066        which reads as *nobody was there*. The caller catches the
1067        request error and records the round as unknown.
1068        """
1069        params: dict[str, Any] = {}
1070        if cursor:
1071            params["cursor"] = cursor
1072        if limit > 0:
1073            params["limit"] = limit
1074        resp = self.request("GET", f"/instances/{instance_id}/traffic-witnesses", params=params)
1075        resp.raise_for_status()
1076        return _parsed(InstanceTrafficWitnesses, resp)
1077
1078    def list_containers(self, instance_id: str) -> list[ComposeContainer]:
1079        """Enumerate every container in an instance's compose project.
1080
1081        Backs the admin-shell feature: the platform forwards a
1082        ``GET /admin/instances/{id}/containers`` to the assigned node,
1083        which returns one row per container (challenge services and
1084        platform-injected sidecars alike). The WebSocket reverse-proxy
1085        is opened separately and does not go through ``NodeClient``.
1086        """
1087        resp = self.request("GET", f"/instances/{instance_id}/containers")
1088        _raise_for_node_status(resp)
1089        return _parsed_list(ComposeContainer, resp)
1090
1091    def run_checker(
1092        self,
1093        instance_id: str,
1094        *,
1095        service: str,
1096        cmd: list[str],
1097        timeout_s: int = DEFAULT_JUDGE_TIMEOUT_S,
1098    ) -> CheckResult:
1099        """Exec an author-supplied checker in a trusted judge sidecar.
1100
1101        Backs the checker / exec-judge feature: the platform forwards a
1102        verify request to the assigned node, which runs ``cmd`` inside the
1103        named sidecar and returns
1104        ``{exit_code, stdout, stderr, timed_out, error}``.
1105
1106        ⚠️ *timeout_s* is the author's, and the node clamps it (#3169): the
1107        clamped budget is what is sent and what the wait is derived from
1108        (`core.judge_budget.platform_wait_s`), so ``0`` no longer waits 10 s
1109        on a node running 30, and ``600`` no longer holds a threadpool slot
1110        for 610 s on a node that stops at 120. Sent once (``RETRY_UNSENT``).
1111        """
1112        exec_s = judge_exec_s(timeout_s)
1113        resp = self.request(
1114            "POST",
1115            f"/instances/{instance_id}/check",
1116            json={"service": service, "cmd": cmd, "timeout_s": exec_s},
1117            timeout=judge_platform_wait_s(exec_s),
1118            retry_on=RETRY_UNSENT,
1119        )
1120        _raise_for_node_status(resp)
1121        return _parsed(CheckResult, resp)
1122
1123    def run_harness(
1124        self,
1125        *,
1126        image: str,
1127        run_id: str = "",
1128        env: dict[str, str],
1129        timeout_s: int = 1800,
1130        untrusted: bool = False,
1131        network: str = "",
1132        own_network: bool = False,
1133        match_network: str = "",
1134        match_address: str = "",
1135        brief: dict[str, Any] | None = None,
1136        target_instance: str = "",
1137        target_aliases: dict[str, str] | None = None,
1138        target_access: str = "",
1139        vpn_config: str = "",
1140        estate_zones: Sequence[str] = (),
1141        target_endpoints: Sequence[str] = (),
1142        target_hosts: Mapping[str, str] | None = None,
1143        target_datagram_endpoints: Sequence[str] = (),
1144        pull_ticket: str = "",
1145        registry_auth: HarnessRegistryAuth | None = None,
1146        limits: HarnessLimits | None = None,
1147        admission_profile: bool = False,
1148        egress_allow: list[str] | None = None,
1149        internet: bool = True,
1150        rootfs_budget_bytes: int = 0,
1151        unpack_budget_bytes: int = 0,
1152        disk_low_watermark_bytes: int = 0,
1153        ingress: bool = True,
1154        local_ingress: bool = False,
1155        platform_image: bool = False,
1156        image_policy: HarnessImagePolicy | None,
1157        attachments: dict[str, str] | None = None,
1158        keep_work: bool = False,
1159        resume_work: bool = False,
1160        exploit: dict[str, Any] | None = None,
1161    ) -> HarnessRun:
1162        """Run a one-shot evaluation-harness container on the node.
1163
1164        ``exploit`` (#2819 W2) runs a smoke's reference exploit in the fence
1165        instead of an agent (``HarnessExploit``'s fields); sent only when set.
1166
1167        ``image_policy`` (#2608) is **required, with no default**: every
1168        caller says whether this image is a user's (the registry rules the
1169        node re-checks before pulling) or the platform's own (``None``). An
1170        untrusted run without one is refused by the node.
1171
1172        The node ``docker run``s ``image`` with ``env`` injected, waits for it
1173        to exit (bounded by ``timeout_s``), and returns
1174        ``{exit_code, stdout, stderr, timed_out, error}`` — the same shape as
1175        :meth:`run_checker`. The harness writes its JSON rollup to stdout. The
1176        per-call HTTP timeout sits above the node's run budget so the client
1177        doesn't abandon the request before the (long-running) harness returns.
1178
1179        ``untrusted`` runs the image under the node's isolation profile —
1180        off host networking, bounded resources, no capabilities — for an
1181        image the platform did not build. It requires a network: the
1182        node answers 400 without one, because "isolate this" and "I will
1183        not say where the target lives" do not compose.
1184
1185        Any of ``network`` (the target instance's own, when the caller
1186        knows it), ``own_network`` or ``target_instance`` satisfies that.
1187        ⚠️ Most eval dispatches need the second: there the harness
1188        *starts* the target partway through its own run, so at ``docker
1189        run`` time no target network exists and there is no name to
1190        send. The node creates a fresh bridge and removes it afterwards.
1191
1192        ``target_instance`` reverses that ordering, and is what an
1193        `engagement` uses: the platform starts the estate first and names
1194        it here, and the node wires its published containers to the
1195        agent's bridge. ⚠️ An **id**, never a network name — the name is
1196        a fact only the node's daemon can read, and ``docker run``
1197        *creates* whatever string it is handed, so a platform guessing
1198        one starts a run that reaches nothing.
1199        """
1200        resp = self.request(
1201            "POST",
1202            "/harness/run",
1203            json={
1204                "image": image,
1205                "env": env,
1206                "timeout_s": timeout_s,
1207                "untrusted": untrusted,
1208                "network": network,
1209                "own_network": own_network,
1210                "match_network": match_network,
1211                # ⚠️ **The platform-assigned arena address.** The node
1212                # connects the agent to the match network with this
1213                # ``--ip``, so the address is known before the round loop
1214                # asks (§7.1). Empty means "let docker's IPAM pick" — an
1215                # older platform's behaviour, and what keeps a node reading
1216                # this field unchanged for every non-arena dispatch.
1217                "match_address": match_address,
1218                # ⚠️ Sent even when empty, so a node reading it gets the
1219                # same shape either way; the node's default is what makes
1220                # an older platform's dispatch unchanged.
1221                "pull_ticket": pull_ticket,
1222                # #3475: the owner's registry login, sent **only when set**,
1223                # so a node that predates the field sees the body it always did.
1224                **(
1225                    {"registry_auth": registry_auth.model_dump()}
1226                    if registry_auth is not None
1227                    else {}
1228                ),
1229                # ⚠️ ``None`` leaves the node's own defaults in force, which is
1230                # what a dispatch said before the ceilings were settings.
1231                "limits": limits.model_dump() if limits is not None else None,
1232                # ⚠️ Same rule, and empty is a real answer here too: a
1233                # dispatch with no run row (the conformance probe, a dry
1234                # run) has nothing to label, and a label with an empty
1235                # value would make every one of them answer one query.
1236                "run_id": run_id,
1237                # ⚠️ Sent as the already-built manifest. The platform is
1238                # the side that knows the challenge, the competition and
1239                # the run mode; a node deriving its own would be a
1240                # second implementation of a published contract.
1241                "brief": brief,
1242                "target_instance": target_instance,
1243                # ⚠️ **The names the brief published, so they resolve on
1244                # the agent's bridge too.** An estate's hosts answer to a
1245                # per-instance FQDN minted by its own resolver; joined to
1246                # a new network a container answers only to its compose
1247                # *service* name. Sending the mapping is what keeps the
1248                # address the agent was told about true under either
1249                # placement — without it the agent is handed a hostname
1250                # that does not resolve, reaches nothing, and reports an
1251                # empty result, which reads as a model that gave up.
1252                #
1253                # ⚠️ The platform's to send: it computed the surface. A
1254                # node re-deriving it from the rendered compose would be
1255                # a second implementation of the address the brief has
1256                # already promised.
1257                "target_aliases": target_aliases or {},
1258                # #2550: ``{name: base64}`` for ``/ctfy/work/attachments/``.
1259                # ⚠️ Sent only when there are some, so every other dispatch
1260                # is byte-identical to before.
1261                **({"attachments": attachments} if attachments else {}),
1262                # #2550 restarts: sent only when asked for, like the files.
1263                **({"keep_work": True} if keep_work else {}),
1264                **({"resume_work": True} if resume_work else {}),
1265                # #2861 slice B: the target access the platform read off the
1266                # target's node. ⚠️ Sent only when there is one, so every
1267                # other dispatch is byte-identical to before; and only to a
1268                # node whose ``capabilities().harness_access`` said it reads
1269                # them, or to the target's own node (where an older node's
1270                # local read is still right).
1271                **(
1272                    {
1273                        "target_access": target_access,
1274                        "vpn_config": vpn_config,
1275                        "estate_zones": list(estate_zones),
1276                        "target_endpoints": list(target_endpoints),
1277                    }
1278                    if target_access
1279                    else {}
1280                ),
1281                # #3020: names → the target node's ingress address, for a
1282                # published target handed over the agent-ingress tunnel. Sent
1283                # only when there are some, and only to a node whose
1284                # ``capabilities().harness_ingress_access`` said it reads them.
1285                **({"target_hosts": dict(target_hosts)} if target_hosts else {}),
1286                # #3446: the published endpoints that are UDP, which the node
1287                # never TCP-probes. Sent only when there are some.
1288                **(
1289                    {"target_datagram_endpoints": list(target_datagram_endpoints)}
1290                    if target_datagram_endpoints
1291                    else {}
1292                ),
1293                # #2356: ⚠️ sent only when asked for, so a dispatch that does
1294                # not ask is byte-identical to an older platform's — and an
1295                # older node, which would ignore the fields, is never sent
1296                # them: the caller checks ``capabilities().harness_admission``.
1297                **(
1298                    {
1299                        "admission_profile": True,
1300                        "egress_allow": list(egress_allow or []),
1301                        "internet": internet,
1302                        "rootfs_budget_bytes": rootfs_budget_bytes,
1303                        "unpack_budget_bytes": unpack_budget_bytes,
1304                        "disk_low_watermark_bytes": disk_low_watermark_bytes,
1305                        "ingress": ingress,
1306                        # #2819 W1, F2: sent only when asked for.
1307                        **({"local_ingress": True} if local_ingress else {}),
1308                    }
1309                    if admission_profile
1310                    else {}
1311                ),
1312                "image_policy": image_policy.model_dump() if image_policy is not None else None,
1313                # #2608 N4: only for a row with no owner; see the node's request.
1314                **({"platform_image": True} if platform_image else {}),
1315                **({"exploit": exploit} if exploit is not None else {}),
1316            },
1317            timeout=(
1318                timeout_s
1319                # #2819 W2: the exploit's `solution/Dockerfile` builds first.
1320                + (SOLUTION_BUILD_TIMEOUT_S if exploit is not None else 0)
1321                + IMAGE_PULL_TIMEOUT_S
1322                + HARNESS_IMAGE_INSPECT_TIMEOUT_S
1323                + HARNESS_CREATE_TIMEOUT_S
1324                + 30
1325            ),
1326            retry_on=RETRY_UNSENT,
1327        )
1328        _raise_for_node_status(resp)
1329        return _parsed(HarnessRun, resp)
1330
1331    def cancel_harness(self, *, run_id: str) -> int:
1332        """Stop whatever this node is running for ``run_id``; returns how many.
1333
1334        Until this existed a dispatched container was reclaimed by
1335        **nothing but its own timeout** — ``_harness_timeout``'s docstring
1336        says exactly that — so an operator pressing cancel on a
1337        match-long run was asking for something the platform had no way
1338        to deliver.
1339
1340        ⚠️ **Zero is an ordinary answer, not a failure.** The run may have
1341        finished on its own, been dispatched to a different node, or been
1342        lost to a node restart. A caller that read zero as an error would
1343        report a broken platform every time a cancel arrived a second
1344        late.
1345        """
1346        resp = self.request("POST", "/harness/cancel", json={"run_id": run_id})
1347        body = resp.json()
1348        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
1349
1350    def verify_patch(
1351        self, challenge_id: str, files: dict[str, bytes], *, deadline_s: int
1352    ) -> PatchVerifyResult:
1353        """Build a submitted patch on the node and return its verdict.
1354
1355        ``files`` maps a challenge-relative path to its new bytes; base64
1356        is applied here because a patch target may legitimately be
1357        binary. Returns ``{verdict, detail, applied, steps, error}``.
1358
1359        **An empty ``verdict`` with a populated ``error`` is not a
1360        judgement** — it means the node refused the submission, the
1361        verifier crashed, or the run's deadline ran out. The caller must
1362        retry or retire the lease rather than write a score, since failing
1363        to judge a patch is not the same as judging it unfixed.
1364
1365        ⚠️ *deadline_s* bounds the node's whole run (#3164,
1366        `ctfy.core.patch_budget`), and the HTTP wait is derived from it —
1367        `patch_client_wait_s`, past the node's own answer — so the client
1368        never abandons a request the node is still working on. Sent once:
1369        a ``ReadTimeout`` is a request the node has (#3162).
1370        """
1371        resp = self.request(
1372            "POST",
1373            "/patches/verify",
1374            json={
1375                "challenge_id": challenge_id,
1376                "files": {
1377                    path: base64.b64encode(content).decode() for path, content in files.items()
1378                },
1379                "deadline_s": deadline_s,
1380            },
1381            timeout=patch_client_wait_s(deadline_s),
1382            retry_on=RETRY_UNSENT,
1383        )
1384        _raise_for_node_status(resp)
1385        return _parsed(PatchVerifyResult, resp)
1386
1387    def collect_patch_files(
1388        self, instance_id: str, service: str, paths: list[str]
1389    ) -> dict[str, bytes]:
1390        """Read a player's SSH edits back out of their running box.
1391
1392        ``paths`` are absolute in-container paths the *platform* derived
1393        from the pristine source tree. Returns only what could be read:
1394        a path missing from the reply means the player did not change it,
1395        and the caller falls back to the shipped file rather than
1396        recording a deletion they never made.
1397        """
1398        resp = self.request(
1399            "POST",
1400            f"/instances/{instance_id}/patch-collect",
1401            json={"service": service, "paths": paths},
1402        )
1403        _raise_for_node_status(resp)
1404        body: dict[str, str] = resp.json()
1405        return {path: base64.b64decode(blob) for path, blob in body.items()}
1406
1407    def write_awd_answers(self, boxes: dict[str, dict[str, str]]) -> dict[str, str]:
1408        """Rotate a round's flags into every box of this node, in one call.
1409
1410        ⚠️ **One call for the whole node, never one per box.** §5.2's
1411        capacity red line is 1000 teams x 3 services injected inside
1412        120 s; per-box that is 3000 round trips, batched it is about
1413        fourteen. The signature is the enforcement — a caller cannot
1414        accidentally loop.
1415
1416        Returns ``{instance_id: ""}`` for the boxes written and a reason
1417        for the ones that were not, because one team's unreachable box
1418        must not cost every other team on the node its rotation. An
1419        unrotated box keeps serving last round's flag, which anyone
1420        holding the old value can replay.
1421        """
1422        resp = self.request("POST", "/awd/answers", json={"boxes": boxes})
1423        _raise_for_node_status(resp)
1424        result: dict[str, str] = resp.json()
1425        return result
1426
1427    def probe_awd_service(
1428        self,
1429        *,
1430        instance_id: str,
1431        service_id: str,
1432        tick: int,
1433        flag: str,
1434        previous_flag: str = "",
1435        previous_flag_id: str = "",
1436    ) -> dict[str, str]:
1437        """Run one round's SLA probes against one box, from outside it.
1438
1439        The flags travel *to* the node: a checker that sourced its
1440        expected value from the box it is checking would pass on any box
1441        that returns whatever it was handed.
1442        """
1443        resp = self.request(
1444            "POST",
1445            "/awd/probe",
1446            json={
1447                "instance_id": instance_id,
1448                "service_id": service_id,
1449                "tick": tick,
1450                "flag": flag,
1451                "previous_flag": previous_flag,
1452                "previous_flag_id": previous_flag_id,
1453            },
1454        )
1455        _raise_for_node_status(resp)
1456        result: dict[str, str] = resp.json()
1457        return result
1458
1459    def attack_awd_box(
1460        self,
1461        *,
1462        instance_id: str,
1463        tick: int,
1464        timeout_s: int = 60,
1465        agent_image: str = "",
1466        counter_flag: str = "",
1467        forensic_marker: str = "",
1468        llm_api_key: str = "",
1469    ) -> AwdAttackResult:
1470        """Run this box's own reference exploit against it, once.
1471
1472        Per **box**, not per service: one `run_exploit` discovers every
1473        answer the challenge declares and prints them all.
1474
1475        ``timeout_s`` is sent because the platform is what knows how long
1476        a round is; the node clamps it, because a container held open for
1477        an hour is the node's problem whoever asked for it.
1478
1479        ``forensic_marker`` is §4.3.1's, carried only by the agent
1480        branch: `build_attack_argv` passes no environment at all, so a
1481        marker on the reference exploit would be the one value that
1482        voids `EXPLOIT007`'s guarantee.
1483
1484        ``agent_image`` + ``counter_flag`` switch the node to §4.2b's
1485        agent attacker. ⚠️ **Both or neither** — the node refuses a half
1486        request rather than falling back to the reference exploit, which
1487        would run a different format than the one configured while the
1488        board looked entirely normal.
1489        """
1490        resp = self.request(
1491            "POST",
1492            "/awd/attack",
1493            json={
1494                "instance_id": instance_id,
1495                "tick": tick,
1496                "timeout_s": timeout_s,
1497                "agent_image": agent_image,
1498                "counter_flag": counter_flag,
1499                "forensic_marker": forensic_marker,
1500                "llm_api_key": llm_api_key,
1501            },
1502        )
1503        _raise_for_node_status(resp)
1504        return _parsed(AwdAttackResult, resp)
1505
1506    def get_container_logs(self, instance_id: str) -> str:
1507        """Fetch combined container stdout/stderr for archival."""
1508        resp = self.request("GET", f"/instances/{instance_id}/container-logs")
1509        resp.raise_for_status()
1510        return str(resp.json().get("logs") or "")
1511
1512    def get_pcap(self, instance_id: str) -> PcapCapture:
1513        """Fetch the merged tcpdump capture for *instance_id*.
1514
1515        ``data`` is empty when no capture exists (sidecar disabled,
1516        instance never reached the running state, etc.) so callers can
1517        persist conditionally without try/except.
1518
1519        ⚠️ **It also says what the per-instance cap already dropped**
1520        (#1714), off the node's response headers, and ``None`` when the
1521        node did not say — an older node, which is *not reported*, never
1522        a clean capture.
1523        """
1524        resp = self.request("GET", f"/instances/{instance_id}/pcap")
1525        if resp.status_code == 404:
1526            return PcapCapture(data=b"")
1527        resp.raise_for_status()
1528        return PcapCapture(
1529            data=resp.content,
1530            dropped_bytes=_header_int(resp, CAPTURE_DROPPED_BYTES_HEADER),
1531            dropped_segments=_header_int(resp, CAPTURE_DROPPED_SEGMENTS_HEADER),
1532        )
1533
1534    def get_harness_artifact(self, artifact_id: str) -> bytes:
1535        """One finished run's collected archive, or ``b""``.
1536
1537        ⚠️ **Empty rather than an exception for a 404**, following
1538        :meth:`get_pcap`. The node reaps uncollected archives, and a
1539        platform that comes back after a restart is exactly the caller
1540        that meets one that is gone — a state its own record already
1541        describes, not a transport failure.
1542        """
1543        resp = self.request("GET", f"/harness/artifacts/{artifact_id}")
1544        if resp.status_code == 404:
1545            return b""
1546        resp.raise_for_status()
1547        return resp.content
1548
1549    def delete_harness_artifact(self, artifact_id: str) -> bool:
1550        """Tell the node the archive has been stored and can go.
1551
1552        ⚠️ **Never raises on a transport failure**, because it is called
1553        after the bytes are safely in object storage: turning a cleanup
1554        error into the run's error would replace a success with a
1555        janitorial complaint. The node's reaper collects whatever this
1556        misses.
1557        """
1558        try:
1559            resp = self.request("DELETE", f"/harness/artifacts/{artifact_id}")
1560        except Exception:
1561            return False
1562        return resp.status_code == 200
1563
1564    def get_agent_runtime(self, instance_id: str) -> AgentRuntime:
1565        """Provider-agnostic runtime hints for attaching an agent (proxy URL,
1566        CA PEM, optional Docker-specific names). Replaces the old
1567        sandbox-network shape."""
1568        resp = self.request("GET", f"/instances/{instance_id}/agent-runtime")
1569        resp.raise_for_status()
1570        return _parsed(AgentRuntime, resp)
1571
1572    # -- per-instance rendered attachments ---------------------------------
1573
1574    def list_instance_attachments(self, instance_id: str) -> AttachmentList:
1575        """List per-instance attachments rendered into the node's workdir.
1576
1577        Backs the platform's per-instance attachment endpoint, which
1578        surfaces the team-specific rendered file listing.
1579
1580        ⚠️ This used to hand back the raw dict, under a docstring saying
1581        "caller projects through :class:`AttachmentList` for typing" —
1582        the node's route already declared that model, so the type existed
1583        and this method's signature was the one place it was dropped.
1584        Projecting *here* means there is one place a malformed answer is
1585        caught rather than one per caller."""
1586        resp = self.request("GET", f"/instances/{instance_id}/attachments")
1587        resp.raise_for_status()
1588        return _parsed(AttachmentList, resp)
1589
1590    def get_vpn_config(self, instance_id: str) -> bytes:
1591        """Fetch the per-instance VPN client config from the node.
1592
1593        The node reads the backend's client config out of the
1594        instance's VPN service container (generated on first boot by
1595        the base image's bootstrap) and rewrites its endpoint to the
1596        node's public host:port. Returns the body verbatim — the
1597        platform-side route adds the ``Content-Disposition`` header
1598        before re-emitting to the player, because only it knows the
1599        backend and therefore the filename the player needs.
1600
1601        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
1602        proxy route catches 404 and re-emits to the player, anything
1603        else is treated as a 502.
1604        """
1605        resp = self.request("GET", f"/instances/{instance_id}/vpn-config")
1606        resp.raise_for_status()
1607        return resp.content
1608
1609    def get_vpn_access(self, instance_id: str) -> tuple[bytes, tuple[str, ...] | None]:
1610        """The player's VPN config plus the estate's resolver zones (#2861 B).
1611
1612        The config is :meth:`get_vpn_config`'s, byte for byte; the zones ride
1613        a response header the node adds only when asked. ``None`` for the
1614        zones is a node that did not say (older, or docker unreadable there)
1615        — never "no zones". Raises like :meth:`get_vpn_config`.
1616        """
1617        resp = self.request(
1618            "GET", f"/instances/{instance_id}/vpn-config", params={"estate_zones": "true"}
1619        )
1620        resp.raise_for_status()
1621        raw = resp.headers.get("X-Ctfy-Estate-Zones")
1622        return resp.content, (tuple(raw.split()) if raw is not None else None)
1623
1624    def download_instance_attachment(self, instance_id: str, filename: str) -> tuple[bytes, str]:
1625        """Download one per-instance attachment as ``(bytes, content_type)``.
1626
1627        Buffers the whole body — attachments are bounded (typical case
1628        is a 10 KB binary or a small text file). For huge captures the
1629        caller should hand the player a CDN URL instead.
1630
1631        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
1632        proxy route catches 404 and re-emits to the player, anything
1633        else is a 502. Invalid names or instance keys raise
1634        ``InvalidAttachmentName`` before any request is sent."""
1635        # The instance is a single route key, never part of the file name.
1636        if component_refusal(instance_id) is not None:
1637            raise InvalidAttachmentName("Invalid attachment instance")
1638        resp = self.request(
1639            "GET",
1640            f"/instances/{quote(instance_id, safe='')}/attachments/{attachment_url_path(filename)}",
1641        )
1642        resp.raise_for_status()
1643        return resp.content, resp.headers.get("content-type", "application/octet-stream")