EndWorker acknowledges termination while unprocessed data sits in per-channel FIFO queues (potential tuple loss)

### What happened?

`EndWorker` can acknowledge termination while unprocessed **data** sits in the worker's per-channel FIFO queues, letting the coordinator `gracefulStop` the actor and silently drop tuples.

The check in `EndHandler` inspects only the raw arrival queue (`dp.inputManager.inputMessageQueue`). But the DP thread's main loop drains that queue wholesale into per-channel `AmberFIFOChannel.fifoQueue`s (step 1, `DPThread.scala:141-160`) **before** it processes a single message (step 3), so a `DataFrame` that arrived slightly before `EndWorker` is invisible to the check:

```
arrival queue: [ EndWorker ] <- checked by EndHandler (empty after step-1 drain => "success")
channel queues: [ DataFrame, ... ] <- never checked; dropped when the actor is stopped
```

On success the coordinator runs `removeActorRef` + `removeControlChannel` + `gracefulStop` (`RegionExecutionManager.scala:190-208`), so any data still queued in a channel is lost. This is the false-negative half of the predicate; the false-positive half (a queued coordinator `ReturnInvocation` failing healthy teardowns) was fixed in #6916, which deliberately left this direction unchanged because it is a behavior change in the strict direction.

A fix needs care: the predicate must cover **data** channels only (e.g. `!inputGateway.getAllDataChannels.exists(_.hasMessage)`) — on the control channel, the reply to the `portCompleted` that triggered termination legitimately sits behind `EndWorker` by construction (see #6916). It should also consider out-of-order stashed messages (`AmberFIFOChannel.ofoMap`) and half-consumed state (`inputManager.hasUnfinishedInput` / `outputManager.hasUnfinishedOutput`), and needs an integration soak to confirm no paused/backpressured-channel scenario can make region termination fail all 150 attempts.

### How to reproduce?

Unit-level (deterministic, follows the `EndHandlerSpec` harness): build a `DataProcessor` over an empty arrival queue, push a message into a data channel via `dp.inputGateway.getChannel(dataChannelId).acceptMessage(WorkflowFIFOMessage(dataChannelId, 0, DataFrame(...)))`, then call `endWorker` — it succeeds even though a data message is queued. In production the window is the race between an upstream flush and region completion.

### Version/Branch

main (observed at 429be110a7; discovered during the investigation for #6916).

Start with EndHandlerSpec and the cited DPThread.scala:141-160 and RegionExecutionManager.scala:190-208; trace how EndWorker checks queues before termination. Inspect AmberFIFOChannel.fifoQueue, ofoMap, and the unfinished input/output state, then run the deterministic harness and an integration soak covering paused or backpressured channels. Done means termination cannot acknowledge while data remains queued or half-consumed, without breaking healthy teardowns.

Written by the indexing model from the issue text.