From 94f995bb05885baa84b30756d6a7bf723507bc24 Mon Sep 17 00:00:00 2001 From: Kosinkadink Date: Thu, 14 May 2026 19:50:47 -0700 Subject: [PATCH] feat(async-polling): advertise ASYNC_EXECUTE capability and derive poll cap from duration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final pieces of Track A (async-polling mode) on the client side. The proxy_node already routes between ``/execute`` and ``/execute_async`` on the descriptor's ``remote.execution.mode``; this commit makes the contract honest in two ways: - ``client.py``: advertise the new ``Capability.ASYNC_EXECUTE`` token so the server can gate ``async_polling`` descriptors on its presence. An older client that only knows ``/execute`` is now visible to the server and won't be silently routed to a long-poll provider whose lifecycle it can't drive. - ``protocol.py`` + ``proxy_node.py``: drop the ``DEFAULT_MAX_POLL_ATTEMPTS`` wire-default in favour of duration-based ``DEFAULT_SOFT_TIMEOUT_S`` / ``DEFAULT_HARD_TIMEOUT_S`` / ``DEFAULT_POLL_GRACE_S``. The poll-loop cap is now a *derived* value (``ceil((hard_timeout_s + grace) / poll_interval_s)``) so changing the poll cadence can never silently shrink or stretch the contract. Industry precedent: OAuth2 device flow §3.5, Kubernetes ``--timeout``, gRPC deadlines all express bounds as durations. Lockstep with the server side of Track A (comfy-rnp-server feat/async-polling-mode), which now enforces ``mode`` dispatch on ``/execute`` vs ``/execute_async`` and migrates existing providers (BFL/Bria/LTXV/Grok video/OpenAIChatNode/Stability UpscaleCreative → ``async_polling``; DALL-E/Ideogram/Recraft/sync Stability → ``request_response``). Co-authored-by: Amp Amp-Thread-ID: https://ampcode.com/threads/T-019e28eb-31ae-76a1-bb45-c0922f43bcc3 --- client.py | 7 +++++++ protocol.py | 13 ++++++++++++- proxy_node.py | 37 ++++++++++++++++++++++++++----------- 3 files changed, 45 insertions(+), 12 deletions(-) diff --git a/client.py b/client.py index fb3a5a6..84a9843 100644 --- a/client.py +++ b/client.py @@ -43,6 +43,13 @@ CLIENT_CAPABILITIES = [ Capability.IMAGE_PNG_BASE64, Capability.IMAGE_PNG_BASE64_BATCH, Capability.MASK_PNG_BASE64, + # Advertises that this client can drive the + # ``execute_async`` + poll-loop lifecycle (submit, poll, cancel, + # TASK_LOST replay). Servers gate ``async_polling``-mode + # descriptors on the presence of this token so an older client + # that only knows ``/execute`` doesn't silently submit a + # long-poll node into the sync endpoint. + Capability.ASYNC_EXECUTE, ] # Reported in the X-RNP-Client-Version header. diff --git a/protocol.py b/protocol.py index 66d263d..2ecc8d6 100644 --- a/protocol.py +++ b/protocol.py @@ -21,7 +21,18 @@ DEFAULT_POLL_INTERVAL_S = 1.5 # Values mirror comfy_api_nodes.util.client.poll_op_raw so RNP-routed # nodes behave identically to a direct upstream call when the descriptor # leaves a field unset. -DEFAULT_MAX_POLL_ATTEMPTS = 160 +# +# Lifetime budget design: the wire format declares ``soft_timeout_s`` +# (server-side stuck threshold) and ``hard_timeout_s`` (server-side +# hard kill) as durations. The client's poll-loop cap is a derived +# value — ``max_poll_attempts = ceil((hard_timeout_s + DEFAULT_POLL_GRACE_S) +# / poll_interval_s)`` — not a separate wire field, so changing the +# poll cadence can never silently shrink or stretch the contract. +# Industry precedent: OAuth2 device flow §3.5, Kubernetes +# ``--timeout``, gRPC deadlines all express bounds as durations. +DEFAULT_SOFT_TIMEOUT_S = 240.0 +DEFAULT_HARD_TIMEOUT_S = 300.0 +DEFAULT_POLL_GRACE_S = 30.0 DEFAULT_TIMEOUT_PER_POLL_S = 120.0 DEFAULT_MAX_RETRIES_PER_POLL = 10 DEFAULT_RETRY_DELAY_PER_POLL_S = 1.0 diff --git a/proxy_node.py b/proxy_node.py index 3e66606..2c0ef9a 100644 --- a/proxy_node.py +++ b/proxy_node.py @@ -22,6 +22,7 @@ from __future__ import annotations import asyncio import contextlib import logging +import math import time import uuid from typing import Any @@ -36,8 +37,9 @@ from . import client as rnp_client from . import serialization from .protocol import ( DEFAULT_CANCEL_TIMEOUT_S, - DEFAULT_MAX_POLL_ATTEMPTS, + DEFAULT_HARD_TIMEOUT_S, DEFAULT_MAX_RETRIES_PER_POLL, + DEFAULT_POLL_GRACE_S, DEFAULT_POLL_INTERVAL_S, DEFAULT_RETRY_BACKOFF_PER_POLL, DEFAULT_RETRY_DELAY_PER_POLL_S, @@ -146,19 +148,32 @@ def _extract_execution_policy(execution: dict[str, Any]) -> dict[str, Any]: node with an empty ``execution`` block behaves identically to a direct upstream call. The schema is the wire contract from ``comfy_rnp_protocol.constants``: ``poll_interval_s``, - ``max_poll_attempts``, ``timeout_per_poll_s``, ``cancel_timeout_s``, - ``max_retries_per_poll``, ``retry_delay_per_poll_s``, - ``retry_backoff_per_poll``, ``retry: {max, delay_s, backoff, - retry_on}``, ``idempotency``. + ``soft_timeout_s``, ``hard_timeout_s``, ``timeout_per_poll_s``, + ``cancel_timeout_s``, ``max_retries_per_poll``, + ``retry_delay_per_poll_s``, ``retry_backoff_per_poll``, + ``retry: {max, delay_s, backoff, retry_on}``, ``idempotency``. + + The client's poll-loop cap (``max_poll_attempts``) is **derived** + from ``hard_timeout_s + DEFAULT_POLL_GRACE_S`` and + ``poll_interval_s`` — there is no wire field for it. Coupling a + contract bound to a cadence value silently breaks every time you + tune the cadence; expressing the bound as a duration keeps the + contract invariant under cadence changes (see protocol.py + "Lifetime budget design" comment). """ retry_block = execution.get("retry") if isinstance(execution.get("retry"), dict) else {} + poll_interval_s = _coerce_pos( + execution.get("poll_interval_s"), DEFAULT_POLL_INTERVAL_S, + ) + hard_timeout_s = _coerce_pos( + execution.get("hard_timeout_s"), + _coerce_pos(execution.get("soft_timeout_s"), DEFAULT_HARD_TIMEOUT_S), + ) + poll_budget_s = hard_timeout_s + DEFAULT_POLL_GRACE_S + max_poll_attempts = max(1, math.ceil(poll_budget_s / poll_interval_s)) return { - "poll_interval_s": _coerce_pos( - execution.get("poll_interval_s"), DEFAULT_POLL_INTERVAL_S, - ), - "max_poll_attempts": _coerce_pos_int( - execution.get("max_poll_attempts"), DEFAULT_MAX_POLL_ATTEMPTS, - ), + "poll_interval_s": poll_interval_s, + "max_poll_attempts": max_poll_attempts, "timeout_per_poll_s": _coerce_pos( execution.get("timeout_per_poll_s"), DEFAULT_TIMEOUT_PER_POLL_S, ),