Fix G004: replace logging f-strings with % formatting, remove G004 from suppression list

Agent-Logs-Url: https://github.com/GitHubSecurityLab/seclab-taskflow-agent/sessions/fc7d4cac-0cf0-4a55-b9a1-b00a172e1c03

Co-authored-by: kevinbackhouse <4358136+kevinbackhouse@users.noreply.github.com>

Author: Copilot

pyproject.toml

@@ -172,7 +172,6 @@ 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

@@ -70,27 +70,27 @@ def __init__(
 
     async def on_agent_start(self, context: RunContextWrapper[TContext], agent: Agent[TContext]) -> None:
         """Called when an agent begins execution."""
-        logging.debug(f"TaskRunHooks on_agent_start: {agent.name}")
+        logging.debug("TaskRunHooks on_agent_start: %s", agent.name)
         if self._on_agent_start:
             await self._on_agent_start(context, agent)
 
     async def on_agent_end(self, context: RunContextWrapper[TContext], agent: Agent[TContext], output: Any) -> None:
         """Called when an agent finishes execution."""
-        logging.debug(f"TaskRunHooks on_agent_end: {agent.name}")
+        logging.debug("TaskRunHooks on_agent_end: %s", agent.name)
         if self._on_agent_end:
             await self._on_agent_end(context, agent, output)
 
     async def on_tool_start(self, context: RunContextWrapper[TContext], agent: Agent[TContext], tool: Tool) -> None:
         """Called before a tool invocation begins."""
-        logging.debug(f"TaskRunHooks on_tool_start: {tool.name}")
+        logging.debug("TaskRunHooks on_tool_start: %s", tool.name)
         if self._on_tool_start:
             await self._on_tool_start(context, agent, tool)
 
     async def on_tool_end(
         self, context: RunContextWrapper[TContext], agent: Agent[TContext], tool: Tool, result: str
     ) -> None:
         """Called after a tool invocation completes."""
-        logging.debug(f"TaskRunHooks on_tool_end: {tool.name} ")
+        logging.debug("TaskRunHooks on_tool_end: %s", tool.name)
         if self._on_tool_end:
             await self._on_tool_end(context, agent, tool, result)
 
@@ -117,33 +117,33 @@ async def on_handoff(
         self, context: RunContextWrapper[TContext], agent: Agent[TContext], source: Agent[TContext]
     ) -> None:
         """Called when control is handed off from one agent to another."""
-        logging.debug(f"TaskAgentHooks on_handoff: {source.name} -> {agent.name}")
+        logging.debug("TaskAgentHooks on_handoff: %s -> %s", source.name, agent.name)
         if self._on_handoff:
             await self._on_handoff(context, agent, source)
 
     async def on_start(self, context: RunContextWrapper[TContext], agent: Agent[TContext]) -> None:
         """Called when the agent starts processing."""
-        logging.debug(f"TaskAgentHooks on_start: {agent.name}")
+        logging.debug("TaskAgentHooks on_start: %s", agent.name)
         if self._on_start:
             await self._on_start(context, agent)
 
     async def on_end(self, context: RunContextWrapper[TContext], agent: Agent[TContext], output: Any) -> None:
         """Called when the agent finishes processing."""
-        logging.debug(f"TaskAgentHooks on_end: {agent.name}")
+        logging.debug("TaskAgentHooks on_end: %s", agent.name)
         if self._on_end:
             await self._on_end(context, agent, output)
 
     async def on_tool_start(self, context: RunContextWrapper[TContext], agent: Agent[TContext], tool: Tool) -> None:
         """Called before a tool invocation begins."""
-        logging.debug(f"TaskAgentHooks on_tool_start: {tool.name}")
+        logging.debug("TaskAgentHooks on_tool_start: %s", tool.name)
         if self._on_tool_start:
             await self._on_tool_start(context, agent, tool)
 
     async def on_tool_end(
         self, context: RunContextWrapper[TContext], agent: Agent[TContext], tool: Tool, result: str
     ) -> None:
         """Called after a tool invocation completes."""
-        logging.debug(f"TaskAgentHooks on_tool_end: {tool.name}")
+        logging.debug("TaskAgentHooks on_tool_end: %s", tool.name)
         if self._on_tool_end:
             await self._on_tool_end(context, agent, tool, result)
 

src/seclab_taskflow_agent/mcp_lifecycle.py

@@ -97,10 +97,10 @@ def build_mcp_servers(
                 if "command" in params:
 
                     def _print_out(line: str) -> None:
-                        logging.info(f"Streamable MCP Server stdout: {line}")
+                        logging.info("Streamable MCP Server stdout: %s", line)
 
                     def _print_err(line: str) -> None:
-                        logging.info(f"Streamable MCP Server stderr: {line}")
+                        logging.info("Streamable MCP Server stderr: %s", line)
 
                     server_proc = StreamableMCPThread(
                         params["command"],
@@ -137,7 +137,7 @@ async def mcp_session_task(
     """
     try:
         for entry in entries:
-            logging.debug(f"Connecting mcp server: {entry.name}")
+            logging.debug("Connecting mcp server: %s", entry.name)
             if entry.process is not None:
                 entry.process.start()
                 await entry.process.async_wait_for_connection(poll_interval=0.1)
@@ -148,17 +148,17 @@ async def mcp_session_task(
 
         for entry in list(reversed(entries)):
             try:
-                logging.debug(f"Starting cleanup for mcp server: {entry.name}")
+                logging.debug("Starting cleanup for mcp server: %s", entry.name)
                 await entry.server.cleanup()
-                logging.debug(f"Cleaned up mcp server: {entry.name}")
+                logging.debug("Cleaned up mcp server: %s", entry.name)
                 if entry.process is not None:
                     entry.process.stop()
                     try:
                         await asyncio.to_thread(entry.process.join_and_raise)
                     except Exception as e:
-                        logging.warning(f"Streamable mcp server process exception: {e}")
+                        logging.warning("Streamable mcp server process exception: %s", e)
             except asyncio.CancelledError:
-                logging.exception(f"Timeout on cleanup for mcp server: {entry.name}")
+                logging.exception("Timeout on cleanup for mcp server: %s", entry.name)
     except RuntimeError:
         logging.exception("RuntimeError in mcp session task")
     except asyncio.CancelledError:

src/seclab_taskflow_agent/mcp_servers/codeql/client.py

@@ -487,7 +487,7 @@ def _get_source_prefix(database_path: Path, strip_leading_slash=True) -> str:
                 source_prefix = source_prefix.lstrip("/")
             return source_prefix
         except (yaml.YAMLError, FileNotFoundError, KeyError) as e:
-            logging.error(f"Error parsing sourceLocationPrefix: {e}")
+            logging.error("Error parsing sourceLocationPrefix: %s", e)
             raise
 
 

src/seclab_taskflow_agent/mcp_utils.py

@@ -166,7 +166,7 @@ def mcp_client_params(
             case "stdio":
                 env = dict(sp.env) if sp.env else None
                 args = list(sp.args) if sp.args else None
-                logging.debug(f"Initializing toolbox: {tb}\nargs:\n{args}\nenv:\n{env}\n")
+                logging.debug("Initializing toolbox: %s\nargs:\n%s\nenv:\n%s\n", tb, args, env)
                 if env:
                     for k, v in list(env.items()):
                         try:
@@ -175,11 +175,11 @@ def mcp_client_params(
                             logging.critical(e)
                             logging.info("Assuming toolbox has default configuration available")
                             del env[k]
-                logging.debug(f"Tool call environment: {env}")
+                logging.debug("Tool call environment: %s", env)
                 if args:
                     for i, v in enumerate(args):
                         args[i] = swap_env(v)
-                logging.debug(f"Tool call args: {args}")
+                logging.debug("Tool call args: %s", args)
                 server_params["command"] = sp.command
                 server_params["args"] = args
                 server_params["env"] = env
@@ -199,7 +199,7 @@ def mcp_client_params(
                 if sp.command is not None:
                     env = dict(sp.env) if sp.env else None
                     args = list(sp.args) if sp.args else None
-                    logging.debug(f"Initializing streamable toolbox: {tb}\nargs:\n{args}\nenv:\n{env}\n")
+                    logging.debug("Initializing streamable toolbox: %s\nargs:\n%s\nenv:\n%s\n", tb, args, env)
                     exe = shutil.which(sp.command)
                     if exe is None:
                         raise FileNotFoundError(f"Could not resolve path to {sp.command}")

src/seclab_taskflow_agent/prompt_parser.py

@@ -52,10 +52,10 @@ def parse_prompt_args(
         args = parser.parse_known_args(user_prompt.split(" ") if user_prompt else None)
     except SystemExit as e:
         if e.code == 2:
-            logging.exception(f"User provided incomplete prompt: {user_prompt}")
+            logging.exception("User provided incomplete prompt: %s", user_prompt)
         return None, None, None, None, "", help_msg
     except Exception:
-        logging.exception(f"Failed to parse prompt: {user_prompt}")
+        logging.exception("Failed to parse prompt: %s", user_prompt)
         return None, None, None, None, "", help_msg
     p = args[0].p.strip() if args[0].p else None
     t = args[0].t.strip() if args[0].t else None
@@ -65,7 +65,7 @@ def parse_prompt_args(
     if args[0].globals:
         for g in args[0].globals:
             if "=" not in g:
-                logging.error(f"Invalid global variable format: {g}. Expected KEY=VALUE")
+                logging.error("Invalid global variable format: %s. Expected KEY=VALUE", g)
                 return None, None, None, None, "", help_msg
             key, value = g.split("=", 1)
             cli_globals[key.strip()] = value.strip()

src/seclab_taskflow_agent/runner.py

@@ -197,14 +197,14 @@ async def _build_prompts_to_run(
             logging.critical("No last MCP tool result available")
             raise
         except json.JSONDecodeError as exc:
-            logging.critical(f"Could not parse tool result as JSON: {last_mcp_tool_results[-1][:200]}")
+            logging.critical("Could not parse tool result as JSON: %s", last_mcp_tool_results[-1][:200])
             raise ValueError("Tool result is not valid JSON") 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}")
+            logging.critical("Could not parse result text: %s", text)
             raise ValueError("Result text is not valid JSON") from exc
         try:
             iter(iterable_result)
@@ -215,7 +215,7 @@ async def _build_prompts_to_run(
         if not iterable_result:
             await render_model_output("** 🤖❗MCP tool result iterable is empty!\n")
         else:
-            logging.debug(f"Rendering templated prompts for results: {iterable_result}")
+            logging.debug("Rendering templated prompts for results: %s", iterable_result)
             for value in iterable_result:
                 try:
                     rendered_prompt = render_template(
@@ -227,7 +227,7 @@ 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}")
+                    logging.error("Error rendering template for result %s: %s", value, e)
                     raise ValueError(f"Template rendering failed: {e}")
 
         # Consume only after all prompts rendered successfully so that
@@ -408,15 +408,15 @@ async def _run_streamed() -> None:
                             rate_limit_backoff = MAX_RATE_LIMIT_BACKOFF
                         else:
                             rate_limit_backoff += rate_limit_backoff
-                        logging.exception(f"Hit rate limit ... holding for {rate_limit_backoff}")
+                        logging.exception("Hit rate limit ... holding for %s", rate_limit_backoff)
                         await asyncio.sleep(rate_limit_backoff)
 
             await _run_streamed()
             complete = True
 
         except MaxTurnsExceeded as e:
             await render_model_output(f"** 🤖❗ Max Turns Reached: {e}\n", async_task=async_task, task_id=task_id)
-            logging.exception(f"Exceeded max_turns: {max_turns}")
+            logging.exception("Exceeded max_turns: %s", max_turns)
         except AgentsException as e:
             await render_model_output(f"** 🤖❗ Agent Exception: {e}\n", async_task=async_task, task_id=task_id)
             logging.exception("Agent Exception")
@@ -576,7 +576,7 @@ async def on_handoff_hook(context: RunContextWrapper[TContext], agent: Agent[TCo
                         inputs_dict=inputs,
                     )
                 except jinja2.TemplateError as e:
-                    logging.error(f"Template rendering error: {e}")
+                    logging.error("Template rendering error: %s", e)
                     raise ValueError(f"Failed to render prompt template: {e}") from e
 
             with TmpEnv(env):
@@ -657,7 +657,7 @@ async def _deploy(ra: dict, pp: str) -> bool:
                     complete = True
                     for result in task_results:
                         if isinstance(result, Exception):
-                            logging.error(f"Caught exception in Gather: {result}")
+                            logging.error("Caught exception in Gather: %s", result)
                             result = False
                         complete = result and complete
                     return complete
@@ -690,13 +690,13 @@ async def _deploy(ra: dict, pp: str) -> bool:
                                 f"** 🤖🔄 Task {task_name!r} failed: {exc}\n"
                                 f"** 🤖🔄 Retrying in {backoff}s ({remaining} attempts left)\n"
                             )
-                            logging.warning(f"Task {task_name!r} attempt {attempt + 1} failed: {exc}")
+                            logging.warning("Task %r attempt %s failed: %s", task_name, attempt + 1, exc)
                             await asyncio.sleep(backoff)
                         else:
-                            logging.error(f"Task {task_name!r} failed after {TASK_RETRY_LIMIT} attempts: {exc}")
+                            logging.error("Task %r failed after %s attempts: %s", task_name, TASK_RETRY_LIMIT, exc)
                     except Exception as exc:
                         last_task_error = exc
-                        logging.error(f"Task {task_name!r} failed (non-retriable): {exc}")
+                        logging.error("Task %r failed (non-retriable): %s", task_name, exc)
                         break
 
                 # If all retries exhausted with an exception, save and re-raise

src/seclab_taskflow_agent/session.py

@@ -80,7 +80,7 @@ def save(self) -> Path:
         self.updated_at = datetime.now(timezone.utc).isoformat()
         path = self.file_path
         path.write_text(self.model_dump_json(indent=2))
-        logging.debug(f"Session checkpoint saved: {path}")
+        logging.debug("Session checkpoint saved: %s", path)
         return path
 
     def record_task(
@@ -132,5 +132,5 @@ def list_sessions(cls) -> list[TaskflowSession]:
             try:
                 sessions.append(cls.model_validate_json(f.read_text()))
             except Exception:
-                logging.warning(f"Skipping corrupt session file: {f}")
+                logging.warning("Skipping corrupt session file: %s", f)
         return sessions

src/seclab_taskflow_agent/shell_utils.py

@@ -18,7 +18,7 @@ def shell_command_to_string(cmd: list[str]) -> str:
     Raises:
         RuntimeError: If the command exits with a non-zero return code.
     """
-    logging.info(f"Executing: {cmd}")
+    logging.info("Executing: %s", cmd)
     p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8")
     stdout, stderr = p.communicate()
     p.wait()