Fix threads in offline deployment (#333)

* feat: add configurable max threads for local and online deployments

- Add max_threads configuration to settings.json with separate values
  for local (4) and online (2) deployments
- Add get_max_threads() helper function in common.py that returns
  appropriate thread count based on deployment mode
- Add Threads number input in parameter_section for local mode UI
- Update CommandExecutor.run_multiple_commands() to use semaphore
  limiting based on max_threads setting
- Update CommandExecutor.run_topp() to pass -threads parameter to
  TOPP tools with intelligently distributed thread allocation

Thread distribution algorithm:
- parallel_commands = min(num_files, max_threads)
- threads_per_command = max(1, max_threads // parallel_commands)

This ensures optimal resource usage: with 4 max_threads and 2 files,
runs 2 parallel commands with 2 threads each; with 4 files, runs 4
parallel commands with 1 thread each.

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* fix: properly initialize session state for Threads number input

Pre-initialize max_threads_override in session state before rendering
the widget, and remove the value parameter to let Streamlit read from
session state. This ensures user changes are properly tracked.

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* fix: persist Threads parameter across page navigation

Use input_widget() which integrates with the parameter manager to
persist the value to params.json. Also copy value to session state
for get_max_threads() to access.

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* refactor: use parameter manager directly in get_max_threads()

- Pass parameter_manager to get_max_threads() to read persisted value
- Remove session state copying workaround in StreamlitUI
- Cleaner design that reads directly from params.json

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* refactor: move _get_max_threads() to CommandExecutor

- Move function from common.py to CommandExecutor as private method
- Simplifies code by using self.parameter_manager directly
- Remove unnecessary import and parameter passing

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* fix: validate _get_max_threads() return value

Coerce to int and clamp to minimum of 1 to prevent:
- Semaphore(0) deadlocks
- Divide-by-zero in threads_per_command calculation
- Issues from non-numeric config values

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* style: move streamlit import to top of file

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* feat: log warning for invalid max_threads values

Logs to minimal log (level 0) when max_threads is non-numeric or < 1.

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* simplify: remove warning logging from _get_max_threads()

UI already enforces min_value=1, keep just the validation.

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

* simplify: remove exception handling, fail loud on invalid config

https://claude.ai/code/session_01XGuw7AxoKRWHr9EZX1SDXi

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: t0mdavid-m

src/common/common.py

@@ -32,29 +32,6 @@
 OS_PLATFORM = sys.platform
 
 
-def get_max_threads() -> int:
-    """
-    Get max threads for current deployment mode.
-
-    In local mode, checks for UI override in session state.
-    In online mode, uses the configured value directly.
-
-    Returns:
-        int: Maximum number of threads to use for parallel processing.
-    """
-    settings = st.session_state.get("settings", {})
-    max_threads_config = settings.get("max_threads", {"local": 4, "online": 2})
-
-    if settings.get("online_deployment", False):
-        return max_threads_config.get("online", 2)
-    else:
-        # Local mode: check for UI override, fallback to config default
-        return st.session_state.get(
-            "max_threads_override",
-            max_threads_config.get("local", 4)
-        )
-
-
 def is_safe_workspace_name(name: str) -> bool:
     """
     Check if a workspace name is safe (no path traversal characters).

src/workflow/CommandExecutor.py

@@ -9,6 +9,7 @@
 import sys
 import importlib.util
 import json
+import streamlit as st
 
 class CommandExecutor:
     """
@@ -25,6 +26,28 @@ def __init__(self, workflow_dir: Path, logger: Logger, parameter_manager: Parame
         self.logger = logger
         self.parameter_manager = parameter_manager
 
+    def _get_max_threads(self) -> int:
+        """
+        Get max threads for current deployment mode.
+
+        In local mode, reads from parameter manager (persisted params.json).
+        In online mode, uses the configured value directly from settings.
+
+        Returns:
+            int: Maximum number of threads to use for parallel processing (minimum 1).
+        """
+        settings = st.session_state.get("settings", {})
+        max_threads_config = settings.get("max_threads", {"local": 4, "online": 2})
+
+        if settings.get("online_deployment", False):
+            value = max_threads_config.get("online", 2)
+        else:
+            default = max_threads_config.get("local", 4)
+            params = self.parameter_manager.get_parameters_from_json()
+            value = params.get("max_threads", default)
+
+        return max(1, int(value))
+
     def run_multiple_commands(
         self, commands: list[str]
     ) -> bool:
@@ -43,10 +66,8 @@ def run_multiple_commands(
         Returns:
             bool: True if all commands succeeded, False if any failed.
         """
-        from src.common.common import get_max_threads
-
         # Get thread settings and calculate distribution
-        max_threads = get_max_threads()
+        max_threads = self._get_max_threads()
         num_commands = len(commands)
         parallel_commands = min(num_commands, max_threads)
 
@@ -234,8 +255,7 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b
             n_processes = max(io_lengths)
 
         # Calculate threads per command based on max_threads setting
-        from src.common.common import get_max_threads
-        max_threads = get_max_threads()
+        max_threads = self._get_max_threads()
         parallel_commands = min(n_processes, max_threads)
         threads_per_command = max(1, max_threads // parallel_commands)
 

src/workflow/StreamlitUI.py

@@ -1112,11 +1112,12 @@ def parameter_section(self, custom_parameter_function) -> None:
         if not st.session_state.settings.get("online_deployment", False):
             max_threads_config = st.session_state.settings.get("max_threads", {})
             default_threads = max_threads_config.get("local", 4)
-            st.number_input(
-                "Threads",
+            self.input_widget(
+                key="max_threads",
+                default=default_threads,
+                name="Threads",
+                widget_type="number",
                 min_value=1,
-                value=default_threads,
-                key="max_threads_override",
                 help="Maximum threads for parallel processing. Threads are distributed between parallel commands and per-tool thread allocation."
             )