Faster request path, history and macOS file watching; incremental transcripts and dry-run shares, protocol 41

Cargo.toml

@@ -103,6 +103,8 @@ flate2 = "1"
 # regex-automata / regex-syntax 已经在依赖树里（tracing 的 EnvFilter），
 # 所以加它的增量很小
 regex = "1"
+# 在整个请求体里找一小段字节（SIMD）。serde_json、regex 本来就依赖它，不多编任何东西
+memchr = "2"
 # 保格式地改 TOML。接管 Codex 要写它的 config.toml，而重新序列化
 # 会把用户的注释和排版一起抹掉
 toml_edit = "0.25"

bin/twcore/src/main.rs

@@ -840,23 +840,28 @@ fn cmd_serve(path: &Path, port: Option<u16>, safe: bool, parent: Option<u32>) ->
 
           每一轮都重新读配置，所以改完期限不用重启 —— 下一轮就按新的算。
           扫的是目录和一条 DELETE，空跑一次的代价可以忽略。
+
+          **在阻塞线程上跑，删正文时不拿记录库的锁**（见 `tw_store::task::gc`）：一次删几个
+          GB 可以要好几秒，以前那期间这把锁一直攥着，记录和查询都在等。
         */
         if let Some(rec) = store.clone() {
             let gw = state.clone();
             tokio::spawn(async move {
                 loop {
                     let r = gw.config().retention.clone();
                     let freed = {
-                        let g = rec.lock().await;
-                        g.gc(
-                            std::time::SystemTime::now()
-                                .duration_since(std::time::UNIX_EPOCH)
-                                .map(|d| d.as_millis() as i64)
-                                .unwrap_or(0),
-                            r.body_days,
-                            r.row_days,
-                            r.body_max_bytes,
-                        )
+                        let rec = rec.clone();
+                        let now = std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .map(|d| d.as_millis() as i64)
+                            .unwrap_or(0);
+                        let (body_days, row_days, max) =
+                            (r.body_days, r.row_days, r.body_max_bytes);
+                        tokio::task::spawn_blocking(move || {
+                            tw_store::task::gc(&rec, now, body_days, row_days, max)
+                        })
+                        .await
+                        .unwrap_or(0)
                     };
                     if freed > 0 {
                         tracing::info!(
@@ -1013,7 +1018,14 @@ fn build_store(
     // 每记下一行请求，交给网关的密钥用量上限结算（见 `tw_control::key_limits`）
     settled: tw_store::SettleHook,
 ) -> Option<std::sync::Arc<tokio::sync::Mutex<tw_store::Recorder>>> {
-    let events = bus.subscribe();
+    // **存储层自己的那条通道，不订阅界面的广播**（见 `EventBus::record_feed`）：广播上
+    // 掉队丢的是界面上的几个点，记录丢的是那几个请求
+    let Some(events) = bus.record_feed() else {
+        tracing::warn!(
+            "request recording is already attached to the event bus; not starting a second one"
+        );
+        return None;
+    };
     let (db, blobs) = match tw_store::open(dir) {
         Ok(opened) => opened,
         Err(e) => {

crates/tw-api/src/ep.rs

@@ -64,8 +64,9 @@ endpoints! {
     Fixture: GET "/request/{id}/fixture" [id], () => String, text;
     Sessions: GET "/sessions", api::ListQuery => Vec<api::SessionView>;
     SessionDetail: GET "/sessions/{id}" [id], () => api::SessionDetail;
-    /// 一次会话读成一段对话：每一轮新说的话、回答、工具调用和结果（已脱敏）
-    SessionTranscript: GET "/sessions/{id}/transcript" [id], () => api::Transcript;
+    /// 一次会话读成一段对话：每一轮新说的话、回答、工具调用和结果（已脱敏）。可以只要
+    /// 从某一轮起的那些
+    SessionTranscript: GET "/sessions/{id}/transcript" [id], api::TranscriptQuery => api::Transcript;
 
     // ─────────────────────────────────────────────── 测速、回放、试路由
     SpeedQuote: POST "/speed/quote", api::SpeedRunRequest => api::SpeedQuote;

crates/tw-api/src/lib.rs

@@ -798,7 +798,14 @@ pub const MSG_CODES: &str = include_str!("../msg-codes.txt");
 /// **40 起模型规格多了会不会推理、收不收图**：[`ModelRow`] 多了 `reasoning`、`image_input`
 /// 和它们的来源，[`ModelSpecSave`] 多了 `reasoning`、`image_input`（四项都空才是删掉）。
 /// 照 39 写的界面保存规格时会把这两项丢掉。
-pub const CONTROL_API_VERSION: u32 = 40;
+///
+/// **41 起会话的对话视图可以只取新的轮次，试算给出每家分到的份额**：
+/// `GET /sessions/{id}/transcript` 的请求从 `()` 变成 [`TranscriptQuery`]（`from_turn`，
+/// 不带就是整份），[`Transcript`] 多了 `total_turns` 和 `settled_turns`（前面这么多轮
+/// 不会再变，下次从这里取）；[`DryRunCandidate`] 多了 `share`（负载均衡组里这一家按
+/// 当前权重和系数分到的份额，别的组和这一轮不参与的是 None）。照 40 写的界面取对话视图
+/// 时传 `null` 会被拒。
+pub const CONTROL_API_VERSION: u32 = 41;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "ts", derive(ts_rs::TS))]
@@ -4697,6 +4704,9 @@ pub struct SessionDetail {
 /// 太大的只留开头，没存下来的也有。读不到的地方，那一轮的 `gaps` 说出来。
 ///
 /// **已脱敏**，和请求详情里的正文同一套打码。图片只说类型和大小，从不带数据。
+///
+/// **可以只要后面的几轮**（[`TranscriptQuery::from_turn`]）：会话开着时每来一轮重读一次，
+/// 前面不会再变的那些（`settled_turns`）不必再传一遍。
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[cfg_attr(feature = "ts", derive(ts_rs::TS))]
 pub struct Transcript {
@@ -4705,10 +4715,31 @@ pub struct Transcript {
     /// Gemini 的 `systemInstruction`，Chat 和 Responses 还有开头连着的 system、developer
     /// 消息，几段之间空一行。没有是 null
     pub system: Option<String>,
-    /// 和 [`SessionDetail::turns`] 同样的请求，同样的顺序
+    /// 这次会话一共几轮，就是 [`SessionDetail::turns`] 有几条。给了
+    /// [`TranscriptQuery::from_turn`] 时，`turns` 是其中从那一轮起的那些
+    pub total_turns: u32,
+    /// 开头这么多轮不会再变了：它们的正文都落了盘（或者不会再来），后面的请求不会再改写
+    /// 它们，这次会话里还在跑的请求落库时也不会排到它们前面。**下一次从这里要起**
+    /// （`from_turn = settled_turns`），留着手上的前这么多轮，换掉后面的。
+    ///
+    /// 后面那几轮还会变：最后一轮的回答可能还没落盘、工具调用的号会被下一轮换成客户端
+    /// 记下的那个、一个开始得早结束得晚的请求会插到它们中间
+    pub settled_turns: u32,
+    /// 和 [`SessionDetail::turns`] 同样的请求，同样的顺序。给了
+    /// [`TranscriptQuery::from_turn`] 的，只有从那一轮起的那些
     pub turns: Vec<TranscriptTurn>,
 }
 
+/// `GET /sessions/{id}/transcript` 的查询串。
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
+pub struct TranscriptQuery {
+    /// 只要从这一轮起的那些（从 0 数，[`Transcript::turns`] 里的第几条）。不给是整段；
+    /// 比总轮数还大的，`turns` 是空的
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub from_turn: Option<u32>,
+}
+
 /// 对话里的一轮，就是会话里的一个请求。
 ///
 /// 客户端每一轮都把整段历史发上来：请求 i 的消息 = 请求 i-1 的消息 + 上一轮的回答 + 新的
@@ -5066,6 +5097,12 @@ pub struct DryRunCandidate {
     /// 分得少，没有样本的那一项算 1。只在 `balance_by` 不是 `weights` 的 `load-balance` 里有
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub balance_factor: Option<f64>,
+    /// 照此刻的数字轮下去，它分到的新对话占这个组的多少，0 到 1：有效权重（权重 × 系数）
+    /// 除以这一轮参加的各家之和，和数据面挑排头是同一份数字。只在 `load-balance` 里有，
+    /// 同一个组里有份额的各家加起来是 1。熔断着、冷却着、并发数满着的这一轮不参加，没有
+    /// 份额 —— 全都这样时都参加
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub share: Option<f64>,
 }
 
 /// 一个要转换格式的候选上游。

crates/tw-api/src/ts.rs

@@ -301,14 +301,21 @@ mod tests {
     #[test]
     fn a_transcript_is_turns_of_messages_and_parts() {
         let ts = typescript();
-        assert!(ts.contains("  SessionTranscript: { req: null; res: Transcript };"));
+        assert!(ts.contains("  SessionTranscript: { req: TranscriptQuery; res: Transcript };"));
+        assert!(
+            decl_of(&ts, "TranscriptQuery").contains("from_turn?: number"),
+            "{}",
+            decl_of(&ts, "TranscriptQuery")
+        );
         assert!(ts.contains(
             "  SessionTranscript: { method: \"GET\", path: \"/sessions/{id}/transcript\", params: [\"id\"], format: \"json\" },"
         ));
         let transcript = decl_of(&ts, "Transcript");
         for field in [
             "session: string",
             "system: string | null",
+            "total_turns: number",
+            "settled_turns: number",
             "turns: Array<TranscriptTurn>",
         ] {
             assert!(transcript.contains(field), "{field}: {transcript}");

crates/tw-control/src/dryrun.rs

@@ -51,29 +51,37 @@ fn order_like_the_data_plane(
 }
 
 /// 一家候选的排序依据：典型的快慢（从发出去到回答的第一段内容）、最近的成功率、按它们算出
-/// 的系数。**只给顺序真用到的那几样**（[`runtime_facts`] 只取用得上的）—— `url-test` 看快慢；
-/// `load-balance` 按 `balance_by` 看快慢、成败，系数乘在权重上。用不着的、没有样本的
-/// 是 None。
+/// 的系数，和 `load-balance` 照这些数字分给它的份额。**只给顺序真用到的那几样**
+/// （[`runtime_facts`] 只取用得上的）—— `url-test` 看快慢；`load-balance` 按 `balance_by`
+/// 看快慢、成败，系数乘在权重上。用不着的、没有样本的是 None。
 struct Basis {
     ttfb_ms: Option<u32>,
     success_rate: Option<f64>,
     balance_factor: Option<f64>,
+    share: Option<f64>,
 }
 
-/// 每一家候选的排序依据，按 `d.candidates` 的名字查。系数和数据面一样按这份候选、
-/// 这份数字算（`tw_engine::balance_factors`）
+/// 每一家候选的排序依据，按 `d.candidates` 的名字查。系数、份额和数据面一样按这份候选、
+/// 这份数字算（`tw_engine::balance_factors`、`tw_engine::weighted::shares`）
 fn bases(
     d: &tw_engine::Decision,
     runtime: Option<&(&tw_engine::Group, tw_engine::Facts)>,
 ) -> std::collections::HashMap<String, Basis> {
     let Some((g, f)) = runtime else {
         return Default::default();
     };
-    let factors = if g.kind == tw_engine::GroupType::LoadBalance && !g.balance_by.is_weights() {
+    let balanced = g.kind == tw_engine::GroupType::LoadBalance;
+    let factors = if balanced && !g.balance_by.is_weights() {
         tw_engine::balance_factors(g.balance_by, &d.candidates, f)
     } else {
         Vec::new()
     };
+    // 份额只按权重分时也有：权重本身就是比例
+    let shares = if balanced {
+        tw_engine::weighted::shares(g, &d.candidates, f)
+    } else {
+        Vec::new()
+    };
     d.candidates
         .iter()
         .enumerate()
@@ -82,6 +90,7 @@ fn bases(
                 ttfb_ms: f.ttfb_ms.get(name).copied(),
                 success_rate: f.success.get(name).copied(),
                 balance_factor: factors.get(i).copied(),
+                share: shares.get(i).copied().flatten(),
             };
             (name.clone(), basis)
         })
@@ -325,7 +334,7 @@ pub async fn dry_run(
             // `load-balance` 的候选带上各自的权重：排头的为什么是它，一半在这个数里
             let balanced = group.filter(|g| g.kind == tw_engine::GroupType::LoadBalance);
             // 每一家收到的模型名，和为什么不是请求里写的那个；顺序看运行时数字的，再加上
-            // 每一家的那几个数字（权重、快慢、成功率、系数）—— 「为什么轮到它」
+            // 每一家的那几个数字（权重、快慢、成功率、系数、份额）—— 「为什么轮到它」
             // 要能从这里看出来
             let mut basis_of = bases(&d, runtime.as_ref());
             out.candidate_models = out
@@ -344,6 +353,7 @@ pub async fn dry_run(
                         ttfb_ms: basis.as_ref().and_then(|b| b.ttfb_ms),
                         success_rate: basis.as_ref().and_then(|b| b.success_rate),
                         balance_factor: basis.as_ref().and_then(|b| b.balance_factor),
+                        share: basis.as_ref().and_then(|b| b.share),
                     }
                 })
                 .collect();

crates/tw-control/src/key_limits.rs

@@ -57,10 +57,16 @@ pub fn rebuild(gw: &tw_gateway::AppState, db: &tw_store::Db) {
 pub fn follow(gw: &tw_gateway::AppState, store: Arc<tokio::sync::Mutex<tw_store::Recorder>>) {
     // 弱引用：网关的账拿着这个办法，办法再拿着账就是一个圈，谁都放不掉
     let limits = Arc::downgrade(&gw.key_limits);
+    // **接上时的运行时，记在这里**：结算在存储层自己的记录线程上，那条线程不在运行时里，在
+    // 那儿问「当前的运行时」问不到，新的一期就永远不会从记录里加回来
+    let here = tokio::runtime::Handle::try_current().ok();
     gw.key_limits
         .reread_with(Arc::new(move |m: tw_gateway::key_limits::Moved| {
             let (limits, store) = (limits.clone(), store.clone());
-            let Ok(rt) = tokio::runtime::Handle::try_current() else {
+            let Some(rt) = here
+                .clone()
+                .or_else(|| tokio::runtime::Handle::try_current().ok())
+            else {
                 return;
             };
             rt.spawn(async move {