[Bug]: Resume triggers (HITL/external tool results) can be duplicated or permanently lost in multi-instance deployments

### Prerequisites

- [x] I have searched the existing [issues](https://github.com/agentscope-ai/agentscope/issues) and [discussions](https://github.com/agentscope-ai/agentscope/discussions), and this is not a duplicate.
- [x] This is a bug, not a usage question. (For questions, please use [Discussions](https://github.com/agentscope-ai/agentscope/discussions/new?category=general) instead.)

### Background / Description

### Summary

In a multi-instance deployment, **resume** triggers — carrying `UserConfirmResultEvent`, `ExternalExecutionResultEvent`, or `UserInterruptEvent` — can be **duplicated or permanently lost** due to the non-atomic consumption semantics of the shared wakeup queue. Unlike `wake` triggers (addressed by #1868 / #2206 via an empty-inbox guard), resume events carry irreplaceable data (HITL approval decisions, external tool execution results) that cannot be recovered by "just skipping if empty" and cannot be safely applied more than once.

This is a **data correctness** defect distinct from #1868: #1868 is about wasted LLM calls from duplicate wake-ups; this issue is about **lost or duplicated tool side-effects** from resume events.

### Root cause

The shared wakeup queue (`agentscope:wakeups`) is consumed via `queue_drain` ([`_redis_message_bus.py:228-267`](https://github.com/agentscope-ai/agentscope/blob/9d1026fa/src/agentscope/app/message_bus/_redis_message_bus.py#L228-L267)), which performs `XRANGE` followed by `XDEL` — a two-step, non-atomic operation. The `WakeupDispatcher` ([`_wakeup_dispatcher.py`](https://github.com/agentscope-ai/agentscope/blob/9d1026fa/src/agentscope/app/_manager/_wakeup_dispatcher.py)) in every API process subscribes to the same `agentscope:wakeup_signal` pub/sub channel and drains the same stream key. This violates the "single-consumer-per-key" invariant documented in the `queue_drain` docstring.

Three failure modes exist for resume events:

**Failure 1 — Duplicate delivery (both processes read before either deletes)**

```
Process A: XRANGE → sees entry E (resume with tool result)
Process B: XRANGE → sees entry E (same resume)
Process A: XDEL E → success
Process A: _dispatch_one → spawns run with tool result
Process B: XDEL E → no-op (already deleted)
Process B: _dispatch_one → spawns ANOTHER run with the same tool result
```

Result: tool side-effects **can be** executed twice (e.g., a file write, an API call, a database mutation confirmed by the user via HITL).

**The distributed session lock does not fully prevent this duplication.** `ChatService._run_impl` ([`_chat.py:469-696`](https://github.com/agentscope-ai/agentscope/blob/9d1026fa/src/agentscope/app/_service/_chat.py#L469-L696)) loads `session_record.state` and constructs the stateful agent (Steps 1–5) **before** acquiring the session lock (Step 7, [line 714](https://github.com/agentscope-ai/agentscope/blob/9d1026fa/src/agentscope/app/_service/_chat.py#L714)). Two processes can therefore load the same parked state, then acquire the lock **sequentially** and both apply the same resume event against stale snapshots — the second run does not see the first run's mutations because it loaded state before the first run committed.

**Failure 2 — Lost after delete, before dispatch**

```
Process A: XRANGE → sees entry E
Process A: XDEL E → success (entry removed from stream)
Process A: *** OOM kill / segfault / pod eviction ***
```

Result: resume event **permanently lost**. The user sees the session hang indefinitely — the HITL confirmation they submitted never reaches the agent. No recovery mechanism exists.

**Failure 3 — Retry task lost on crash**

`_schedule_resume_retry` ([`_wakeup_dispatcher.py:323-375`](https://github.com/agentscope-ai/agentscope/blob/9d1026fa/src/agentscope/app/_manager/_wakeup_dispatcher.py#L323-L375)) stores the retry as an in-process `asyncio.Task`. If the process crashes, the retry (and the resume data it carries) is lost with it.

### Impact

| Scenario | Duplicate delivery | Lost delivery |
|----------|-------------------|---------------|
| User confirms a destructive tool call (e.g., `rm`, `DROP TABLE`) | May be executed **twice** against stale state | User waits forever; session stuck |
| External executor returns tool result | Agent processes result twice, may hallucinate or corrupt state | Result permanently lost; session cannot proceed |
| User sends interrupt to a parked session | Two interrupt handlers race | Interrupt lost; session remains parked |

The empty-inbox guard (#2206) does **not** help here because resume triggers carry `input_msg != None` — the guard explicitly skips them.

### Proposed solution

**Phase 1 — Redis Streams consumer group for resume triggers**

Replace the `XRANGE`+`XDEL` pattern for resume events with a proper consumer group protocol on a **dedicated** stream key (e.g. `agentscope:wakeups:resume:v2`). Old `XRANGE`/`XDEL` consumers and new group consumers must **never** share a key — an old consumer can `XDEL` entries that are already in the new group's PEL, silently losing them.

1. Create a consumer group on the dedicated resume stream key.
2. Each `WakeupDispatcher` instance joins the group as a named consumer (`XREADGROUP GROUP dispatchers BLOCK ... STREAMS key >`).
3. Each new entry is initially assigned to **one** consumer in the group. Delivery remains **at-least-once** because pending entries may be reclaimed via `XAUTOCLAIM` (e.g., when the original consumer crashes), so application-level idempotency is still required.
4. `XACK` only after either:
- the resume has been handed off to a **durable** run/inbox record; or
- the resume has completed and the updated session state, reply message, and idempotency marker have been durably committed.

Creating an `asyncio.Task` or spawning `ChatService.run` is **not** a durable handoff.
5. Crashed consumers' pending entries are reclaimed via `XAUTOCLAIM` with a configurable `min-idle-time`.
6. While an entry is actively being processed, the consumer must periodically refresh its claim idle time (e.g., via `XCLAIM` with the current entry ID to reset the idle counter). Otherwise, a legitimate long-running resume (e.g., waiting on an LLM call or external tool execution that exceeds `min-idle-time`) may be concurrently reclaimed by another consumer via `XAUTOCLAIM`, leading to duplicate processing even without a crash.

**Phase 2 — Application-layer idempotency key**

Even with at-least-once delivery from consumer groups (redelivery after crash/reclaim), the application must reject duplicates:

0. **Reload state inside the lock.** Before deduplication and application, acquire the session lock and reload the latest mutable session state inside the lock. A stateful agent built from a pre-lock snapshot must not be reused for resume processing — this is part of fixing Failure 1's stale-state problem.

1. Derive a stable idempotency key per resume event:

```
{session_id}:{reply_id}:{resume_event_id}
```

Resume events already contain `EventBase.id`; callers must preserve that ID across retries. Alternatively, the API should accept a stable `Idempotency-Key` header. Store a **payload hash** alongside the key, so reuse of the same key with a different payload is rejected as a conflict (409).

> **Note:** `{session_id}:{reply_id}:{event_type}` is insufficient — the same `reply_id` may legitimately receive multiple `ExternalExecutionResultEvent`s (one per offloaded tool call), all sharing the same `event_type`. The discriminator must be the event's own stable ID.

2. Before applying a resume event, check a durable idempotency record **within the session's persistence boundary**. The idempotency marker and updated session state must be committed atomically where the storage backend supports transactions (e.g., within the same SQL transaction if session state is in PostgreSQL). A separate Redis SET is insufficient when session state is stored in another backend — a crash between the two writes can leave them inconsistent.
3. If already present, log a duplicate-rejection metric and `XACK` immediately. If absent, persist the key atomically with applying the event.

**Phase 3 — Separate channels by reliability requirement**

- `wake` / message triggers: remain on the current best-effort `XRANGE`+`XDEL` path (tolerable because #2206 makes them idempotent via the empty-inbox guard).
- `resume` triggers: routed to the reliable consumer-group channel.

This avoids forcing the heavier protocol on triggers that don't need it.

**Guarantee boundary**

Consumer groups and resume-level deduplication provide **effectively-once resume application** at the session state layer. However, preventing duplicate **external** side effects also requires tool-call-level idempotency, for example using `session_id + reply_id + resume_event_id + tool_call_id` as the effect key. Without idempotency support from the external system itself, strict exactly-once side effects cannot be guaranteed. This issue scopes its acceptance criteria to resume-level deduplication; tool-call-level idempotency may be addressed separately.

**Migration**

- Producers (`enqueue_run_trigger`) route by `kind`: `wake` → existing key, `resume` → new dedicated key.
- Rollback: stop routing new resume events to the v2 key (revert producer routing), but keep the v2 consumer group running until both the stream and its PEL are fully drained. Only after all in-flight entries are processed and ACKed, remove the v2 consumer group and stream key. Do **not** replay v2 entries back into the old unreliable queue.

### Acceptance criteria

- [ ] With N >= 2 API instances sharing one Redis, a single resume event (HITL confirm / external tool result) is applied to the session state **exactly once**.
- [ ] Force-killing a process at each of the following critical boundaries does not lose or duplicate the resume event — it is eventually processed by a surviving consumer:
- After `XREADGROUP`, before processing begins
- After session lock acquisition, before state mutation
- After session state and idempotency marker are committed, before `XACK` (verifies that redelivery hits the idempotent reject path and ACKs cleanly)
- [ ] A consumer that goes offline for an extended period has its pending entries reclaimed by healthy consumers via `XAUTOCLAIM`.
- [ ] A healthy resume run that lasts longer than `min-idle-time` is **not** concurrently reclaimed by another consumer (active claim heartbeat prevents spurious reclaim).
- [ ] Submitting the same resume event twice (e.g., user double-clicks confirm, or `XAUTOCLAIM` redelivers) results in idempotent handling: one state application, one log/metric entry noting the duplicate.
- [ ] Observable metrics/logs: group lag (`XINFO GROUPS`), PEL size/oldest idle time (`XPENDING`), redelivery count, and idempotent-reject count.

### Related issues

- #1868 — duplicate wake-up execution (same root cause, wake-side manifestation)
- #2206 — fix for #1868 via empty-inbox guard (does not cover resume)
- #1870 — earlier PR attempt for #1868 (stale)
- #1722 — multi-process / distributed deployment roadmap
- #1848 — scheduler manager distributed refactor
- #2223 — scheduler N× execution per replica

### Error Messages

```shell
# No single error message — the defect is silent.
# Failure 1 (duplicate): two ChatService.run invocations with the same
# resume event.id appear in logs on different instances; the tool
# side-effect executes twice.
#
# Failure 2 (loss): session remains in PARKED state indefinitely after
# the user submits confirmation; no error logged because the process
# that held the entry died before dispatch.
#
# Failure 3 (retry loss): same as Failure 2 but the entry was
# re-queued into an asyncio.Task that died with the process.
```

### Steps to Reproduce

The race window between `XRANGE` and `XDEL` is narrow under normal load, making natural reproduction timing-dependent. The following deterministic and stress-test approaches confirm the defect:

**Approach A — Injected barrier (deterministic, test-only):**

```python
# 1. Deploy 2 app instances with RedisMessageBus pointing at the same Redis.
# 2. Patch queue_drain in one instance to insert a sleep(1) between
# XRANGE and XDEL.
# 3. Create a session, trigger a tool call that parks for HITL confirmation.
# 4. Submit the UserConfirmResultEvent via POST /chat/.
# 5. Assert that both instances invoke ChatService.run with input_msg
# carrying the same event.id. This deterministically demonstrates
# duplicate delivery.
# 6. To deterministically demonstrate duplicate *application*
# (stale-state problem), add a second barrier: after both runs load
# session_record.state but before either acquires the session lock,
# then release both. The second run applies the resume against state
# that does not reflect the first run's mutations.
```

**Approach B — Tool invocation counter (stress/integration test):**

```python
# 1. Register a tool that increments a persistent counter on each
# invocation.
# 2. Trigger resume in a multi-instance deployment.
# 3. Observe that the counter *may reach 2* with the current
# implementation. The fixed implementation must keep it at 1.
```

**Approach C — Crash simulation (deterministic, Failure 2):**

```python
# 1. Single instance. Patch queue_drain to call os._exit(1) immediately
# after XDEL returns.
# 2. Submit a resume event.
# 3. Assert: the resume entry is gone from the stream, the session
# remains parked indefinitely, and no mechanism recovers the lost
# event.
```

### Environment

- AgentScope commit: main branch at `9d1026fa`
- Python Version: 3.11+
- OS: macOS
- Deployment: any multi-instance topology sharing a single Redis