[Refactor/Chore] Separate response-listener completion from app execution terminal state

**AI disclosure**: This issue was drafted and analyzed with Codex. I have reviewed the analysis, and I am responsible for the content.

### Self Checks

- [X] I have read the [Contributing Guide]() and [Language Policy]().
- [X] This is only for refactors or chores; if you would like to ask a question, please head to [Discussions]().
- [X] I have searched for existing issues [search for existing issues](), including closed ones.
- [X] I confirm that I am using English to submit this report, otherwise it will be closed.
- [X] 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
- [X] Please do not modify this template :) and fill in all the required fields.

### Description

`AppQueueManager` currently conflates two independent lifecycles:

1. the lifecycle of an app or workflow execution; and
2. the lifecycle of the current response-listener segment.

This became observable after client-disconnect cancellation was added in langgenius/dify#39186. That change introduced `_execution_terminal` and made `AppQueueManager.listen()` abort the underlying execution when the response generator closes before that flag is set:

```python
finally:
if not self._execution_terminal.is_set():
self._abort_execution(
"Client response stream closed before app execution completed"
)
```

Normal terminal events call `stop_listen(execution_terminal=True)`, which both closes the listener and suppresses the disconnect abort.

A paused workflow does not fit that model:

| Lifecycle | State after a workflow pause |
| -- | -- |
| Workflow execution | `PAUSED`, unfinished, and resumable |
| Current response-listener segment | Complete |

The persistence layer deliberately records `WorkflowExecutionStatus.PAUSED` with `update_finished=False`. The workflow response pipeline emits the pause response and ends the current listener segment. Before [#39485](), `QueueWorkflowPausedEvent` was not recognized by `WorkflowAppQueueManager` as an expected listener boundary, so generator cleanup was mistaken for a client disconnect and sent cancellation signals to the paused execution. This caused langgenius/dify#39448.

[#39485]() is behaviorally correct as a focused regression fix: it adds `QueueWorkflowPausedEvent` to the event-type check that closes the listener. However, the implementation must call:

```python
self.stop_listen(execution_terminal=True)
```

That statement is semantically false for a paused execution. The new concept—"the current listener segment completed normally"—exists only in the PR description and surrounding explanation, not in the lifecycle model exposed by `AppQueueManager`.

#### Current ownership problem

* `WorkflowAppQueueManager` should own the contextual mapping from workflow queue events to listener-segment boundaries.
* `AppQueueManager` should own listener mechanics and detect whether a consumer detached unexpectedly.
* GraphEngine and the persistence layer should own workflow execution state.
* An execution coordinator or cancellation port should own the policy for translating client detachment into distributed cancellation.

The event-type check in `WorkflowAppQueueManager` is therefore in the correct layer. Moving `QueueWorkflowPausedEvent` into the base queue manager, or adding a universal `is_terminal` / `ends_stream` property to the event, would leak app-specific response policy into the event model. For example, Advanced Chat performs additional message persistence before ending its response stream.

#### Proposed bounded refactor

Represent the listener lifecycle directly and remove execution-state knowledge from the queue manager:

1. Rename `_execution_terminal` to a listener-specific state such as `_listener_segment_completed`.
2. Replace `stop_listen(execution_terminal=True)` with an operation whose contract is explicit, such as `complete_listener_segment()`.
3. Remove the boolean parameter. All current production callers pass `execution_terminal=True`; the no-argument/non-terminal form is only exercised by a unit test.
4. In `listen()` cleanup, abort only when the listener segment was not completed by the producer.
5. Keep the app-specific event classification in each queue manager. In particular, `WorkflowAppQueueManager` should continue to classify `QueueWorkflowPausedEvent` as the end of its current listener segment.
6. Add concise interface documentation explaining that listener completion does not imply execution completion.

Conceptually:

```python
self._listener_segment_completed = threading.Event()

def complete_listener_segment(self) -> None:
self._listener_segment_completed.set()
self._clear_task_belong_cache()
self._q.put(None)
```

The cleanup path would then test the state it actually owns:

```python
finally:
if not self._listener_segment_completed.is_set():
self._abort_execution(
"Client response stream closed before app execution completed"
)
```

#### Acceptance criteria

* A workflow pause completes the current response-listener segment without setting the legacy stop flag or sending a GraphEngine abort command.
* Successful, partially successful, failed, errored, and explicitly stopped executions still close their listener normally without a duplicate abort.
* Closing a response generator before its producer completes the listener segment still cancels the underlying execution exactly once.
* Timeout and manual-stop behavior remains unchanged.
* Advanced Chat pause handling remains unchanged; its stream still ends only after its message-specific completion processing.
* Names and docstrings distinguish listener-segment completion from workflow execution terminality.
* Focused unit tests cover pause, normal terminal completion, unexpected consumer close, timeout, and idempotent cancellation.

#### Out of scope

A later change may inject an execution-cancellation port so that `AppQueueManager` no longer directly knows about both the legacy Redis stop flag and GraphEngine commands. That is a related boundary improvement, but it is not required for this bounded lifecycle refactor.

### Motivation

The current boolean creates a hidden cross-module invariant: every new legitimate response boundary must be added to an event-type allowlist and then described as an execution terminal state, even when the execution remains resumable.

This produces:

* **change amplification**: new pause, handoff, or reconnect behavior requires coordinated changes across event handling, listener cleanup, and cancellation;
* **cognitive load**: understanding a small event-type check requires tracing the response generator, Redis stop flag, GraphEngine command channel, and persistence status;
* **unknown unknowns**: omitting one expected listener boundary can silently turn normal control flow into a destructive distributed cancellation.

Modeling the listener lifecycle explicitly makes the common behavior obvious and prevents future resumable execution states from repeating the langgenius/dify#39448 failure mode.

### Additional Context

Related work:

* langgenius/dify#39186 introduced response-stream cleanup cancellation and the `execution_terminal` flag.
* langgenius/dify#39448 reported HITL resume being aborted after a legitimate workflow pause.
* [#39485]() adds `QueueWorkflowPausedEvent` to the current terminal-event type check as a focused regression fix.

Primary affected code:

* `api/core/app/apps/base_app_queue_manager.py`
* `api/core/app/apps/workflow/app_queue_manager.py`
* `api/core/app/apps/message_based_app_queue_manager.py`
* `api/core/app/apps/pipeline/pipeline_queue_manager.py`
* corresponding queue-manager unit tests

Start by reading api/core/app/apps/base_app_queue_manager.py and compare its listener cleanup with the workflow, message-based, and pipeline queue managers. Then inspect the corresponding queue-manager unit tests and the handling of QueueWorkflowPausedEvent. Done means listener completion is distinct from execution terminality, pause remains resumable, unexpected generator closure still cancels once, and the listed terminal, timeout, and manual-stop cases remain covered.

Written by the indexing model from the issue text.