# A2A跨设备Trace存储方案

**问题：** 现有Trace存储在本地文件系统，跨设备时如何访问？

## 场景分析

### 场景1：云端Agent调用终端Agent

```
云端Agent (云端存储)
    ↓ 调用
终端Agent (终端存储)
    ↓ 返回 sub_trace_id
云端Agent 想 continue_from sub_trace_id
    ↓ 问题：sub_trace在终端，云端访问不到！
```

### 场景2：终端Agent调用云端Agent

```
终端Agent (终端存储)
    ↓ 调用
云端Agent (云端存储)
    ↓ 返回 sub_trace_id
终端Agent 想 continue_from sub_trace_id
    ↓ 问题：sub_trace在云端，终端访问不到！
```

## 方案对比

### 方案1：远程Trace访问（推荐）

#### 核心思想

**Trace ID包含位置信息，通过HTTP API访问远程Trace**

#### Trace ID格式

```
本地Trace: abc-123
远程Trace: agent://terminal-agent-456/abc-123
          ^^^^^^^^ ^^^^^^^^^^^^^^^^^ ^^^^^^^
          协议      Agent地址          本地ID
```

#### 实现

```python
# agent/trace/remote_store.py

class RemoteTraceStore:
    """远程Trace存储代理"""

    def __init__(self, agent_url: str, api_key: str):
        self.agent_url = agent_url
        self.api_key = api_key

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        """通过HTTP API获取远程Trace"""
        response = await self._get(f"/api/traces/{trace_id}")
        if response:
            return Trace(**response)
        return None

    async def get_main_path_messages(
        self, trace_id: str, head_sequence: int
    ) -> List[Message]:
        """获取远程Trace的消息"""
        response = await self._get(
            f"/api/traces/{trace_id}/messages",
            params={"mode": "main_path", "head_sequence": head_sequence}
        )
        return [Message(**m) for m in response["messages"]]

    async def add_message(self, message: Message) -> str:
        """向远程Trace追加消息（续跑）"""
        response = await self._post(
            f"/api/traces/{message.trace_id}/messages",
            data=message.to_dict()
        )
        return response["message_id"]


# agent/trace/hybrid_store.py

class HybridTraceStore:
    """混合存储：本地 + 远程"""

    def __init__(self, local_store: FileSystemTraceStore):
        self.local_store = local_store
        self.remote_stores = {}  # agent_url -> RemoteTraceStore

    def _parse_trace_id(self, trace_id: str) -> tuple[str, str]:
        """
        解析Trace ID

        返回: (location, local_id)
        - location: "local" 或 agent_url
        - local_id: 本地Trace ID
        """
        if trace_id.startswith("agent://"):
            # agent://terminal-agent-456/abc-123
            parts = trace_id[8:].split("/", 1)
            return parts[0], parts[1]
        else:
            return "local", trace_id

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        """获取Trace（自动路由到本地或远程）"""
        location, local_id = self._parse_trace_id(trace_id)

        if location == "local":
            return await self.local_store.get_trace(local_id)
        else:
            # 远程Trace
            if location not in self.remote_stores:
                # 创建远程存储代理
                self.remote_stores[location] = RemoteTraceStore(
                    agent_url=f"https://{location}",
                    api_key=self._get_api_key(location)
                )
            return await self.remote_stores[location].get_trace(local_id)

    async def add_message(self, message: Message) -> str:
        """添加消息（自动路由）"""
        location, local_id = self._parse_trace_id(message.trace_id)

        if location == "local":
            return await self.local_store.add_message(message)
        else:
            # 向远程Trace追加消息
            return await self.remote_stores[location].add_message(message)
```

#### agent工具修改

```python
@tool(description="创建 Agent 执行任务")
async def agent(
    task: Union[str, List[str]],
    continue_from: Optional[str] = None,  # 支持远程Trace ID
    agent_url: Optional[str] = None,      # 新增：远程Agent地址
    context: Optional[dict] = None,
) -> Dict[str, Any]:
    """
    创建 Agent 执行任务

    Args:
        task: 任务描述
        continue_from: 继续已有trace（支持远程Trace ID）
        agent_url: 远程Agent地址（如果调用远程Agent）
        context: 框架注入的上下文
    """
    store = context.get("store")  # HybridTraceStore

    if agent_url:
        # 调用远程Agent
        result = await _call_remote_agent(agent_url, task, continue_from)
        # 返回远程Trace ID
        remote_trace_id = f"agent://{agent_url}/{result['sub_trace_id']}"
        return {
            **result,
            "sub_trace_id": remote_trace_id,
            "remote": True
        }
    else:
        # 本地执行（现有逻辑）
        if continue_from:
            # 可能是远程Trace，HybridStore会自动处理
            existing = await store.get_trace(continue_from)
            if not existing:
                return {"status": "failed", "error": "Trace not found"}

        # ... 现有逻辑
```

#### 使用示例

```python
# 云端Agent调用终端Agent
result1 = await agent(
    task="分析本地项目",
    agent_url="terminal-agent-456.local"
)
# 返回: {"sub_trace_id": "agent://terminal-agent-456.local/abc-123"}

# 继续对话（自动访问远程Trace）
result2 = await agent(
    task="重点分析core模块",
    continue_from=result1["sub_trace_id"],  # 远程Trace ID
    agent_url="terminal-agent-456.local"
)
# HybridStore自动通过HTTP API访问远程Trace
```

#### 优势

1. **透明** - Agent不需要关心Trace在哪里
2. **灵活** - 支持本地和远程Trace
3. **简单** - 只需要在Trace ID中编码位置信息
4. **兼容** - 现有本地Trace不受影响

#### 劣势

1. **网络延迟** - 访问远程Trace需要HTTP请求
2. **依赖网络** - 远程Agent必须在线

---

### 方案2：中心化Trace存储

#### 核心思想

**所有Agent共享同一个Trace存储（数据库）**

#### 架构

```
云端Agent ──┐
            ├──> 中心化Trace存储（PostgreSQL/MongoDB）
终端Agent ──┘
```

#### 实现

```python
# agent/trace/db_store.py

class DatabaseTraceStore:
    """数据库Trace存储"""

    def __init__(self, db_url: str):
        self.db = connect(db_url)

    async def create_trace(self, trace: Trace) -> str:
        await self.db.traces.insert_one(trace.to_dict())
        return trace.trace_id

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        doc = await self.db.traces.find_one({"trace_id": trace_id})
        if doc:
            return Trace(**doc)
        return None

    # ... 其他方法
```

#### 配置

```yaml
# config/storage.yaml
trace_store:
  type: database
  url: postgresql://user:pass@db.cloud/traces
  # 或
  url: mongodb://db.cloud/traces
```

#### 优势

1. **简单** - 所有Agent访问同一个存储，无需特殊处理
2. **一致性** - 数据强一致性
3. **查询能力** - 可以跨Trace查询和分析

#### 劣势

1. **依赖中心** - 需要中心化数据库
2. **网络依赖** - 终端Agent必须能访问数据库
3. **隐私问题** - 终端数据存储在云端
4. **迁移成本** - 需要从文件系统迁移到数据库

---

### 方案3：Trace同步/缓存

#### 核心思想

**按需同步远程Trace到本地**

#### 实现

```python
class CachedRemoteStore:
    """带缓存的远程存储"""

    def __init__(self, local_store, remote_url):
        self.local_store = local_store
        self.remote_url = remote_url
        self.cache = {}  # trace_id -> Trace

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        # 1. 检查本地
        trace = await self.local_store.get_trace(trace_id)
        if trace:
            return trace

        # 2. 检查缓存
        if trace_id in self.cache:
            return self.cache[trace_id]

        # 3. 从远程获取并缓存
        trace = await self._fetch_remote(trace_id)
        if trace:
            self.cache[trace_id] = trace
            # 可选：持久化到本地
            await self.local_store.create_trace(trace)
        return trace

    async def add_message(self, message: Message) -> str:
        # 同时写入本地和远程
        local_id = await self.local_store.add_message(message)
        await self._sync_to_remote(message)
        return local_id
```

#### 优势

1. **性能** - 本地缓存减少网络请求
2. **离线能力** - 缓存后可以离线访问

#### 劣势

1. **一致性** - 缓存可能过期
2. **复杂度** - 需要处理同步和冲突
3. **存储开销** - 需要额外的本地存储

---

### 方案4：混合模式（推荐）

#### 核心思想

**结合方案1和方案2的优点**

- **组织内部**：使用中心化存储（数据库）
- **跨组织**：使用远程访问（HTTP API）

#### 架构

```
组织内部：
云端Agent ──┐
            ├──> 组织数据库
职能Agent ──┘

跨组织：
组织A Agent <──HTTP API──> 组织B Agent
```

#### 配置

```yaml
# config/storage.yaml
trace_store:
  # 组织内部使用数据库
  internal:
    type: database
    url: postgresql://org-db/traces

  # 跨组织使用远程访问
  external:
    type: remote
    agents:
      - url: https://partner-org.com
        api_key: xxx
```

#### 优势

1. **灵活** - 根据场景选择最优方案
2. **性能** - 组织内部低延迟
3. **隐私** - 跨组织数据不共享存储

---

## 推荐方案

### MVP阶段（Phase 1-2）：方案1 - 远程Trace访问

**理由：**
1. **最小改动** - 只需要添加RemoteTraceStore和HybridStore
2. **灵活** - 支持任意拓扑
3. **隐私友好** - 数据不离开设备
4. **渐进式** - 可以逐步迁移到方案4

**实现步骤：**
1. 实现RemoteTraceStore（通过HTTP API访问）
2. 实现HybridTraceStore（路由到本地或远程）
3. 修改agent工具支持agent_url参数
4. 在Trace ID中编码位置信息

### 长期（Phase 3+）：方案4 - 混合模式

**理由：**
1. **组织内部高效** - 数据库存储，低延迟
2. **跨组织灵活** - HTTP API，保护隐私
3. **可扩展** - 支持复杂场景

## 实现细节

### 1. Trace ID格式

```python
# 本地Trace
"abc-123"

# 远程Trace（完整格式）
"agent://terminal-agent-456.local:8000/abc-123"

# 远程Trace（简化格式，使用注册的agent_id）
"@terminal-agent-456/abc-123"
```

### 2. Agent注册表

```yaml
# config/agents.yaml
agents:
  terminal-agent-456:
    url: https://terminal-agent-456.local:8000
    api_key: ak_xxx
    type: terminal
```

### 3. API端点

```python
# 必需的API端点（用于远程访问）
GET  /api/traces/{trace_id}                    # 获取Trace
GET  /api/traces/{trace_id}/messages           # 获取消息
POST /api/traces/{trace_id}/run                # 续跑
POST /api/traces/{trace_id}/messages           # 追加消息
```

### 4. 认证和授权

```python
# 每个Agent有自己的API Key
headers = {
    "Authorization": f"Bearer {api_key}"
}

# 权限检查
if trace.uid != request.user_id:
    raise Forbidden("Cannot access other user's trace")
```

### 5. 性能优化

```python
# 批量获取消息
GET /api/traces/{trace_id}/messages?limit=100&offset=0

# 增量同步
GET /api/traces/{trace_id}/messages?since_sequence=50

# 压缩传输
headers = {"Accept-Encoding": "gzip"}
```

## 总结

**推荐路线：**

1. **Phase 1（MVP）** - 实现方案1（远程Trace访问）
   - 最小改动
   - 快速验证跨设备A2A
   - 2-3周

2. **Phase 2** - 优化和增强
   - 添加缓存
   - 批量API
   - 性能优化
   - 2-3周

3. **Phase 3（可选）** - 迁移到方案4（混合模式）
   - 组织内部使用数据库
   - 跨组织使用远程访问
   - 4-6周

**关键优势：**
- 渐进式实现
- 最小化风险
- 保持灵活性
