[Bug]: Session SSE silently loses team HITL events between replay and live subscribe

### Prerequisites

- [x] I searched existing issues, discussions, and open/closed PRs. #1890 / #1918 introduced leader-session HITL projection, while #2110 concerns missing terminal events after an LLM failure; neither covers this replay/live handoff race.
- [x] This is a correctness bug, not a usage question.

### Background / Description

`GET /sessions/{session_id}/stream` reads the replay log and performs team HITL projection **before** subscribing to the live channel. An event published in that window is silently absent from the already-open SSE connection.

This affects the leader-session HITL projection introduced by #1890 / #1918; it does not dispute that feature's intended behavior.

This reproduces through the production `stream_session_events()` route function using `RedisMessageBus` over a real Redis TCP connection (`redis-py -> TCP -> Redis 6.0.16`). Redis remains healthy and accepts the event.

Expected behavior: once the endpoint starts serving a healthy session stream, it must not silently omit events published while replay or team projection is awaiting storage; replay/live overlap should not produce duplicates.

Actual behavior: the event exists in the Redis Stream and is published successfully, but the SSE client never receives it.

### Root Cause

At base commit [`30ca3ef`](https://github.com/agentscope-ai/agentscope/commit/30ca3ef33189ce11cbd04a3475e009d2b879f021), the route performs:

1. [`log_read()` replay](https://github.com/agentscope-ai/agentscope/blob/30ca3ef33189ce11cbd04a3475e009d2b879f021/src/agentscope/app/_router/_session.py#L760-L766)
2. [`projection.list()` and per-worker storage awaits](https://github.com/agentscope-ai/agentscope/blob/30ca3ef33189ce11cbd04a3475e009d2b879f021/src/agentscope/app/_router/_session.py#L776-L806)
3. [live `subscribe()`](https://github.com/agentscope-ai/agentscope/blob/30ca3ef33189ce11cbd04a3475e009d2b879f021/src/agentscope/app/_router/_session.py#L824-L827)

The HITL projector independently performs [`upsert()` followed by `publish()`](https://github.com/agentscope-ai/agentscope/blob/30ca3ef33189ce11cbd04a3475e009d2b879f021/src/agentscope/app/_service/_projectors/_subagent_hitl.py#L170-L176). Therefore normal multi-worker concurrency is sufficient:

1. SSE captures the projection list and suspends on worker A's storage query.
2. Worker B upserts a new HITL projection and publishes its event.
3. Worker B is absent from the captured projection list, and Redis reports zero subscribers at publish time.
4. The route subscribes later and waits indefinitely; Pub/Sub does not replay the missed event.

This is a replay-then-subscribe TOCTOU. Its window grows with projection/storage latency and concurrent worker activity.

### Impact

For `subagent_require_user_confirm`, the leader UI is never shown the approval card, while the worker remains parked waiting for input. The SSE connection stays open and continues heartbeats, so reconnect-based recovery is not triggered.

The base cleanup also [deletes the entire replay Stream after persistence](https://github.com/agentscope-ai/agentscope/blob/30ca3ef33189ce11cbd04a3475e009d2b879f021/src/agentscope/app/_service/_chat.py#L653-L658), making a later reconnect unable to recover the event. This is a correctness failure rather than a rendering delay.

### Error Messages

There is normally no application or Redis error; the failure is silent. A deterministic probe adds a 1.5-second observation timeout and reports:

```text
replay= data: {... "name": "warmup" ...}
pubsub_numsub= [('agentscope:session:events:leader', 0)]
entry_id= 1784353855842-0 stream_len= 2
RESULT=LOST TimeoutError after 1.5s
```

The assigned entry ID and `stream_len=2` show that Redis accepted and retained worker B's event. The timeout occurs because the still-open SSE connection never delivers it.

Invoking the exact base cleanup operation against the same real Redis backend produces:

```text
BASE_LOG_TRIM entry_id=1784353757737-0 stream_len_before_trim=1
BASE_LOG_TRIM stream_len_after_trim=0 key_exists=0
```

### Steps to Reproduce

1. Start a real Redis server and verify TCP connectivity:

```bash
sudo apt-get install -y redis-server redis-tools
sudo systemctl start redis-server
redis-cli -h 127.0.0.1 -p 6379 ping
# PONG
```

2. Create a clean worktree at the current base commit and use the repository's development environment:

```bash
git fetch upstream main
git worktree add ../agentscope-sse-repro 30ca3ef33189ce11cbd04a3475e009d2b879f021
cd ../agentscope-sse-repro
python -m pip install -e '.[service,storage]'
```

3. Save the following as `/tmp/repro_sse_handoff.py`:

```python
import asyncio

from agentscope.app._router._session import stream_session_events
from agentscope.app._service import SessionProjection, SubagentHitlProjector
from agentscope.app.message_bus import MessageBusKeys, RedisMessageBus

class Storage:
def __init__(self):
self.worker_lookup_started = asyncio.Event()
self.release_worker_lookup = asyncio.Event()

async def get_session(self, _user_id, _agent_id, session_id):
if session_id == "leader":
return object()
self.worker_lookup_started.set()
await self.release_worker_lookup.wait()
return None

def hitl(worker):
return {
"worker_session_id": worker,
"worker_agent_id": f"{worker}-agent",
"worker_agent_name": worker,
"reply_id": f"{worker}-reply",
"event_type": "require_user_confirm",
"event": {},
"created_at": "2026-07-18T00:00:00",
}

async def main():
bus = RedisMessageBus(host="127.0.0.1", port=6379, db=15)
await bus.__aenter__()
redis = bus.get_client()
await redis.flushdb()
projection = SessionProjection(bus)
storage = Storage()
key = MessageBusKeys.session_events("leader")

worker_a = hitl("worker-a")
await projection.upsert(
"leader",
SubagentHitlProjector.KIND,
SubagentHitlProjector.entry_id("worker-a", "worker-a-reply"),
worker_a,
)
await projection.publish("leader", "warmup", {"worker": "warmup"})

response = await stream_session_events(
"leader",
agent_id="leader-agent",
user_id="user",
storage=storage,
message_bus=bus,
)
stream = response.body_iterator
try:
print("replay=", (await asyncio.wait_for(anext(stream), 2)).strip())
next_event = asyncio.create_task(anext(stream))
await asyncio.wait_for(storage.worker_lookup_started.wait(), 2)
print("pubsub_numsub=", await redis.pubsub_numsub(key))

worker_b = hitl("worker-b")
await projection.upsert(
"leader",
SubagentHitlProjector.KIND,
SubagentHitlProjector.entry_id("worker-b", "worker-b-reply"),
worker_b,
)
await projection.publish(
"leader",
SubagentHitlProjector.EVT_REQUIRE,
worker_b,
)
entries = await bus.log_read(key, max_count=1000)
print("entry_id=", entries[-1][0], "stream_len=", await redis.xlen(key))
await asyncio.sleep(0.25)
storage.release_worker_lookup.set()

try:
print("live=", (await asyncio.wait_for(next_event, 1.5)).strip())
except asyncio.TimeoutError:
print("RESULT=LOST TimeoutError after 1.5s")
finally:
storage.release_worker_lookup.set()
await stream.aclose()
await redis.flushdb()
await bus.aclose()

asyncio.run(main())
```

4. Run it against the base source tree:

```bash
PYTHONPATH=src python /tmp/repro_sse_handoff.py
```

The output matches the failure shown above. The 250 ms controlled await is only a deterministic scheduling mechanism; any non-zero storage await creates the same ordering window.

For comparison, a local subscribe-first proof-of-fix produces:

```text
pubsub_numsub= [('agentscope:session:events:leader', 1)]
entry_id= 1784353864916-0 stream_len= 2
live= data: {... "name": "subagent_require_user_confirm", "value": {"worker_session_id": "worker-b", ...}}
```

The same race also reproduces with fakeredis, confirming that it is ordering logic rather than a Redis server failure.

### Proposed Scoped Fix

- Start the live subscription before replay and wait for Redis's subscribe acknowledgement through the existing `on_ready` barrier.
- Buffer live events during replay/projection and deduplicate the replay/live overlap by Redis Stream entry ID.
- Keep the scoped fix focused on the replay/live handoff; handle run-end trim and retention together in the separate cursor proposal.
- Add deterministic regression coverage for the projection window and cancellation before subscription readiness.

A longer-term follow-up can replace the Stream/Pub/Sub dual path with a single cursor-based Stream tail using SSE `id:` / `Last-Event-ID`, explicit retention-gap recovery, and bounded slow-consumer semantics. That larger protocol change is not required to close this correctness bug.

### Environment

- AgentScope Version: `2.0.4.post1` (`main` at `30ca3ef33189ce11cbd04a3475e009d2b879f021`)
- Python Version: `3.11.15`
- Redis: `6.0.16`, real TCP connection via `redis-py`; identical ordering result with fakeredis
- OS: Windows development environment with Redis on Ubuntu 22.04 / WSL2