[pre-commit.ci] pre-commit autoupdate

.pre-commit-config.yaml

@@ -14,7 +14,7 @@ ci:
     - check-ruff-baselined
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v5.0.0  # Use the latest stable version
+    rev: v6.0.0  # Use the latest stable version
     hooks:
       - id: check-ast
       - id: debug-statements
@@ -114,7 +114,7 @@ repos:
         pass_filenames: false
         args: []
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.14.14
+    rev: v0.15.21
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

src/aiperf/api/api_service.py

@@ -151,11 +151,13 @@ async def _start_api_server(self) -> None:
 
         self.info(f"AIPerf FastAPI started at {self._base_url}/")
         self.info(
-            lambda: "  Routes: "
-            + " | ".join(
-                r.path
-                for r in self.app.routes
-                if hasattr(r, "methods") and r.path not in ("/openapi.json",)
+            lambda: (
+                "  Routes: "
+                + " | ".join(
+                    r.path
+                    for r in self.app.routes
+                    if hasattr(r, "methods") and r.path not in ("/openapi.json",)
+                )
             )
         )
 

src/aiperf/common/base_component_service.py

@@ -67,7 +67,9 @@ async def _heartbeat_task(self) -> None:
     async def _register_service_on_start(self) -> None:
         """Register the service with the system controller on startup."""
         self.debug(
-            lambda: f"Attempting to register service {self} ({self.service_id}) with system controller"
+            lambda: (
+                f"Attempting to register service {self} ({self.service_id}) with system controller"
+            )
         )
         result = None
         command_message = RegisterServiceCommand(
@@ -97,7 +99,9 @@ async def _register_service_on_start(self) -> None:
                     )
                 else:
                     self.debug(
-                        lambda: f"Service {self.service_id} registered with system controller"
+                        lambda: (
+                            f"Service {self.service_id} registered with system controller"
+                        )
                     )
                 break
 

src/aiperf/common/base_service.py

@@ -181,7 +181,9 @@ async def _kill(self) -> None:
             )
         except Exception as publish_error:
             self.debug(
-                lambda e=publish_error: f"Failed to publish BaseServiceErrorMessage during _kill (comms may already be down): {e!r}"
+                lambda e=publish_error: (
+                    f"Failed to publish BaseServiceErrorMessage during _kill (comms may already be down): {e!r}"
+                )
             )
         self.stop_requested = True
         self.stopped_event.set()

src/aiperf/common/mixins/command_handler_mixin.py

@@ -74,12 +74,16 @@ async def _process_command_message(self, message: CommandMessage) -> None:
         """
         self.trace_or_debug(
             lambda: f"Received command message: {message}",
-            lambda: f"Received '{message.command}' command from '{message.service_id}' (id: {message.command_id})",
+            lambda: (
+                f"Received '{message.command}' command from '{message.service_id}' (id: {message.command_id})"
+            ),
         )
         if message.command_id in self._processed_command_ids:
             self.trace_or_debug(
                 lambda: f"Received duplicate command message: {message}. Ignoring.",
-                lambda: f"Received duplicate command message '{message.command}' from '{message.service_id}' (id: {message.command_id}). Ignoring.",
+                lambda: (
+                    f"Received duplicate command message '{message.command}' from '{message.service_id}' (id: {message.command_id}). Ignoring."
+                ),
             )
             # If we receive a duplicate command message, we just send an acknowledged response.
             await self._publish_command_acknowledged_response(message)
@@ -91,7 +95,9 @@ async def _process_command_message(self, message: CommandMessage) -> None:
             # In the case of a broadcast command, you will receive a command message from yourself.
             # We ignore these messages.
             self.debug(
-                lambda: f"Received broadcast command message from self: {message}. Ignoring."
+                lambda: (
+                    f"Received broadcast command message from self: {message}. Ignoring."
+                )
             )
             return
 
@@ -317,7 +323,9 @@ async def _process_command_response_message(self, message: CommandResponse) -> N
         """
         self.trace_or_debug(
             lambda: f"Received command response message: {message}",
-            lambda: f"Received {message.status} response for command '{message.command}' from '{message.service_id}' (id: {message.command_id})",
+            lambda: (
+                f"Received {message.status} response for command '{message.command}' from '{message.service_id}' (id: {message.command_id})"
+            ),
         )
 
         # If the command ID is in the single response futures, we set the result of the future.
@@ -328,7 +336,9 @@ async def _process_command_response_message(self, message: CommandResponse) -> N
                 future.set_result(message)
             else:
                 self.debug(
-                    lambda: f"Already received response for command {message.command_id}, ignoring duplicate from {message.service_id}"
+                    lambda: (
+                        f"Already received response for command {message.command_id}, ignoring duplicate from {message.service_id}"
+                    )
                 )
             return
 
@@ -343,7 +353,9 @@ async def _process_command_response_message(self, message: CommandResponse) -> N
                     future.set_result(message)
                 else:
                     self.debug(
-                        lambda: f"Already received response for command {message.command_id} from {message.service_id}, ignoring duplicate"
+                        lambda: (
+                            f"Already received response for command {message.command_id} from {message.service_id}, ignoring duplicate"
+                        )
                     )
             else:
                 if message.command != CommandType.PROFILE_CANCEL:
@@ -355,5 +367,7 @@ async def _process_command_response_message(self, message: CommandResponse) -> N
         # If we reach here, we received a command response that we were not tracking. It is
         # safe to ignore.
         self.debug(
-            lambda: f"Received command response for untracked command: {message}. Ignoring."
+            lambda: (
+                f"Received command response for untracked command: {message}. Ignoring."
+            )
         )

src/aiperf/common/mixins/hooks_mixin.py

@@ -69,7 +69,9 @@ def __init__(self, **kwargs) -> None:
                     )
 
         self.trace(
-            lambda: f"Provided hook types: {self._provided_hook_types} for {self.__class__.__name__}"
+            lambda: (
+                f"Provided hook types: {self._provided_hook_types} for {self.__class__.__name__}"
+            )
         )
 
     def _check_hook_type_is_provided(self, hook_type: HookType) -> None:
@@ -152,8 +154,9 @@ def for_each_hook_param(
                 )
             for param in params:
                 self.trace(
-                    lambda param=param,
-                    type=param_type: f"param: {param}, param_type: {type}"
+                    lambda param=param, type=param_type: (
+                        f"param: {param}, param_type: {type}"
+                    )
                 )
                 if not isinstance(param, param_type):
                     raise ValueError(

src/aiperf/common/mixins/message_bus_mixin.py

@@ -50,7 +50,9 @@ def _add_to_subscription_map(hook: Hook, message_type: MessageTypeT) -> None:
             subscribe_all for efficiency
             """
             self.debug(
-                lambda: f"Adding subscription for message type: '{message_type}' for hook: {hook}"
+                lambda: (
+                    f"Adding subscription for message type: '{message_type}' for hook: {hook}"
+                )
             )
             subscription_map.setdefault(message_type, []).append(hook.func)
 

src/aiperf/common/mixins/progress_tracker_mixin.py

@@ -80,7 +80,9 @@ def _update_phase_progress(
         pct = getattr(stats, f"{prefix}_progress_percent")
 
         _logger.debug(
-            lambda: f"Updating {prefix} stats for phase '{stats.phase.title()}': progress_percent: {pct}, finished: {finished}"
+            lambda: (
+                f"Updating {prefix} stats for phase '{stats.phase.title()}': progress_percent: {pct}, finished: {finished}"
+            )
         )
 
         if not pct or finished == 0:

src/aiperf/common/mixins/pull_client_mixin.py

@@ -46,7 +46,9 @@ async def _setup_pull_handler_hooks(self) -> None:
 
         def _register_pull_callback(hook: Hook, message_type: MessageTypeT) -> None:
             self.debug(
-                lambda: f"Registering pull callback for message type: {message_type} for hook: {hook}"
+                lambda: (
+                    f"Registering pull callback for message type: {message_type} for hook: {hook}"
+                )
             )
             self.pull_client.register_pull_callback(
                 message_type=message_type,

src/aiperf/common/mixins/reply_client_mixin.py

@@ -40,7 +40,9 @@ async def _setup_request_handler_hooks(self) -> None:
 
         def _register_request_handler(hook: Hook, message_type: MessageTypeT) -> None:
             self.debug(
-                lambda: f"Registering request handler for message type: {message_type} for hook: {hook}"
+                lambda: (
+                    f"Registering request handler for message type: {message_type} for hook: {hook}"
+                )
             )
             self.reply_client.register_request_handler(
                 service_id=self.id,

src/aiperf/config/comm/base.py

@@ -121,25 +121,25 @@ def get_address(self, address_type: CommAddress) -> str:
 # CommAddress to a resolver callable keeps BaseZMQCommunicationConfig.get_address
 # flat (no match ladder) and makes the mapping easy to extend.
 _ADDRESS_RESOLVERS: dict[CommAddress, Callable[[BaseZMQCommunicationConfig], str]] = {
-    CommAddress.EVENT_BUS_PROXY_FRONTEND: lambda c: c.event_bus_proxy_config.resolve_frontend(
-        c._remote_host
+    CommAddress.EVENT_BUS_PROXY_FRONTEND: lambda c: (
+        c.event_bus_proxy_config.resolve_frontend(c._remote_host)
     ),
-    CommAddress.EVENT_BUS_PROXY_BACKEND: lambda c: c.event_bus_proxy_config.resolve_backend(
-        c._remote_host
+    CommAddress.EVENT_BUS_PROXY_BACKEND: lambda c: (
+        c.event_bus_proxy_config.resolve_backend(c._remote_host)
     ),
-    CommAddress.DATASET_MANAGER_PROXY_FRONTEND: lambda c: c.dataset_manager_proxy_config.resolve_frontend(
-        c._remote_host
+    CommAddress.DATASET_MANAGER_PROXY_FRONTEND: lambda c: (
+        c.dataset_manager_proxy_config.resolve_frontend(c._remote_host)
     ),
-    CommAddress.DATASET_MANAGER_PROXY_BACKEND: lambda c: c.dataset_manager_proxy_config.resolve_backend(
-        c._remote_host
+    CommAddress.DATASET_MANAGER_PROXY_BACKEND: lambda c: (
+        c.dataset_manager_proxy_config.resolve_backend(c._remote_host)
     ),
     # Raw inference proxy is always local (within-pod IPC). Workers and record
     # processors are co-located in the same pod, so remote_host is ignored.
-    CommAddress.RAW_INFERENCE_PROXY_FRONTEND: lambda c: c.raw_inference_proxy_config.resolve_frontend(
-        None
+    CommAddress.RAW_INFERENCE_PROXY_FRONTEND: lambda c: (
+        c.raw_inference_proxy_config.resolve_frontend(None)
     ),
-    CommAddress.RAW_INFERENCE_PROXY_BACKEND: lambda c: c.raw_inference_proxy_config.resolve_backend(
-        None
+    CommAddress.RAW_INFERENCE_PROXY_BACKEND: lambda c: (
+        c.raw_inference_proxy_config.resolve_backend(None)
     ),
     CommAddress.CREDIT_ROUTER: lambda c: c.credit_router_address,
     CommAddress.CREDIT_RETURN_ROUTER: lambda c: c.credit_return_router_address,

src/aiperf/controller/multiprocess_service_manager.py

@@ -95,8 +95,9 @@ async def run_service(
             process.start()
 
             self.debug(
-                lambda pid=process.pid,
-                type=service_type: f"Service {type} started as process (pid: {pid})"
+                lambda pid=process.pid, type=service_type: (
+                    f"Service {type} started as process (pid: {pid})"
+                )
             )
 
             self.multi_process_info.append(
@@ -269,7 +270,9 @@ async def _wait_for_process(self, info: MultiProcessRunInfo) -> None:
             info.process.kill()
         else:
             self.debug(
-                lambda: f"Service {info.service_type} process stopped (pid: {info.process.pid})"
+                lambda: (
+                    f"Service {info.service_type} process stopped (pid: {info.process.pid})"
+                )
             )
 
     async def wait_for_all_services_start(

src/aiperf/dataset/composer/base.py

@@ -330,7 +330,9 @@ def _inject_context_prompts(self, conversations: list[Conversation]) -> None:
             if shared_system_prompt:
                 conversation.system_message = shared_system_prompt
                 self.trace(
-                    lambda conv=conversation: f"Set system_message on conversation {conv.session_id}"
+                    lambda conv=conversation: (
+                        f"Set system_message on conversation {conv.session_id}"
+                    )
                 )
 
             # Set user context prompt (unique per session)
@@ -340,9 +342,10 @@ def _inject_context_prompts(self, conversations: list[Conversation]) -> None:
                 )
                 conversation.user_context_message = user_context
                 self.trace(
-                    lambda idx=session_index,
-                    conv=conversation: f"Set user_context_message for session {idx} "
-                    f"(conversation {conv.session_id})"
+                    lambda idx=session_index, conv=conversation: (
+                        f"Set user_context_message for session {idx} "
+                        f"(conversation {conv.session_id})"
+                    )
                 )
 
     @staticmethod

src/aiperf/dataset/composer/synthetic_rankings.py

@@ -109,6 +109,8 @@ def _create_turn(self, num_passages: int) -> Turn:
         self._finalize_turn(turn)
 
         self.debug(
-            lambda: f"[rankings] query_len={len(query_text)} chars, passages={num_passages}"
+            lambda: (
+                f"[rankings] query_len={len(query_text)} chars, passages={num_passages}"
+            )
         )
         return turn

src/aiperf/dataset/dataset_manager.py

@@ -293,8 +293,10 @@ def _generate_input_payloads(
         )
         endpoint: EndpointProtocol = EndpointClass(model_endpoint=model_endpoint)
         self.debug(
-            lambda: f"Created endpoint protocol for {model_endpoint.endpoint.type}, "
-            f"class: {endpoint.__class__.__name__}",
+            lambda: (
+                f"Created endpoint protocol for {model_endpoint.endpoint.type}, "
+                f"class: {endpoint.__class__.__name__}"
+            ),
         )
         session_payloads_map: dict[str, list] = {}
         for conversation in self.dataset.values():

src/aiperf/dataset/generator/image.py

@@ -116,7 +116,9 @@ def generate(self, *args, **kwargs) -> str:
 
         image = self._create_source_image(width, height)
         self.debug(
-            lambda: f"Generated image from {self.config.source} with width={width}, height={height}"
+            lambda: (
+                f"Generated image from {self.config.source} with width={width}, height={height}"
+            )
         )
         base64_image = utils.encode_image(image, image_format)
         return f"data:image/{image_format.name.lower()};base64,{base64_image}"

src/aiperf/dataset/generator/prompt.py

@@ -148,8 +148,10 @@ def tokenize_chunk(chunk):
         ]
         self._corpus_size = len(self._tokenized_corpus)
         self.debug(
-            lambda: f"Initialized corpus with {self._corpus_size} tokens "
-            f"from {len(chunks)} chunks using {num_threads} thread(s)"
+            lambda: (
+                f"Initialized corpus with {self._corpus_size} tokens "
+                f"from {len(chunks)} chunks using {num_threads} thread(s)"
+            )
         )
 
     def _create_prefix_prompt_pool(self) -> None:
@@ -164,7 +166,9 @@ def _create_prefix_prompt_pool(self) -> None:
         pool_size = self.prefix_prompts.pool_size or 0
         self._prefix_prompts = [self.generate_prompt(length) for _ in range(pool_size)]
         self.debug(
-            lambda: f"Initialized prefix prompts pool with {len(self._prefix_prompts)} prompts"
+            lambda: (
+                f"Initialized prefix prompts pool with {len(self._prefix_prompts)} prompts"
+            )
         )
 
     def generate(
@@ -433,8 +437,10 @@ def generate_user_context_prompt(self, session_index: int) -> str:
             new_prompt = self.generate_prompt(length)
             self._user_context_prompts.append(new_prompt)
             self.debug(
-                lambda: f"Generated user context prompt #{len(self._user_context_prompts) - 1} "
-                f"for session {len(self._user_context_prompts) - 1}"
+                lambda: (
+                    f"Generated user context prompt #{len(self._user_context_prompts) - 1} "
+                    f"for session {len(self._user_context_prompts) - 1}"
+                )
             )
 
         return self._user_context_prompts[session_index]

src/aiperf/dataset/loader/base_hf_dataset.py

@@ -178,7 +178,9 @@ def _extract_audio(self, row: dict[str, Any], audio_column: str) -> list[Audio]:
             return [Audio(name="", contents=[f"wav,{b64}"])]
         except (ImportError, OSError, ValueError, RuntimeError) as e:
             self.debug(
-                lambda exc=e: f"Failed to encode WAV from column '{audio_column}': {exc.__class__.__name__}: {exc}"
+                lambda exc=e: (
+                    f"Failed to encode WAV from column '{audio_column}': {exc.__class__.__name__}: {exc}"
+                )
             )
             return []