Keep conversations with encrypted history on the upstream that encrypted it

crates/tw-api/msg-codes.txt

@@ -422,6 +422,7 @@ gw.route.failed
 gw.route.no_upstream_alive
 gw.route.protocol_mismatch
 gw.route.rule_failed
+gw.route.sealed_upstream_gone
 gw.route.selected_upstream_missing
 gw.route.upstream_missing
 gw.toolcall.connection_cut
@@ -439,6 +440,7 @@ gw.upstream.idle_timeout_mid_stream
 gw.upstream.paused
 gw.upstream.period_used_up
 gw.upstream.rate_limited
+gw.upstream.sealed_elsewhere
 gw.upstream.sign_failed
 gw.upstream.status
 gw.upstream.status_message

crates/tw-api/src/lib.rs

@@ -2075,6 +2075,9 @@ slug_enum! {
         Turn = "turn",
         /// 新的一轮，而上一次回答读写的缓存值得留、又没凉
         Cache = "cache",
+        /// 对话里带着上次回答的那一家封存的内容（Responses 的 `encrypted_content`），别家解不开。
+        /// 封存的不只是推理时，候选里也只有它
+        Sealed = "sealed",
     }
 }
 

crates/tw-dialect/src/lib.rs

@@ -9,7 +9,8 @@
 //! `usage()`），拼上游地址（[`url`]），去掉 DeepSeek Harness 只发给 DeepSeek 的
 //! 扩展（[`harness`]），在原文上找调用方的正文（[`caller`]，内容过滤读它、删它），
 //! 读写各格式里名字不同的请求参数（[`params`]），Codex 的远程压缩转给别家时怎么做
-//! （[`compaction`]），以及请求体上的提示缓存断点（[`cache`]）。
+//! （[`compaction`]），请求体上的提示缓存断点（[`cache`]），以及上游封存、只有那个账号读得出来
+//! 的内容（[`sealed`]）。
 
 pub mod anthropic;
 pub mod bedrock;
@@ -25,6 +26,7 @@ pub mod ir;
 pub mod official;
 pub mod params;
 pub mod responses;
+pub mod sealed;
 pub mod think;
 pub mod url;
 pub mod usage;

crates/tw-dialect/src/sealed.rs

@@ -0,0 +1,174 @@
+//! 上游封存的内容：只有封存它的那个账号读得出来的密文。
+//!
+//! OpenAI 的 Responses 不在服务端存对话时（`store: false`，Codex 一律这样发），把下一轮还要
+//! 用的东西加密交给客户端，客户端每一轮原样带回：推理（`reasoning` 条目的
+//! `encrypted_content`）、远程压缩（`compaction`）、Codex 多代理之间传的话（`agent_message`
+//! 里 `encrypted_content` 那一段）、工具结果里加密的那段。**密钥在 OpenAI 那边、按账号分**：
+//! 换了一个账号或者一家中转，对方解不开，整个请求被拒（`invalid_encrypted_content`）。客户端
+//! 每一轮都带着整段历史，于是这段对话从此每一轮都被拒。
+//!
+//! 能去掉的只有推理：那是模型写给自己的笔记，去掉了对话照样接得上。别的去掉就丢了内容 ——
+//! 多代理之间传的话，加密的就是正文本身。所以分两种（[`Sealed`]）。
+//!
+//! 网关自己写的「密文」（[`crate::ir::CARRIED`] 开头：转换时替别家带着的签名、压缩摘要）
+//! 不算：那是网关读得懂的，换到哪一家都能接着转。
+
+use serde_json::Value;
+
+use crate::ir::{CARRIED, Dialect};
+
+/// 请求里带着上游封存的内容。
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Sealed {
+    /// 只有封存的推理：别家解不开时去掉就能接着发
+    Reasoning,
+    /// 还有去不掉的：压缩的前文、代理之间传的话、加密的工具结果。只有封存它的那个账号接得下去
+    Content,
+}
+
+/// Responses 请求体里带着的上游封存的内容。没带的是 None。
+///
+/// 看 `input` 里每一项的每一层：哪儿有不是网关写的 `encrypted_content`，这一项就是封存的。
+/// 不按条目类型列举 —— 新出来的一种封存（多代理的话就是后来才有的）照样认得出。
+pub fn sealed(body: &Value) -> Option<Sealed> {
+    let mut found = None;
+    for item in body.get("input")?.as_array()? {
+        if !carries(item) {
+            continue;
+        }
+        if item.get("type").and_then(Value::as_str) == Some("reasoning") {
+            found = Some(Sealed::Reasoning);
+        } else {
+            return Some(Sealed::Content);
+        }
+    }
+    found
+}
+
+fn carries(v: &Value) -> bool {
+    match v {
+        Value::Object(o) => o.iter().any(|(k, v)| {
+            (k == "encrypted_content"
+                && v.as_str()
+                    .is_some_and(|s| !s.is_empty() && !s.starts_with(CARRIED)))
+                || carries(v)
+        }),
+        Value::Array(a) => a.iter().any(carries),
+        _ => false,
+    }
+}
+
+/// 上游这个错误是在拒绝别的账号封存的内容吗。`wire` 是发给它的那一份的格式，`said` 是
+/// 错误正文或流里那一帧错误。
+///
+/// 看原话，不看错误外壳的结构：同一句话，OpenAI 放在 `error.code`，Codex 后端和中转各有
+/// 各的包法，有的只有一句话。
+///
+/// - OpenAI：`invalid_encrypted_content`（「The encrypted content for item rs_… could not be
+///   verified」「Encrypted function output content could not be decrypted or decoded」）；
+/// - Anthropic：「Invalid `signature` in `thinking` block」。
+pub fn refused(wire: Dialect, said: &[u8]) -> bool {
+    let text = String::from_utf8_lossy(said).to_ascii_lowercase();
+    match wire {
+        Dialect::Responses => {
+            text.contains("invalid_encrypted_content")
+                || (text.contains("encrypted")
+                    && [
+                        "could not be verified",
+                        "could not be decrypted",
+                        "could not decrypt",
+                    ]
+                    .iter()
+                    .any(|w| text.contains(w)))
+        }
+        Dialect::Anthropic => {
+            text.contains("invalid") && text.contains("signature") && text.contains("thinking")
+        }
+        _ => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn input(items: Value) -> Value {
+        json!({"model": "gpt-5", "input": items})
+    }
+
+    #[test]
+    fn sealed_reasoning_alone_can_be_left_out() {
+        let body = input(json!([
+            {"role": "user", "content": "hi"},
+            {"type": "reasoning", "id": "rs_1", "summary": [], "encrypted_content": "gAAAAB"},
+            {"type": "function_call", "call_id": "c1", "name": "ls", "arguments": "{}"},
+        ]));
+        assert_eq!(sealed(&body), Some(Sealed::Reasoning));
+    }
+
+    #[test]
+    fn what_the_agents_said_to_each_other_cannot() {
+        // Codex 多代理：子代理交回来的话整段加密，发信人和任务名是明文
+        let body = input(json!([
+            {"type": "reasoning", "encrypted_content": "gAAAAB"},
+            {"type": "agent_message", "author": "/root/worker", "recipient": "/root", "content": [
+                {"type": "input_text", "text": "Message Type: MESSAGE\nPayload:\n"},
+                {"type": "encrypted_content", "encrypted_content": "gAAAAC"}
+            ]},
+        ]));
+        assert_eq!(sealed(&body), Some(Sealed::Content));
+        // 远程压缩的前文、加密的工具结果也是
+        for item in [
+            json!({"type": "compaction", "encrypted_content": "gAAAAD"}),
+            json!({"type": "function_call_output", "call_id": "c1", "output": [
+                {"type": "encrypted_content", "encrypted_content": "gAAAAE"}
+            ]}),
+        ] {
+            assert_eq!(sealed(&input(json!([item]))), Some(Sealed::Content));
+        }
+    }
+
+    #[test]
+    fn what_the_gateway_wrote_itself_and_empty_fields_are_not_sealed() {
+        let body = input(json!([
+            {"type": "reasoning", "encrypted_content": "tw1.a.sig"},
+            {"type": "reasoning", "encrypted_content": ""},
+            {"type": "compaction", "encrypted_content": "tw1.c.aGk_"},
+            {"role": "user", "content": "hi"},
+        ]));
+        assert_eq!(sealed(&body), None);
+        assert_eq!(sealed(&json!({"messages": []})), None);
+    }
+
+    #[test]
+    fn refusals_to_read_sealed_content_are_recognised() {
+        for said in [
+            json!({"error": {"message": "The encrypted content for item rs_68 could not be verified.",
+                "type": "invalid_request_error", "code": "invalid_encrypted_content"}})
+            .to_string(),
+            // Codex 后端在流里报的，只有这一句
+            "Encrypted function output content could not be decrypted or decoded.".into(),
+            r#"{"detail":"Could not decrypt the provided encrypted_content."}"#.into(),
+        ] {
+            assert!(refused(Dialect::Responses, said.as_bytes()), "{said}");
+        }
+        let other = json!({"error": {"message": "max_output_tokens is too large", "code": "invalid_value"}});
+        assert!(!refused(Dialect::Responses, other.to_string().as_bytes()));
+        let anthropic = json!({"type": "error", "error": {"type": "invalid_request_error",
+            "message": "messages.1.content.0: Invalid `signature` in `thinking` block"}});
+        assert!(refused(
+            Dialect::Anthropic,
+            anthropic.to_string().as_bytes()
+        ));
+        let budget = json!({"type": "error", "error": {"type": "invalid_request_error",
+            "message": "max_tokens: must be greater than thinking.budget_tokens"}});
+        assert!(!refused(Dialect::Anthropic, budget.to_string().as_bytes()));
+        // 格式对不上：Anthropic 的那句不算 Responses 的，别的格式没有封存
+        assert!(!refused(
+            Dialect::Responses,
+            anthropic.to_string().as_bytes()
+        ));
+        assert!(!refused(Dialect::Chat, other.to_string().as_bytes()));
+    }
+}

crates/tw-gateway/src/affinity.rs

@@ -20,8 +20,14 @@
 //! 对话按「路由 + 对话标识」记（见 [`Conversation`]），回答的那一家还要对得上策略组：
 //! 同一段对话这一轮被规则送去另一个组，上次那一家的缓存和这一组无关。
 //!
-//! **只在内存里**，最多记 [`CAP`] 段、每段最多 [`KEEP_MS`]：core 重启之后第一轮
-//! 按策略重新排，最坏是一次缓存重建。**跨配置重载存活**，但沿用的路由决定只在同一
+//! **带着上游封存内容的对话另当别论**（见 [`tw_dialect::sealed`]）：那段历史只有封存它的
+//! 那个账号读得出来，换到别家整个请求被拒，而且之后每一轮都被拒。所以每段对话还记着上次
+//! 回答它的那一家（[`Affinity::holder`]），不管策略组、不管缓存凉没凉、配置改没改、那一家
+//! 中途没没了声音。怎么用它见 `server::pipeline::route`。
+//!
+//! **只在内存里**，最多记 [`CAP`] 段、每段最多 [`KEEP_MS`]（带过封存内容的 [`SEALED_KEEP_MS`]）：
+//! core 重启之后第一轮按策略重新排，最坏是一次缓存重建 —— 带着封存内容的对话这时认不出
+//! 是谁封存的，可能被送去解不开的那一家。**跨配置重载存活**，但沿用的路由决定只在同一
 //! 份配置里有效（配置改了，规则可能已经不是那条了）。
 
 use std::collections::HashMap;
@@ -44,6 +50,10 @@ pub const CACHE_COLD_MS: u64 = 5 * 60 * 1000;
 /// 一段对话记多久。一天没动静的对话，缓存早就没了，沿用的路由决定也不再有意义
 pub const KEEP_MS: u64 = 24 * 60 * 60 * 1000;
 
+/// 带过封存内容的对话记多久。缓存早就凉了，可那段历史还是只有那一家读得出来：客户端
+/// 隔几天接着这段对话说话（Codex 的 `resume`）也得送回那一家
+pub const SEALED_KEEP_MS: u64 = 30 * KEEP_MS;
+
 /// 最多记几段对话。满了先清掉超期的，还满就清最久没动静的那段
 pub const CAP: usize = 4096;
 
@@ -142,6 +152,11 @@ struct Entry {
     held: Option<Held>,
     /// 上次回答它的那一家
     answer: Option<Answer>,
+    /// 上次回答它的那一家，**只给封存的内容用**（见 [`Affinity::holder`]）：和 `answer` 不同，
+    /// 那一家没了声音也不放开 —— 换到别家也解不开
+    holder: Option<String>,
+    /// 这段对话带过封存的内容：记得久一些（[`SEALED_KEEP_MS`]）
+    sealed: bool,
 }
 
 struct Held {
@@ -180,7 +195,7 @@ impl Affinity {
             return None;
         }
         let map = self.lock();
-        let e = map.get(&c.key).filter(|e| fresh(e, now_ms))?;
+        let e = map.get(&c.key).filter(|e| recent(e, now_ms))?;
         let h = e.held.as_ref()?;
         (h.turn == c.turn.number && Weak::ptr_eq(&h.engine, &Arc::downgrade(engine)))
             .then(|| h.decision.clone())
@@ -216,7 +231,7 @@ impl Affinity {
         let map = self.lock();
         let a = map
             .get(&c.key)
-            .filter(|e| fresh(e, now_ms))?
+            .filter(|e| recent(e, now_ms))?
             .answer
             .as_ref()?;
         if a.group.as_deref() != group {
@@ -262,7 +277,22 @@ impl Affinity {
             at_ms: now_ms,
             cache,
         };
-        self.upsert(&t.key, now_ms, |e| e.answer = Some(answer));
+        self.upsert(&t.key, now_ms, |e| {
+            e.holder = Some(answer.provider.clone());
+            e.answer = Some(answer);
+        });
+    }
+
+    /// 这段对话带着上游封存的内容：上次回答它的那一家 —— 请求里最新的那几段是它封存的，前面
+    /// 的它也读得出来（不然它回答不了）。认不出是谁的（没回答过、core 重启过、记录超期了）是
+    /// None。
+    ///
+    /// 记下这段对话带过封存的内容：之后记得久一些（[`SEALED_KEEP_MS`]）
+    pub fn holder(&self, c: &Conversation, now_ms: u64) -> Option<String> {
+        let mut map = self.lock();
+        let e = map.get_mut(&c.key).filter(|e| fresh(e, now_ms))?;
+        e.sealed = true;
+        e.holder.clone()
     }
 
     /// `provider` 在这段对话上没了声音（无响应超时，见 `server::pipeline::idle`）：上次回答它
@@ -298,12 +328,19 @@ impl Affinity {
             touched_ms: now_ms,
             held: None,
             answer: None,
+            holder: None,
+            sealed: false,
         });
-        // 超期的整段作废：一天前那一轮的决定和回答，对现在的请求都没有意义
-        if !fresh(e, now_ms) {
+        // 超期的作废：一天前那一轮的决定和回答，对现在的请求都没有意义。封存的内容是谁的，
+        // 记到它自己的期限
+        if !recent(e, now_ms) {
             e.held = None;
             e.answer = None;
         }
+        if !fresh(e, now_ms) {
+            e.holder = None;
+            e.sealed = false;
+        }
         e.touched_ms = e.touched_ms.max(now_ms);
         f(e);
     }
@@ -314,10 +351,17 @@ impl Affinity {
     }
 }
 
-fn fresh(e: &Entry, now_ms: u64) -> bool {
+/// 这一轮的决定、上次的回答还算数：一天之内动过
+fn recent(e: &Entry, now_ms: u64) -> bool {
     now_ms.saturating_sub(e.touched_ms) <= KEEP_MS
 }
 
+/// 这段对话还记着：带过封存内容的记 [`SEALED_KEEP_MS`]，别的和 [`recent`] 一样
+fn fresh(e: &Entry, now_ms: u64) -> bool {
+    let keep = if e.sealed { SEALED_KEEP_MS } else { KEEP_MS };
+    now_ms.saturating_sub(e.touched_ms) <= keep
+}
+
 /// 这一次由谁回答。**跟着请求的结局走**：上游回了成功、响应走完了才记下它和它读写
 /// 了多少缓存（见 [`crate::ending::Ending`]）。
 pub struct Ticket {
@@ -599,6 +643,33 @@ mod tests {
         assert_eq!(a.stay(&conv(1, true), Some("池子"), &mut c, up, MIN), None);
     }
 
+    #[test]
+    fn the_holder_of_sealed_content_outlasts_what_the_cache_would_keep() {
+        let a = Arc::new(Affinity::default());
+        let c = conv(1, false);
+        assert_eq!(a.holder(&c, 0), None, "没回答过：认不出是谁的");
+        answer(&a, &c, "乙", 0, 0);
+        // 没了声音也照记：换到别家也解不开
+        a.left(&c, "乙");
+        assert_eq!(a.holder(&c, MIN).as_deref(), Some("乙"));
+        // 带过封存内容的对话记一个月：隔几天接着说话也送得回去，缓存那一份早就作废了
+        let later = 3 * KEEP_MS;
+        assert_eq!(a.holder(&c, later).as_deref(), Some("乙"));
+        let mut cands = pool();
+        assert_eq!(
+            a.stay(&conv(1, true), Some("池子"), &mut cands, up, later),
+            None
+        );
+        // 之后回答它的是丙：从此是丙的
+        answer(&a, &c, "丙", 0, later);
+        assert_eq!(a.holder(&c, later + MIN).as_deref(), Some("丙"));
+        assert_eq!(a.holder(&c, later + SEALED_KEEP_MS + 1), None);
+        // 没带过封存内容的对话照旧只记一天
+        let plain = Conversation::new("default", "别的", Turn::default());
+        answer(&a, &plain, "甲", 0, 0);
+        assert_eq!(a.holder(&plain, KEEP_MS + 1), None);
+    }
+
     #[test]
     fn a_day_old_conversation_is_forgotten_and_the_table_stays_bounded() {
         let a = Arc::new(Affinity::default());