Fix ECS monitoring and scroll validation issues

Update ecs create-express-gateway-service, ecs update-express-gateway-service,
and ecs delete-express-gateway-service commands to not output API response
when run with the --monitor-resources flag.

Fix scrolling bounds calculations when line wrapping is present.

Author: jutyler1

.changes/next-release/enhancement-ecs-89147.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "``ecs``",
+  "description": "Update ``ecs create-express-gateway-service``, ``ecs update-express-gateway-service``, and ``ecs delete-express-gateway-service`` commands to not output API response when run with the ``--monitor-resources`` flag. Also fixes scrolling bounds calculations when line wrapping is present."
+}

awscli/customizations/ecs/monitorexpressgatewayservice.py

@@ -182,7 +182,6 @@ class ECSExpressGatewayServiceWatcher:
         service_arn (str): ARN of the service to monitor
         mode (str): Monitoring mode - 'RESOURCE' or 'DEPLOYMENT'
         timeout_minutes (int): Maximum monitoring time in minutes (default: 30)
-        exit_hook (callable): Optional callback function on exit
     """
 
     def __init__(
@@ -191,24 +190,24 @@ def __init__(
         service_arn,
         mode,
         timeout_minutes=30,
-        exit_hook=None,
         display=None,
         use_color=True,
     ):
         self._client = client
         self.service_arn = service_arn
         self.mode = mode
         self.timeout_minutes = timeout_minutes
-        self.exit_hook = exit_hook or self._default_exit_hook
         self.last_described_gateway_service_response = None
         self.last_execution_time = 0
         self.cached_monitor_result = None
         self.start_time = time.time()
         self.use_color = use_color
         self.display = display or Display()
 
-    def _default_exit_hook(self, x):
-        return x
+    @staticmethod
+    def is_monitoring_available():
+        """Check if monitoring is available (requires TTY)."""
+        return sys.stdout.isatty()
 
     def exec(self):
         """Start monitoring the express gateway service with progress display."""
@@ -226,7 +225,13 @@ async def _execute_with_progress_async(
         """Execute monitoring loop with animated progress display."""
         spinner_chars = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
         spinner_index = 0
-        current_output = "Waiting for initial data"
+
+        # Initialize with basic service resource
+        service_resource = ManagedResource("Service", self.service_arn)
+        initial_output = service_resource.get_status_string(
+            spinner_char="{SPINNER}", use_color=self.use_color
+        )
+        current_output = initial_output
 
         async def update_data():
             nonlocal current_output

awscli/customizations/ecs/monitormutatinggatewayservice.py

@@ -171,6 +171,14 @@ def after_call(self, parsed, context, http_response, **kwargs):
         ).get('serviceArn'):
             return
 
+        # Check monitoring availability
+        if not self._watcher_class.is_monitoring_available():
+            uni_print(
+                "Monitoring is not available (requires TTY). Skipping monitoring.\n",
+                out_file=sys.stderr,
+            )
+            return
+
         if not self.session or not self.parsed_globals:
             uni_print(
                 "Unable to create ECS client. Skipping monitoring.",
@@ -188,22 +196,15 @@ def after_call(self, parsed, context, http_response, **kwargs):
         # Get service ARN from response
         service_arn = parsed.get('service', {}).get('serviceArn')
 
-        # Define exit hook to replace parsed response
-        def exit_hook(new_response):
-            if new_response:
-                parsed.clear()
-                parsed.update(new_response)
+        # Clear output when monitoring is invoked
+        parsed.clear()
 
         try:
-            # Determine if color should be used
-            use_color = self._should_use_color(self.parsed_globals)
-
             self._watcher_class(
                 ecs_client,
                 service_arn,
                 self.effective_resource_view,
-                exit_hook=exit_hook,
-                use_color=use_color,
+                use_color=self._should_use_color(self.parsed_globals),
             ).exec()
         except Exception as e:
             uni_print(

awscli/customizations/ecs/prompt_toolkit_display.py

@@ -1,4 +1,5 @@
 import asyncio
+import re
 
 from prompt_toolkit.application import Application
 from prompt_toolkit.formatted_text import ANSI
@@ -8,6 +9,12 @@
 from prompt_toolkit.widgets import Frame
 
 
+def _get_visual_line_length(line):
+    """Get the visual length of a line, excluding ANSI escape codes."""
+    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
+    return len(ansi_escape.sub('', line))
+
+
 class Display:
     def __init__(self):
         self.control = FormattedTextControl(text="")
@@ -21,6 +28,7 @@ def __init__(self):
             text="up/down to scroll, q to quit"
         )
         self.content_lines = 0
+        self.raw_text = ""
         kb = KeyBindings()
 
         @kb.add('q')
@@ -35,12 +43,24 @@ def scroll_up(event):
 
         @kb.add('down')
         def scroll_down(event):
-            window_height = (
-                event.app.output.get_size().rows - 3
-            )  # Frame top, frame bottom, status bar
-            if self.window.vertical_scroll < max(
-                0, self.content_lines - window_height
-            ):
+            # Frame top, frame bottom, status bar
+            window_height = event.app.output.get_size().rows - 3
+
+            # Account for frame borders and scrollbar
+            window_width = event.app.output.get_size().columns - 4
+            total_display_lines = 0
+
+            for line in self.raw_text.split('\n'):
+                visual_length = _get_visual_line_length(line)
+                if visual_length == 0:
+                    total_display_lines += 1
+                else:
+                    total_display_lines += max(
+                        1, (visual_length + window_width - 1) // window_width
+                    )
+
+            max_scroll = max(0, total_display_lines - window_height)
+            if self.window.vertical_scroll < max_scroll:
                 self.window.vertical_scroll += 1
 
         self.app = Application(
@@ -58,12 +78,42 @@ def scroll_down(event):
 
     def display(self, text, status_text=""):
         """Update display with ANSI colored text."""
+        self.raw_text = text
         self.control.text = ANSI(text)
         self.content_lines = len(text.split('\n'))
+
+        self._validate_scroll_position()
+
         if status_text:
             self.status_control.text = status_text
         self.app.invalidate()
 
+    def _validate_scroll_position(self):
+        """Ensure scroll position is valid for current content."""
+        if not hasattr(self.app, 'output') or not self.app.output:
+            return
+
+        try:
+            window_height = self.app.output.get_size().rows - 3
+            window_width = self.app.output.get_size().columns - 4
+
+            total_display_lines = 0
+            for line in self.raw_text.split('\n'):
+                visual_length = _get_visual_line_length(line)
+                if visual_length == 0:
+                    total_display_lines += 1
+                else:
+                    total_display_lines += max(
+                        1, (visual_length + window_width - 1) // window_width
+                    )
+
+            max_scroll = max(0, total_display_lines - window_height)
+            if self.window.vertical_scroll > max_scroll:
+                self.window.vertical_scroll = max_scroll
+        except (AttributeError, OSError):
+            # If we can't determine terminal size, leave scroll position unchanged
+            pass
+
     async def run(self):
         """Run the display app."""
         await self.app.run_async()

tests/functional/ecs/test_monitormutatinggatewayservice.py

@@ -32,8 +32,12 @@ def test_add_to_parser(self):
 
 class TestMonitorMutatingGatewayService:
     def setup_method(self):
+        self.mock_watcher_class = Mock()
+        self.mock_watcher_class.is_monitoring_available.return_value = True
         self.handler = MonitorMutatingGatewayService(
-            'create-gateway-service', 'DEPLOYMENT'
+            'create-gateway-service',
+            'DEPLOYMENT',
+            watcher_class=self.mock_watcher_class,
         )
 
     def test_init(self):

tests/unit/customizations/ecs/test_monitorexpressgatewayservice.py

@@ -105,6 +105,22 @@ def test_interactive_mode_requires_tty(self, mock_isatty, capsys):
 class TestECSExpressGatewayServiceWatcher:
     """Test the watcher class through public interface"""
 
+    @patch('sys.stdout.isatty')
+    def test_is_monitoring_available_with_tty(self, mock_isatty):
+        """Test is_monitoring_available returns True when TTY is available"""
+        mock_isatty.return_value = True
+        assert (
+            ECSExpressGatewayServiceWatcher.is_monitoring_available() is True
+        )
+
+    @patch('sys.stdout.isatty')
+    def test_is_monitoring_available_without_tty(self, mock_isatty):
+        """Test is_monitoring_available returns False when TTY is not available"""
+        mock_isatty.return_value = False
+        assert (
+            ECSExpressGatewayServiceWatcher.is_monitoring_available() is False
+        )
+
     def setup_method(self):
         self.mock_client = Mock()
         self.service_arn = (

tests/unit/customizations/ecs/test_monitormutatinggatewayservice.py

@@ -72,8 +72,12 @@ class TestMonitorMutatingGatewayService:
     """Test the event handler for monitoring gateway service mutations."""
 
     def setup_method(self):
+        self.mock_watcher_class = Mock()
+        self.mock_watcher_class.is_monitoring_available.return_value = True
         self.handler = MonitorMutatingGatewayService(
-            'create-gateway-service', 'DEPLOYMENT'
+            'create-gateway-service',
+            'DEPLOYMENT',
+            watcher_class=self.mock_watcher_class,
         )
 
     def test_init(self):
@@ -242,6 +246,7 @@ def test_after_call_successful_monitoring(self):
             'https://ecs.us-west-2.amazonaws.com'
         )
         mock_parsed_globals.verify_ssl = True
+        mock_parsed_globals.color = 'off'
         handler.session = mock_session
         handler.parsed_globals = mock_parsed_globals
 
@@ -272,18 +277,17 @@ def test_after_call_successful_monitoring(self):
             mock_client,
             service_arn,
             'DEPLOYMENT',
-            exit_hook=ANY,
             use_color=False,
         )
         mock_watcher.exec.assert_called_once()
+        # Verify parsed response was cleared
+        assert parsed == {}
 
-    def test_after_call_exception_handling(self, capsys):
-        """Test exception handling in after_call method."""
+    def test_after_call_monitoring_not_available(self, capsys):
+        """Test that monitoring is skipped when not available (no TTY)."""
         # Setup handler state
         mock_watcher_class = Mock()
-        mock_watcher = Mock()
-        mock_watcher.exec.side_effect = Exception("Test exception")
-        mock_watcher_class.return_value = mock_watcher
+        mock_watcher_class.is_monitoring_available.return_value = False
 
         handler = MonitorMutatingGatewayService(
             'create-gateway-service',
@@ -299,6 +303,7 @@ def test_after_call_exception_handling(self, capsys):
             'https://ecs.us-west-2.amazonaws.com'
         )
         mock_parsed_globals.verify_ssl = True
+        mock_parsed_globals.color = 'off'
         handler.session = mock_session
         handler.parsed_globals = mock_parsed_globals
 
@@ -309,57 +314,67 @@ def test_after_call_exception_handling(self, capsys):
         # Setup call parameters
         service_arn = 'arn:aws:ecs:us-west-2:123456789012:service/test-service'
         parsed = {'service': {'serviceArn': service_arn}}
+        original_parsed = dict(parsed)
         context = Mock()
         http_response = Mock()
         http_response.status_code = 200
 
-        # Execute - should not raise exception
+        # Execute
         handler.after_call(parsed, context, http_response)
 
+        # Verify parsed response was not cleared
+        assert parsed == original_parsed
+
+        # Verify warning message was printed
         captured = capsys.readouterr()
-        assert "Encountered an error, terminating monitoring" in captured.err
-        assert "Test exception" in captured.err
+        assert (
+            "Monitoring is not available (requires TTY). Skipping monitoring.\n"
+            in captured.err
+        )
 
-    def test_exit_hook_functionality(self):
-        """Test that exit hook properly updates parsed response."""
+    def test_after_call_exception_handling(self, capsys):
+        """Test exception handling in after_call method."""
         # Setup handler state
-        self.handler.effective_resource_view = 'DEPLOYMENT'
+        mock_watcher_class = Mock()
+        mock_watcher = Mock()
+        mock_watcher.exec.side_effect = Exception("Test exception")
+        mock_watcher_class.return_value = mock_watcher
+
+        handler = MonitorMutatingGatewayService(
+            'create-gateway-service',
+            'DEPLOYMENT',
+            watcher_class=mock_watcher_class,
+        )
+        handler.effective_resource_view = 'DEPLOYMENT'
+
         mock_session = Mock()
         mock_parsed_globals = Mock()
         mock_parsed_globals.region = 'us-west-2'
         mock_parsed_globals.endpoint_url = (
             'https://ecs.us-west-2.amazonaws.com'
         )
         mock_parsed_globals.verify_ssl = True
-        self.handler.session = mock_session
-        self.handler.parsed_globals = mock_parsed_globals
+        mock_parsed_globals.color = 'off'
+        handler.session = mock_session
+        handler.parsed_globals = mock_parsed_globals
 
         # Setup mocks
         mock_client = Mock()
         mock_session.create_client.return_value = mock_client
 
-        # Test exit hook functionality by directly calling it
+        # Setup call parameters
         service_arn = 'arn:aws:ecs:us-west-2:123456789012:service/test-service'
         parsed = {'service': {'serviceArn': service_arn}}
+        context = Mock()
+        http_response = Mock()
+        http_response.status_code = 200
 
-        # Create a simple exit hook function
-        def test_exit_hook(new_response):
-            if new_response:
-                parsed.clear()
-                parsed.update(new_response)
-
-        # Test with new response
-        new_response = {
-            'service': {'serviceArn': service_arn, 'status': 'ACTIVE'}
-        }
-        test_exit_hook(new_response)
-
-        assert parsed == new_response
+        # Execute - should not raise exception
+        handler.after_call(parsed, context, http_response)
 
-        # Test with None response (should not update)
-        original_parsed = dict(parsed)
-        test_exit_hook(None)
-        assert parsed == original_parsed
+        captured = capsys.readouterr()
+        assert "Encountered an error, terminating monitoring" in captured.err
+        assert "Test exception" in captured.err
 
     def test_events(self):
         """Test that correct events are returned for CLI integration."""