Add F3 payload layer: receiver-confirmed completion, receiver budgets, awaitable transfer facade

nvflare/client/cell/defs.py

@@ -56,6 +56,11 @@ class Topic:
     # Per task (every round)
     TASK_READY = "client_api.task_ready"
     TASK_ACCEPTED = "client_api.task_accepted"
+    # Sent when the trainer has materialized (or lazily bound) the task payload and hands
+    # it to user code. Ends the payload-materialization phase: the forward-path heartbeat
+    # exemption stops here and heartbeats govern the session lease while user code trains
+    # (design: Heartbeat and Liveness, Revision 2.2).
+    TASK_PAYLOAD_READY = "client_api.task_payload_ready"
     TASK_FAILED = "client_api.task_failed"
     RESULT_READY = "client_api.result_ready"
     RESULT_ACCEPTED = "client_api.result_accepted"

nvflare/fuel/f3/streaming/obj_downloader.py

@@ -28,6 +28,10 @@ def __init__(
         progress_cb=None,
         progress_interval: float = 30.0,
         outcome_cb=None,
+        receiver_ids=None,
+        min_receivers=None,
+        receiver_acquire_timeout=None,
+        receiver_idle_timeout=None,
         **cb_kwargs,
     ):
         """Constructor of ObjectDownloader.
@@ -67,6 +71,10 @@ def __init__(
             progress_cb=progress_cb,
             progress_interval=progress_interval,
             outcome_cb=outcome_cb,
+            receiver_ids=receiver_ids,
+            min_receivers=min_receivers,
+            receiver_acquire_timeout=receiver_acquire_timeout,
+            receiver_idle_timeout=receiver_idle_timeout,
             **cb_kwargs,
         )
 
@@ -87,6 +95,15 @@ def add_object(self, obj: Downloadable, ref_id=None) -> str:
         )
         return rid
 
+    def get_waiter(self):
+        """Returns the awaitable facade (TransferWaiter) over this transaction's terminal outcome.
+
+        waiter.wait(timeout) blocks until the aggregate TransferOutcome is recorded --
+        COMPLETED only when every expected receiver succeeded. This is the primitive
+        upper layers use to guarantee "send returns only when the payload is delivered".
+        """
+        return DownloadService.get_transfer_waiter(self.tx_id)
+
     def delete_transaction(self):
         """Delete the download transaction forcefully.
         You call this method only if you want to stop the downloading process prematurely.

nvflare/fuel/f3/streaming/transfer_outcome.py

@@ -31,20 +31,24 @@
 Outcome status values reuse the TransferProgressState terminal vocabulary
 (completed / failed / aborted) rather than introducing another status set.
 
-Known limits, resolved by later PRs of the same design (see
-docs/design/client_api_execution_modes_plan.md):
-- per-receiver statuses are producer-served (recorded when produce() returns EOF),
-  not receiver-confirmed — a receiver-side finalization failure after the last chunk
-  is not visible here until receiver-confirmed completion (plan PR F3-2) lands;
-- num_receivers is a count without receiver identity — expected-receiver identity
-  checks arrive with per-receiver budgets (plan PR F3-3);
+Semantics with the full payload layer (receiver-confirmed completion, per-receiver
+budgets, awaitable transfer facade):
+- per-receiver statuses are receiver-confirmed where the receiver supports it (a served
+  EOF is provisional until the receiver confirms its finalization succeeded); legacy
+  receivers remain producer-served — both skews and the runtime kill-switch degrade to
+  producer-served semantics (download_service.py, receiver-confirmed completion);
+- expected receiver identities and per-receiver acquire/idle budgets bound the outcome's
+  resolution time (a stalled or never-pulling receiver is finalized FAILED without
+  waiting the whole-transaction TTL); min_receivers surfaces the optional k-of-N quorum
+  via quorum_met while `completed` stays the strict all-receivers certificate;
 - the outcome covers the refs present at termination; adding objects to a
   transaction after receivers already finished the earlier ones is not supported.
 """
 
 import time
 from dataclasses import dataclass
-from typing import Dict, List, Optional
+from types import MappingProxyType
+from typing import Mapping, Optional, Sequence, Tuple
 
 from nvflare.fuel.f3.streaming.transfer_progress import TransferProgressState
 
@@ -91,13 +95,22 @@ class TransferOutcomeReason:
 class RefOutcome:
     """Per-object terminal outcome.
 
-    receiver_statuses maps receiver FQCN to a DownloadStatus value (success / failed)
-    as recorded by the producer side. Treat the dict as read-only: the same instance
-    is shared with every consumer of the outcome.
+    receiver_statuses maps receiver FQCN to a DownloadStatus value (success / failed) --
+    receiver-confirmed where the receiver supports it, producer-served for legacy peers
+    or when the receiver-confirm kill-switch is off, and budget-FAILED for receivers that
+    exhausted their acquire/idle budget. It is deep-frozen at construction (a
+    MappingProxyType over a private copy): the same instance is recorded in the
+    service outcome table and handed to outcome_cb consumers, so a callback must
+    not be able to mutate the recorded per-receiver truth. If outcomes ever cross
+    a process boundary, the serializer must materialize it (dict(...)).
     """
 
     ref_id: str
-    receiver_statuses: Dict[str, str]
+    receiver_statuses: Mapping[str, str]
+
+    def __post_init__(self):
+        # frozen=True only blocks attribute rebinding; freeze the container too
+        object.__setattr__(self, "receiver_statuses", MappingProxyType(dict(self.receiver_statuses)))
 
 
 @dataclass(frozen=True)
@@ -116,20 +129,68 @@ class TransferOutcome:
     reason: str  # a TransferOutcomeReason value
     done_status: str  # the raw TransactionDoneStatus value
     num_receivers: int
-    refs: List[RefOutcome]
+    refs: Tuple[RefOutcome, ...]
     timestamp: float
+    # optional k-of-N quorum declared by the workflow: informational for quorum_met;
+    # `completed` deliberately stays the strict all-receivers certificate
+    min_receivers: Optional[int] = None
+    # expected receiver identities declared by the workflow. When present, completion and
+    # quorum are judged against THESE identities: a status from an unexpected receiver can
+    # neither complete the transfer nor count toward the quorum.
+    receiver_ids: Optional[Tuple[str, ...]] = None
+
+    def __post_init__(self):
+        # frozen=True only blocks attribute rebinding; freeze the containers too so
+        # outcome_cb consumers cannot mutate the recorded outcome
+        object.__setattr__(self, "refs", tuple(self.refs))
+        if self.receiver_ids is not None:
+            object.__setattr__(self, "receiver_ids", tuple(self.receiver_ids))
 
     @property
     def completed(self) -> bool:
         return self.status == TransferProgressState.COMPLETED
 
+    @property
+    def quorum_met(self) -> bool:
+        """True if every ref reached at least min_receivers confirmed successes.
+
+        The k-of-N surface for fan-out workflows (min_responses-style policies): `completed`
+        stays the strict all-receivers certificate; a workflow that accepts partial fan-out
+        checks quorum_met (or thresholds refs itself). Falls back to `completed` when no
+        min_receivers was declared; fails closed with no refs.
+        """
+        if self.min_receivers is None:
+            return self.completed
+        if not self.refs:
+            return False
+        # a receiver counts toward the quorum only if it succeeded on EVERY ref: counting
+        # per-ref successes independently would let different receiver subsets satisfy each
+        # ref while no single receiver holds the complete payload
+        quorum_receivers = None
+        for r in self.refs:
+            ref_successes = {rcv for rcv, v in r.receiver_statuses.items() if v == DownloadStatus.SUCCESS}
+            quorum_receivers = ref_successes if quorum_receivers is None else quorum_receivers & ref_successes
+        if self.receiver_ids is not None:
+            # only declared receivers count toward the quorum
+            quorum_receivers &= set(self.receiver_ids)
+        return len(quorum_receivers) >= self.min_receivers
+
     def expired(self, now: float, ttl: float) -> bool:
         return now - self.timestamp > ttl
 
 
-def _all_receivers_succeeded(num_receivers: int, refs: List[RefOutcome]) -> bool:
+def _all_receivers_succeeded(num_receivers: int, refs: Sequence[RefOutcome], receiver_ids=None) -> bool:
     if num_receivers <= 0 or not refs:
         return False
+    if receiver_ids:
+        # identity mode: every DECLARED receiver must have succeeded on every ref.
+        # Statuses from unexpected receivers are ignored -- they can never certify
+        # a transfer that a declared receiver did not actually get.
+        for r in refs:
+            for expected in receiver_ids:
+                if r.receiver_statuses.get(expected) != DownloadStatus.SUCCESS:
+                    return False
+        return True
     for r in refs:
         if len(r.receiver_statuses) < num_receivers:
             return False
@@ -149,8 +210,10 @@ def compute_transfer_outcome(
     tx_id: str,
     done_status: str,
     num_receivers: int,
-    refs: List[RefOutcome],
+    refs: Sequence[RefOutcome],
     timestamp: Optional[float] = None,
+    min_receivers: Optional[int] = None,
+    receiver_ids=None,
 ) -> TransferOutcome:
     """Compute the aggregate terminal outcome for a terminated transaction.
 
@@ -175,7 +238,7 @@ def compute_transfer_outcome(
 
     if done_status not in _KNOWN_DONE_STATUSES:
         status, reason = TransferProgressState.FAILED, TransferOutcomeReason.UNKNOWN_DONE_STATUS
-    elif _all_receivers_succeeded(num_receivers, refs):
+    elif _all_receivers_succeeded(num_receivers, refs, receiver_ids):
         status, reason = TransferProgressState.COMPLETED, TransferOutcomeReason.ALL_RECEIVERS_SUCCEEDED
     elif done_status == TransactionDoneStatus.DELETED:
         status, reason = TransferProgressState.ABORTED, TransferOutcomeReason.DELETED
@@ -200,4 +263,6 @@ def compute_transfer_outcome(
         num_receivers=num_receivers,
         refs=refs,
         timestamp=timestamp,
+        min_receivers=min_receivers,
+        receiver_ids=receiver_ids,
     )

nvflare/fuel/utils/fobs/decomposers/via_downloader.py

@@ -380,7 +380,9 @@ def decompose(self, target: Any, manager: DatumManager = None) -> Any:
         self.logger.debug(f"ViaDownloader: created ref for target {target_id}: {item_id}")
         return {EncKey.TYPE: EncType.REF, EncKey.DATA: item_id}
 
-    def _create_downloader(self, fobs_ctx: dict, progress_cb=None, timeout_override=None, num_receivers_override=None):
+    def _create_downloader(
+        self, fobs_ctx: dict, progress_cb=None, timeout_override=None, num_receivers_override=None, receiver_ids=None
+    ):
         # Transaction lifecycle is managed solely by _monitor_tx() (download_service.py).
         # We deliberately do NOT subscribe to msg_root deletion here.  The msg_root is
         # deleted as soon as all blobs are delivered, but blob_cb fires asynchronously —
@@ -445,6 +447,9 @@ def _create_downloader(self, fobs_ctx: dict, progress_cb=None, timeout_override=
                 transaction_done_cb=on_complete_cb,
                 progress_cb=progress_cb,
                 progress_interval=RESULT_UPLOAD_PROGRESS_INTERVAL,
+                # expected receiver identities: enables the transaction's per-receiver
+                # acquire budget; None when any identity is unknown
+                receiver_ids=receiver_ids,
             )
 
         return downloader
@@ -603,11 +608,15 @@ def _finalize_download_tx(self, mgr: DatumManager):
                 progress_context = fobs_ctx.get(RESULT_UPLOAD_PROGRESS_CTX_KEY) or {}
                 timeout_override = progress_context.get(ResultUploadProgressContextKey.STREAMING_IDLE_TIMEOUT)
 
+            # forward receiver identities to the transaction only when every identity is
+            # actually known (a (None,) placeholder means unknown-single-receiver)
+            known_receiver_ids = receiver_ids if receiver_ids and all(r is not None for r in receiver_ids) else None
             downloader = self._create_downloader(
                 fobs_ctx,
                 progress_cb=progress_cb if progress_trackable else None,
                 timeout_override=timeout_override,
                 num_receivers_override=download_num_receivers,
+                receiver_ids=known_receiver_ids,
             )
             if downloader is None:
                 self.logger.warning("download transaction was not created because FOBS context has no cell")

tests/unit_test/client/cell/defs_test.py

@@ -33,6 +33,7 @@ def test_expected_topics_present(self):
             "HELLO_REJECTED",
             "TASK_READY",
             "TASK_ACCEPTED",
+            "TASK_PAYLOAD_READY",
             "TASK_FAILED",
             "RESULT_READY",
             "RESULT_ACCEPTED",
@@ -77,6 +78,7 @@ def test_topic_wire_values(self):
             "HELLO_REJECTED": "client_api.hello_rejected",
             "TASK_READY": "client_api.task_ready",
             "TASK_ACCEPTED": "client_api.task_accepted",
+            "TASK_PAYLOAD_READY": "client_api.task_payload_ready",
             "TASK_FAILED": "client_api.task_failed",
             "RESULT_READY": "client_api.result_ready",
             "RESULT_ACCEPTED": "client_api.result_accepted",

tests/unit_test/fuel/f3/streaming/conftest.py

@@ -0,0 +1,35 @@
+# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""DIRECTORY-WIDE fixtures: the autouse fixture below applies to EVERY test in
+tests/unit_test/fuel/f3/streaming/, including tests that never touch confirms.
+A new test here never exercises the real ConfigService read of the confirm
+kill-switch; patch _receiver_confirm_cached explicitly if you need the legacy
+(confirm-off) path or the real config read."""
+
+from unittest.mock import patch
+
+import pytest
+
+from nvflare.fuel.f3.streaming import download_service as ds_module
+
+
+@pytest.fixture(autouse=True)
+def confirm_switch_on():
+    """Pin the receiver-confirm kill-switch ON for all streaming tests.
+
+    Individual tests patch it OFF explicitly; legacy-path tests are unaffected (their
+    requests carry no capability key, so the switch is never consulted for them).
+    """
+    with patch.object(ds_module, "_receiver_confirm_cached", True):
+        yield