MAINT: Restore scenario-declared CLI parameters (pyrit_scan + pyrit_shell)

The frontend refactor regressed scenario-declared parameter handling on
both CLI entry points - flags like `--max-turns 7` were rejected before
the second pass could fetch metadata from the server.

pyrit_scan
- `parse_args` now uses `parse_known_args` (pass 1 is tolerant of
  scenario-specific flags) and stashes leftovers + the raw arg list on
  the Namespace for the second pass.
- `_reparse_with_scenario_params` threads the original `args` list
  instead of reading `sys.argv[1:]`, so explicit-args callers work.
- `main` does a strict re-parse when there are unknown args but no
  scenario is specified, preserving the original `exit 2` error for
  truly invalid flags.

pyrit_shell
- `do_run` now fetches `get_scenario_async` first, builds
  `Parameter` objects from the response's `supported_parameters`,
  and threads `declared_params` into `parse_run_arguments`.
- Calls `extract_scenario_args` and propagates `scenario_params`
  on the REST request payload.

Tests
- New `TestScenarioParamFlow` (4 tests) and `TestShellScenarioParamFlow`
  (4 tests) regression cases covering forward, invalid-flag, no-params,
  and metadata-fetch failure paths.

Diff coverage remains at 96%.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: rlundeen2

pyrit/cli/pyrit_scan.py

@@ -283,8 +283,14 @@ def _extract_scenario_args(*, parsed: Namespace) -> dict[str, Any]:
 
 def parse_args(args: Optional[list[str]] = None) -> Namespace:
     """
-    Parse command-line arguments (pass 1 only — scenario-specific flags
-    are added via a second parse after fetching scenario metadata from server).
+    Parse command-line arguments (pass 1 — tolerant of scenario-declared flags).
+
+    Pass 1 uses ``parse_known_args`` so scenario-specific flags (e.g.
+    ``--max-turns 7``) don't cause an error before we've had a chance to
+    fetch the scenario's declared parameters from the server. The unknown
+    leftovers are stashed on the returned Namespace as ``_unknown_args``
+    so :func:`_reparse_with_scenario_params` can detect truly unknown flags
+    when no scenario was specified.
 
     Args:
         args: Argument list (``sys.argv[1:]`` when None).
@@ -293,7 +299,10 @@ def parse_args(args: Optional[list[str]] = None) -> Namespace:
         Namespace: Parsed command-line arguments.
     """
     parser = _build_base_parser(add_help=True)
-    return parser.parse_args(args)
+    parsed, unknown = parser.parse_known_args(args)
+    parsed._unknown_args = unknown
+    parsed._raw_args = list(args) if args is not None else list(sys.argv[1:])
+    return parsed
 
 
 async def _resolve_server_url_async(*, parsed_args: Namespace) -> str | None:
@@ -443,17 +452,32 @@ def _reparse_with_scenario_params(
     *, parsed_args: Namespace, supported_params: list[dict[str, Any]]
 ) -> Namespace | None:
     """
-    Re-parse ``sys.argv`` with scenario-declared flags added to the base parser.
+    Re-parse the original args with scenario-declared flags added to the base parser.
+
+    The original argument list is read from ``parsed_args._raw_args`` (populated
+    by :func:`parse_args`). If no scenario-declared parameters are supplied but
+    pass 1 left unknown args behind, surface the error now via strict re-parse.
 
     Returns:
         Namespace | None: The re-parsed Namespace, or ``None`` on argparse ``SystemExit``.
     """
+    raw_args: list[str] = getattr(parsed_args, "_raw_args", sys.argv[1:] if len(sys.argv) > 1 else [])
+
     if not supported_params:
-        return parsed_args
+        unknown = getattr(parsed_args, "_unknown_args", None)
+        if not unknown:
+            return parsed_args
+        # Re-parse strictly so argparse prints the standard "unrecognized arguments" error
+        strict_parser = _build_base_parser(add_help=True)
+        try:
+            return strict_parser.parse_args(raw_args)
+        except SystemExit:
+            return None
+
     pass2_parser = _build_base_parser(add_help=True)
     _add_scenario_params_from_api(parser=pass2_parser, params=supported_params)
     try:
-        return pass2_parser.parse_args(sys.argv[1:] if len(sys.argv) > 1 else [])
+        return pass2_parser.parse_args(raw_args)
     except SystemExit:
         return None
 
@@ -682,6 +706,16 @@ def main(args: Optional[list[str]] = None) -> int:
     except SystemExit as e:
         return e.code if isinstance(e.code, int) else 1
 
+    # If there are leftover unknown flags AND no scenario was specified,
+    # there's no chance for pass 2 to recognize them - fail loudly now.
+    unknown = getattr(parsed_args, "_unknown_args", [])
+    if unknown and not parsed_args.scenario_name:
+        strict_parser = _build_base_parser(add_help=True)
+        try:
+            strict_parser.parse_args(parsed_args._raw_args)
+        except SystemExit as e:
+            return e.code if isinstance(e.code, int) else 1
+
     logging.basicConfig(level=parsed_args.log_level)
 
     return asyncio.run(_run_async(parsed_args=parsed_args))

pyrit/cli/pyrit_shell.py

@@ -242,16 +242,40 @@ def do_run(self, line: str) -> None:
             print("Usage: run <scenario_name> --target <name> [options]")
             return
 
-        from pyrit.cli._cli_args import parse_run_arguments
+        from pyrit.cli._cli_args import extract_scenario_args, parse_run_arguments
         from pyrit.cli._output import (
             print_scenario_result_async,
             print_scenario_run_progress,
             print_scenario_run_summary,
         )
+        from pyrit.common.parameter import Parameter
+
+        # Fetch scenario metadata so the parser recognizes scenario-declared flags.
+        scenario_name_token = line.split(maxsplit=1)[0]
+        declared_params: list[Parameter] | None = None
+        try:
+            scenario_meta = asyncio.run(self._api_client.get_scenario_async(scenario_name=scenario_name_token))
+        except Exception as exc:
+            print(f"Error fetching scenario metadata: {exc}")
+            return
+        if scenario_meta is None:
+            print(f"Error: Scenario '{scenario_name_token}' not found on server.")
+            return
+        supported = scenario_meta.get("supported_parameters") or []
+        if supported:
+            declared_params = [
+                Parameter(
+                    name=p["name"],
+                    description=p.get("description", ""),
+                    param_type=str,
+                    default=p.get("default"),
+                )
+                for p in supported
+            ]
 
         # Parse arguments
         try:
-            args = parse_run_arguments(args_string=line, declared_params=None)
+            args = parse_run_arguments(args_string=line, declared_params=declared_params)
         except ValueError as e:
             print(f"Error: {e}")
             return
@@ -294,6 +318,10 @@ def do_run(self, line: str) -> None:
         if args.get("memory_labels"):
             request["labels"] = args["memory_labels"]
 
+        scenario_params = extract_scenario_args(parsed=args)
+        if scenario_params:
+            request["scenario_params"] = scenario_params
+
         # Start run
         total_strategies = len(request.get("strategies") or [])
         print(f"\nRunning scenario: {scenario_name}")

tests/unit/cli/test_pyrit_scan.py

@@ -623,3 +623,75 @@ def test_main_add_initializer_server_disabled(self, mock_client_class, _mock_pro
         result = pyrit_scan.main(["--add-initializer", str(script)])
         assert result == 1
         assert "disabled" in capsys.readouterr().out
+
+
+class TestScenarioParamFlow:
+    """Regression tests for scenario-declared parameters flowing through the CLI."""
+
+    @staticmethod
+    def _build_mock_client(supported_params=None, status="COMPLETED"):
+        from unittest.mock import AsyncMock
+
+        client = AsyncMock()
+        client.list_scenarios_async.return_value = {"items": [{"scenario_name": "foo"}]}
+        client.get_scenario_async.return_value = {
+            "scenario_name": "foo",
+            "supported_parameters": supported_params or [],
+        }
+        client.start_scenario_run_async.return_value = {"scenario_result_id": "rid", "status": "CREATED"}
+        client.get_scenario_run_async.return_value = {"scenario_result_id": "rid", "status": status}
+        client.get_scenario_run_results_async.return_value = {"items": []}
+        client.close_async = AsyncMock()
+        client.__aenter__ = AsyncMock(return_value=client)
+        client.__aexit__ = AsyncMock(return_value=None)
+        return client
+
+    @patch("pyrit.cli._server_launcher.ServerLauncher.probe_health_async", new_callable=AsyncMock, return_value=True)
+    @patch("pyrit.cli.api_client.PyRITApiClient")
+    @patch("pyrit.cli._output.print_scenario_result_async", new_callable=AsyncMock)
+    @patch("pyrit.cli._output.print_scenario_run_progress")
+    def test_scenario_declared_flag_is_forwarded(self, _mock_prog, _mock_print, mock_client_class, _mock_probe):
+        client = self._build_mock_client(supported_params=[{"name": "max_turns", "description": "..."}])
+        mock_client_class.return_value = client
+
+        result = pyrit_scan.main(["foo", "--target", "t", "--max-turns", "7"])
+
+        assert result == 0
+        sent_request = client.start_scenario_run_async.call_args.kwargs["request"]
+        assert sent_request["scenario_params"] == {"max_turns": "7"}
+
+    @patch("pyrit.cli._server_launcher.ServerLauncher.probe_health_async", new_callable=AsyncMock, return_value=True)
+    @patch("pyrit.cli.api_client.PyRITApiClient")
+    @patch("pyrit.cli._output.print_scenario_result_async", new_callable=AsyncMock)
+    @patch("pyrit.cli._output.print_scenario_run_progress")
+    def test_unknown_flag_after_valid_scenario_errors(
+        self, _mock_prog, _mock_print, mock_client_class, _mock_probe
+    ):
+        client = self._build_mock_client(supported_params=[{"name": "max_turns", "description": "..."}])
+        mock_client_class.return_value = client
+
+        result = pyrit_scan.main(["foo", "--target", "t", "--max-turns", "7", "--unknown-flag"])
+
+        assert result == 1
+        client.start_scenario_run_async.assert_not_called()
+
+    @patch("pyrit.cli._server_launcher.ServerLauncher.probe_health_async", new_callable=AsyncMock, return_value=True)
+    @patch("pyrit.cli.api_client.PyRITApiClient")
+    @patch("pyrit.cli._output.print_scenario_result_async", new_callable=AsyncMock)
+    @patch("pyrit.cli._output.print_scenario_run_progress")
+    def test_no_scenario_params_passes_through_cleanly(self, _mock_prog, _mock_print, mock_client_class, _mock_probe):
+        client = self._build_mock_client(supported_params=[])
+        mock_client_class.return_value = client
+
+        result = pyrit_scan.main(["foo", "--target", "t"])
+
+        assert result == 0
+        sent_request = client.start_scenario_run_async.call_args.kwargs["request"]
+        assert "scenario_params" not in sent_request
+
+    def test_parse_args_tolerates_scenario_specific_flags(self):
+        # Pass 1 must not error on scenario-declared flags (they're recognized in pass 2).
+        parsed = pyrit_scan.parse_args(["foo", "--target", "t", "--max-turns", "7"])
+        assert parsed.scenario_name == "foo"
+        assert parsed.target == "t"
+        assert parsed._unknown_args == ["--max-turns", "7"]

tests/unit/cli/test_pyrit_shell.py

@@ -21,6 +21,8 @@ def mock_api_client():
     client.list_initializers_async.return_value = {"items": [], "pagination": {"total": 0}}
     client.list_targets_async.return_value = {"items": [], "pagination": {"total": 0}}
     client.list_scenario_runs_async.return_value = {"items": []}
+    # Default: scenario fetch returns no declared params (back-compat for older tests)
+    client.get_scenario_async.return_value = {"scenario_name": "foo", "supported_parameters": []}
     client.close_async = AsyncMock()
     client.__aenter__ = AsyncMock(return_value=client)
     client.__aexit__ = AsyncMock(return_value=None)
@@ -607,3 +609,49 @@ def test_stop_server_close_client_swallows_errors(self, shell):
         client.close_async = AsyncMock(side_effect=RuntimeError("ignored"))
         s.do_stop_server("")
         assert s._api_client is None
+
+
+class TestShellScenarioParamFlow:
+    """Regression tests: shell.do_run must forward scenario-declared parameters."""
+
+    def test_run_passes_scenario_declared_params(self, shell):
+        s, client = shell
+        client.get_scenario_async.return_value = {
+            "scenario_name": "foo",
+            "supported_parameters": [{"name": "max_turns", "description": "..."}],
+        }
+        client.start_scenario_run_async = AsyncMock(return_value={"scenario_result_id": "rid", "status": "CREATED"})
+        client.get_scenario_run_async = AsyncMock(return_value={"scenario_result_id": "rid", "status": "COMPLETED"})
+        client.get_scenario_run_results_async = AsyncMock(return_value={"items": []})
+
+        with (
+            patch("pyrit.cli._output.print_scenario_result_async", new_callable=AsyncMock),
+            patch("pyrit.cli._output.print_scenario_run_progress"),
+            patch("time.sleep"),
+        ):
+            s.do_run("foo --target t --max-turns 7")
+
+        sent_request = client.start_scenario_run_async.call_args.kwargs["request"]
+        assert sent_request["scenario_params"] == {"max_turns": "7"}
+
+    def test_run_metadata_fetch_failure_aborts(self, shell, capsys):
+        s, client = shell
+        client.get_scenario_async = AsyncMock(side_effect=RuntimeError("net down"))
+        s.do_run("foo --target t")
+        assert "Error fetching scenario metadata" in capsys.readouterr().out
+
+    def test_run_unknown_scenario_aborts(self, shell, capsys):
+        s, client = shell
+        client.get_scenario_async.return_value = None
+        s.do_run("foo --target t")
+        assert "not found on server" in capsys.readouterr().out
+
+    def test_run_unknown_flag_for_scenario_with_declared_params_errors(self, shell, capsys):
+        s, client = shell
+        client.get_scenario_async.return_value = {
+            "scenario_name": "foo",
+            "supported_parameters": [{"name": "max_turns", "description": "..."}],
+        }
+        s.do_run("foo --target t --not-a-real-flag x")
+        captured = capsys.readouterr().out
+        assert "Unknown argument" in captured or "Error" in captured