feat(usage): migrate legacy records to durable subscription storage

- back up and transactionally migrate legacy usage databases
- add durable inbox with retry, recovery, and cleanup
- prefer RESP usage subscription with HTTP fallback
- preserve existing usage APIs, queries, and deduplication behavior
- add migration, inbox, and RESP protocol tests

README.ja.md

@@ -81,6 +81,8 @@ OpenAI、Claude、Gemini、およびその他の互換形式の間で変換で
 - 時間、モデル、プロバイダー、ソース、キー、結果でデータを絞り込み。
 - リクエストと Token の推移、入力・出力・推論・キャッシュの使用量を確認。
 - リクエスト詳細、分析画面、価格統計を閲覧。
+- CPA のリアルタイム使用量購読と永続ローカル inbox で収集し、購読できない場合は HTTP に自動フォールバック。
+- 起動時に旧使用履歴データベースを一度だけ移行し、移行前のバックアップを `usage-records/backups` に保存。
 
 ### エージェントクライアント設定
 

README.md

@@ -82,6 +82,8 @@ The Usage page helps you understand local request activity and token consumption
 - Filter usage by time, model, provider, source, key, and result.
 - Inspect request/token trends and input, output, reasoning, and cache usage.
 - Browse request details, analysis views, and price statistics.
+- Collect through CPA's real-time usage subscription with a durable local inbox and automatic HTTP fallback.
+- Upgrade legacy usage databases once at startup after saving a backup under `usage-records/backups`.
 
 ### Agent client configuration
 

README.zh-CN.md

@@ -79,6 +79,8 @@ API 接入页面按照协议或 Provider 管理上游 API 凭证和服务地址
 - 按时间、模型、Provider、来源、密钥和结果筛选数据。
 - 查看请求与 Token 趋势，以及输入、输出、思考和缓存用量构成。
 - 浏览请求明细、分析视图和价格统计。
+- 通过 CPA 实时订阅和本地持久化 inbox 采集记录，订阅不可用时自动降级到 HTTP。
+- 启动时一次性迁移旧版使用记录数据库，并在 `usage-records/backups` 下保留迁移前备份。
 
 ### 智能体客户端配置
 

src-tauri/Cargo.toml

@@ -29,7 +29,7 @@ tar = "0.4"
 tauri = { version = "~2.11", features = ["tray-icon"] }
 tauri-plugin-autostart = "2"
 tauri-plugin-opener = "2"
-tokio = { version = "1", features = ["macros", "sync", "time"] }
+tokio = { version = "1", features = ["io-util", "macros", "net", "sync", "time"] }
 tokio-util = "0.7"
 toml = "0.8"
 toml_edit = "0.20"

src-tauri/src/usage/resp.rs

@@ -0,0 +1,320 @@
+use std::time::Duration;
+
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpStream,
+    time::timeout,
+};
+
+const USAGE_CHANNEL: &str = "usage";
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
+const IO_TIMEOUT: Duration = Duration::from_secs(10);
+const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
+const MAX_ARRAY_LENGTH: usize = 10_000;
+const MAX_NESTING_DEPTH: usize = 8;
+
+pub(super) struct UsageSubscription {
+    stream: TcpStream,
+    read_buffer: Vec<u8>,
+}
+
+impl UsageSubscription {
+    pub(super) async fn connect(port: u16, management_key: &str) -> Result<Self, String> {
+        let address = format!("127.0.0.1:{port}");
+        let stream = timeout(CONNECT_TIMEOUT, TcpStream::connect(&address))
+            .await
+            .map_err(|_| format!("连接 CPA usage 订阅超时: {address}"))?
+            .map_err(|error| format!("连接 CPA usage 订阅失败 {address}: {error}"))?;
+        let mut subscription = Self {
+            stream,
+            read_buffer: Vec::new(),
+        };
+        subscription.send_command(&["AUTH", management_key]).await?;
+        match subscription.read_frame().await? {
+            RespValue::Simple(value) if value.eq_ignore_ascii_case("OK") => {}
+            RespValue::Error(error) => return Err(format!("CPA usage 订阅认证失败: {error}")),
+            value => return Err(format!("CPA usage 订阅认证响应无效: {}", value.kind())),
+        }
+        subscription
+            .send_command(&["SUBSCRIBE", USAGE_CHANNEL])
+            .await?;
+        let acknowledgement = subscription.read_frame().await?;
+        if !is_subscription_ack(&acknowledgement) {
+            return Err("CPA 未确认 usage 订阅".to_string());
+        }
+        Ok(subscription)
+    }
+
+    pub(super) async fn next_message(&mut self) -> Result<String, String> {
+        loop {
+            let frame = self.read_frame().await?;
+            if let Some(payload) = subscription_payload(&frame) {
+                return Ok(payload);
+            }
+            if let RespValue::Error(error) = frame {
+                return Err(format!("CPA usage 订阅返回错误: {error}"));
+            }
+        }
+    }
+
+    async fn send_command(&mut self, parts: &[&str]) -> Result<(), String> {
+        let mut command = format!("*{}\r\n", parts.len()).into_bytes();
+        for part in parts {
+            command.extend_from_slice(format!("${}\r\n", part.len()).as_bytes());
+            command.extend_from_slice(part.as_bytes());
+            command.extend_from_slice(b"\r\n");
+        }
+        timeout(IO_TIMEOUT, self.stream.write_all(&command))
+            .await
+            .map_err(|_| "写入 CPA usage 订阅命令超时".to_string())?
+            .map_err(|error| format!("写入 CPA usage 订阅命令失败: {error}"))
+    }
+
+    async fn read_frame(&mut self) -> Result<RespValue, String> {
+        loop {
+            match parse_resp_frame(&self.read_buffer, 0, 0)? {
+                ParseResult::Complete(value, consumed) => {
+                    self.read_buffer.drain(..consumed);
+                    return Ok(value);
+                }
+                ParseResult::Incomplete => {}
+            }
+            if self.read_buffer.len() >= MAX_FRAME_BYTES {
+                return Err("CPA usage 订阅响应超过大小限制".to_string());
+            }
+            let mut chunk = [0_u8; 8192];
+            let read = timeout(IO_TIMEOUT, self.stream.read(&mut chunk))
+                .await
+                .map_err(|_| "读取 CPA usage 订阅响应超时".to_string())?
+                .map_err(|error| format!("读取 CPA usage 订阅响应失败: {error}"))?;
+            if read == 0 {
+                return Err("CPA usage 订阅连接已关闭".to_string());
+            }
+            self.read_buffer.extend_from_slice(&chunk[..read]);
+        }
+    }
+}
+
+#[derive(Debug)]
+enum RespValue {
+    Simple(String),
+    Error(String),
+    Integer(i64),
+    Bulk(Option<Vec<u8>>),
+    Array(Option<Vec<RespValue>>),
+}
+
+impl RespValue {
+    fn kind(&self) -> &'static str {
+        match self {
+            Self::Simple(_) => "simple string",
+            Self::Error(_) => "error",
+            Self::Integer(_) => "integer",
+            Self::Bulk(_) => "bulk string",
+            Self::Array(_) => "array",
+        }
+    }
+
+    fn text(&self) -> Option<String> {
+        match self {
+            Self::Simple(value) | Self::Error(value) => Some(value.clone()),
+            Self::Bulk(Some(value)) => String::from_utf8(value.clone()).ok(),
+            Self::Integer(value) => Some(value.to_string()),
+            Self::Bulk(None) | Self::Array(_) => None,
+        }
+    }
+}
+
+enum ParseResult {
+    Complete(RespValue, usize),
+    Incomplete,
+}
+
+fn parse_resp_frame(input: &[u8], offset: usize, depth: usize) -> Result<ParseResult, String> {
+    if depth > MAX_NESTING_DEPTH {
+        return Err("CPA usage 订阅响应嵌套过深".to_string());
+    }
+    let Some(prefix) = input.get(offset).copied() else {
+        return Ok(ParseResult::Incomplete);
+    };
+    match prefix {
+        b'+' | b'-' | b':' => {
+            let Some((line, next)) = resp_line(input, offset + 1) else {
+                return Ok(ParseResult::Incomplete);
+            };
+            let text = String::from_utf8(line.to_vec())
+                .map_err(|_| "CPA usage 订阅文本响应不是 UTF-8".to_string())?;
+            let value = match prefix {
+                b'+' => RespValue::Simple(text),
+                b'-' => RespValue::Error(text),
+                _ => RespValue::Integer(
+                    text.parse::<i64>()
+                        .map_err(|_| "CPA usage 订阅整数响应无效".to_string())?,
+                ),
+            };
+            Ok(ParseResult::Complete(value, next - offset))
+        }
+        b'$' => {
+            let Some((line, data_start)) = resp_line(input, offset + 1) else {
+                return Ok(ParseResult::Incomplete);
+            };
+            let length = parse_resp_length(line)?;
+            if length < 0 {
+                return Ok(ParseResult::Complete(
+                    RespValue::Bulk(None),
+                    data_start - offset,
+                ));
+            }
+            let length =
+                usize::try_from(length).map_err(|_| "CPA usage 订阅字符串长度无效".to_string())?;
+            if length > MAX_FRAME_BYTES {
+                return Err("CPA usage 订阅字符串超过大小限制".to_string());
+            }
+            let data_end = data_start.saturating_add(length);
+            let frame_end = data_end.saturating_add(2);
+            if input.len() < frame_end {
+                return Ok(ParseResult::Incomplete);
+            }
+            if input.get(data_end..frame_end) != Some(b"\r\n") {
+                return Err("CPA usage 订阅字符串结尾无效".to_string());
+            }
+            Ok(ParseResult::Complete(
+                RespValue::Bulk(Some(input[data_start..data_end].to_vec())),
+                frame_end - offset,
+            ))
+        }
+        b'*' => {
+            let Some((line, mut next)) = resp_line(input, offset + 1) else {
+                return Ok(ParseResult::Incomplete);
+            };
+            let length = parse_resp_length(line)?;
+            if length < 0 {
+                return Ok(ParseResult::Complete(RespValue::Array(None), next - offset));
+            }
+            let length =
+                usize::try_from(length).map_err(|_| "CPA usage 订阅数组长度无效".to_string())?;
+            if length > MAX_ARRAY_LENGTH {
+                return Err("CPA usage 订阅数组超过大小限制".to_string());
+            }
+            let mut values = Vec::with_capacity(length);
+            for _ in 0..length {
+                match parse_resp_frame(input, next, depth + 1)? {
+                    ParseResult::Complete(value, consumed) => {
+                        values.push(value);
+                        next = next.saturating_add(consumed);
+                    }
+                    ParseResult::Incomplete => return Ok(ParseResult::Incomplete),
+                }
+            }
+            Ok(ParseResult::Complete(
+                RespValue::Array(Some(values)),
+                next - offset,
+            ))
+        }
+        _ => Err("CPA usage 订阅响应类型无效".to_string()),
+    }
+}
+
+fn resp_line(input: &[u8], start: usize) -> Option<(&[u8], usize)> {
+    let relative_end = input
+        .get(start..)?
+        .windows(2)
+        .position(|pair| pair == b"\r\n")?;
+    let end = start + relative_end;
+    Some((&input[start..end], end + 2))
+}
+
+fn parse_resp_length(value: &[u8]) -> Result<i64, String> {
+    std::str::from_utf8(value)
+        .map_err(|_| "CPA usage 订阅长度不是 UTF-8".to_string())?
+        .parse::<i64>()
+        .map_err(|_| "CPA usage 订阅长度无效".to_string())
+}
+
+fn is_subscription_ack(value: &RespValue) -> bool {
+    let RespValue::Array(Some(values)) = value else {
+        return false;
+    };
+    values.len() >= 2
+        && values[0]
+            .text()
+            .is_some_and(|value| value.eq_ignore_ascii_case("subscribe"))
+        && values[1].text().as_deref() == Some(USAGE_CHANNEL)
+}
+
+fn subscription_payload(value: &RespValue) -> Option<String> {
+    let RespValue::Array(Some(values)) = value else {
+        return None;
+    };
+    if values.len() < 3
+        || !values[0]
+            .text()
+            .is_some_and(|value| value.eq_ignore_ascii_case("message"))
+        || values[1].text().as_deref() != Some(USAGE_CHANNEL)
+    {
+        return None;
+    }
+    values[2].text()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::net::TcpListener;
+
+    #[test]
+    fn parses_usage_subscription_message() {
+        let input = b"*3\r\n$7\r\nmessage\r\n$5\r\nusage\r\n$16\r\n{\"request_id\":1}\r\n";
+        let ParseResult::Complete(value, consumed) = parse_resp_frame(input, 0, 0).unwrap() else {
+            panic!("expected complete RESP frame");
+        };
+        assert_eq!(consumed, input.len());
+        assert_eq!(
+            subscription_payload(&value).as_deref(),
+            Some("{\"request_id\":1}")
+        );
+    }
+
+    #[test]
+    fn rejects_oversized_arrays() {
+        let error = match parse_resp_frame(b"*10001\r\n", 0, 0) {
+            Err(error) => error,
+            _ => panic!("expected oversized RESP array to fail"),
+        };
+        assert!(error.contains("数组超过大小限制"));
+    }
+
+    #[tokio::test]
+    async fn authenticates_subscribes_and_receives_usage_payload() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let expected_auth = b"*2\r\n$4\r\nAUTH\r\n$6\r\nsecret\r\n";
+            let mut auth = vec![0_u8; expected_auth.len()];
+            stream.read_exact(&mut auth).await.unwrap();
+            assert_eq!(auth, expected_auth);
+            stream.write_all(b"+OK\r\n").await.unwrap();
+
+            let expected_subscribe = b"*2\r\n$9\r\nSUBSCRIBE\r\n$5\r\nusage\r\n";
+            let mut subscribe = vec![0_u8; expected_subscribe.len()];
+            stream.read_exact(&mut subscribe).await.unwrap();
+            assert_eq!(subscribe, expected_subscribe);
+            stream
+                .write_all(b"*3\r\n$9\r\nsubscribe\r\n$5\r\nusage\r\n:1\r\n")
+                .await
+                .unwrap();
+            stream
+                .write_all(b"*3\r\n$7\r\nmessage\r\n$5\r\nusage\r\n$16\r\n{\"request_id\":1}\r\n")
+                .await
+                .unwrap();
+        });
+
+        let mut subscription = UsageSubscription::connect(port, "secret").await.unwrap();
+        assert_eq!(
+            subscription.next_message().await.unwrap(),
+            "{\"request_id\":1}"
+        );
+        server.await.unwrap();
+    }
+}