Fix ECS monitoring and scroll issues

.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,14 @@ 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 = self._calculate_display_lines(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 +68,49 @@ 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 _calculate_display_lines(self, window_width):
+        """Calculate total display lines accounting for line wrapping."""
+        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:
+                # Calculate how many display lines this text line will occupy
+                # when wrapped: ceil(visual_length / window_width)
+                # Using integer math: (visual_length + window_width - 1) // window_width
+                total_display_lines += max(
+                    1, (visual_length + window_width - 1) // window_width
+                )
+        return total_display_lines
+
+    def _validate_scroll_position(self):
+        """Ensure scroll position is valid for current content."""
+        if not getattr(self.app, 'output', None):
+            return
+
+        try:
+            window_height = self.app.output.get_size().rows - 3
+            window_width = self.app.output.get_size().columns - 4
+
+            total_display_lines = self._calculate_display_lines(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

@@ -10,6 +10,8 @@
 
 import pytest
 from botocore.exceptions import ClientError
+from prompt_toolkit.application import create_app_session
+from prompt_toolkit.output import DummyOutput
 
 from awscli.customizations.ecs.exceptions import MonitoringError
 from awscli.customizations.ecs.monitorexpressgatewayservice import (
@@ -105,12 +107,34 @@ 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.app_session = create_app_session(output=DummyOutput())
+        self.app_session.__enter__()
         self.mock_client = Mock()
         self.service_arn = (
             "arn:aws:ecs:us-west-2:123456789012:service/my-cluster/my-service"
         )
 
+    def teardown_method(self):
+        if hasattr(self, 'app_session'):
+            self.app_session.__exit__(None, None, None)
+
     def _create_watcher_with_mocks(self, resource_view="RESOURCE", timeout=1):
         """Helper to create watcher with mocked display"""
         mock_display = Mock()
@@ -724,6 +748,14 @@ def test_monitoring_error_creation(self):
 class TestColorSupport:
     """Test color support functionality"""
 
+    def setup_method(self):
+        self.app_session = create_app_session(output=DummyOutput())
+        self.app_session.__enter__()
+
+    def teardown_method(self):
+        if hasattr(self, 'app_session'):
+            self.app_session.__exit__(None, None, None)
+
     def test_should_use_color_on(self):
         """Test _should_use_color returns True when color is 'on'"""
         command = ECSMonitorExpressGatewayService(Mock())