한 번에 한 턴

@asynccontextmanager
async def turn(conversation_id: str, *, store: ConversationStore | None = None,
               origin_conversation_id: str | None = None, cancellable: bool = False):
    """FIFO admission before history/parent selection, held through persistence.

    asyncio.Lock is fair. App bindings are task-local, so an arriving message
    cannot replace the active turn's tools while it waits for admission.
    """
    store = store or ConversationStore()
    key = (str(Path(store.db_path).resolve()), conversation_id)
    lock = _turn_locks.setdefault(key, asyncio.Lock())
    task = asyncio.current_task()
    waiting = _waiting.setdefault(conversation_id, set()) if cancellable else set()
    if cancellable:
        waiting.add(task)
    from app.streaming import chief_turn
    try:
        async with AsyncExitStack() as stack:
            await stack.enter_async_context(lock)
            await stack.enter_async_context(chief_turn(conversation_id))
            waiting.discard(task)
            token = _origin.set((conversation_id, origin_conversation_id))
            try:
                yield
            finally:
                _origin.reset(token)
    finally:
        waiting.discard(task)
        if cancellable and not waiting:
            _waiting.pop(conversation_id, None)

from __future__ import annotations

import asyncio

LOCKED = True                  # set to False to watch the room tangle

rows: list[dict] = []          # the conversation: {"id", "role", "text", "parent"}
locks: dict[str, asyncio.Lock] = {}


class NoLock:
    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False


def newest_id() -> str | None:
    return rows[-1]["id"] if rows else None


async def model(text: str) -> str:
    await asyncio.sleep(0.2 if text == "from the phone" else 0.01)  # one slow reply
    return "reply to " + text


async def turn(conversation: str, text: str, client_parent: str | None) -> None:
    lock = locks.setdefault(conversation, asyncio.Lock()) if LOCKED else NoLock()
    async with lock:                          # fair: waiters enter in arrival order
        parent = newest_id()                  # the client's parent is ignored
        user = {"id": f"u{len(rows)}", "role": "user", "text": text, "parent": parent}
        rows.append(user)
        answer = await model(text)            # held through the reply...
        rows.append({"id": f"a{len(rows)}", "role": "assistant", "text": answer,
                     "parent": user["id"]})   # ...and through persistence


async def main() -> None:
    await asyncio.gather(
        turn("chief", "from the phone", client_parent=None),
        turn("chief", "from telegram", client_parent="stale"),
        turn("chief", "from the browser", client_parent="stale"),
    )
    for row in rows:
        print(f'{row["id"]:>3} <- {str(row["parent"]):>4}  {row["role"]:<9} {row["text"]}')


asyncio.run(main())