[Bug]: InMemoryMessageBus.try_lock ignores ttl_secs, so a crashed holder blocks the key forever

Prerequisites

• I have searched the existing issues and discussions, and this is not a duplicate.
• This is a bug, not a usage question.

Background / Description

MessageBus.try_lock documents ttl_secs as a lease, in _base.py:373-392:

Used to make an at-least-once queue drain effectively once — the node that wins the claim for a key does the work; the others skip.

ttl_secs (int, defaults to 600): Lease duration; the claim expires automatically so a crashed holder cannot block the key forever.

RedisMessageBus implements exactly that (_redis_message_bus.py:716-720):

async def try_lock(self, key: str, *, ttl_secs: int = 600) -> bool:
    """Non-blocking claim via ``SET key NX EX``. See base."""
    return bool(
        await self._client.set(key, "1", nx=True, ex=ttl_secs),
    )

InMemoryMessageBus accepts ttl_secs and never reads it (_in_memory_message_bus.py:410-418 on main @ 0b157a31):

async def try_lock(self, key: str, *, ttl_secs: int = 600) -> bool:
    """Non-blocking claim on ``key``. See base."""
    if key in self._lock_holders:
        return False
    self._lock_holders[key] = "1"
    return True

So on the in-memory bus the claim never expires. The caller that wins a key and dies before reaching unlock — an exception on a path that does not unwind through a finally, a cancelled task, a worker that is shut down mid-drain — holds that key for the lifetime of the process. Every later try_lock on it returns False forever, so the work behind it is not "done once", it is never done again, and is_locked keeps reporting True.

This is not the documented deviation the class already carries. acquire_lock does say so explicitly ("ttl_secs is accepted for API compatibility but does not expire the lock automatically", ttl_secs: "Ignored (no automatic expiry)") and that is fine — it holds an asyncio.Lock for the duration of a context manager, so the release is structural. try_lock's docstring says only "See base", and the base is what it does not do.

The backend is documented for single-process use (local development, unit tests, examples that want to avoid a Redis dependency), and that is precisely where it hurts: code written and tested against InMemoryMessageBus behaves differently once it is pointed at Redis, and vice versa. A test that exercises a lease-expiry path passes on Redis and hangs on the in-memory bus.

Error Messages

No exception is raised; the claim is silently permanent.

Steps to Reproduce

1. Code:

import asyncio
import time

from agentscope.app.message_bus import InMemoryMessageBus


async def main() -> None:
    async with InMemoryMessageBus() as bus:
        key = "agentscope:drain:doc-42"
        print("winner claims (ttl_secs=1):", await bus.try_lock(key, ttl_secs=1))
        print("second node, same moment:  ", await bus.try_lock(key, ttl_secs=1))
        # the winner dies here: unlock() is never called
        time.sleep(1.2)
        print("is_locked after the lease: ", await bus.is_locked(key))
        print("second node retries:       ", await bus.try_lock(key, ttl_secs=1))


asyncio.run(main())

1. Run: python repro.py

2. Observed on main @ 0b157a31:

winner claims (ttl_secs=1): True
second node, same moment:   False
is_locked after the lease:  True     <- the 1 second lease has passed
second node retries:        False    <- the key is stuck for the whole process

Expected, and what the same script prints against RedisMessageBus:

is_locked after the lease:  False
second node retries:        True

Environment

• AgentScope Version: 2.0.8 (main @ 0b157a31)
• Python Version: 3.11
• OS: macOS 15 (the backend is pure Python, so this is platform independent)