Flaky loop e2e tests: EndWorker fails on in-flight RPC replies during region termination

### What happened?

`LoopIntegrationSpec` fails intermittently in `amber-integration` (roughly 1 in 10 runs, both OSes), with region terminations logging:

```
java.lang.IllegalStateException: worker still has unprocessed messages
at ...promisehandlers.EndHandler.endWorker(EndHandler.scala:51)
[WARN] [RegionExecutionManager] Failed to terminate region N on attempt 1 of 150. Retrying in 200 ms.
```

This is the same fail-fast race previously seen in `DataProcessingSpec` (#5614). Loops amplify it: every loop iteration terminates and re-executes regions, so each iteration is another roll of the dice — and the suite's single retry doesn't save runs where it strikes twice.

### Root cause

The "unprocessed" message is always a **`ReturnInvocation`** — the coordinator's reply to an RPC the *worker itself* sent fire-and-forget (e.g. `workerExecutionCompleted`, `portCompleted`). The race is inherent: that very RPC is what makes the coordinator decide to end the worker, so that reply and `EndWorker` are concurrently in flight. The worker never awaits these replies; a reply-only backlog carries no work.

(Terminology note: these are RPC **returns/replies** (`ReturnInvocation`), not acks — `NetworkAck` is a separate transport-layer mechanism carrying flow-control credits.)

| Side | Behavior at `EndWorker` with a queued reply |
|---|---|
| Scala `EndHandler` | throws `IllegalStateException` → termination retry loop → flaky/slow CI |
| Python `EndWorkerHandler` | **consumes one message as a side effect of logging it**, then `assert empty` — accidentally tolerates exactly one reply; with two queued it crashes, and if the queued message were real work it would be silently swallowed |

### Expected behavior

`EndWorker` should distinguish replies from work: a reply-only backlog is safe to leave at termination (warn + proceed); any other queued message (control invocation, data, ECM, actor command) must still fail loudly so the coordinator's retry lets the worker drain it — **without** consuming it.

Start with EndHandler.scala:51 and the Python EndWorkerHandler, then reproduce the race in LoopIntegrationSpec in amber-integration and compare with DataProcessingSpec. Done means reply-only ReturnInvocation backlog is warned about and allowed during termination, while other queued messages remain unconsumed and still fail loudly.

Written by the indexing model from the issue text.