rpi-server 0.2.0

# rpi-server

TCP JSONL server for rpi with streaming subscriptions — 基于 **Transport trait** 抽象，支持多种传输协议。

## 特性

- ✅ **TCP JSONL 协议** — 简单、高效、易于调试
- ✅ **Transport trait 抽象** — 可扩展 WebSocket/Unix Socket 等传输方式
- ✅ **流式订阅** — 通过 `subscribe` 方法订阅实时事件流
- ✅ **会话管理** — 支持多个 rpi 会话并行
- ✅ **连接管理** — 内置重试、心跳、延迟检测、在线状态追踪
- ✅ **标准化** — JSON-RPC 2.0 消息格式，跨语言兼容

## 架构

```
┌─────────────────────────────────────────┐
│         Transport trait                 │
│  ┌─────────────────────────────────┐   │
│  │   TcpTransport (已实现)          │   │
│  │   - TCP JSONL 协议               │   │
│  │   - 服务间通信首选               │   │
│  └─────────────────────────────────┘   │
│  ┌─────────────────────────────────┐   │
│  │   WebSocketTransport (未来)      │   │
│  │   - 浏览器/Web 客户端            │   │
│  │   - HTTP 升级协议                │   │
│  └─────────────────────────────────┘   │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│      ManagedConnection                  │
│  - 自动重试（最多 10 次）               │
│  - 心跳检测（30s 间隔）                 │
│  - 延迟测量                             │
│  - 在线状态追踪                         │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│         AppState + Sessions             │
│  - 会话生命周期管理                     │
│  - 子进程管理                           │
│  - 事件广播                             │
└─────────────────────────────────────────┘
```

## 安装

```bash
cargo build -p rpi-server --release
```

## 使用方式

### 独立运行

```bash
# 启动服务器（默认 127.0.0.1:9800）
rpi-server

# 无 TUI 模式（明确指定）
rpi-server --server

# 指定端口
rpi-server --port 8080

# 传递 rpi 参数
rpi-server -- --provider anthropic --model claude-3-sonnet
```

### 作为 rpi 插件

```json
// 启动服务器
{ "action": "start", "port": 9800 }

// 响应
{
  "id": "rpi-rpc-1",
  "address": "127.0.0.1:9800",
  "token": "a1b2c3d4...",
  "state": "running",
  "protocol": "jsonl",
  "transport": "tcp"
}
```

### 客户端连接

使用任何 TCP 客户端连接到服务器，发送 JSON-RPC 2.0 格式的请求：

```bash
# 使用 nc 测试
nc 127.0.0.1 9800

# 发送启动会话请求
{"jsonrpc":"2.0","id":1,"method":"start_session","params":{}}

# 响应
{"jsonrpc":"2.0","id":1,"result":{"sessionId":"session-1","session":{...}}}
```

### 使用 ManagedConnection（推荐）

```rust
use std::sync::Arc;
use rpi_server::transport::{ManagedConnection, TcpTransport};

let transport = Arc::new(TcpTransport::new());
let conn = ManagedConnection::new(transport);

// 自动重试连接（最多 10 次）
conn.connect_with_retry("127.0.0.1:9800").await?;

// 发送请求
conn.send(json!({
    "jsonrpc": "2.0",
    "id": 1,
    "method": "start_session",
    "params": {}
})).await?;

// 接收响应
if let Some(response) = conn.recv().await? {
    println!("response: {}", response);
}

// 获取连接状态
let status = conn.status().await;
println!("connected: {}, latency: {}ms", status.connected, status.latency_ms);

// 启动心跳（在独立 task 中）
let conn_clone = conn.clone();
tokio::spawn(async move {
    let _ = conn_clone.start_heartbeat().await;
});
```

## JSON-RPC 方法

### `start_session`

启动新的 rpi 会话。

```json
{
  "jsonrpc": "2.0",
  "method": "start_session",
  "params": {},
  "id": 1
}
```

**响应：**
```json
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "sessionId": "session-1",
    "session": {
      "id": "session-1",
      "state": "running",
      "pid": 12345,
      "event_count": 0
    }
  }
}
```

### `send`

向会话发送命令。

```json
{
  "jsonrpc": "2.0",
  "method": "send",
  "params": {
    "sessionId": "session-1",
    "command": {"type": "prompt", "content": "hello"}
  },
  "id": 2
}
```

### `subscribe`

订阅会话的事件流（流式响应）。

```json
{
  "jsonrpc": "2.0",
  "method": "subscribe",
  "params": {"sessionId": "session-1"},
  "id": 3
}
```

**响应：**
```json
{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "subscriptionId": 1,
    "sessionId": "session-1",
    "status": "subscribed"
  }
}
```

**后续事件流：**
```json
{"jsonrpc":"2.0","method":"event","params":{"subscriptionId":1,"event":{"type":"delta","content":"Hello"}}}
{"jsonrpc":"2.0","method":"event","params":{"subscriptionId":1,"event":{"type":"delta","content":" world"}}}
{"jsonrpc":"2.0","method":"event","params":{"subscriptionId":1,"event":{"type":"session_end"}}}
```

### `stop_session`

停止会话。

```json
{
  "jsonrpc": "2.0",
  "method": "stop_session",
  "params": {"sessionId": "session-1"},
  "id": 4
}
```

### `list_sessions`

列出所有会话。

```json
{
  "jsonrpc": "2.0",
  "method": "list_sessions",
  "params": {},
  "id": 5
}
```

### `server_status`

获取服务器状态。

```json
{
  "jsonrpc": "2.0",
  "method": "server_status",
  "params": {},
  "id": 6
}
```

### `ping`

心跳检测。

```json
{
  "jsonrpc": "2.0",
  "method": "ping",
  "params": {},
  "id": 7
}
```

**响应：**
```json
{
  "jsonrpc": "2.0",
  "id": 7,
  "result": {"type": "pong"}
}
```

## 跨语言客户端

由于使用标准 JSON-RPC 2.0 over TCP，可以用任何语言调用：

### Python

```python
import socket
import json

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 9800))

# 发送请求
request = {
    "jsonrpc": "2.0",
    "method": "start_session",
    "params": {},
    "id": 1
}
sock.sendall((json.dumps(request) + '\n').encode())

# 接收响应
buffer = b''
while True:
    chunk = sock.recv(4096)
    if not chunk:
        break
    buffer += chunk
    if b'\n' in buffer:
        line, buffer = buffer.split(b'\n', 1)
        response = json.loads(line)
        print(response)
        break
```

### Node.js

```javascript
const net = require('net');

const client = new net.Socket();
client.connect(9800, '127.0.0.1', () => {
    const request = {
        jsonrpc: '2.0',
        method: 'start_session',
        params: {},
        id: 1
    };
    client.write(JSON.stringify(request) + '\n');
});

client.on('data', (data) => {
    const lines = data.toString().split('\n').filter(l => l.trim());
    for (const line of lines) {
        const response = JSON.parse(line);
        console.log(response);
    }
});
```

## Transport trait 扩展

要添加新的传输协议（如 WebSocket），只需实现 `Transport` 和 `Connection` trait：

```rust
use async_trait::async_trait;
use rpi_server::transport::{Transport, Connection};

pub struct WebSocketTransport {
    // ...
}

#[async_trait]
impl Transport for WebSocketTransport {
    type Connection = WebSocketConnection;
    
    async fn listen(&self, addr: &str) -> Result<(), String> {
        // 实现 WebSocket 监听
    }
    
    async fn accept(&self) -> Result<WebSocketConnection, String> {
        // 实现接受连接
    }
    
    async fn connect(&self, addr: &str) -> Result<WebSocketConnection, String> {
        // 实现客户端连接
    }
    
    async fn shutdown(&self) -> Result<(), String> {
        // 实现关闭
    }
}

pub struct WebSocketConnection {
    // ...
}

#[async_trait]
impl Connection for WebSocketConnection {
    async fn send(&mut self, msg: serde_json::Value) -> Result<(), String> {
        // 实现发送
    }
    
    async fn recv(&mut self) -> Result<Option<serde_json::Value>, String> {
        // 实现接收
    }
    
    fn id(&self) -> &str {
        // 返回连接 ID
    }
    
    async fn close(&mut self) -> Result<(), String> {
        // 实现关闭
    }
}
```

然后在 `run_server` 中使用：

```rust
let transport = Arc::new(WebSocketTransport::new());
run_server(transport, "127.0.0.1", 9800, token, executable, args).await?;
```

## 连接管理

`ManagedConnection` 提供以下功能：

### 自动重试

```rust
// 连接失败时自动重试，最多 10 次，间隔递增
conn.connect_with_retry("127.0.0.1:9800").await?;
```

### 心跳检测

```rust
// 启动心跳任务（30s 间隔）
let conn_clone = conn.clone();
tokio::spawn(async move {
    let _ = conn_clone.start_heartbeat().await;
});
```

### 状态追踪

```rust
let status = conn.status().await;
println!("connected: {}", status.connected);
println!("latency: {}ms", status.latency_ms);
println!("messages sent: {}", status.messages_sent);
println!("messages received: {}", status.messages_received);
```

## 性能

- **TCP JSONL** 比 WebSocket 少一层协议开销，但需要自己处理消息边界（`\n`）
- **broadcast channel** 支持多个订阅者同时接收事件
- **异步 I/O** — 基于 tokio，高并发性能优秀
- **零拷贝** — 消息直接序列化到 TCP 流，无中间缓冲

## 与 WebSocket 对比

| 特性 | TCP JSONL | WebSocket |
|------|-----------|-----------|
| 协议开销 | 低（仅 `\n` 分隔） | 中（帧协议） |
| 浏览器支持 | ❌ 需要代理 | ✅ 原生支持 |
| 调试难度 | 低（nc/telnet） | 中（需要工具） |
| 心跳检测 | 应用层实现 | 协议自带 |
| 适用场景 | 服务间通信 | Web 客户端 |

## License

MIT