test: pass action name to dirty client and stabilize after TTOU spam

- /unlimited and /limited handlers passed the data dict where the dirty
  client expected the action (method) name, surfacing as a 500 from
  getattr(self, action) on the dirty worker. Pass 'process' as the
  action so the call routes to DirtyApp.process(data).
- TestUnlimitedApps now bumps worker count via TTIN and polls both apps
  for readiness before each test. The preceding TTOU-spam test pins the
  worker count at the LimitedTask floor (2) and the arbiter takes a
  moment to rebind apps to the surviving workers; the previous tests
  raced that rebind and saw 'No workers available'.

tests/docker/dirty_ttin_ttou/app.py

@@ -45,10 +45,14 @@ def app(environ, start_response):
         start_response('200 OK', [('Content-Type', 'text/plain')])
         return [b'OK']
 
+    # client.execute(app_path, action, *args, **kwargs) — action is the
+    # method name on the DirtyApp.  The original fixture passed the data
+    # dict where ``action`` belongs, which surfaced as a 500 from
+    # ``getattr(self, action)`` on the dirty worker.
     if path == '/unlimited':
         try:
             client = get_dirty_client()
-            result = client.execute('app:UnlimitedTask', {'test': 'data'})
+            result = client.execute('app:UnlimitedTask', 'process', {'test': 'data'})
             start_response('200 OK', [('Content-Type', 'application/json')])
             return [json.dumps(result).encode()]
         except Exception as e:
@@ -59,7 +63,7 @@ def app(environ, start_response):
     if path == '/limited':
         try:
             client = get_dirty_client()
-            result = client.execute('app:LimitedTask', {'test': 'data'})
+            result = client.execute('app:LimitedTask', 'process', {'test': 'data'})
             start_response('200 OK', [('Content-Type', 'application/json')])
             return [json.dumps(result).encode()]
         except Exception as e:

tests/docker/dirty_ttin_ttou/test_ttin_ttou_docker.py

@@ -103,6 +103,30 @@ def send_signal_to_dirty_arbiter(sig):
     )
 
 
+def wait_for_apps_ready(*paths, timeout=10):
+    """Poll the given app endpoints until each returns 200.
+
+    The dirty arbiter rebalances apps across workers asynchronously after
+    TTIN/TTOU signals.  Tests that care about app availability — rather
+    than worker counts — should call this between scaling and the request
+    so they don't race the rebalance.
+    """
+    deadline = time.time() + timeout
+    pending = list(paths)
+    while pending and time.time() < deadline:
+        for path in list(pending):
+            try:
+                resp = requests.get(f"{BASE_URL}{path}", timeout=2)
+                if resp.status_code == 200:
+                    pending.remove(path)
+            except requests.RequestException:
+                pass
+        if pending:
+            time.sleep(0.5)
+    if pending:
+        raise RuntimeError(f"Apps did not become ready: {pending}")
+
+
 class TestTTINSignal:
     """Test SIGTTIN increases dirty workers."""
 
@@ -162,14 +186,24 @@ class TestTTOUSignal:
 class TestUnlimitedApps:
     """Test apps with worker_count=None work correctly."""
 
-    def test_unlimited_app_works(self, docker_services):
+    @pytest.fixture(autouse=True)
+    def _ready(self, docker_services):
+        # The TTOU-spam test before this class may leave the arbiter at
+        # the floor (2 workers).  Bump the count back up so LimitedTask
+        # has spare capacity, then wait for both apps to be reachable.
+        for _ in range(2):
+            send_signal_to_dirty_arbiter("TTIN")
+            time.sleep(0.5)
+        wait_for_apps_ready("/unlimited", "/limited", timeout=30)
+
+    def test_unlimited_app_works(self):
         """UnlimitedTask should work."""
         resp = requests.get(f"{BASE_URL}/unlimited", timeout=10)
         assert resp.status_code == 200
         data = resp.json()
         assert data["task"] == "unlimited"
 
-    def test_limited_app_works(self, docker_services):
+    def test_limited_app_works(self):
         """LimitedTask should work."""
         resp = requests.get(f"{BASE_URL}/limited", timeout=10)
         assert resp.status_code == 200