Fix EM102: extract f-strings from exception constructors

pyproject.toml

@@ -171,7 +171,6 @@ target-version = "py310"
 ignore = [
   # Style choices — deliberate project conventions
   "EM101",    # Exception string literals
-  "EM102",    # Exception f-strings
   "G004",     # Logging f-strings
   "T201",     # print() used for user output
   "TRY003",   # Raise with inline message strings

src/seclab_taskflow_agent/agent.py

@@ -182,7 +182,8 @@ def __init__(
         if token:
             resolved_token = os.getenv(token, "")
             if not resolved_token:
-                raise RuntimeError(f"Token env var {token!r} is not set")
+                msg = f"Token env var {token!r} is not set"
+                raise RuntimeError(msg)
         else:
             resolved_token = get_AI_token()
 

src/seclab_taskflow_agent/available_tools.py

@@ -108,17 +108,21 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
         # Resolve package and filename from dotted path
         components = toolname.rsplit(".", 1)
         if len(components) != 2:
-            raise BadToolNameError(
+            msg = (
                 f'Not a valid toolname: "{toolname}". '
                 f'Expected format: "packagename.filename"'
             )
+            raise BadToolNameError(
+                msg
+            )
         package, filename = components
 
         try:
             pkg_dir = importlib.resources.files(package)
             if not pkg_dir.is_dir():
+                msg = f"Cannot load {toolname} because {pkg_dir} is not a valid directory."
                 raise BadToolNameError(
-                    f"Cannot load {toolname} because {pkg_dir} is not a valid directory."
+                    msg
                 )
             filepath = pkg_dir.joinpath(filename + ".yaml")
             with filepath.open() as fh:
@@ -128,16 +132,20 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
             header = raw.get("seclab-taskflow-agent", {})
             filetype = header.get("filetype", "")
             if filetype != tooltype.value:
-                raise FileTypeException(
+                msg = (
                     f"Error in {filepath}: expected filetype {tooltype.value!r}, "
                     f"got {filetype!r}."
                 )
+                raise FileTypeException(
+                    msg
+                )
 
             # Parse into the appropriate Pydantic model
             model_cls = DOCUMENT_MODELS.get(filetype)
             if model_cls is None:
+                msg = f"Unknown filetype {filetype!r} in {toolname}"
                 raise BadToolNameError(
-                    f"Unknown filetype {filetype!r} in {toolname}"
+                    msg
                 )
 
             try:
@@ -147,8 +155,9 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
                 for err in exc.errors():
                     if "Unsupported version" in str(err.get("msg", "")):
                         raise VersionException(str(err["msg"])) from exc
+                msg = f"Validation error loading {toolname}: {exc}"
                 raise BadToolNameError(
-                    f"Validation error loading {toolname}: {exc}"
+                    msg
                 ) from exc
 
             # Cache and return
@@ -158,10 +167,13 @@ def _load(self, tooltype: AvailableToolType, toolname: str) -> DocumentModel:
             return doc
 
         except ModuleNotFoundError as exc:
-            raise BadToolNameError(f"Cannot load {toolname}: {exc}") from exc
+            msg = f"Cannot load {toolname}: {exc}"
+            raise BadToolNameError(msg) from exc
         except FileNotFoundError:
+            msg = f"Cannot load {toolname} because {filepath} is not a valid file."
             raise BadToolNameError(
-                f"Cannot load {toolname} because {filepath} is not a valid file."
+                msg
             )
         except ValueError as exc:
-            raise BadToolNameError(f"Cannot load {toolname}: {exc}") from exc
+            msg = f"Cannot load {toolname}: {exc}"
+            raise BadToolNameError(msg) from exc

src/seclab_taskflow_agent/capi.py

@@ -38,7 +38,8 @@ def to_url(self) -> str:
             case AI_API_ENDPOINT_ENUM.AI_API_OPENAI:
                 return f"https://{self}/v1"
             case _:
-                raise ValueError(f"Unsupported endpoint: {self}")
+                msg = f"Unsupported endpoint: {self}"
+                raise ValueError(msg)
 
 
 COPILOT_INTEGRATION_ID = "vscode-chat"

src/seclab_taskflow_agent/cli.py

@@ -36,7 +36,8 @@
 def _parse_global(value: str) -> tuple[str, str]:
     """Parse a ``KEY=VALUE`` string into a (key, value) pair."""
     if "=" not in value:
-        raise typer.BadParameter(f"Invalid global variable format: {value!r}. Expected KEY=VALUE.")
+        msg = f"Invalid global variable format: {value!r}. Expected KEY=VALUE."
+        raise typer.BadParameter(msg)
     key, _, val = value.partition("=")
     return key.strip(), val.strip()
 

src/seclab_taskflow_agent/mcp_lifecycle.py

@@ -116,7 +116,8 @@ def _print_err(line: str) -> None:
                     client_session_timeout_seconds=client_session_timeout,
                 )
             case _:
-                raise ValueError(f"Unsupported MCP transport: {params['kind']}")
+                msg = f"Unsupported MCP transport: {params['kind']}"
+                raise ValueError(msg)
 
         entries.append(MCPServerEntry(MCPNamespaceWrap(confirms, mcp_server), server_proc, name=tb))
 

src/seclab_taskflow_agent/mcp_servers/codeql/client.py

@@ -467,7 +467,8 @@ def _file_uri_to_path(uri):
     # note: don't try to parse paths like "file://a/b" because that returns "/b", should be "file:///a/b"
     parsed = urlparse(uri)
     if parsed.scheme != "file":
-        raise ValueError(f"Not a file:// uri: {uri}")
+        msg = f"Not a file:// uri: {uri}"
+        raise ValueError(msg)
     path = unquote(parsed.path)
     region = None
     if ":" in path:
@@ -605,7 +606,8 @@ def run_query(
     if target:
         target_pos = get_query_position(query_path, target)
         if not target_pos:
-            raise ValueError(f"Could not resolve quick eval target for {target}")
+            msg = f"Could not resolve quick eval target for {target}"
+            raise ValueError(msg)
     try:
         with (
             QueryServer(database, keep_alive=keep_alive, log_stderr=log_stderr) as server,
@@ -635,5 +637,6 @@ def run_query(
                 case _:
                     raise ValueError("Unsupported output format {fmt}")
     except Exception as e:
-        raise RuntimeError(f"Error in run_query: {e}") from e
+        msg = f"Error in run_query: {e}"
+        raise RuntimeError(msg) from e
     return result

src/seclab_taskflow_agent/mcp_servers/codeql/jsonrpyc/__init__.py

@@ -86,7 +86,8 @@ def check_id(cls, id: str | int | None, *, allow_empty: bool = False) -> None:
         :raises TypeError: When *id* is invalid.
         """
         if (id is not None or not allow_empty) and not isinstance(id, (int, str)):
-            raise TypeError(f"id must be an integer or string, got {id} ({type(id)})")
+            msg = f"id must be an integer or string, got {id} ({type(id)})"
+            raise TypeError(msg)
 
     @classmethod
     def check_method(cls, method: str, /) -> None:
@@ -98,7 +99,8 @@ def check_method(cls, method: str, /) -> None:
         :raises TypeError: When *method* is invalid.
         """
         if not isinstance(method, str):
-            raise TypeError(f"method must be a string, got {method} ({type(method)})")
+            msg = f"method must be a string, got {method} ({type(method)})"
+            raise TypeError(msg)
 
     @classmethod
     def check_code(cls, code: int, /) -> None:
@@ -113,7 +115,8 @@ def check_code(cls, code: int, /) -> None:
         try:
             get_error(code)
         except Exception:
-            raise TypeError(f"invalid error code, got {code} ({type(code)})")
+            msg = f"invalid error code, got {code} ({type(code)})"
+            raise TypeError(msg)
 
     @classmethod
     def request(
@@ -502,7 +505,8 @@ def _handle_method(self, req: dict[str, Any]) -> None:
         try:
             method = req["method"]
             if method not in self.method_handlers:
-                raise ValueError(f"No handler defined for method: {method}")
+                msg = f"No handler defined for method: {method}"
+                raise ValueError(msg)
             result = self.method_handlers[method](req["params"])
             if "id" in req:
                 res = Spec.response(req["id"], result)
@@ -752,7 +756,8 @@ def run(self) -> None:
                         print(f"Incoming jsonrpc message: {decoded_msg}")
                     self.rpc._handle(decoded_msg)
                 else:
-                    raise ValueError(f"Don't know how to handle: {decoded_line}")
+                    msg = f"Don't know how to handle: {decoded_line}"
+                    raise ValueError(msg)
             else:
                 self._stopper.wait(self.interval)
 
@@ -836,13 +841,16 @@ class MyCustomRPCError(RPCError):
             title = "My custom error"
     """
     if not issubclass(cls, RPCError):
-        raise TypeError(f"'{cls}' is not a subclass of RPCError")
+        msg = f"'{cls}' is not a subclass of RPCError"
+        raise TypeError(msg)
 
     # check duplicates
     if cls.code in error_map_code:
-        raise AttributeError(f"duplicate RPC error code {cls.code}")
+        msg = f"duplicate RPC error code {cls.code}"
+        raise AttributeError(msg)
     if cls.code_range in error_map_code_range:
-        raise AttributeError(f"duplicate RPC error code range {cls.code_range}")
+        msg = f"duplicate RPC error code range {cls.code_range}"
+        raise AttributeError(msg)
 
     # register
     error_map_code[cls.code] = cls
@@ -867,7 +875,8 @@ def get_error(code: int) -> type[RPCError]:
         if lower <= code <= upper:
             return cls
 
-    raise ValueError(f"unknown error code '{code}' ({type(code)})")
+    msg = f"unknown error code '{code}' ({type(code)})"
+    raise ValueError(msg)
 
 
 @register_error

src/seclab_taskflow_agent/mcp_servers/codeql/mcp_server.py

@@ -53,10 +53,12 @@
 def _resolve_query_path(language: str, query: str) -> Path:
     global TEMPLATED_QUERY_PATHS
     if language not in TEMPLATED_QUERY_PATHS:
-        raise RuntimeError(f"Error: Language `{language}` not supported!")
+        msg = f"Error: Language `{language}` not supported!"
+        raise RuntimeError(msg)
     query_path = TEMPLATED_QUERY_PATHS[language].get(query)
     if not query_path:
-        raise RuntimeError(f"Error: query `{query}` not supported for `{language}`!")
+        msg = f"Error: query `{query}` not supported for `{language}`!"
+        raise RuntimeError(msg)
     return Path(query_path)
 
 
@@ -69,7 +71,8 @@ def _resolve_db_path(relative_db_path: str | Path):
     absolute_path = CODEQL_DBS_BASE_PATH / relative_db_path
     if not absolute_path.is_dir():
         _debug_log(f"Database path not found: {absolute_path}")
-        raise RuntimeError(f"Error: Database not found at {absolute_path}!")
+        msg = f"Error: Database not found at {absolute_path}!"
+        raise RuntimeError(msg)
     return absolute_path
 
 

src/seclab_taskflow_agent/mcp_transport.py

@@ -109,7 +109,8 @@ async def async_wait_for_connection(
         host = parsed.hostname
         port = parsed.port
         if host is None or port is None:
-            raise ValueError(f"URL must include a host and port: {self.url}")
+            msg = f"URL must include a host and port: {self.url}"
+            raise ValueError(msg)
         deadline = asyncio.get_event_loop().time() + timeout
         while True:
             try:
@@ -119,7 +120,8 @@ async def async_wait_for_connection(
                 return
             except (OSError, ConnectionRefusedError):
                 if asyncio.get_event_loop().time() > deadline:
-                    raise TimeoutError(f"Could not connect to {host}:{port} after {timeout} seconds")
+                    msg = f"Could not connect to {host}:{port} after {timeout} seconds"
+                    raise TimeoutError(msg)
                 await asyncio.sleep(poll_interval)
 
     def wait_for_connection(
@@ -139,15 +141,17 @@ def wait_for_connection(
         host = parsed.hostname
         port = parsed.port
         if host is None or port is None:
-            raise ValueError(f"URL must include a host and port: {self.url}")
+            msg = f"URL must include a host and port: {self.url}"
+            raise ValueError(msg)
         deadline = time.time() + timeout
         while True:
             try:
                 with socket.create_connection((host, port), timeout=2):
                     return
             except OSError:
                 if time.time() > deadline:
-                    raise TimeoutError(f"Could not connect to {host}:{port} after {timeout} seconds")
+                    msg = f"Could not connect to {host}:{port} after {timeout} seconds"
+                    raise TimeoutError(msg)
                 time.sleep(poll_interval)
 
     def run(self) -> None:

src/seclab_taskflow_agent/mcp_utils.py

@@ -202,7 +202,8 @@ def mcp_client_params(
                     logging.debug(f"Initializing streamable toolbox: {tb}\nargs:\n{args}\nenv:\n{env}\n")
                     exe = shutil.which(sp.command)
                     if exe is None:
-                        raise FileNotFoundError(f"Could not resolve path to {sp.command}")
+                        msg = f"Could not resolve path to {sp.command}"
+                        raise FileNotFoundError(msg)
                     start_cmd = [exe]
                     if args:
                         for i, v in enumerate(args):
@@ -220,7 +221,8 @@ def mcp_client_params(
                     server_params["env"] = env
 
             case _:
-                raise ValueError(f"Unsupported MCP transport {kind}")
+                msg = f"Unsupported MCP transport {kind}"
+                raise ValueError(msg)
 
         client_params[tb] = (
             server_params,

src/seclab_taskflow_agent/models.py

@@ -62,8 +62,9 @@ def _normalise_version(cls, v: Any) -> str:
     @classmethod
     def _validate_version(cls, v: str) -> str:
         if v != SUPPORTED_VERSION:
+            msg = f"Unsupported version: {v}. Only version {SUPPORTED_VERSION} is supported."
             raise ValueError(
-                f"Unsupported version: {v}. Only version {SUPPORTED_VERSION} is supported."
+                msg
             )
         return v
 

src/seclab_taskflow_agent/runner.py

@@ -78,8 +78,9 @@ def _resolve_model_config(
     models_params: dict[str, dict[str, Any]] = m_config.model_settings or {}
     unknown = set(models_params) - set(model_keys)
     if unknown:
+        msg = f"Settings section of model_config file {model_config_ref} contains models not in the model section: {unknown}"
         raise ValueError(
-            f"Settings section of model_config file {model_config_ref} contains models not in the model section: {unknown}"
+            msg
         )
     return model_keys, model_dict, models_params, m_config.api_type
 
@@ -103,7 +104,8 @@ def _merge_reusable_task(
     """
     reusable_doc = available_tools.get_taskflow(task.uses)
     if reusable_doc is None:
-        raise ValueError(f"No such reusable taskflow: {task.uses}")
+        msg = f"No such reusable taskflow: {task.uses}"
+        raise ValueError(msg)
     if len(reusable_doc.taskflow) > 1:
         raise ValueError("Reusable taskflows can only contain 1 task")
     parent_task = reusable_doc.taskflow[0].task
@@ -147,7 +149,8 @@ def _resolve_task_model(
 
     task_model_settings: dict[str, Any] | Any = task.model_settings or {}
     if not isinstance(task_model_settings, dict):
-        raise ValueError(f"model_settings in task {task.name or ''} needs to be a dictionary")
+        msg = f"model_settings in task {task.name or ''} needs to be a dictionary"
+        raise ValueError(msg)
 
     # Task-level overrides can also set engine keys
     task_settings = dict(task_model_settings)
@@ -228,7 +231,8 @@ async def _build_prompts_to_run(
                     prompts_to_run.append(rendered_prompt)
                 except jinja2.TemplateError as e:
                     logging.error(f"Error rendering template for result {value}: {e}")
-                    raise ValueError(f"Template rendering failed: {e}")
+                    msg = f"Template rendering failed: {e}"
+                    raise ValueError(msg)
 
         # Consume only after all prompts rendered successfully so that
         # the result remains available for retry/resume on failure.
@@ -577,7 +581,8 @@ async def on_handoff_hook(context: RunContextWrapper[TContext], agent: Agent[TCo
                     )
                 except jinja2.TemplateError as e:
                     logging.error(f"Template rendering error: {e}")
-                    raise ValueError(f"Failed to render prompt template: {e}") from e
+                    msg = f"Failed to render prompt template: {e}"
+                    raise ValueError(msg) from e
 
             with TmpEnv(env):
                 prompts_to_run: list[str] = await _build_prompts_to_run(
@@ -611,7 +616,8 @@ async def run_prompts(async_task: bool = False, max_concurrent_tasks: int = 5) -
                         for agent_name in current_agents:
                             personality = available_tools.get_personality(agent_name)
                             if personality is None:
-                                raise ValueError(f"No such personality: {agent_name}")
+                                msg = f"No such personality: {agent_name}"
+                                raise ValueError(msg)
                             resolved_agents[agent_name] = personality
 
                         if not resolved_agents:

src/seclab_taskflow_agent/session.py

@@ -121,7 +121,8 @@ def load(cls, session_id: str) -> TaskflowSession:
         """
         path = session_dir() / f"{session_id}.json"
         if not path.exists():
-            raise FileNotFoundError(f"No session checkpoint found: {session_id}")
+            msg = f"No session checkpoint found: {session_id}"
+            raise FileNotFoundError(msg)
         return cls.model_validate_json(path.read_text())
 
     @classmethod