test: SSE conformance leg for the gateway e2e suite (+ graceful 501 degradation)

src/comfy_sdk/jobs.py

@@ -16,6 +16,7 @@
 
 import httpx
 
+from comfy_low.errors import ApiError
 from comfy_low.models import Job as LowJob
 from comfy_low.models import Output as LowOutput
 from comfy_low.transport import AsyncComfyLow, ComfyLow
@@ -94,6 +95,10 @@ def cancel(self) -> Job:
     def events(self) -> Iterator[Event]:
         """Typed live event iterator. Auto-reconnects with no replay; falls back
         to polling to detect terminal status if the stream ends early.
+
+        A surface without SSE (501 from the events endpoint — contract-legal)
+        ends the iteration silently: streaming is an enhancement over the
+        poll-authoritative ``wait``/``result``, never a requirement.
         """
         events_url = self._model.urls.events or self._model.id
         while True:
@@ -108,6 +113,10 @@ def events(self) -> Iterator[Event]:
                         yield ev
                         return
                     yield ev
+            except ApiError as exc:
+                if exc.http_status == 501:
+                    return  # surface has no SSE — poll paths remain authoritative
+                raise
             except (httpx.HTTPError, httpx.StreamError):
                 pass  # connection dropped mid-stream — reconnect below
             if terminal_seen:
@@ -200,6 +209,10 @@ async def events(self) -> AsyncIterator[Event]:
                         yield ev
                         return
                     yield ev
+            except ApiError as exc:
+                if exc.http_status == 501:
+                    return  # surface has no SSE — poll paths remain authoritative
+                raise
             except (httpx.HTTPError, httpx.StreamError):
                 pass
             if terminal_seen:

tests/conftest.py

@@ -44,6 +44,8 @@ class ServerState:
     # (a "zombie": no terminal, no close) for `stall_seconds`.
     sse_mode: str = "normal"
     stall_seconds: float = 2.0
+    # GET /jobs/{id}/events answers 501 not_implemented — a surface without SSE.
+    events_not_implemented: bool = False
     # If set, GET /assets/{id}/content responds 302 to this URL instead of
     # serving bytes directly (simulates a signed-URL redirect to another host).
     redirect_content_to: str | None = None
@@ -241,6 +243,9 @@ def _serve_job(self, job_id: str) -> None:
 
         def _serve_events(self, job_id: str) -> None:
             state.events_connect_count += 1
+            if state.events_not_implemented:
+                self._err(501, "not_implemented", "SSE is not supported on this surface")
+                return
             self.send_response(200)
             self.send_header("Content-Type", "text/event-stream")
             self.end_headers()

tests/integration/test_gateway_e2e.py

@@ -24,18 +24,23 @@
 import json
 import os
 import struct
+import threading
+import time
 import zlib
 from pathlib import Path
 
 import pytest
 
-from comfy_sdk import Comfy
+from comfy_sdk import Comfy, OutputReady, StatusChange
+from comfy_sdk.events import Event
 
 BASE_URL = os.environ.get("COMFY_BASE_URL")
 API_KEY = os.environ.get("COMFY_API_KEY")
 INPUT_NAME = "sdk_e2e_input.png"
 WORKFLOW_FILE = os.environ.get("COMFY_WORKFLOW_FILE")
 JOB_TIMEOUT_S = 600  # cold start on a scale-to-zero deployment takes minutes
+STREAM_CAP_S = 120  # hard wall-clock cap on stream consumption (never hang CI)
+FIRST_FRAME_S = 30  # generous bound on connect-to-first-snapshot (~0.4s live)
 
 pytestmark = pytest.mark.skipif(
     not (BASE_URL and API_KEY),
@@ -125,6 +130,61 @@ def test_image_edit_by_name(client: Comfy, input_asset, tmp_path) -> None:
         assert (width, height) == (512, 512)
 
 
+def test_stream_delivers_lifecycle(client: Comfy, input_asset) -> None:
+    """SSE conformance: ``job.events()`` delivers the v1 lifecycle live.
+
+    The gateway's v1 stream emits ``status`` (snapshot on connect, then on
+    change) and ``output`` frames. Asserted contract: a StatusChange snapshot
+    arrives promptly after connect, an OutputReady arrives before the terminal
+    frame, iteration ends on its own at a terminal StatusChange, and the
+    poll-authoritative state agrees. Duplicate frames are legal (snapshot
+    semantics), so counts are never asserted — only presence and order.
+
+    Runs after the plain submit-and-poll test, so the deployment is warm and
+    only the connect-to-first-frame gap (served by the gateway itself, not the
+    worker) is asserted tight.
+    """
+    graph, _ = _edit_workflow(INPUT_NAME)
+    job = client.submit(client.workflows.from_json(graph))
+
+    arrivals: list[tuple[float, Event]] = []
+    errors: list[BaseException] = []
+    t0 = time.monotonic()
+
+    def consume() -> None:
+        try:
+            for ev in job.events():
+                arrivals.append((time.monotonic() - t0, ev))
+        except BaseException as exc:  # surfaced in the main thread below
+            errors.append(exc)
+
+    # Daemon-thread watchdog: a wedged stream fails the test instead of
+    # hanging CI, and the abandoned thread cannot block interpreter exit.
+    consumer = threading.Thread(target=consume, daemon=True)
+    consumer.start()
+    consumer.join(timeout=STREAM_CAP_S)
+    kinds = [type(e).__name__ for _, e in arrivals]
+    assert not consumer.is_alive(), (
+        f"events() did not end on its own within {STREAM_CAP_S}s; frames so far: {kinds}"
+    )
+    if errors:
+        raise errors[0]
+
+    events = [e for _, e in arrivals]
+    status_arrivals = [(gap, e) for gap, e in arrivals if isinstance(e, StatusChange)]
+    assert status_arrivals, f"no StatusChange frames: {kinds}"
+    first_gap = status_arrivals[0][0]
+    assert first_gap < FIRST_FRAME_S, f"snapshot status took {first_gap:.1f}s to arrive"
+
+    assert isinstance(events[-1], StatusChange), f"stream did not end on a status frame: {kinds}"
+    assert events[-1].status == "succeeded", f"terminal status {events[-1].status}: {kinds}"
+    assert any(isinstance(e, OutputReady) for e in events[:-1]), (
+        f"no OutputReady before the terminal frame: {kinds}"
+    )
+
+    assert job.refresh().status == "succeeded"
+
+
 @pytest.mark.xfail(
     reason="gateway does not yet resolve core/ASSET references in the graph; "
     "it resolves string filename references only",

tests/test_events.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from comfy_sdk import Comfy, OutputReady, Progress, StatusChange
+from comfy_sdk import AsyncComfy, Comfy, OutputReady, Progress, StatusChange
 
 
 def _wf(client: Comfy):
@@ -49,6 +49,47 @@ def test_sse_reconnect_with_no_replay(server) -> None:
     assert len(progresses) == 2  # one per connection, not a replayed backlog
 
 
+def test_events_polls_to_terminal_when_stream_ends_without_terminal(server) -> None:
+    # The stream closes cleanly (no error) after one progress frame, without a
+    # terminal status. The documented backstop: poll the authoritative state —
+    # it already reports succeeded, so events() ends on a synthetic terminal
+    # StatusChange instead of reconnecting.
+    server.state.sse_mode = "reconnect"  # 1st connection: one progress frame, clean close
+    server.state.polls_to_succeed = 1
+    with Comfy(server.base_url) as client:
+        job = client.submit(_wf(client))
+        seen = list(job.events())
+
+    assert server.state.events_connect_count == 1  # backstop polled; no reconnect
+    assert isinstance(seen[-1], StatusChange)
+    assert seen[-1].status == "succeeded"
+
+
+def test_events_end_silently_when_events_endpoint_not_implemented(server) -> None:
+    """Graceful degradation on a surface without SSE: a 501 from the events
+    endpoint (contract-legal) ends the iteration with no events, and the
+    poll-authoritative ``wait`` stays fully functional. Before this behavior
+    was introduced, ``events()`` raised the protocol-level ``ApiError``
+    (code ``not_implemented``, http_status 501)."""
+    server.state.events_not_implemented = True
+    with Comfy(server.base_url) as client:
+        job = client.submit(_wf(client))
+        assert list(job.events()) == []
+        assert job.wait().status == "succeeded"
+
+    assert server.state.events_connect_count == 1  # no reconnect loop on 501
+
+
+async def test_async_events_end_silently_when_events_endpoint_not_implemented(server) -> None:
+    server.state.events_not_implemented = True
+    async with AsyncComfy(server.base_url) as client:
+        job = await client.submit(_wf(client))
+        assert [e async for e in job.events()] == []
+        assert (await job.wait()).status == "succeeded"
+
+    assert server.state.events_connect_count == 1
+
+
 def test_preview_event_decodes_base64(server) -> None:
     from base64 import b64encode