diff --git a/CHANGELOG.md b/CHANGELOG.md index be76134..54fe888 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,9 @@ the fuller account of each version, including verification notes. id and a live handle, so the generation stays collectable. - `SubscribeTimeout` — a `TimeoutError` subclass raised by the `models.subscribe` timeouts that still raise, carrying `.request_id`, `.model`, `.cancelled` and `.cancel_error`. +- `IN_QUEUE` / `IN_PROGRESS` — the two live values of the contract's closed `RouterQueueStatus` + enum, exported alongside `COMPLETED` so a caller comparing `QueueUpdate.status` uses the + contract's own spelling. ### Changed @@ -32,9 +35,18 @@ the fuller account of each version, including verification notes. process. - **A 2xx on the cleanup cancel is no longer taken as proof the run stopped.** The cancel's own answer is read: terminal with a bucket is a cancellation, terminal without one is a run that - finished and is collected, and a live status (a `202`/`CANCELING`, or a request that won the race - into flight) is a detach. Only a body-less accepted cancel still reports a cancellation - unconfirmed, which is the shape that carries nothing to read. + finished and is collected, and a live status is a detach. Only a body-less accepted cancel still + reports a cancellation unconfirmed, which is the shape that carries nothing to read. +- **The cancel route's `202 CANCELLATION_REQUESTED` is read per the contract: accepted, then + confirmed by one status read.** It says the ask was taken and nothing more, so the request's own + state decides the ending — `COMPLETED` carrying the `cancelled` bucket is a cancellation + (`SubscribeTimeout` with `.cancelled` `True`, rather than the `Cancelled` router exception a run + that failed on its own raises); `IN_PROGRESS` or an unrecognised live status is a + `DetachedRequest`; and a row still `IN_QUEUE` — which the route's write order forbids — reports + that the cancel never applied, as `SubscribeTimeout` with `.cancelled` `False` and a + `.cancel_error` coded `cancel_not_applied`. A `DetachedRequest` can no longer be built carrying + `IN_QUEUE` at all: that status means the request was never dispatched and cannot be charged, + which is the opposite of what a detach claims. - **Only a `409` that names no bucket is read as the in-flight refusal.** A typed `CancelRefused` is recognised by its class; a `409` carrying a documented bucket (`invalid_input`, `concurrency_limit_exceeded`) is not the state refusal and surfaces on diff --git a/README.md b/README.md index 337903e..b964a70 100644 --- a/README.md +++ b/README.md @@ -536,7 +536,7 @@ The handle carries four operations: |---|---| | `handle.status()` | one authoritative poll, returned as a `QueueUpdate` (`status`, `queue_position`, `error_type`, `retry_after`, `raw`) | | `handle.get(timeout=None)` | poll to completion, then return the provider's own payload — the same value `run` would have returned | -| `handle.cancel()` | ask the server to cancel. A request, not a guarantee: the queue honours one only while the request is still waiting to be dispatched, and a run it has already started is served, billed, and its cancel *refused* — which reaches you as an exception rather than as an update | +| `handle.cancel()` | ask the server to cancel. A request, not a guarantee: the route takes a request in either live state and answers `202 CANCELLATION_REQUESTED`, which says the ask landed and not that the run stopped — read `handle.status()` afterwards. Only an already-terminal request is *refused*, which reaches you as an exception rather than as an update | | `handle.iter_events(timeout=None)` | the poll loop with its updates exposed — yields the first observation, every change of status or queue position, and the completion | Polling is **poll-authoritative**: there is no stream to reconcile against on @@ -581,10 +581,11 @@ outlive the caller's patience. #### A `timeout` detaches; it does not reliably cancel `timeout=` is a **client-side** bound with no server-side meaning. When it runs -out, `subscribe` makes a best-effort `cancel()` — and **the queue honours a -cancel only while the request is still waiting to be dispatched.** A run it has -already started is served to the end and **billed in full**, whatever the -caller does. So a timeout on an in-flight run is a *detach*, not a +out, `subscribe` makes a best-effort `cancel()` — and **accepting a cancel is +not the same as stopping the run.** The cancel route takes a request in either +live state, but a generation already on the wire at a partner may complete +anyway, and one that completes is **billed** whether or not anyone collected +it. So a timeout on a run the cancel did not stop is a *detach*, not a cancellation: you stop waiting, the generation carries on, it completes, and **you pay for it**. @@ -599,9 +600,10 @@ from comfy_sdk import DetachedRequest, SubscribeTimeout try: outcome = client.models.subscribe("fal-ai/flux-pro", {"prompt": "a cat"}, timeout=300) except SubscribeTimeout as exc: - # The request was still queued, the cancel was accepted, nothing ran and - # nothing is billed. `exc.cancelled` is False if the cancel itself failed, - # in which case the run may still be going — `exc.request_id` reaches it. + # The cancel landed and the request's row says it was withdrawn. + # `exc.cancelled` is False if the cancel itself failed or never applied, in + # which case the run may still be going — `exc.request_id` reaches it, and + # `exc.cancel_error` says what went wrong. raise if isinstance(outcome, DetachedRequest): @@ -616,16 +618,29 @@ The three ways the timeout can end, told apart by type and never by a message: | | | |---|---| -| **cancelled** | the request was still queued and the cancel really stopped it. Nothing ran, nothing is billed. Raises `SubscribeTimeout` (a `TimeoutError`) with `.cancelled` `True` | -| **detached** | the run was already in flight, so the cancel did not stop it. It continues, completes and **is billed**. Returns a `DetachedRequest` | -| **completed during teardown** | the run finished while the timeout was being torn down. The result exists and is paid for, so it is collected and returned like any other result | - -Which one you get is read off the cancel's **answer**, not off the bare fact -that it returned a 2xx: a cancel the route accepts while reporting a live -status (a `202`/`CANCELING`, or a request that won the race into flight) is a -**detach**, because the run did not stop. Only a body-less accepted cancel — -the shape that carries nothing to read — is reported as a cancellation -unconfirmed. +| **cancelled** | the cancel landed and the request's row came back `COMPLETED` carrying the `cancelled` bucket. Raises `SubscribeTimeout` (a `TimeoutError`) with `.cancelled` `True`. A request cancelled while still `IN_QUEUE` was never dispatched and cannot be charged; one cancelled after it was admitted **may still be charged**, and this ending is not a claim that it was not | +| **detached** | the cancel did not stop the run: the confirming poll found it still `IN_PROGRESS` (or reporting some live status this SDK cannot place). It continues, completes and **is billed**. Returns a `DetachedRequest` | +| **completed during teardown** | the run finished on its own while the timeout was being torn down. The result exists and is paid for, so it is collected and returned like any other result | + +Which one you get is read off the cancel's **answer and then off the request's +own state**, never off the bare fact that a 2xx came back. The route's own +answer is `202 CANCELLATION_REQUESTED`, which says the ask was accepted and +nothing more — so it is **confirmed by one status read**: + +* the row is `COMPLETED` carrying `cancelled` → `SubscribeTimeout` with + `.cancelled` `True`; +* the row is still `IN_QUEUE` → the cancel never applied (the route writes the + row terminal *before* it answers, so this is a server out of its own + documented order). `SubscribeTimeout` with `.cancelled` `False` and a + `.cancel_error` coded `cancel_not_applied`. Not a detach: an `IN_QUEUE` + request cannot have been charged, which is the opposite of what a detach + says; +* the row is `IN_PROGRESS`, or an unrecognised live status → a + `DetachedRequest`. + +A body-less accepted cancel — the legacy `204`, the shape that carries nothing +to read — is still reported as a cancellation unconfirmed, and a 2xx echoing a +live queue state is still a detach. A cancel that fails for some *other* reason — a transport failure, a rejected credential, a `500`, or a `409` naming a bucket the contract documents as diff --git a/src/comfy_sdk/__init__.py b/src/comfy_sdk/__init__.py index 9f6a231..a222fd2 100644 --- a/src/comfy_sdk/__init__.py +++ b/src/comfy_sdk/__init__.py @@ -71,6 +71,8 @@ from .jobs import AsyncJob, Job, JobWorkflow from .model_requests import ( COMPLETED, + IN_PROGRESS, + IN_QUEUE, AsyncDetachedRequest, AsyncRequestHandle, DetachedRequest, @@ -118,6 +120,8 @@ "AsyncRequestHandle", "QueueUpdate", "COMPLETED", + "IN_QUEUE", + "IN_PROGRESS", "DetachedRequest", "AsyncDetachedRequest", "SubscribeTimeout", diff --git a/src/comfy_sdk/model_requests.py b/src/comfy_sdk/model_requests.py index cb33d9d..246e864 100644 --- a/src/comfy_sdk/model_requests.py +++ b/src/comfy_sdk/model_requests.py @@ -53,7 +53,7 @@ from . import _core from .exceptions import ComfyError, translating from .retry import DEFAULT_RETRY, NO_RETRY, Retrier, RetryPolicy, error_bucket_of -from .router_exceptions import CancelRefused, RouterError, error_from_completion +from .router_exceptions import Cancelled, CancelRefused, RouterError, error_from_completion #: The one terminal queue status. Deliberately a single value rather than a #: :data:`comfy_sdk._core.TERMINAL`-style set: the server does not express a @@ -62,6 +62,48 @@ #: strand a caller whose request really did reach ``COMPLETED``. COMPLETED = "COMPLETED" +#: A request admitted to the queue that has not been dispatched to a partner +#: yet. One of the three values of the vendored contract's ``RouterQueueStatus`` +#: (``spec/router-openapi.yaml``, the ``RouterQueueStatus`` schema), which is a +#: **closed** ``enum`` on purpose: the spec says so in its own description, +#: because a lifecycle that grows a fourth state is a breaking change to every +#: polling loop written against it whether it is declared as an enum or not. +#: It is exported because a caller comparing :attr:`QueueUpdate.status` wants +#: the contract's own spelling rather than a literal of their own -- and +#: because it is the one live status with a billing consequence attached: the +#: ``cancelled`` bucket's meaning pins that a request cancelled while it was +#: still ``IN_QUEUE`` "was never dispatched and cannot be charged". +IN_QUEUE = "IN_QUEUE" + +#: A request a partner is generating for. The second of +#: :data:`IN_QUEUE`/:data:`IN_PROGRESS`/:data:`COMPLETED`, exported for the +#: same reason. Unlike :data:`IN_QUEUE` it carries no unbilled guarantee: the +#: ``cancelled`` bucket says a request cancelled after it was admitted "may +#: still be charged", so a cancel landing here is a detach, not a free stop. +IN_PROGRESS = "IN_PROGRESS" + +#: What the cancel route's ``202`` names itself (``spec/router-openapi.yaml``, +#: the ``RouterCancelStatus`` schema and the ``cancelRouterModelRequest`` +#: ``202`` description). Deliberately NOT exported and deliberately not part of +#: :data:`IN_QUEUE`/:data:`IN_PROGRESS`/:data:`COMPLETED`: it is a value of the +#: cancel body's own vocabulary, not a queue state, and a caller who compared +#: :attr:`QueueUpdate.status` against it would be comparing against something +#: the status route can never answer. It means the ask was accepted for a +#: request that had not yet reached a terminal state -- not that the run +#: stopped -- so this SDK confirms it with one status read; see +#: :func:`_reading_of_accepted_cancel`. +_CANCELLATION_REQUESTED = "CANCELLATION_REQUESTED" + +#: The code the SDK raises from the timeout teardown when the cancel route +#: answered its ``202`` and the confirming poll still found the request +#: :data:`IN_QUEUE`. The contract does not allow that ordering -- the route +#: writes the row terminal *before* it answers -- so it is a server that +#: violated its own write order, and the honest report is that the cancel did +#: not land. Reported rather than detached because the spec pins an +#: ``IN_QUEUE`` request as never dispatched and unchargeable, and a +#: :class:`DetachedRequest` asserts the opposite. +_CANCEL_NOT_APPLIED_CODE = "cancel_not_applied" + #: Failures a poll is retried through, matching ``models.run``'s tuple exactly. #: Being listed is not "retryable" — :meth:`~comfy_sdk.retry.RetryPolicy.should_retry` #: decides that. What it buys is that anything else propagates untouched. @@ -92,28 +134,36 @@ _CANCEL_TIMEOUT = 10.0 #: The status a cancel is refused with when the REQUEST'S OWN STATE is what -#: refuses it, rather than the caller, the credential or the server. The queue -#: honours a cancel only while the request is still waiting to be dispatched: a -#: run already in flight is served and billed, and one that has already finished -#: cannot be un-finished. Both refusals are the plain HTTP reading of ``409`` -#: -- the target's current state conflicts with the operation -- and that -#: reading is what this keys on, deliberately rather than on the refusal's -#: prose, which no contract pins and which differs between the two. +#: refuses it, rather than the caller, the credential or the server. It is the +#: plain HTTP reading of ``409`` -- the target's current state conflicts with +#: the operation -- and that reading is what this keys on, deliberately rather +#: than on the refusal's prose, which no contract pins. #: -#: It is the FALLBACK reading, not the first one. Cancel-after-completion is -#: now typed -- it reaches this SDK as +#: **The route emits exactly one such ``409``, and it is typed.** +#: ``ALREADY_COMPLETED`` is the whole of it, and the vendored contract says so: +#: ``cancelRouterModelRequest`` in ``spec/router-openapi.yaml`` declares a +#: ``202`` and that one ``409``, and describes the ``409`` as "the request had +#: already reached a terminal state, so there was nothing to cancel". The +#: cancellation write is guarded on the request being NON-terminal, so a +#: request in either live state is CANCELLED rather than refused and the +#: ``409`` is reached only when that guard matched nothing. **There is no +#: in-flight ``409``.** It reaches this SDK as #: :class:`~comfy_sdk.router_exceptions.AlreadyCompleted` under the -#: :class:`~comfy_sdk.router_exceptions.CancelRefused` base -- and -#: :func:`_refused_on_state` asks that question first, because branching on the -#: class is strictly better than branching on a status. This status is what -#: catches the refusal the contract has NOT named yet: the in-flight one, which -#: today carries no bucket on the header or in the body and so arrives as a -#: bare :class:`~comfy_sdk.exceptions.ComfyError`. It is matched only when the -#: response names no bucket at all, so a ``409`` the contract DOES name -- -#: ``invalid_input``, ``concurrency_limit_exceeded`` -- is not swept in with -#: it. When the in-flight refusal is given a bucket of its own, add the -#: subclass to :data:`~comfy_sdk.router_exceptions.CANCEL_REFUSALS` and this -#: status clause can go. +#: :class:`~comfy_sdk.router_exceptions.CancelRefused` base, and +#: :func:`_refused_on_state` asks that typed question FIRST, because branching +#: on the class is strictly better than branching on a status. +#: +#: The status clause below it therefore has no producer today. It is kept, and +#: it is kept NARROW: it matches a ``409`` naming no bucket at all, so a +#: ``409`` the contract DOES name -- ``invalid_input``, +#: ``concurrency_limit_exceeded`` -- is not swept in with it. What it buys is +#: failing closed on a deployment that refuses on state without a bucket, in +#: either direction: such a refusal reads as "the run did not stop" and is +#: confirmed by a poll, rather than escaping as an unexplained cancel failure. +#: If the route is ever given a second state refusal with a bucket of its own, +#: add the subclass to +#: :data:`~comfy_sdk.router_exceptions.CANCEL_REFUSALS` and this status clause +#: can go. _CANCEL_REFUSED_STATUS = 409 #: The code :mod:`comfy_low.errors` synthesises for a ``409`` that identified @@ -407,12 +457,17 @@ def __init__( self.request_id = request_id #: The canonical ``{provider}/{model}`` id the request was submitted to. self.model = model - #: ``True`` only when the queue ACCEPTED the cleanup cancel, which it - #: does only while the request is still waiting to be dispatched. The - #: run is then gone and there is nothing left to collect or be billed - #: for. ``False`` means the cancel failed and the run's fate is - #: unknown -- read :attr:`cancel_error`, and treat the request as - #: possibly still running. + #: ``True`` only when the request's own row confirmed the stop -- + #: ``COMPLETED`` carrying the ``cancelled`` bucket -- or when a + #: body-less accepted cancel left nothing to confirm against. The run + #: is then gone and there is nothing left to collect. It is not by + #: itself a statement about the charge: a request cancelled while + #: still ``IN_QUEUE`` was never dispatched and cannot be charged, + #: while one cancelled after admission may still be. + #: + #: ``False`` means the cancel failed, or was accepted and did not + #: apply, so the run's fate is unknown -- read :attr:`cancel_error`, + #: and treat the request as possibly still running. self.cancelled = cancelled #: The failure the cleanup cancel raised, when it raised one. Also on #: ``__cause__``. ``None`` when the cancel was accepted. It is surfaced @@ -452,9 +507,11 @@ def _refused_on_state(exc: BaseException) -> bool: is :class:`~comfy_sdk.router_exceptions.AlreadyCompleted`. Branching on the class is the honest test, and it costs nothing to ask first. * failing that, a :data:`_CANCEL_REFUSED_STATUS` that names **no bucket at - all** -- the shape the in-flight refusal has today, which carries no - ``error_type`` on the header or in the body and so reaches this SDK as a - bare :class:`~comfy_sdk.exceptions.ComfyError`. + all**, reaching this SDK as a bare + :class:`~comfy_sdk.exceptions.ComfyError`. The shipped route emits no + such refusal -- ``ALREADY_COMPLETED`` is its only ``409`` and it is typed + -- so this clause is a fallback with no producer today, kept because it + fails closed: see :data:`_CANCEL_REFUSED_STATUS`. The second clause **fails closed**, which is the whole of the narrowing: a ``409`` that DOES name a bucket is something the contract already has a @@ -517,11 +574,16 @@ class _CancelReading(enum.Enum): #: The run finished on its own. There IS a result, and it is already paid #: for, so it is collected rather than discarded. FINISHED = "finished" - #: The route took the message but the run has NOT stopped -- a ``202`` - #: naming a ``CANCELING`` status, or a ``200`` on a request that won the - #: race into flight. Confirmed against the server, then reported as a - #: detach. + #: The route took the message but the run has NOT stopped -- a ``200`` or + #: ``202`` echoing a live queue state, which is not a shape the shipped + #: route produces. Confirmed against the server, then reported as a detach. UNSTOPPED = "unstopped" + #: The contract's own answer: a ``202`` naming + #: :data:`_CANCELLATION_REQUESTED`. The route has ALREADY written the row + #: terminal by the time it answers this, but the body says nothing about + #: which terminal state, so it is confirmed by one status read -- and + #: unlike :attr:`UNSTOPPED` that read is expected to find a stop. + ACCEPTED = "accepted" def _reading_of_accepted_cancel(accepted: QueueUpdate) -> _CancelReading: @@ -535,30 +597,94 @@ def _reading_of_accepted_cancel(accepted: QueueUpdate) -> _CancelReading: assert ``cancelled=True`` from the bare fact of a 2xx, which is the one reading the response does not support. - Three readings, from the update the cancel answered with: + Four readings, from the update the cancel answered with: - * no status at all (the body-less ``204`` an accepted cancel usually is) -- + * no status at all (the body-less ``204`` a legacy accepted cancel is) -- :attr:`~_CancelReading.STOPPED`. There is nothing to read, and an accepted cancel on an undispatched request is what that shape means; - spending another round trip to re-confirm the ordinary case would charge - every timeout for the rare one. + spending another round trip to re-confirm it would charge every timeout + for a shape the current contract does not even emit. + * :data:`_CANCELLATION_REQUESTED` -- :attr:`~_CancelReading.ACCEPTED`, the + contract's own ``202``. The route only answers it once its guarded + ``UPDATE`` has already written the row terminal, so the ask DID land; + what the body does not say is which terminal state, and the ``cancelled`` + bucket's meaning makes that a billing question. So it is confirmed by one + status read rather than asserted. Compared case-sensitively against the + raw string, because :func:`_update_from` neither trims nor folds it and a + value that differs in case is a different value. * a terminal :data:`COMPLETED` carrying an ``error_type`` -- - :attr:`~_CancelReading.STOPPED` too. Terminal, and the bucket is how this + :attr:`~_CancelReading.STOPPED`. Terminal, and the bucket is how this queue expresses a stop (it has no ``CANCELED`` status of its own; see :data:`COMPLETED`). * a terminal :data:`COMPLETED` carrying NO bucket -- :attr:`~_CancelReading.FINISHED`. It ran to completion and was billed, so there is a real result behind it. * anything else, which is a non-terminal status on a live request -- - :attr:`~_CancelReading.UNSTOPPED`. + :attr:`~_CancelReading.UNSTOPPED`. No shipped deployment answers this; + it is what a deployment that echoed a queue state would land on. """ if not accepted.status: return _CancelReading.STOPPED + if accepted.status == _CANCELLATION_REQUESTED: + return _CancelReading.ACCEPTED if not accepted.is_completed: return _CancelReading.UNSTOPPED return _CancelReading.STOPPED if accepted.error_type is not None else _CancelReading.FINISHED +def _is_cancelled_completion(update: QueueUpdate) -> bool: + """Whether a terminal update is a request the cancel route STOPPED. + + The queue has no ``CANCELED`` status of its own (see :data:`COMPLETED`), so + a stop is :data:`COMPLETED` plus the contract's ``cancelled`` bucket -- + which is precisely what the cancel route writes before it answers its + ``202`` (``error_type=cancelled`` on the guarded ``UPDATE``). + + It matters that this is told apart from every other terminal + ``error_type``: a run that failed on its own should raise its typed error, + which is what :meth:`RequestHandle._collect_or_detach` does. A stop the SDK + itself asked for is not a failure to report -- it is the answer to the ask + -- so it becomes the cancelled ending instead. + + One function rather than the expression inline in both handles, because + ``tests/test_sync_async_parity.py`` walks the sync and async surfaces + together and a predicate spelled twice is a predicate that drifts once. + """ + return update.is_completed and update.error_type == Cancelled.error_type + + +def _cancel_not_applied(request_id: str, status: str) -> ComfyError: + """The failure a cancel that did not take effect reports as. + + Raised from :meth:`RequestHandle._detach_report` whenever the confirming + poll finds the request still :data:`IN_QUEUE`, however that poll was + reached. On the contract path it means a server answered its ``202`` out + of its own documented order -- the cancellation write is guarded on the + request being non-terminal and lands BEFORE the ``202``; on the + bucket-less-``409`` fallback it means a refusal that left the request + exactly where it was. Either way the ask did not take effect, which is not + something to paper over with a detach: the contract pins an ``IN_QUEUE`` + request as unbilled, and a detach claims the opposite. + + The message names the request id and quotes the status VERBATIM rather + than describing it, because the enum is closed and a value outside it is + the most useful thing the message could carry -- and ``subscribe`` folds + this onto :attr:`SubscribeTimeout.cancel_error`, where it is the caller's + only account of what the cancel did. + + :attr:`~comfy_sdk.exceptions.ComfyError.request_id` is deliberately left + unset: that field is the server-minted ``X-Comfy-Request-Id`` of one HTTP + call, which is not the same identifier as the queued request's own id, and + filling it with the latter would make the two indistinguishable. The queued + id reaches the caller on :attr:`SubscribeTimeout.request_id`, and in this + message. + """ + return ComfyError( + f"the cancel for request {request_id} did not take effect: the request is still {status!r}", + code=_CANCEL_NOT_APPLIED_CODE, + ) + + def _subscribe_timed_out( handle: _RequestHandleBase, timed_out: BaseException, @@ -792,12 +918,16 @@ def cancel(self) -> QueueUpdate: gives an update whose ``status`` is ``""``; the authoritative state is the next :meth:`status`. - The queue honours a cancel only while the request is still waiting to - be dispatched. A run it has already started is served to the end and - **billed**, and asking to cancel it is *refused* rather than ignored — - which reaches you as an exception, not as an update. ``models.subscribe`` - reads that refusal as a detach; here it is the caller's to handle, since - an explicit ``cancel()`` is not a timeout cleaning up after itself. + The route takes a request in **either** live state, answering ``202`` + with :data:`_CANCELLATION_REQUESTED`; it is only the terminal one it + refuses, with a ``409`` that reaches you as + :class:`~comfy_sdk.router_exceptions.AlreadyCompleted` — an exception, + not an update. What "taken" does not settle is whether the generation + stopped: one already on the wire at a partner may complete anyway, and + a partner generation that completes is **billed** whether or not + anyone collected it. ``models.subscribe``'s timeout confirms the + difference with a poll; here that is the caller's to do, since an + explicit ``cancel()`` is not a timeout cleaning up after itself. """ with translating(): payload, headers = self._call( @@ -836,17 +966,17 @@ def _after_subscribe_timeout(self) -> DetachedRequest | QueueUpdate | None: outcomes and a caller who has just lost their wait needs to tell them apart: - * ``None`` — the run is gone. Either the queue ACCEPTED the cancel on - a request it had not dispatched yet, or the cancel's own body came - back terminal; nothing will be billed further and there is nothing - to collect, so ``subscribe`` raises, exactly as it always has. + * ``None`` — the run is gone. Either the cancel's own body came back + terminal, or the contract's ``202`` was confirmed by a poll that + found the row ``COMPLETED``/``cancelled``; there is nothing to + collect, so ``subscribe`` raises, exactly as it always has. * a :class:`DetachedRequest` — the run did NOT stop, and a confirming poll found it still going. Either the queue REFUSED the cancel on the request's own state (:func:`_refused_on_state`), or it took the - cancel but answered with a live status - (:attr:`~_CancelReading.UNSTOPPED`). It was already in flight, so it - is served and billed whatever the caller does; the honest report is - that the caller detached from a run that is still theirs to collect. + cancel and the poll still found a live status. It is in flight, so + it is served and billed whatever the caller does; the honest report + is that the caller detached from a run that is still theirs to + collect. * a :class:`QueueUpdate` — the request is ``COMPLETED``, found either by the confirming poll or on the cancel's own answer. It finished while the teardown was running, so there IS a result, and @@ -855,7 +985,11 @@ def _after_subscribe_timeout(self) -> DetachedRequest | QueueUpdate | None: A 2xx is deliberately NOT read as proof the run stopped; see :func:`_reading_of_accepted_cancel` for what the body has to say - before this reports a cancellation. + before this reports a cancellation. The contract's own ``202`` + (:attr:`~_CancelReading.ACCEPTED`) falls through to + :meth:`_detach_report` exactly as :attr:`~_CancelReading.UNSTOPPED` + does — one poll settles which of the three endings it is, because the + ``202`` body says the ask landed and nothing else. Any other cancel failure propagates. That is the narrowing this method exists for: the timeout path used to swallow every cancel failure @@ -878,28 +1012,54 @@ def _after_subscribe_timeout(self) -> DetachedRequest | QueueUpdate | None: return None if reading is _CancelReading.FINISHED: return accepted + # ACCEPTED and UNSTOPPED both fall through: neither says what the + # request's state IS, which is the one thing that decides the + # ending, so both are settled by the same confirming poll. return self._detach_report() - def _detach_report(self) -> DetachedRequest | QueueUpdate: - """What a cancel that did not stop the run left behind, confirmed. + def _detach_report(self) -> DetachedRequest | QueueUpdate | None: + """What one confirming poll found after a cancel that was not terminal. - Reached two ways, which establish the same thing: the queue REFUSED - the cancel on the request's state (:func:`_refused_on_state`), or it - accepted the message and answered with a live status - (:attr:`~_CancelReading.UNSTOPPED`). Either way the run did not stop. + Reached three ways: the route ACCEPTED the ask with the contract's + ``202`` (:attr:`~_CancelReading.ACCEPTED`), it REFUSED the cancel on + the request's state (:func:`_refused_on_state`), or it accepted the + message and echoed a live status + (:attr:`~_CancelReading.UNSTOPPED`). None of the three says what the + request's state actually is, so this reads it. One unretried, short-bounded poll: the caller's deadline has already - run out, so this is not the place to spend a retry budget. The poll is - what separates "still running" from "finished while we were tearing - down", which neither answer says on its own — and the authoritative - state of a request is always the next status read, never what a cancel - answered. + run out, so this is not the place to spend a retry budget. The + authoritative state of a request is always the next status read, never + what a cancel answered. + + Four answers, from that read: + + * terminal and carrying the ``cancelled`` bucket + (:func:`_is_cancelled_completion`) — ``None``, the cancelled ending. + This is what the contract's ``202`` normally resolves to, because the + route writes ``COMPLETED``/``cancelled`` before it answers. It is + deliberately NOT routed through :meth:`_collect_or_detach`, whose + ``error_type is not None`` branch re-raises the typed error: that is + right for a run that failed on its own and wrong for a stop this SDK + itself asked for. + * terminal for any other reason, or none at all — the update, which + ``subscribe`` collects (or raises the run's own typed failure from). + * still :data:`IN_QUEUE` — a ``ComfyError`` coded + :data:`_CANCEL_NOT_APPLIED_CODE`. The route's guard covers + ``IN_QUEUE``, so a request still sitting there after an accepted + cancel means the ask did not land, and the contract says such a + request "was never dispatched and cannot be charged" — which is the + one claim a :class:`DetachedRequest` must never make. ``subscribe``'s + caller turns it into ``SubscribeTimeout(cancelled=False, + cancel_error=...)``. + * any other live status — :data:`IN_PROGRESS`, or a value the closed + enum does not name — a detach, as before. A poll that fails does not undo what the cancel's answer established: - the request was NOT cancelled. So it still reports a detach, with - :attr:`DetachedRequest.status` left empty to say the state was not + the request was NOT confirmed stopped. So it still reports a detach, + with :attr:`DetachedRequest.status` left empty to say the state was not confirmed, rather than raising and stranding the caller without the - handle to the run they are now paying for. An empty ``status`` is + handle to the run they may now be paying for. An empty ``status`` is therefore the one value that leaves the billing claim UNVERIFIED: the poll may have failed on a rejected credential or a ``404``, and this does not tell those apart from a transient blip. @@ -908,9 +1068,15 @@ def _detach_report(self) -> DetachedRequest | QueueUpdate: update = self._status(budget=_CANCEL_TIMEOUT, policy=NO_RETRY) except _CANCEL_FAILURES: return self._detached("") + if _is_cancelled_completion(update): + return None if update.is_completed: return update - return self._detached(update.status) + if update.status != IN_QUEUE: + return self._detached(update.status) + # OUTSIDE the handler above, so a cancel that never landed is not + # chained onto whatever the poll happened to fail with first. + raise _cancel_not_applied(self._request_id, update.status) def _collect_or_detach( self, completion: QueueUpdate, *, budget: float | None = None @@ -1085,15 +1251,19 @@ async def _after_subscribe_timeout(self) -> AsyncDetachedRequest | QueueUpdate | return accepted return await self._detach_report() - async def _detach_report(self) -> AsyncDetachedRequest | QueueUpdate: - """Async :meth:`RequestHandle._detach_report`.""" + async def _detach_report(self) -> AsyncDetachedRequest | QueueUpdate | None: + """Async :meth:`RequestHandle._detach_report` — the same four answers.""" try: update = await self._status(budget=_CANCEL_TIMEOUT, policy=NO_RETRY) except _CANCEL_FAILURES: return self._detached("") + if _is_cancelled_completion(update): + return None if update.is_completed: return update - return self._detached(update.status) + if update.status != IN_QUEUE: + return self._detached(update.status) + raise _cancel_not_applied(self._request_id, update.status) async def _collect_or_detach( self, completion: QueueUpdate, *, budget: float | None = None @@ -1144,25 +1314,49 @@ class _DetachedBase: #: addressed by both. model: str #: The status the confirming poll reported for the request, verbatim — - #: ``"IN_PROGRESS"`` on the ordinary detach. ``""`` when that poll itself - #: failed, which says the state was not confirmed, never that the request - #: stopped: the refused cancel had already established that it did not. + #: :data:`IN_PROGRESS` on the ordinary detach. It is one of the contract's + #: **live** states, or a live value its closed enum does not name yet; + #: :data:`COMPLETED` never reaches here (a terminal poll is collected or + #: raised instead), and neither does :data:`IN_QUEUE`, which is the one + #: state the contract pins as never dispatched and unchargeable — a detach + #: asserts the opposite, so that combination is refused outright below. #: - #: An empty value is the one case where this report's "still running, and - #: still billing" reading is UNVERIFIED. The poll behind it is allowed to - #: fail for any reason and they are not told apart — a rejected credential - #: or a ``404`` saying the request is gone reads the same as a transient - #: blip. Re-read :meth:`RequestHandle.status` before acting on the charge. + #: ``""`` when the poll itself failed, which says the state was not + #: confirmed, never that the request stopped. That empty value is the one + #: case where this report's "still running, and still billing" reading is + #: UNVERIFIED: the poll is allowed to fail for any reason and they are not + #: told apart — a rejected credential or a ``404`` saying the request is + #: gone reads the same as a transient blip. Re-read + #: :meth:`RequestHandle.status` before acting on the charge. status: str + def __post_init__(self) -> None: + """Refuse the one status a detach report cannot honestly carry. + + A detach says "this run is in flight, and you are being billed for + it". :data:`IN_QUEUE` says the exact opposite — the contract's + ``cancelled`` meaning pins a request cancelled in that state as never + dispatched and unchargeable — so a report carrying it would be a + billing claim that contradicts the spec. Every producer in this module + already routes that case elsewhere; this is the invariant stated where + it cannot be bypassed by a new one. + """ + if self.status == IN_QUEUE: + raise ValueError( + f"a detach report cannot carry {IN_QUEUE!r}: a request in that state " + "was never dispatched and cannot be charged" + ) + @dataclass(frozen=True) class DetachedRequest(_DetachedBase): - """What ``models.subscribe`` RETURNS when its timeout could not cancel the run. + """What ``models.subscribe`` RETURNS when its timeout could not stop the run. - The queue honours a cancel only before it dispatches a request. Past that - point the run is served and **billed** whatever the caller does, so a - ``subscribe`` timeout is a *detach* and not a cancellation: the caller has + The cancel route takes a request in either live state, but taking it is not + the same as the run stopping: a partner generation already on the wire may + complete anyway, and one that completes is **billed** whether or not + anyone collected it. So when the confirming poll finds the request still + running, the timeout is a *detach* and not a cancellation: the caller has stopped waiting, and the generation carries on without them. It is returned rather than raised because nothing has gone wrong — the run @@ -1178,9 +1372,10 @@ class DetachedRequest(_DetachedBase): Branch on the class, never on a message: that is the whole point of the type. The three ways a ``subscribe`` timeout can end are this, a - :class:`SubscribeTimeout` with ``cancelled=True`` (the request was still - queued and really is gone), and an ordinary result (it completed while the - timeout was being torn down). + :class:`SubscribeTimeout` (``cancelled=True`` when the row came back + ``COMPLETED``/``cancelled`` — it really is gone; ``cancelled=False`` with + a :attr:`~SubscribeTimeout.cancel_error` when the cancel did not land), and + an ordinary result (it completed while the timeout was being torn down). """ #: The live request, ready to poll or collect. The same object @@ -1199,6 +1394,8 @@ class AsyncDetachedRequest(_DetachedBase): __all__ = [ "COMPLETED", + "IN_PROGRESS", + "IN_QUEUE", "AsyncDetachedRequest", "AsyncRequestHandle", "DetachedRequest", diff --git a/src/comfy_sdk/models.py b/src/comfy_sdk/models.py index 95a7b27..2c22b7b 100644 --- a/src/comfy_sdk/models.py +++ b/src/comfy_sdk/models.py @@ -697,35 +697,47 @@ def subscribe( no server-side meaning: the queue's own timeouts are the server's. When it runs out this makes a best-effort :meth:`~comfy_sdk.model_requests.RequestHandle.cancel`, and **what that - cancel is answered decides how the call ends.** The queue honours a - cancel only while a request is still waiting to be dispatched; a run it - has already started is served and **billed** whatever the caller does. - So there are three endings, told apart by type and never by a message: - - * **Cancelled** — the cancel was accepted, the request never ran, and - nothing will be billed for it. Raises + cancel is answered — and then what the request's own state says — + decides how the call ends.** The route takes a cancel in either live + state and answers ``202 CANCELLATION_REQUESTED``, which says the ask + landed and nothing more; a generation already on the wire may complete + anyway, and one that completes is **billed** whatever the caller does. + So one confirming status read settles it, and there are three endings, + told apart by type and never by a message: + + * **Cancelled** — the request's row came back ``COMPLETED`` carrying + the ``cancelled`` bucket. Raises :class:`~comfy_sdk.model_requests.SubscribeTimeout` (a - ``TimeoutError``) with ``cancelled=True``. - * **Detached** — the queue refused the cancel because the run was - already in flight. Nothing has gone wrong: the generation continues, - completes and is billed, and it stays collectable. **Returns** a + ``TimeoutError``) with ``cancelled=True``. A request cancelled while + still ``IN_QUEUE`` was never dispatched and cannot be charged; one + cancelled after admission may still be, and this ending does not + claim otherwise. + * **Detached** — the confirming poll found the run still going + (``IN_PROGRESS``, or a live status this SDK cannot place). Nothing + has gone wrong: the generation continues, completes and is billed, + and it stays collectable. **Returns** a :class:`~comfy_sdk.model_requests.DetachedRequest` carrying the ``request_id``, the model id and a live handle, rather than raising. - * **Completed during teardown** — the refusal turned out to be a - request that had just finished. The result exists and has been paid - for, so it is collected and returned like any other result. + * **Completed during teardown** — the run had just finished on its + own. The result exists and has been paid for, so it is collected and + returned like any other result. A cancel that fails for any *other* reason — a transport failure, a rejected credential, a ``500`` — is not benign and is not swallowed: the ``SubscribeTimeout`` is raised with ``cancelled=False`` and the cancel's own failure on ``.cancel_error`` and ``__cause__``, because a - cancel that never landed means the run may still be going. + cancel that never landed means the run may still be going. The same + goes for a cancel the route ACCEPTED whose row is nevertheless still + ``IN_QUEUE`` — an ordering the route's own write order forbids: the + ``cancel_error`` is coded ``cancel_not_applied``, because an + ``IN_QUEUE`` request cannot have been charged and a detach would claim + it was. **The teardown itself is not inside the bound.** ``timeout`` bounds the *wait*; deciding how it ended costs up to three further round trips beyond it, each bounded at ``_CANCEL_TIMEOUT`` (10s): the - cleanup cancel, the confirming status read the refusal needs, and — - on the completed-during-teardown ending only — the result fetch. So a + cleanup cancel, the status read that confirms what it did, and — on + the completed-during-teardown ending only — the result fetch. So a ``subscribe(timeout=N)`` that ends in a detach can return up to ~30s after ``N`` in the worst case. Overrunning is the deliberate trade: the alternative is telling a caller their run was cancelled without diff --git a/tests/conftest.py b/tests/conftest.py index 3f8ab06..99dd188 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -237,8 +237,10 @@ class ServerState: "seed": 7, } ) - # Status code for the cancel response; 204 exercises the empty-body path. - queue_cancel_status: int = 200 + # Status code for the cancel response. 202 is the contract's own answer + # (`spec/router-openapi.yaml`, `cancelRouterModelRequest`); 204 exercises + # the legacy empty-body path a deployment may still answer with. + queue_cancel_status: int = 202 # Cancels that answer a transient failure (status, code) before one is # accepted — for proving the cleanup cancel after a timeout does not ride # the client's retry policy. @@ -261,19 +263,34 @@ class ServerState: # set it to a `401` or a `500` and the refusal must surface rather than read # as a detach. queue_cancel_refusal: tuple[int, dict[str, Any]] | None = None - # The bucket a cancelled request's completion carries. `None` models the - # cancel that LOST the race: the answer is still terminal, but the run - # finished on its own, so there is a real result behind it rather than a - # stop. - queue_cancel_error_type: str | None = "client_disconnected" - # The queue `status` an ACCEPTED cancel answers with. "COMPLETED" is the - # ordinary one -- this queue has no `CANCELED` status and expresses a stop - # as COMPLETED plus a bucket. A live value like "IN_PROGRESS" models the - # 2xx a request that won the race into flight is answered with: the route - # took the message, but the run did not stop. - queue_cancel_accept_status: str = "COMPLETED" - # Set by a cancel; makes every later status poll report the cancellation. + # The bucket the CANCEL RESPONSE's own body carries. `None` is the contract + # default: the `202` answers `{request_id, status}` and nothing else, and + # the bucket lives on the status route. Set it to model a deployment whose + # cancel answers terminal directly -- "client_disconnected" for a stop, + # `None` alongside `queue_cancel_accept_status="COMPLETED"` for the cancel + # that LOST the race to a run that finished on its own. + queue_cancel_error_type: str | None = None + # The `status` an ACCEPTED cancel answers with. "CANCELLATION_REQUESTED" is + # the contract's own value: the ask was accepted for a request that had not + # reached a terminal state, which the route writes BEFORE it answers. + # "COMPLETED" models a deployment answering terminal directly, and a live + # value models one that echoes a queue state. + queue_cancel_accept_status: str = "CANCELLATION_REQUESTED" + # Set by a cancel that the server treated as landing -- which the real route + # does before it answers its `202`, so both the contract's accept status and + # a directly-terminal one set it. Makes every later status poll report the + # cancelled row. queue_canceled: bool = False + # The bucket the status route reports for a cancelled row. "cancelled" is + # what the route writes (`error_type=cancelled` on its guarded UPDATE); a + # knob so a test can model a deployment that writes a different one. + queue_cancelled_error_type: str | None = "cancelled" + # When False, an ACCEPTED cancel does NOT mark the row cancelled -- a + # server that answered its `202` without the guarded UPDATE having landed, + # which the contract's write order forbids. The only way to drive the SDK's + # "accepted but did not apply" reading, since a conforming stub cannot + # produce it. + queue_cancel_applies: bool = True # --- counters the tests assert on --- upload_count: int = 0 @@ -740,8 +757,8 @@ def _serve_queue_status(self, request_id: str) -> None: return if state.queue_canceled: body["status"] = "COMPLETED" - if state.queue_cancel_error_type is not None: - body["error_type"] = state.queue_cancel_error_type + if state.queue_cancelled_error_type is not None: + body["error_type"] = state.queue_cancelled_error_type body["detail"] = "the request was cancelled" self._json(200, body, headers=headers) return @@ -795,12 +812,16 @@ def _put_queue_cancel(self, request_id: str) -> None: status, refusal = state.queue_cancel_refusal self._json(status, refusal) return - # Only a cancel whose own answer is TERMINAL actually stopped the - # request. One answering a live status took the message and left - # the run going, so later polls must still see it running -- - # otherwise the stub decides the very thing the SDK is under test - # for reading correctly. - state.queue_canceled = state.queue_cancel_accept_status == "COMPLETED" + # The real route writes the row terminal (`error_type=cancelled`) + # under a guard covering both live states BEFORE it answers its + # `202`, so the contract's accept status marks the row cancelled + # exactly as a directly-terminal answer does. A cancel echoing a + # LIVE queue state took the message and left the run going, so + # later polls must still see it running -- otherwise the stub + # decides the very thing the SDK is under test for reading. + state.queue_canceled = state.queue_cancel_applies and ( + state.queue_cancel_accept_status in ("CANCELLATION_REQUESTED", "COMPLETED") + ) if state.queue_cancel_status == 204: self.send_response(204) self.send_header("Content-Length", "0") diff --git a/tests/test_models_queue.py b/tests/test_models_queue.py index 0b015c4..6d9f2d5 100644 --- a/tests/test_models_queue.py +++ b/tests/test_models_queue.py @@ -43,16 +43,21 @@ from comfy_sdk.exceptions import ComfyError from comfy_sdk.model_requests import ( COMPLETED, + IN_PROGRESS, + IN_QUEUE, AsyncDetachedRequest, AsyncRequestHandle, DetachedRequest, RequestHandle, SubscribeTimeout, + _CancelReading, + _reading_of_accepted_cancel, _refused_on_state, ) from comfy_sdk.retry import NO_RETRY from comfy_sdk.router_exceptions import ( AlreadyCompleted, + Cancelled, ContentPolicyViolation, NotEnabled, RouterError, @@ -463,13 +468,25 @@ def test_the_key_rides_out_on_a_failure(server) -> None: def test_cancel_asks_the_server_and_reports_what_it_said(server, fast_poll) -> None: + """The contract's own answer reaches the caller verbatim, bucket and all. + + `202 CANCELLATION_REQUESTED` says the ask was accepted and carries no + `error_type`, and `cancel()` reports exactly that rather than editorialising + it into a stop — its docstring tells callers to read the status afterwards. + """ + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + with _client() as client: handle = client.models.submit(MODEL, ARGS) update = handle.cancel() assert server.state.queue_cancel_count == 1 assert update.request_id == server.state.queue_request_id - assert update.error_type == server.state.queue_cancel_error_type + assert update.status == "CANCELLATION_REQUESTED" + assert update.error_type is None + assert update.is_completed is False def test_a_cancel_answered_with_no_body_still_identifies_the_request(server, fast_poll) -> None: @@ -847,16 +864,18 @@ async def _pause(delay: float) -> None: # --- the timeout's three endings: cancelled, detached, completed ------------ # -# The queue honours a cancel only before it dispatches a request: a run already -# in flight is served and billed whatever the caller does, and the cancel -# `subscribe` fires on its own timeout is refused. These pin the three ways -# that timeout can now end, and — just as load-bearing — that everything which -# is NOT that benign refusal still raises. +# The cancel route takes a request in either live state, so "accepted" is not +# "stopped": a partner generation already on the wire may complete anyway, and +# one that completes is billed. These pin the ways that timeout can now end, +# and — just as load-bearing — that everything which is NOT a benign refusal +# still raises. -#: The refusal a dispatched run's cancel is answered with. A `409` carrying -#: prose and no error bucket, which is the shape the queue's state refusals -#: have today. -IN_FLIGHT_REFUSAL = (409, {"detail": "in-flight tasks cannot be cancelled"}) +#: A HYPOTHETICAL refusal, modelling a `409` that carries prose and no error +#: bucket. **No shipped deployment emits it**: the route's only `409` is +#: `ALREADY_COMPLETED`, which arrives typed. It is kept because the SDK's +#: bucket-less-`409` clause exists to fail closed on a deployment that grew +#: one, and a fallback with no test is a fallback that rots. +UNSHIPPED_BUCKETLESS_REFUSAL = (409, {"detail": "in-flight tasks cannot be cancelled"}) def _no_sleep(monkeypatch) -> None: @@ -867,7 +886,11 @@ def test_a_refused_in_flight_cancel_detaches_instead_of_raising(server, monkeypa """Acceptance: the timeout hands the run back rather than erroring out.""" _no_sleep(monkeypatch) server.state.queue_polls_to_complete = 10_000 - server.state.queue_cancel_refusal = IN_FLIGHT_REFUSAL + # Dispatched: the confirming poll has to find a LIVE status for a detach to + # be the honest report. `IN_QUEUE` would mean the request was never + # dispatched and cannot be charged, which is not a detach at all. + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_refusal = UNSHIPPED_BUCKETLESS_REFUSAL with _client() as client: outcome = client.models.subscribe(MODEL, ARGS, timeout=0.0) @@ -876,7 +899,7 @@ def test_a_refused_in_flight_cancel_detaches_instead_of_raising(server, monkeypa assert outcome.request_id == server.state.queue_request_id assert outcome.model == MODEL # Confirmed against the server rather than inferred from the refusal. - assert outcome.status == server.state.queue_pending_status + assert outcome.status == IN_PROGRESS assert isinstance(outcome.handle, RequestHandle) assert outcome.handle.request_id == outcome.request_id # The cancel was attempted exactly as before; only its refusal is read @@ -894,7 +917,8 @@ async def _sleep(_delay: float) -> None: monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) server.state.queue_polls_to_complete = 10_000 - server.state.queue_cancel_refusal = IN_FLIGHT_REFUSAL + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_refusal = UNSHIPPED_BUCKETLESS_REFUSAL async with AsyncComfy(api_key="comfyui-test-key") as client: outcome = await client.models.subscribe(MODEL, ARGS, timeout=0.0) @@ -902,14 +926,28 @@ async def _sleep(_delay: float) -> None: assert isinstance(outcome, AsyncDetachedRequest) assert outcome.request_id == server.state.queue_request_id assert outcome.model == MODEL + assert outcome.status == IN_PROGRESS assert isinstance(outcome.handle, AsyncRequestHandle) assert server.state.queue_cancel_count == 1 def test_an_accepted_cancel_keeps_the_cancelled_semantics(server, monkeypatch) -> None: - """Acceptance: a request still in the queue really is cancelled, and raises.""" + """Acceptance, the contract path on a QUEUED request: `202` then `cancelled`. + + The wire shape the vendored contract pins: the route answers + `202 CANCELLATION_REQUESTED`, having already written the row `COMPLETED` + with `error_type=cancelled` under its `status IN (IN_QUEUE, IN_PROGRESS)` + guard. One confirming poll reads that row, and a stop the SDK itself asked + for is the cancelled ending — not the `Cancelled` router exception a run + that failed on its own would raise. + """ _no_sleep(monkeypatch) server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_QUEUE + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + server.state.queue_cancelled_error_type = "cancelled" with _client() as client: with pytest.raises(SubscribeTimeout) as excinfo: @@ -921,22 +959,94 @@ def test_an_accepted_cancel_keeps_the_cancelled_semantics(server, monkeypatch) - assert excinfo.value.request_id == server.state.queue_request_id assert excinfo.value.model == MODEL assert server.state.queue_cancel_count == 1 - # No confirming poll: an accepted cancel needs no second opinion. - assert server.state.queue_status_count == 1 + # `subscribe`'s own single poll, plus the one that confirms the `202`. + assert server.state.queue_status_count == 2 + # Nothing is collected: the row is terminal, and it is terminal because we + # stopped it. + assert server.state.queue_result_count == 0 async def test_async_accepted_cancel_keeps_the_cancelled_semantics(server, monkeypatch) -> None: + """The awaitable half of the contract path on a queued request.""" + async def _sleep(_delay: float) -> None: return None monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_QUEUE + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + server.state.queue_cancelled_error_type = "cancelled" async with AsyncComfy(api_key="comfyui-test-key") as client: with pytest.raises(SubscribeTimeout) as excinfo: await client.models.subscribe(MODEL, ARGS, timeout=0.0) assert excinfo.value.cancelled is True + assert excinfo.value.cancel_error is None + assert server.state.queue_cancel_count == 1 + assert server.state.queue_status_count == 2 + assert server.state.queue_result_count == 0 + + +def test_the_contract_path_on_an_in_flight_request_is_the_same_cancelled_ending( + server, monkeypatch +) -> None: + """Acceptance, the contract path on a DISPATCHED request: the same shape. + + The route's guard covers `IN_PROGRESS` as well as `IN_QUEUE`, so a + mid-flight cancel gets the identical wire shape and the identical ending. + Per the spec's `cancelled` meaning such a cancel **may still be charged** — + a partner generation that completes is charged whether or not anyone + collected it. The SDK does not adjudicate that: it reports what the row + says, and the row says the request was withdrawn. + """ + _no_sleep(monkeypatch) + server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + + with _client() as client: + with pytest.raises(SubscribeTimeout) as excinfo: + client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert excinfo.value.cancelled is True + assert excinfo.value.cancel_error is None + assert server.state.queue_cancel_count == 1 + assert server.state.queue_status_count == 2 + assert server.state.queue_result_count == 0 + + +async def test_async_contract_path_on_an_in_flight_request_is_the_same_cancelled_ending( + server, monkeypatch +) -> None: + """The awaitable half: a mid-flight cancel reports what the row says. + + Same caveat as the sync twin — the spec's `cancelled` meaning allows a + mid-flight cancel to be charged, and this ending is not a claim that it was + not. + """ + + async def _sleep(_delay: float) -> None: + return None + + monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) + server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + + async with AsyncComfy(api_key="comfyui-test-key") as client: + with pytest.raises(SubscribeTimeout) as excinfo: + await client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert excinfo.value.cancelled is True + assert server.state.queue_result_count == 0 def test_cancelled_and_detached_are_told_apart_without_reading_a_message( @@ -949,6 +1059,8 @@ def test_cancelled_and_detached_are_told_apart_without_reading_a_message( """ _no_sleep(monkeypatch) server.state.queue_polls_to_complete = 10_000 + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" with _client() as client: with pytest.raises(SubscribeTimeout) as excinfo: @@ -956,7 +1068,9 @@ def test_cancelled_and_detached_are_told_apart_without_reading_a_message( cancelled = excinfo.value server.state.queue_canceled = False - server.state.queue_cancel_refusal = IN_FLIGHT_REFUSAL + # A dispatched run whose cancel is refused: the detaching half. + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_refusal = UNSHIPPED_BUCKETLESS_REFUSAL detached = client.models.subscribe(MODEL, ARGS, timeout=0.0) assert cancelled.cancelled is True @@ -1022,7 +1136,8 @@ def test_a_detached_request_is_collectable_by_request_id(server, monkeypatch, fa # One poll inside the subscribe, one confirming poll after the refusal, and # the run completes on the one after that. server.state.queue_polls_to_complete = 3 - server.state.queue_cancel_refusal = IN_FLIGHT_REFUSAL + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_refusal = UNSHIPPED_BUCKETLESS_REFUSAL with _client() as client: detached = client.models.subscribe(MODEL, ARGS, timeout=0.0) @@ -1058,7 +1173,7 @@ def test_a_confirming_poll_that_fails_still_reports_a_detach(server, monkeypatch """ _no_sleep(monkeypatch) server.state.queue_polls_to_complete = 10_000 - server.state.queue_cancel_refusal = IN_FLIGHT_REFUSAL + server.state.queue_cancel_refusal = UNSHIPPED_BUCKETLESS_REFUSAL with _client(retry=NO_RETRY) as client: handle = client.models.submit(MODEL, ARGS) @@ -1119,7 +1234,8 @@ async def _sleep(_delay: float) -> None: monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) server.state.queue_polls_to_complete = 3 - server.state.queue_cancel_refusal = IN_FLIGHT_REFUSAL + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_refusal = UNSHIPPED_BUCKETLESS_REFUSAL async with AsyncComfy(api_key="comfyui-test-key") as client: detached = await client.models.subscribe(MODEL, ARGS, timeout=0.0) @@ -1132,19 +1248,21 @@ async def _sleep(_delay: float) -> None: # --- a 2xx cancel is not proof the run stopped ------------------------------ -def test_a_cancel_accepted_on_a_live_status_detaches_rather_than_claiming_a_stop( +def test_a_cancel_accepted_on_an_unknown_live_status_detaches_rather_than_claiming_a_stop( server, monkeypatch ) -> None: - """A 2xx says the route took the message, not that the work stopped. - - The binding accepts ``200``, ``202`` and ``204`` alike, so a request that - won the race into flight — answered ``202``/``CANCELING``, or ``200`` on a - run already dispatched — used to come back as ``cancelled=True``: the one - reading the response does not support, and the exact claim (nothing ran, - nothing is billed) that is most expensive to get wrong. + """A 2xx echoing a status the SDK cannot place is a detach, not a stop. + + ``CANCELING`` is **not** a value this route sends — its cancel body says + ``CANCELLATION_REQUESTED`` and ``RouterQueueStatus`` is a closed enum that + does not contain it. It stands in for any live status a future or + non-conforming deployment might echo, and the point is that such a 2xx + never comes back as ``cancelled=True``: that is the one claim (nothing ran, + nothing is billed) most expensive to get wrong. """ _no_sleep(monkeypatch) server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_PROGRESS server.state.queue_cancel_status = 202 server.state.queue_cancel_accept_status = "CANCELING" server.state.queue_cancel_error_type = None @@ -1153,18 +1271,21 @@ def test_a_cancel_accepted_on_a_live_status_detaches_rather_than_claiming_a_stop outcome = client.models.subscribe(MODEL, ARGS, timeout=0.0) assert isinstance(outcome, DetachedRequest) - assert outcome.status == server.state.queue_pending_status + assert outcome.status == IN_PROGRESS assert server.state.queue_cancel_count == 1 -async def test_async_cancel_accepted_on_a_live_status_detaches(server, monkeypatch) -> None: - """The async half reads a live accept the same way.""" +async def test_async_cancel_accepted_on_an_unknown_live_status_detaches( + server, monkeypatch +) -> None: + """The async half reads an unplaceable live accept the same way.""" async def _sleep(_delay: float) -> None: return None monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_PROGRESS server.state.queue_cancel_status = 202 server.state.queue_cancel_accept_status = "CANCELING" server.state.queue_cancel_error_type = None @@ -1173,7 +1294,7 @@ async def _sleep(_delay: float) -> None: outcome = await client.models.subscribe(MODEL, ARGS, timeout=0.0) assert isinstance(outcome, AsyncDetachedRequest) - assert outcome.status == server.state.queue_pending_status + assert outcome.status == IN_PROGRESS def test_a_cancel_that_lost_the_race_returns_the_result_it_was_billed_for( @@ -1188,6 +1309,10 @@ def test_a_cancel_that_lost_the_race_returns_the_result_it_was_billed_for( """ _no_sleep(monkeypatch) server.state.queue_polls_to_complete = 10_000 + # A deployment that answers the cancel terminally rather than with the + # contract's `202`: COMPLETED, and carrying no bucket at all. + server.state.queue_cancel_status = 200 + server.state.queue_cancel_accept_status = COMPLETED server.state.queue_cancel_error_type = None with _client() as client: @@ -1213,6 +1338,252 @@ def test_a_body_less_accepted_cancel_still_reports_a_cancellation(server, monkey assert server.state.queue_cancel_count == 1 +def test_a_202_whose_row_is_still_queued_reports_that_the_cancel_did_not_apply( + server, monkeypatch +) -> None: + """A server that answered `202` out of its documented write order. + + The route's guarded UPDATE covers `IN_QUEUE` and runs BEFORE the `202` is + written, so a request still sitting there after an accepted cancel means + the ask never landed. It is NOT a detach: the spec pins an `IN_QUEUE` + request as never dispatched and unchargeable, and a `DetachedRequest` + claims the opposite. It surfaces as the cancel's own failure. + """ + _no_sleep(monkeypatch) + server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_QUEUE + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + server.state.queue_cancel_applies = False + + with _client(retry=NO_RETRY) as client: + with pytest.raises(SubscribeTimeout) as excinfo: + client.models.subscribe(MODEL, ARGS, timeout=0.0) + + exc = excinfo.value + assert not isinstance(exc, DetachedRequest) + assert exc.cancelled is False + assert isinstance(exc.cancel_error, ComfyError) + assert exc.cancel_error.code == "cancel_not_applied" + # The ids reach the caller through the message as well as the fields: this + # is what a support request quotes. + assert server.state.queue_request_id in str(exc) + assert IN_QUEUE in str(exc.cancel_error) + # Raised OUTSIDE the poll's own `except`, so nothing reads as "during + # handling of" a cancel failure: the only context is the timeout that + # started the teardown. + assert isinstance(exc.cancel_error.__context__, TimeoutError) + assert not isinstance(exc.cancel_error.__context__, ComfyError) + + +async def test_async_202_whose_row_is_still_queued_reports_that_the_cancel_did_not_apply( + server, monkeypatch +) -> None: + """The awaitable half of the violated-write-order reading.""" + + async def _sleep(_delay: float) -> None: + return None + + monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) + server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_QUEUE + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + server.state.queue_cancel_applies = False + + async with AsyncComfy(api_key="comfyui-test-key", retry=NO_RETRY) as client: + with pytest.raises(SubscribeTimeout) as excinfo: + await client.models.subscribe(MODEL, ARGS, timeout=0.0) + + exc = excinfo.value + assert not isinstance(exc, AsyncDetachedRequest) + assert exc.cancelled is False + assert isinstance(exc.cancel_error, ComfyError) + assert exc.cancel_error.code == "cancel_not_applied" + assert server.state.queue_request_id in str(exc) + + +def test_a_202_whose_row_is_still_in_progress_detaches(server, monkeypatch) -> None: + """The same violated write order on a dispatched run IS a detach. + + `IN_PROGRESS` carries no unbilled guarantee — the spec's `cancelled` + meaning says a request cancelled after admission may still be charged — so + the honest report is the one that hands the run back. + """ + _no_sleep(monkeypatch) + server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + server.state.queue_cancel_applies = False + + with _client(retry=NO_RETRY) as client: + outcome = client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert isinstance(outcome, DetachedRequest) + assert outcome.status == IN_PROGRESS + assert server.state.queue_cancel_count == 1 + + +async def test_async_202_whose_row_is_still_in_progress_detaches(server, monkeypatch) -> None: + """The awaitable half: a dispatched run the cancel did not stop.""" + + async def _sleep(_delay: float) -> None: + return None + + monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) + server.state.queue_polls_to_complete = 10_000 + server.state.queue_pending_status = IN_PROGRESS + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + server.state.queue_cancel_applies = False + + async with AsyncComfy(api_key="comfyui-test-key", retry=NO_RETRY) as client: + outcome = await client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert isinstance(outcome, AsyncDetachedRequest) + assert outcome.status == IN_PROGRESS + + +def test_a_second_cancel_racing_the_first_still_reports_the_cancelled_ending( + server, monkeypatch +) -> None: + """`409 ALREADY_COMPLETED` over a row an earlier cancel already stopped. + + The refusal says only "there was nothing left to cancel"; the confirming + poll is what says why. Finding `COMPLETED`/`cancelled` there, the answer is + the cancelled ending — **not** the `Cancelled` router exception, which is + what `_collect_or_detach` would have raised for a run that failed on its + own. + """ + _no_sleep(monkeypatch) + # `subscribe`'s own poll is the last pending one; the confirming poll after + # the refusal finds the row the first cancel left behind. + server.state.queue_polls_to_complete = 1 + server.state.queue_error_type = "cancelled" + server.state.queue_cancel_refusal = (409, {"status": "ALREADY_COMPLETED"}) + + with _client(retry=NO_RETRY) as client: + with pytest.raises(SubscribeTimeout) as excinfo: + client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert not isinstance(excinfo.value, Cancelled) + assert excinfo.value.cancelled is True + assert excinfo.value.cancel_error is None + assert server.state.queue_result_count == 0 + + +async def test_async_second_cancel_racing_the_first_still_reports_the_cancelled_ending( + server, monkeypatch +) -> None: + """The awaitable half of the racing-cancel reading.""" + + async def _sleep(_delay: float) -> None: + return None + + monkeypatch.setattr("comfy_sdk.model_requests.asyncio.sleep", _sleep) + server.state.queue_polls_to_complete = 1 + server.state.queue_error_type = "cancelled" + server.state.queue_cancel_refusal = (409, {"status": "ALREADY_COMPLETED"}) + + async with AsyncComfy(api_key="comfyui-test-key", retry=NO_RETRY) as client: + with pytest.raises(SubscribeTimeout) as excinfo: + await client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert excinfo.value.cancelled is True + assert server.state.queue_result_count == 0 + + +def test_a_completion_that_failed_on_its_own_still_raises_its_typed_error( + server, monkeypatch +) -> None: + """The other side of the cancelled-completion reading, and the narrow one. + + Only the `cancelled` bucket becomes the cancelled ending. Every other + terminal bucket is the RUN's outcome rather than an answer to the SDK's + ask, so it still raises the typed router exception through + `_collect_or_detach` — which is what keeps this narrowing from swallowing a + real failure. + """ + _no_sleep(monkeypatch) + server.state.queue_polls_to_complete = 1 + server.state.queue_error_type = "content_policy_violation" + server.state.queue_cancel_refusal = (409, {"status": "ALREADY_COMPLETED"}) + + with _client(retry=NO_RETRY) as client: + with pytest.raises(ContentPolicyViolation): + client.models.subscribe(MODEL, ARGS, timeout=0.0) + + +@pytest.mark.parametrize( + ("status", "error_type", "expected"), + [ + ("", None, _CancelReading.STOPPED), + ("CANCELLATION_REQUESTED", None, _CancelReading.ACCEPTED), + ("cancellation_requested", None, _CancelReading.UNSTOPPED), + (COMPLETED, "cancelled", _CancelReading.STOPPED), + (COMPLETED, None, _CancelReading.FINISHED), + (IN_PROGRESS, None, _CancelReading.UNSTOPPED), + ], + ids=["body-less", "contract-202", "wrong-case", "terminal-bucket", "terminal-bare", "live"], +) +def test_the_reading_of_an_accepted_cancel_body(status, error_type, expected) -> None: + """Each 2xx body shape, and which reading it earns. + + The wrong-case row is the point of comparing raw: `RouterCancelStatus` is a + closed enum of upper-case values, so a lower-case echo is a DIFFERENT value + and must not be read as the contract's accept. + """ + raw: dict[str, Any] = {"request_id": "req-1", "status": status} + if error_type is not None: + raw["error_type"] = error_type + update = QueueUpdate( + request_id="req-1", status=status, error_type=error_type, queue_position=None, raw=raw + ) + assert _reading_of_accepted_cancel(update) is expected + + +def test_no_detach_report_can_carry_the_unbilled_status(server) -> None: + """The invariant, stated where no future producer can route around it. + + A detach asserts "in flight, and billing". `IN_QUEUE` asserts the opposite + — never dispatched, cannot be charged — so the two cannot be combined, and + the report refuses rather than leaving the contradiction to a reader. + """ + with _client() as client: + handle = client.models.submit(MODEL, ARGS) + with pytest.raises(ValueError, match=IN_QUEUE): + handle._detached(IN_QUEUE) + # Every other live status, and the unconfirmed empty one, are fine. + assert handle._detached(IN_PROGRESS).status == IN_PROGRESS + assert handle._detached("").status == "" + + +def test_the_spike_reproduction_reports_a_cancellation(server, monkeypatch) -> None: + """The exact stub configuration the investigation reproduced against. + + Before this reading it raised `Cancelled` ("the model refused the + request") — the run's own typed failure, reported for a stop the SDK + itself asked for. + """ + _no_sleep(monkeypatch) + server.state.queue_polls_to_complete = 1 + server.state.queue_error_type = "cancelled" + server.state.queue_cancel_status = 202 + server.state.queue_cancel_accept_status = "CANCELLATION_REQUESTED" + server.state.queue_cancel_error_type = None + + with _client(retry=NO_RETRY) as client: + with pytest.raises(SubscribeTimeout) as excinfo: + client.models.subscribe(MODEL, ARGS, timeout=0.0) + + assert excinfo.value.cancelled is True + + # --- which 409s are the state refusal ---------------------------------------