fix: address code scanning findings from PR review

- fix uninitialized args and inconsistent return shape in prompt_parser
- add explicit returns after exhaustive match statements in capi
- narrow BaseException to Exception in mcp_transport thread
- add comment for empty except in cleanup shutdown path
- iterate over copy in mcp_lifecycle cleanup to avoid mutation during iteration
- remove unused _ENGINE_SETTING_KEYS constant from runner
- remove unused import in test_session
- default TaskDefinition.name/description to empty string for per-index naming
- use explicit re-export pattern (as X) in __main__ and cli

Author: anticomputer

src/seclab_taskflow_agent/__main__.py

@@ -18,8 +18,9 @@
 load_dotenv(find_dotenv(usecwd=True))
 
 # Re-export for backwards compatibility — some tests import from __main__
-from .prompt_parser import parse_prompt_args  # noqa: E402, F401
-from .runner import deploy_task_agents, run_main  # noqa: E402, F401
+from .prompt_parser import parse_prompt_args as parse_prompt_args  # noqa: E402
+from .runner import deploy_task_agents as deploy_task_agents  # noqa: E402
+from .runner import run_main as run_main  # noqa: E402
 
 if __name__ == "__main__":
     from .cli import app

src/seclab_taskflow_agent/capi.py

@@ -37,8 +37,7 @@ def to_url(self) -> str:
                 return f"https://{self}/inference"
             case AI_API_ENDPOINT_ENUM.AI_API_OPENAI:
                 return f"https://{self}/v1"
-            case _:
-                raise ValueError(f"Unsupported endpoint: {self}")
+        raise ValueError(f"Unsupported endpoint: {self}")
 
 
 COPILOT_INTEGRATION_ID = "vscode-chat"
@@ -114,25 +113,18 @@ def list_capi_models(token: str) -> dict[str, dict]:
 def supports_tool_calls(model: str, models: dict[str, dict]) -> bool:
     """Check whether the given model supports tool calls."""
     api_endpoint = get_AI_endpoint()
-    match urlparse(api_endpoint).netloc:
+    netloc = urlparse(api_endpoint).netloc
+    match netloc:
         case AI_API_ENDPOINT_ENUM.AI_API_GITHUBCOPILOT:
             return models.get(model, {}).get("capabilities", {}).get("supports", {}).get("tool_calls", False)
         case AI_API_ENDPOINT_ENUM.AI_API_MODELS_GITHUB:
             return "tool-calling" in models.get(model, {}).get("capabilities", [])
         case AI_API_ENDPOINT_ENUM.AI_API_OPENAI:
-            # OpenAI doesn't expose capabilities in the models list
-            # Check if model name indicates function calling support
-            model_lower = model.lower()
-            return any(
-                [
-                    "gpt-" in model_lower,
-                ]
-            )
-        case _:
-            raise ValueError(
-                f"Unsupported Model Endpoint: {api_endpoint}\n"
-                f"Supported endpoints: {[e.to_url() for e in AI_API_ENDPOINT_ENUM]}"
-            )
+            return "gpt-" in model.lower()
+    raise ValueError(
+        f"Unsupported Model Endpoint: {api_endpoint}\n"
+        f"Supported endpoints: {[e.to_url() for e in AI_API_ENDPOINT_ENUM]}"
+    )
 
 
 def list_tool_call_models(token: str) -> dict[str, dict]:

src/seclab_taskflow_agent/cli.py

@@ -178,4 +178,4 @@ def main(
 # Legacy compatibility shim — implementation moved to prompt_parser.py
 # ---------------------------------------------------------------------------
 
-from .prompt_parser import parse_prompt_args  # noqa: F401, E402
+from .prompt_parser import parse_prompt_args as parse_prompt_args  # noqa: E402

src/seclab_taskflow_agent/mcp_lifecycle.py

@@ -145,7 +145,7 @@ async def mcp_session_task(
         connected.set()
         await cleanup.wait()
 
-        for entry in reversed(entries):
+        for entry in list(reversed(entries)):
             try:
                 logging.debug(f"Starting cleanup for mcp server: {entry.server._name}")
                 await entry.server.cleanup()
@@ -158,8 +158,6 @@ async def mcp_session_task(
                         logging.warning(f"Streamable mcp server process exception: {e}")
             except asyncio.CancelledError:
                 logging.exception(f"Timeout on cleanup for mcp server: {entry.server._name}")
-            finally:
-                entries.remove(entry)
     except RuntimeError:
         logging.exception("RuntimeError in mcp session task")
     except asyncio.CancelledError:

src/seclab_taskflow_agent/mcp_transport.py

@@ -166,7 +166,7 @@ def run(self) -> None:
             if self.exit_code not in _EXPECTED_EXIT_CODES:
                 self.exception = subprocess.CalledProcessError(self.exit_code, self.cmd)
 
-        except BaseException as e:
+        except Exception as e:
             self.exception = e
 
     def _read_stream(
@@ -248,7 +248,7 @@ async def cleanup(self, *args: Any, **kwargs: Any) -> None:
         try:
             asyncio.run_coroutine_threadsafe(super().cleanup(*args, **kwargs), self.t.loop).result()
         except asyncio.CancelledError:
-            pass
+            pass  # Swallow cancellation during cleanup shutdown
         finally:
             self.t.loop.stop()
             self.t.join()

src/seclab_taskflow_agent/models.py

@@ -81,8 +81,8 @@ class TaskDefinition(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    name: str = "taskflow"
-    description: str = "taskflow"
+    name: str = ""
+    description: str = ""
     agents: list[str] = Field(default_factory=list)
     user_prompt: str = ""
     run: str = ""

src/seclab_taskflow_agent/prompt_parser.py

@@ -22,7 +22,7 @@
 
 def parse_prompt_args(
     available_tools: AvailableTools, user_prompt: str | None = None
-) -> tuple[str | None, str | None, bool, dict[str, str], str, str] | tuple[None, None, None, None, str]:
+) -> tuple[str | None, str | None, bool, dict[str, str], str, str] | tuple[None, None, None, None, str, str]:
     """Legacy CLI parser kept for backwards compatibility with tests.
 
     Returns:
@@ -53,7 +53,7 @@ def parse_prompt_args(
     except SystemExit as e:
         if e.code == 2:
             logging.exception(f"User provided incomplete prompt: {user_prompt}")
-            return None, None, None, None, help_msg
+        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
     list_models = args[0].l

src/seclab_taskflow_agent/runner.py

@@ -119,10 +119,6 @@ def _merge_reusable_task(
     return TaskDefinition.model_validate(merged)
 
 
-# Keys in model_settings that are handled by the engine, not ModelSettings.
-_ENGINE_SETTING_KEYS = {"api_type", "endpoint", "token"}
-
-
 def _resolve_task_model(
     task: TaskDefinition,
     model_keys: list[str],

tests/test_session.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from seclab_taskflow_agent.session import CompletedTask, TaskflowSession, session_dir
+from seclab_taskflow_agent.session import CompletedTask, TaskflowSession
 
 
 class TestTaskflowSession: