Agent/gateway/core/channels/backends/memory_trace.py

from __future__ import annotations

import asyncio
import uuid
from typing import Any


class MemoryTraceBackend:
    """进程内 (channel, user_id) → trace_id；接入 Lifecycle 后可替换为 TraceManager。"""

    def __init__(self) -> None:
        self._map: dict[tuple[str, str], str] = {}
        self._lock = asyncio.Lock()

    async def get_or_create_trace(
        self,
        *,
        channel: str,
        user_id: str,
        workspace_id: str,
        agent_type: str,
        metadata: dict[str, Any],
    ) -> str:
        _ = workspace_id, agent_type, metadata
        key = (channel, user_id)
        async with self._lock:
            if key not in self._map:
                self._map[key] = str(uuid.uuid4())
            return self._map[key]

    def clear(self) -> None:
        self._map.clear()