Fix EM101: assign exception string literals to variables

pyproject.toml

@@ -170,7 +170,6 @@ target-version = "py310"
 # deliberate style choices.  Rules can be tightened incrementally.
 ignore = [
   # Style choices — deliberate project conventions
-  "EM101",    # Exception string literals
   "EM102",    # Exception f-strings
   "G004",     # Logging f-strings
   "T201",     # print() used for user output

src/seclab_taskflow_agent/capi.py

@@ -167,7 +167,8 @@ def get_AI_token() -> str:
     token = os.getenv("COPILOT_TOKEN")
     if token:
         return token
-    raise RuntimeError("AI_API_TOKEN environment variable is not set.")
+    msg = "AI_API_TOKEN environment variable is not set."
+    raise RuntimeError(msg)
 
 
 # ---------------------------------------------------------------------------

src/seclab_taskflow_agent/mcp_servers/codeql/client.py

@@ -194,10 +194,12 @@ def _server_request_run(
         template_values: dict | None = None,
     ):
         if not self.active_database:
-            raise RuntimeError("No Active Database")
+            msg = "No Active Database"
+            raise RuntimeError(msg)
 
         if not self.active_connection:
-            raise RuntimeError("No Active Connection")
+            msg = "No Active Connection"
+            raise RuntimeError(msg)
 
         if isinstance(quick_eval_pos, dict):
             # A quick eval position contains:
@@ -302,7 +304,8 @@ def _format(self, query):
     def _resolve_query_server(self):
         help_msg = shell_command_to_string(self.codeql_cli + ["excute", "--help"])
         if not re.search("query-server2", help_msg):
-            raise RuntimeError("Legacy server not supported!")
+            msg = "Legacy server not supported!"
+            raise RuntimeError(msg)
         return "query-server2"
 
     def _resolve_library_paths(self, query_path):
@@ -463,7 +466,8 @@ def _file_uri_to_path(uri):
     # internally the codeql client will resolve both relative and full paths
     # regardless of root directory differences
     if not uri.startswith("file:///"):
-        raise ValueError("URI path should be formatted as absolute")
+        msg = "URI path should be formatted as absolute"
+        raise ValueError(msg)
     # 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":
@@ -633,7 +637,8 @@ def run_query(
                 case "sarif":
                     result = server._bqrs_to_sarif(bqrs_path, server._query_info(query_path))
                 case _:
-                    raise ValueError("Unsupported output format {fmt}")
+                    msg = f"Unsupported output format {fmt}"
+                    raise ValueError(msg)
     except Exception as e:
         raise RuntimeError(f"Error in run_query: {e}") from e
     return result

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

@@ -459,7 +459,8 @@ def call(
                 if timeout:
                     elapsed = time.monotonic() - starting_time
                     if elapsed > timeout:
-                        raise TimeoutError("RPC Request timed out")
+                        msg = "RPC Request timed out"
+                        raise TimeoutError(msg)
 
                 time.sleep(block)
         return id

src/seclab_taskflow_agent/mcp_transport.py

@@ -216,7 +216,8 @@ def join_and_raise(self, timeout: float | None = None) -> None:
         """
         self.join(timeout)
         if self.is_alive():
-            raise RuntimeError("Process thread did not exit within timeout.")
+            msg = "Process thread did not exit within timeout."
+            raise RuntimeError(msg)
         if self.exception is not None:
             raise self.exception
 

src/seclab_taskflow_agent/models.py

@@ -106,7 +106,8 @@ class TaskDefinition(BaseModel):
     @model_validator(mode="after")
     def _run_xor_prompt(self) -> TaskDefinition:
         if self.run and self.user_prompt:
-            raise ValueError("shell task ('run') and prompt task ('user_prompt') are mutually exclusive")
+            msg = "shell task ('run') and prompt task ('user_prompt') are mutually exclusive"
+            raise ValueError(msg)
         return self
 
 

src/seclab_taskflow_agent/runner.py

@@ -105,7 +105,8 @@ def _merge_reusable_task(
     if reusable_doc is None:
         raise ValueError(f"No such reusable taskflow: {task.uses}")
     if len(reusable_doc.taskflow) > 1:
-        raise ValueError("Reusable taskflows can only contain 1 task")
+        msg = "Reusable taskflows can only contain 1 task"
+        raise ValueError(msg)
     parent_task = reusable_doc.taskflow[0].task
     merged: dict[str, Any] = parent_task.model_dump(by_alias=True, exclude_defaults=True)
     current: dict[str, Any] = task.model_dump(by_alias=True, exclude_defaults=True)
@@ -198,14 +199,16 @@ async def _build_prompts_to_run(
             raise
         except json.JSONDecodeError as exc:
             logging.critical(f"Could not parse tool result as JSON: {last_mcp_tool_results[-1][:200]}")
-            raise ValueError("Tool result is not valid JSON") from exc
+            msg = "Tool result is not valid JSON"
+            raise ValueError(msg) from exc
 
         text = last_result.get("text", "")
         try:
             iterable_result = json.loads(text)
         except json.JSONDecodeError as exc:
             logging.critical(f"Could not parse result text: {text}")
-            raise ValueError("Result text is not valid JSON") from exc
+            msg = "Result text is not valid JSON"
+            raise ValueError(msg) from exc
         try:
             iter(iterable_result)
         except TypeError:
@@ -403,7 +406,8 @@ async def _run_streamed() -> None:
                         max_retry -= 1
                     except RateLimitError:
                         if rate_limit_backoff == MAX_RATE_LIMIT_BACKOFF:
-                            raise APITimeoutError("Max rate limit backoff reached")
+                            msg = "Max rate limit backoff reached"
+                            raise APITimeoutError(msg)
                         if rate_limit_backoff > MAX_RATE_LIMIT_BACKOFF:
                             rate_limit_backoff = MAX_RATE_LIMIT_BACKOFF
                         else:
@@ -556,7 +560,8 @@ async def on_handoff_hook(context: RunContextWrapper[TContext], agent: Agent[TCo
             inputs = task.inputs or {}
             task_prompt = task.user_prompt or ""
             if run and task_prompt:
-                raise ValueError("shell task and prompt task are mutually exclusive!")
+                msg = "shell task and prompt task are mutually exclusive!"
+                raise ValueError(msg)
             must_complete = task.must_complete
             max_turns = task.max_steps or DEFAULT_MAX_TURNS
             toolboxes_override = task.toolboxes or []
@@ -615,10 +620,11 @@ async def run_prompts(async_task: bool = False, max_concurrent_tasks: int = 5) -
                             resolved_agents[agent_name] = personality
 
                         if not resolved_agents:
-                            raise ValueError(
+                            msg = (
                                 "No agents resolved for this task. "
                                 "Specify a personality with -p or provide an agents list."
                             )
+                            raise ValueError(msg)
 
                         async def _deploy(ra: dict, pp: str) -> bool:
                             async with semaphore: