Agent/agent/execution/trace_id.py

"""
Trace ID 生成和解析工具

提供 Trace ID 的生成、解析等功能。

Trace ID 格式：
- 主 Trace: {uuid} (标准 UUID)
- Sub-Trace: {parent_id}@{mode}-{timestamp}-{seq}
  例如: 2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001
"""

import uuid
from datetime import datetime
from threading import Lock
from typing import Dict, Optional


# 全局计数器（线程安全）
_seq_lock = Lock()
_seq_counter: Dict[str, int] = {}  # key: "{parent_id}@{mode}-{timestamp}"


def generate_trace_id() -> str:
    """
    生成主 Trace ID

    Returns:
        标准 UUID 字符串
    """
    return str(uuid.uuid4())


def generate_sub_trace_id(parent_id: str, mode: str) -> str:
    """
    生成 Sub-Trace ID

    格式: {parent_id}@{mode}-{timestamp}-{seq}

    使用完整的 parent_id（不截断），避免 ID 冲突风险。
    同一秒内多次调用会递增序号。

    Args:
        parent_id: 父 Trace ID（完整 UUID）
        mode: 运行模式（explore, delegate, compaction 等）

    Returns:
        Sub-Trace ID

    Examples:
        >>> generate_sub_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", "explore")
        '2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001'

        >>> generate_sub_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d", "delegate")
        '2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@delegate-20260204220030-001'
    """
    # 直接使用完整 UUID，不截断
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")

    # 生成序号（同一秒内递增）
    prefix = f"{parent_id}@{mode}-{timestamp}"
    with _seq_lock:
        seq = _seq_counter.get(prefix, 0) + 1
        _seq_counter[prefix] = seq

    return f"{prefix}-{seq:03d}"


def parse_parent_trace_id(trace_id: str) -> Optional[str]:
    """
    从 trace_id 解析出 parent_trace_id

    Args:
        trace_id: Trace ID

    Returns:
        父 Trace ID，如果是主 Trace 则返回 None

    Examples:
        >>> parse_parent_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001")
        '2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d'

        >>> parse_parent_trace_id("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d")
        None
    """
    if '@' in trace_id:
        return trace_id.split('@')[0]
    return None


def is_sub_trace(trace_id: str) -> bool:
    """
    判断是否为 Sub-Trace

    Args:
        trace_id: Trace ID

    Returns:
        True 表示是 Sub-Trace，False 表示是主 Trace

    Examples:
        >>> is_sub_trace("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001")
        True

        >>> is_sub_trace("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d")
        False
    """
    return '@' in trace_id


def extract_mode(trace_id: str) -> Optional[str]:
    """
    从 Sub-Trace ID 中提取运行模式

    Args:
        trace_id: Trace ID

    Returns:
        运行模式（explore, delegate 等），如果是主 Trace 则返回 None

    Examples:
        >>> extract_mode("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@explore-20260204220012-001")
        'explore'

        >>> extract_mode("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d@delegate-20260204220030-001")
        'delegate'

        >>> extract_mode("2f8d3a1c-4b6e-4f9a-8c2d-1e5b7a9f3c4d")
        None
    """
    if '@' not in trace_id:
        return None

    # 格式: parent@mode-timestamp-seq
    parts = trace_id.split('@')[1]  # "mode-timestamp-seq"
    mode = parts.split('-')[0]
    return mode


def reset_seq_counter():
    """
    重置序号计数器

    主要用于测试，生产环境不应调用此函数。
    """
    global _seq_counter
    with _seq_lock:
        _seq_counter.clear()