[Bug]: interrupt during _run_impl assembly phase cancels nothing — the run completes and a full "abandoned" reply is persisted

### 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

### What I was trying to do

Call `ChatService.interrupt(user_id, session_id, agent_id)` to stop an
in-progress chat run **while the run is still in its assembly phase** — i.e. the
window between `chat_service.run(...)` spawning `_run_impl` and the run acquiring
the session lock. This window is noticeable on the **first message of a session**,
where `workspace_manager.get_workspace(...)` / `get_tts_model` / middleware / KB /
toolkit / model assembly (steps 1–6 of `_run_impl`) can take on the order of
seconds when materializing a cold workspace.

### Expected behavior

The run should be cancelled, exactly as interrupting during the locked generation
phase (step 7) does: the agent's `CancelledError` cleanup runs, the client receives
a `ReplyEndEvent(INTERRUPTED)` terminal event, and **no full reply is persisted**.
During the window the 409 guard (run already in `ChatRunRegistry`) should keep
working so a re-send is correctly rejected until the cancelled run releases its slot.

### What actually happens

`interrupt` returns normally (no exception), but the run is **not** stopped. It
proceeds to step 7, generates the complete reply, and the `finally` block's
**shielded** persistence (`_chat.py:963-985`) writes the **full reply** to storage.
The transcript gains an assistant message the user never wanted (a "ghost/abandoned"
reply). The user's interrupt intent, instead of cancelling the run, gets enqueued as
a `UserInterruptEvent` resume trigger that only runs *after* the abandoned run
finishes and releases the lock — by which time the session is idle and the trigger
no-ops.

### Root cause

`ChatService.interrupt` (`src/agentscope/app/_service/_chat.py:388`) uses
`message_bus.is_locked(session_lock)` as the **sole** "is a run active" judge
(lines 429-432):

```python
# _chat.py:429-445
if await self._message_bus.is_locked(
MessageBusKeys.session_lock(session_id),
):
await self._message_bus.publish( # cancel path
MessageBusKeys.session_interrupt_channel(),
{"session_id": session_id},
)
return

await enqueue_run_trigger( # resume-trigger path
self._message_bus,
user_id=user_id, session_id=session_id, agent_id=agent_id,
kind=MessageBusKeys.WAKEUP_KIND_RESUME,
inputs=UserInterruptEvent(reply_id=session.state.reply_id),
)
```

But `_run_impl` (`_chat.py:503`) has an assembly phase (steps 1-6, lines 524-780)
that runs **before** `acquire_lock` (step 7, line 798):

```python
# _chat.py:524- (assembly, steps 1-6)
try:
agent_record = await self._access.resolve_agent(...) # 536
session_record = await self._storage.get_session(...) # 551
workspace = await self._workspace_manager.get_workspace(...) # 556 — slow on cold build
...middlewares / TTS / KB / toolkit / model / assemble...
except Exception as e: # 781 — catches only Exception
async with self._message_bus.acquire_lock(...): # 786 — only the failure path
await self._report_failure(...)
return
# step 7:
async with self._message_bus.acquire_lock( # 798 — lock acquired here
lock_key, ttl_secs=MessageBusKeys.SESSION_RUN_TTL_SECS,
):
... actual generation ...
```

During assembly `is_locked` is `False`, so `interrupt` takes the **resume-trigger**
branch and never publishes on `session_interrupt_channel`. The assembly task is
therefore never cancelled.

### Key point: the cancel path *can* cancel an in-assembly task — the gap is branch selection, not capability

`CancelDispatcher._interrupt_session`
(`src/agentscope/app/_manager/_cancel_dispatcher.py:227-246`) keys off the
**registry** (the in-process asyncio task handle), **not** `is_locked`:

```python
# _cancel_dispatcher.py:227-246
def _interrupt_session(self, session_id: str) -> None:
task = self._registry.get(session_id)
if task is not None and not task.done():
logger.info("CancelDispatcher: interrupting local chat run for session %s", session_id)
task.cancel()
```

The assembly task is already in the registry (it occupies the 409 slot), so
publishing on the interrupt channel *would* cancel it. And an assembly-phase
`CancelledError` **does not persist** — `reply_msg` is not yet bound, and
`CancelledError` is a `BaseException` not caught by the `except Exception` at line
781, so it bubbles out with no `finally` upsert (`reply_msg is None`). The
capability is present; `interrupt` simply chooses not to invoke it when
`is_locked` is `False`.

### Why the full reply survives

```python
# _chat.py:963-985 (finally, inside the step-7 lock)
async def _persist() -> None:
if reply_msg is not None:
await self._storage.upsert_message(user_id, session_id, reply_msg) # full reply
await self._storage.update_session_state(...)
await self._message_bus.log_trim(events_key)
persist_task = asyncio.create_task(_persist())
try:
await asyncio.shield(persist_task) # shielded — survives outer cancel
except asyncio.CancelledError:
await persist_task # await persistence, then propagate
```

Even a cancellation that arrives in step 7 is shielded so the full `reply_msg`
lands before the lock is released. For the assembly-phase case the run isn't
cancelled at all, so the complete reply is persisted as a matter of course.

### Suggested fix directions (least → most invasive)

1. **Judge by "active run", not only `is_locked`** (recommended, targeted). Before
falling to the resume-trigger branch, check whether a run is active: in-process,
`ChatRunRegistry.get(session_id)` (task not done) → treat as active and publish on
`session_interrupt_channel` (same as the `is_locked=True` branch). For multi-replica
deployments the registry is per-process, so a distributed "run alive" marker is
needed (set at spawn, **before** `acquire_lock`; cleared at task done) and the judge
becomes `is_locked OR run_alive`. This realizes the intent of "lock acquired at
spawn" via a lightweight marker instead of actually holding the lock early. The
assembly task is already in the registry, so the cancel reaches it, and an
assembly-phase cancel persists nothing (see above).
2. **Synthesize an INTERRUPTED reply on assembly-phase cancel** (complements #1).
Add `except asyncio.CancelledError:` to the assembly try block (parallel to the
`except Exception` at line 781; `CancelledError` is not caught by `Exception`).
Mirror `_report_failure` (`_chat.py:315-386`) but emit
`finished_reason=ReplyFinishedReason.INTERRUPTED` (no `error` field) with
`AssistantMsg(content=[])`, publish + upsert best-effort, then `raise` to honour
asyncio semantics. Gives the client a clean terminal event and an empty (not full)
assistant message — the same shape as an early mid-reply cancel. `_report_failure`
should not be parameterized for this (it carries `ERROR` + `error` semantics); use a
separate sibling. Only meaningful once #1 lets the cancel reach the assembly task.
3. **Move `acquire_lock` to `_run_impl` entry** (thorough, most invasive). Acquire the
lock before the assembly try so `is_locked` is `True` throughout assembly and the
existing judge works unchanged. Caveats: the lock is non-reentrant (Redis `SET NX` /
in-memory `asyncio.Lock`), so the nested `acquire_lock` at lines 786-790 (the
assembly failure path) must be removed (call `_report_failure` directly — lock already
held); confirm no assembly step takes `session_lock` (lines 786-790 are the only
re-entry point). Lock is held longer (cold `get_workspace` build under lock); rely on
heartbeat (TTL/2) renewal, and let the build emit heartbeats or raise
`SESSION_RUN_TTL_SECS` if cold builds can exceed it. Direction #2 is still needed for
the terminal event.

### Boundaries / notes

- The **409 busy guard** ("already has an active chat run") is normal concurrency
protection (in-process registry + Redis distributed lock, two layers) and is **not**
part of this bug; it continues to work during the affected window.
- **`404 ⇔ no run`**: when the session does not exist, `trigger_chat_run` itself 404s
before spawning, so a 404 cannot coexist with in-flight generation. (A 404 observed
here belongs to a different path — session ownership / id resolution — and is unrelated
to the persisted-reply issue.)
- A **downstream platform cannot cleanly fix this**: if it merely "publishes cancel after
an `is_locked` timeout", it **loses the user message** — the Case A user-input upsert
(`_chat.py:823-834`) happens inside the step-7 lock, *after* assembly, so cancelling
the assembly task never reaches it. The root-cause fix belongs upstream (interrupt
judge / lock timing).

### Error Messages

```shell
**None.** This is a silent logic/race bug: `interrupt()` returns normally, no exception
is raised and nothing is logged as an error. The run simply isn't cancelled, runs to
completion, and the full reply is persisted by the shielded `finally`. The only
observable artifact is an unexpected assistant message in the session transcript and
the absence of a `REPLY_END(INTERRUPTED)` terminal event on the event stream.
```

### Steps to Reproduce

The trigger is timing-based: `interrupt` must land during the assembly phase (before
`acquire_lock` at `_chat.py:798`). To make it deterministic, slow down
`workspace_manager.get_workspace` so the assembly phase lasts long enough to interrupt
into. Sketch (adapt the fakes to your test harness; the essential point is a slow
`get_workspace` so the interrupt lands before the lock appears):

1. Code:

```python
import asyncio
import agentscope # 2.0.6
from agentscope.app.message_bus import InMemoryMessageBus
from agentscope.app.message_bus._keys import MessageBusKeys
# ...your ChatService construction: storage / workspace_manager /
# resource_access / ChatRunRegistry / WakeupDispatcher / CancelDispatcher...

class SlowWorkspaceManager:
"""Fake that makes the assembly phase deterministic-slow."""
async def get_workspace(self, user_id, agent_id, session_id, *a, **kw):
await asyncio.sleep(5) # hold the run in assembly (pre-lock)
return await self._real.get_workspace(user_id, agent_id, session_id, *a, **kw)

# --- wire SlowWorkspaceManager into ChatService ---

# 1. spawn the run (first user message → cold/slow assembly)
run_task = asyncio.create_task(
chat_service.run(user_id, agent_id, session_id, user_msg)
)

# 2. while the run is still assembling (is_locked is False), interrupt
await asyncio.sleep(1) # ensure we're inside the 5s assembly window
assert not await message_bus.is_locked(MessageBusKeys.session_lock(session_id))
await chat_service.interrupt(user_id, session_id, agent_id) # returns normally

# 3. let the un-cancelled run finish
await run_task

# 4. observe: an unwanted FULL assistant reply was persisted
messages = await storage.list_messages(user_id, agent_id, session_id)
assistant_replies = [m for m in messages if m.name == agent_id]
assert assistant_replies and assistant_replies[-1].content # ← BUG: full abandoned reply

# and the event log has NO REPLY_END(INTERRUPTED) — the interrupt never took effect
```

2. Run: `python repro_assembly_interrupt.py`
3. See: `interrupt()` returns, the run completes, a full assistant message is
persisted, no `REPLY_END(INTERRUPTED)` is emitted.

**Control / expected (interrupt during step-7 locked generation)**: if you move the
`interrupt` call to *after* `is_locked` becomes `True` (i.e. past `acquire_lock` at
line 798), the cancel path fires, the agent's `CancelledError` cleanup runs, an
`INTERRUPTED` terminal event is emitted, and the full reply is not persisted — this is
the behaviour the assembly-phase case should match.

### Environment

- AgentScope Version: 2.0.6
- Python Version: 3.11.15
- OS: Linux 6.2.0-39-generic x86_64