@@ -67,6 +67,14 @@ class AgentJetJob:
                 whose ``num_repeat`` episodes do *not* all share the same reward. Tasks with uniform
                 reward (e.g. all 0 or all 1) produce zero advantage under GRPO and are skipped —
                 useful when the dataset contains many too-easy or too-hard prompts.
+              - "rollout_until_any_client_agree_sync_weight": defer the stop decision to the swarm
+                clients themselves. Stops as soon as **any** active swarm client invokes
+                ``SwarmClient.agree_sync_weight()``. A client is "active" once it has successfully
+                ``end_episode``'d at least one rewarded (non-abort) episode since the last weight
+                sync, and falls off the active list after 10 minutes of no chat-completion or
+                ``begin_episode`` activity.
+              - "rollout_until_all_clients_agree_sync_weight": like the above, but stops only when
+                **every** active swarm client has agreed (and there is at least one active client).
         max_env_worker: an estimation about how many episodes will be running in parallel (all swarm clients combined).
         backbone: Training backbone framework (e.g., 'verl').
         max_prompt_length: Maximum token length for input prompts (token length before the first llm-generated token, default 3000).
 
@@ -345,6 +345,12 @@ ajet:
   # "rollout_until_finish_enough_non_dummy_tasks":
   #   AgentJet will identify the **task_id** of each episode, and stop when it has collected [>= ajet.data.train_batch_size] unique & FINISHED & NON-DUMMY **task_id**.
   #     (Hint: a **task_id** is considered "NON-DUMMY" at least one of **episodes** of **task_id** has **different** reward value.)
+  # "rollout_until_any_client_agree_sync_weight":
+  #   AgentJet defers the stop decision to swarm clients: stop as soon as ANY active swarm client has called `SwarmClient.agree_sync_weight()`.
+  #     (Hint: a swarm client becomes "active" once it has successfully `end_episode`'d a rewarded (non-abort) episode since the last weight sync,
+  #      and falls off the active list if it does no chat-completion / begin_episode for 10 minutes.)
+  # "rollout_until_all_clients_agree_sync_weight":
+  #   Like the above, but stop only when EVERY active swarm client has agreed (and there is at least one active client).
   swarm_mode_sample_collection_method: "rollout_until_finish_enough_tasks"
   swarm_mode_sample_collection_max_cached_episodes: 9999
 
 
@@ -53,6 +53,12 @@ ajet:
   # "rollout_until_finish_enough_non_dummy_tasks":
   #   AgentJet will identify the **task_id** of each episode, and stop when it has collected [>= ajet.data.train_batch_size] unique & FINISHED & NON-DUMMY **task_id**.
   #     (Hint: a **task_id** is considered "NON-DUMMY" at least one of **episodes** of **task_id** has **different** reward value.)
+  # "rollout_until_any_client_agree_sync_weight":
+  #   AgentJet defers the stop decision to swarm clients: stop as soon as ANY active swarm client has called `SwarmClient.agree_sync_weight()`.
+  #     (Hint: a swarm client becomes "active" once it has successfully `end_episode`'d a rewarded (non-abort) episode since the last weight sync,
+  #      and falls off the active list if it does no chat-completion / begin_episode for 10 minutes.)
+  # "rollout_until_all_clients_agree_sync_weight":
+  #   Like the above, but stop only when EVERY active swarm client has agreed (and there is at least one active client).
   swarm_mode_sample_collection_method: "rollout_until_finish_enough_tasks"
 
   data:
 
@@ -26,8 +26,9 @@
 from ajet.context_tracker.single_agent_tracking import SingleAgentContextTracker
 from ajet.tuner_lib.experimental.interchange_utils import (
     http_change_engine_status,
-    http_update_rollout_pool_information,
+    http_update_rollout_pool_information_and_fetch_instruction,
     CurrentBatchRolloutPoolInformation,
+    SwarmClientInstruction,
 )
 
 
@@ -292,6 +293,14 @@ def rollout_swarm(  # noqa: C901
         completed_task_id_map_ct: Dict[str, List[SingleAgentContextTracker]] = IterationSafeDict()
         executor_lock = threading.Lock()
 
+        accept_client_control = ("client" in self.config.ajet.swarm_mode_sample_collection_method)
+        if accept_client_control:
+            # Latest active-client / agreed-sync-weight snapshot from the swarm server. Refreshed on every pool-information update;
+            # consumed by the `rollout_until_*_agree_sync_weight` stop conditions.
+            latest_swarm_client_instructions: Dict[str, SwarmClientInstruction | None] = {"swarm_clients": None}
+        else:
+            latest_swarm_client_instructions = None
+
         # count tasks to see whether we have reach the finish line for next weight update
         def count_tasks(completed_task_id_map_ct):
             total_completed_episodes = 0
@@ -340,6 +349,20 @@ def enough_finished_task_stop_condition(completed_task_id_map_ct) -> bool:
                 completed_task_id_map_ct.clear()
             return (total_completed_tasks >= n_batch_task)
 
+        def any_client_agree_sync_weight_stop_condition(completed_task_id_map_ct) -> bool:
+            # ajet.swarm_mode_sample_collection_method == "rollout_until_any_client_agree_sync_weight"
+            instr = latest_swarm_client_instructions["swarm_clients"]
+            if instr is None:
+                return False
+            return any(c.allowed_sync_weight for c in instr.active_clients)
+
+        def all_clients_agree_sync_weight_stop_condition(completed_task_id_map_ct) -> bool:
+            # ajet.swarm_mode_sample_collection_method == "rollout_until_all_clients_agree_sync_weight"
+            instr = latest_swarm_client_instructions["swarm_clients"]
+            if instr is None or not instr.active_clients:
+                return False
+            return all(c.allowed_sync_weight for c in instr.active_clients)
+
         def enough_non_dummy_task_stop_condition(completed_task_id_map_ct) -> bool:
             # ajet.swarm_mode_sample_collection_method == "rollout_until_finish_enough_non_dummy_tasks"
             counts = count_tasks(completed_task_id_map_ct)
@@ -372,6 +395,10 @@ def enough_non_dummy_task_stop_condition(completed_task_id_map_ct) -> bool:
             stop_condition = enough_finished_task_stop_condition
         elif self.config.ajet.swarm_mode_sample_collection_method == "rollout_until_finish_enough_non_dummy_tasks":
             stop_condition = enough_non_dummy_task_stop_condition
+        elif self.config.ajet.swarm_mode_sample_collection_method == "rollout_until_any_client_agree_sync_weight":
+            stop_condition = any_client_agree_sync_weight_stop_condition
+        elif self.config.ajet.swarm_mode_sample_collection_method == "rollout_until_all_clients_agree_sync_weight":
+            stop_condition = all_clients_agree_sync_weight_stop_condition
         else:
             logger.error(f"Invalid swarm_mode_sample_collection_method: {self.config.ajet.swarm_mode_sample_collection_method}, fallback to default method: rollout_until_finish_enough_tasks")
             stop_condition = enough_finished_task_stop_condition
@@ -442,9 +469,16 @@ def update_rollout_result_array_preview(observation_window, completed_task_id_ma
             buffer += f"Total completed tasks: {counts['total_completed_tasks']} (target {n_batch_task})\n"
             buffer += f"Total completed non-dummy tasks: {counts['total_completed_non_dummy_tasks']} (target {n_batch_task})\n"
             buffer += f"Current stop condition: {self.config.ajet.swarm_mode_sample_collection_method}\n"
+            if accept_client_control:
+                sc_inst = latest_swarm_client_instructions["swarm_clients"]
+                if sc_inst is not None:
+                    n_active = len(sc_inst.active_clients)
+                    n_agreed = sum(1 for c in sc_inst.active_clients if c.allowed_sync_weight)
+                    buffer += f"Active clients: {n_active} (agreed: {n_agreed})\n"
             observation_window["info"][-1] = buffer
 
-            # Update rollout pool information via API
+            # Update rollout pool information via API and pull the latest
+            # active-client / agreed-sync-weight instruction from the server.
             pool_info = CurrentBatchRolloutPoolInformation(
                 sample_collection_method=self.config.ajet.swarm_mode_sample_collection_method,
                 completed_episodes=counts['total_completed_episodes'],
@@ -457,7 +491,10 @@ def update_rollout_result_array_preview(observation_window, completed_task_id_ma
                 completed_tasks_details=completed_tasks_details,
                 completed_tasks_rewards=completed_tasks_rewards,
             )
-            http_update_rollout_pool_information(self.config, pool_info)
+            if accept_client_control:
+                instruction = http_update_rollout_pool_information_and_fetch_instruction(self.config, pool_info)
+                if instruction is not None:
+                    latest_swarm_client_instructions["swarm_clients"] = instruction
             return
 
         update_rollout_result_array_preview(observation_window, completed_task_id_map_ct)
 
@@ -118,6 +118,138 @@ class VerboseLogsResponse(BaseModel):
     entries: List[VerboseLogEntry] = []
 
 
+class AgreeSyncWeightRequest(BaseModel):
+    client_uuid: str
+
+
+class ActiveSwarmClient(BaseModel):
+    """Server-tracked record for one active swarm client.
+
+    A swarm client enters this list once it has successfully `end_episode`'d
+    a rewarded (non-abort) episode since the last weight sync, and falls off
+    after `CLIENT_ACTIVE_TIMEOUT` seconds of no chat-completion /
+    `begin_episode` activity. The whole list is reset whenever the engine
+    leaves ROLLING/ROLLING_POST.
+
+    Used both as the swarm server's authoritative storage (single
+    `shared_mem_dict["active_swarm_clients"]: List[ActiveSwarmClient]` key)
+    and as the wire payload sent back to the trainer in
+    `SwarmClientInstruction`. Add future per-client signals (e.g.
+    `requested_pause`, custom metrics) here -- pydantic field defaults keep
+    the wire format backwards-compatible across server/trainer versions.
+
+    Fields:
+        client_uuid: the client_uuid as generated in `SwarmClient.__init__`.
+        last_activity_at: unix timestamp of the most recent chat-completion,
+            `begin_episode`, or `end_episode` from this client. Used by the
+            server's expiry sweep.
+        allowed_sync_weight: True iff this client has explicitly agreed to
+            the next weight sync via `SwarmClient.agree_sync_weight()`.
+    """
+    client_uuid: str
+    last_activity_at: float
+    allowed_sync_weight: bool = False
+
+
+class SwarmClientInstruction(BaseModel):
+    """Server -> trainer instruction returned alongside pool-info updates.
+
+    Fields:
+        active_clients: list of `ActiveSwarmClient` records, one per
+            currently active client.
+
+    Example wire payload:
+        ```json
+        {
+            "active_clients": [
+                {"client_uuid": "9f3c-...-aaaa", "last_activity_at": 1746513900.1, "allowed_sync_weight": true},
+                {"client_uuid": "9f3c-...-bbbb", "last_activity_at": 1746513912.4, "allowed_sync_weight": false},
+                {"client_uuid": "9f3c-...-cccc", "last_activity_at": 1746513918.7, "allowed_sync_weight": false}
+            ]
+        }
+        ```
+
+    Example trainer-side use (matches DynamicRolloutManager.rollout_swarm):
+        ```python
+        # rollout_until_any_client_agree_sync_weight
+        if any(c.allowed_sync_weight for c in instr.active_clients):
+            stop()
+
+        # rollout_until_all_clients_agree_sync_weight
+        if instr.active_clients and all(
+            c.allowed_sync_weight for c in instr.active_clients
+        ):
+            stop()
+        ```
+
+        For the payload above:
+          - "any" stop-condition evaluates True (one client agreed).
+          - "all" stop-condition evaluates False (two of three not yet agreed).
+    """
+    active_clients: List[ActiveSwarmClient] = []
+
+
+# Active-client tracking timeout (seconds): a client falls off the active list
+# if it has done no chat-completion or begin_episode call within this window.
+CLIENT_ACTIVE_TIMEOUT = 10 * 60
+
+
+# --------------------------------------------------------------------
+# active-client tracking helpers
+# --------------------------------------------------------------------
+# All active-client state lives behind a single shared_mem_dict key:
+#   "active_swarm_clients": List[ActiveSwarmClient]
+# (See `ActiveSwarmClient` for field semantics and lifecycle.) The helpers
+# below are imported by the swarm server's FastAPI routes and by the
+# OAI-mode chat-completion handler.
+
+
+def _refresh_client_activity(client_uuid: str, shared_mem_dict) -> None:
+    """If client is in the active list, refresh its last-activity timestamp.
+
+    Called on chat-completion and begin_episode (claim_episode). Does NOT
+    add the client to the list -- only end_episode (success, non-abort) does.
+    """
+    if not client_uuid:
+        return
+    clients: List[ActiveSwarmClient] = list(shared_mem_dict.get("active_swarm_clients", []))
+    for i, c in enumerate(clients):
+        if c.client_uuid == client_uuid:
+            clients[i] = c.model_copy(update={"last_activity_at": time.time()})
+            shared_mem_dict["active_swarm_clients"] = clients
+            return
+
+
+def _register_active_client(client_uuid: str, shared_mem_dict) -> None:
+    """Add client to the active list (idempotent) and refresh its timestamp."""
+    if not client_uuid:
+        return
+    clients: List[ActiveSwarmClient] = list(shared_mem_dict.get("active_swarm_clients", []))
+    now = time.time()
+    for i, c in enumerate(clients):
+        if c.client_uuid == client_uuid:
+            clients[i] = c.model_copy(update={"last_activity_at": now})
+            shared_mem_dict["active_swarm_clients"] = clients
+            return
+    clients.append(ActiveSwarmClient(client_uuid=client_uuid, last_activity_at=now))
+    shared_mem_dict["active_swarm_clients"] = clients
+
+
+def _expire_inactive_clients(shared_mem_dict) -> None:
+    """Drop clients whose last activity is older than CLIENT_ACTIVE_TIMEOUT."""
+    now = time.time()
+    clients: List[ActiveSwarmClient] = list(shared_mem_dict.get("active_swarm_clients", []))
+    if not clients:
+        return
+    kept = [c for c in clients if (now - c.last_activity_at) <= CLIENT_ACTIVE_TIMEOUT]
+    if len(kept) != len(clients):
+        shared_mem_dict["active_swarm_clients"] = kept
+
+
+def _reset_active_client_tracking(shared_mem_dict) -> None:
+    """Clear all active-client state."""
+    shared_mem_dict["active_swarm_clients"] = []
+
 
 DEBUG = False
 # DEBUG = True
@@ -233,24 +365,34 @@ def http_push_verbose_log(message: str, tag: str = "", config=None):
             logger.warning(f"Failed to push verbose log: {e}")
 
 
-def http_update_rollout_pool_information(config, pool_info: CurrentBatchRolloutPoolInformation):
+def http_update_rollout_pool_information_and_fetch_instruction(
+    config, pool_info: CurrentBatchRolloutPoolInformation
+) -> SwarmClientInstruction | None:
     """
-    Update the rollout pool information on the interchange server.
+    Update the rollout pool information on the interchange server, and fetch
+    the swarm server's view of currently-active clients and their
+    agree-to-sync-weight state.
 
     Args:
         config: The configuration object
         pool_info: CurrentBatchRolloutPoolInformation object with rollout statistics
+
+    Returns:
+        SwarmClientInstruction with `active_clients` (List[ActiveSwarmClient]),
+        or None if the request failed.
     """
     try:
         resp = httpx.post(
-            f"{get_interchange_server_url(config)}/update_current_batch_rollout_pool_information",
+            f"{get_interchange_server_url(config)}/update_current_batch_rollout_pool_information_and_fetch_instruction",
             json=pool_info.model_dump(),
             timeout=5
         )
         resp.raise_for_status()
+        return SwarmClientInstruction.model_validate(resp.json())
     except Exception as e:
         if DEBUG:
             logger.warning(f"Failed to update rollout pool information: {e}")
+        return None
 
 
 def get_zmq_socket(config, episode_uuid: str, tag: str = ""):
 
@@ -301,6 +301,7 @@ async def chat_completions(request: Request, authorization: str = Header(None)):
         # enable_swarm_mode
         if enable_swarm_mode:
             from ajet.tuner_lib.experimental.swarm_server import ep_key
+            from ajet.tuner_lib.experimental.interchange_utils import _refresh_client_activity
             assert shared_mem_dict is not None
             assert shared_mem_dict_lock is not None
 
@@ -319,6 +320,9 @@ async def chat_completions(request: Request, authorization: str = Header(None)):
                 shared_mem_dict[ep_key(episode_uuid)] = es
             if es.episode_type == "eval":
                 preserve_sampling_params = True
+            # chat-completion counts as activity for keeping the owning client
+            # in the swarm-server active list (no-op if it's not active yet).
+            _refresh_client_activity(es.client_uuid, shared_mem_dict)
 
         # For streaming, we process as non-streaming but return in streaming format
         original_stream = new_req.stream