diff --git a/deploy/helm/cheers/templates/frontend.yaml b/deploy/helm/cheers/templates/frontend.yaml
index a3c14a13..6e1203b6 100644
--- a/deploy/helm/cheers/templates/frontend.yaml
+++ b/deploy/helm/cheers/templates/frontend.yaml
@@ -16,13 +16,15 @@ data:
       index index.html;
 
       # Security headers (audit M2): clickjacking + MIME-sniff + referrer + a
-      # minimal CSP. CSP is intentionally narrow (no script-src lockdown) to
-      # avoid breaking the Vite SPA; tighten with nonces later. No location here
-      # defines its own add_header, so these inherit to every response.
+      # hardened CSP. The production Vite build emits only external hashed module
+      # scripts (no inline <script>), so script-src 'self' is safe; style-src keeps
+      # 'unsafe-inline' because the SPA injects runtime styles. API + WS are proxied
+      # same-origin (/api, /ws) so connect-src 'self' ws: wss:' covers them. No
+      # location here defines its own add_header, so these inherit to every response.
       add_header X-Content-Type-Options "nosniff" always;
       add_header X-Frame-Options "DENY" always;
       add_header Referrer-Policy "strict-origin-when-cross-origin" always;
-      add_header Content-Security-Policy "frame-ancestors 'none'; object-src 'none'; base-uri 'self'" always;
+      add_header Content-Security-Policy "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob: https:; media-src 'self' blob:; font-src 'self' data:; connect-src 'self' ws: wss:; frame-ancestors 'none'; object-src 'none'; base-uri 'self'" always;
 
       location / {
         try_files $uri $uri/ /index.html;
@@ -40,7 +42,7 @@ data:
         add_header X-Content-Type-Options "nosniff" always;
         add_header X-Frame-Options "DENY" always;
         add_header Referrer-Policy "strict-origin-when-cross-origin" always;
-        add_header Content-Security-Policy "frame-ancestors 'none'; object-src 'none'; base-uri 'self'" always;
+        add_header Content-Security-Policy "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob: https:; media-src 'self' blob:; font-src 'self' data:; connect-src 'self' ws: wss:; frame-ancestors 'none'; object-src 'none'; base-uri 'self'" always;
       }
 
       location /api {
diff --git a/frontend/src/components/MarkdownRenderer.tsx b/frontend/src/components/MarkdownRenderer.tsx
index e63f7115..fbe9a5f6 100644
--- a/frontend/src/components/MarkdownRenderer.tsx
+++ b/frontend/src/components/MarkdownRenderer.tsx
@@ -1,4 +1,4 @@
-import { useContext, useMemo } from "react";
+import { memo, useContext, useMemo } from "react";
 import ReactMarkdown from "react-markdown";
 import remarkGfm from "remark-gfm";
 import { PathOpenContext, looksLikePath } from "@/features/chat/workspaceLink";
@@ -46,7 +46,7 @@ function CodeBlock({
   );
 }
 
-export function MarkdownRenderer({ content, className }: Props) {
+export const MarkdownRenderer = memo(function MarkdownRenderer({ content, className }: Props) {
   const onPath = useContext(PathOpenContext);
   // react-markdown v10 dropped the `className` prop (passing it throws). Wrap in a styled
   // div instead so "prose" + caller classes still apply.
@@ -92,4 +92,4 @@ export function MarkdownRenderer({ content, className }: Props) {
     </ReactMarkdown>
     </div>
   );
-}
+});
diff --git a/frontend/src/features/chat/ChannelView.tsx b/frontend/src/features/chat/ChannelView.tsx
index 1a788275..e5a2bf69 100644
--- a/frontend/src/features/chat/ChannelView.tsx
+++ b/frontend/src/features/chat/ChannelView.tsx
@@ -145,6 +145,17 @@ export function ChannelView({ channel, onBack, sidebarOpen, onToggleSidebar }: P
 
   // Channel file index (id + name) built from loaded messages' attachments — no
   // extra fetch. Powers F3 filename suggestions (draft names a file → offer it).
+  // Attachments never change during streaming, but `flushDeltas` swaps the whole
+  // `messages` array identity every rAF, so keying this off `messages` would rebuild
+  // the nested O(messages × attachments) index on every token delta. Instead key off a
+  // cheap signature of just the messages that HAVE files (id + count) — it only changes
+  // when attachments actually change, keeping both the work AND the value identity stable
+  // across deltas. (Cheap enough to compute inline each render.)
+  let channelFilesKey = `${messages.length}`;
+  for (const m of messages) {
+    const n = m.files?.length ?? 0;
+    if (n) channelFilesKey += `|${m.msg_id}:${n}`;
+  }
   const channelFiles = useMemo(() => {
     const byId = new Map<string, { file_id: string; filename: string }>();
     for (const m of messages) {
@@ -156,7 +167,8 @@ export function ChannelView({ channel, onBack, sidebarOpen, onToggleSidebar }: P
       }
     }
     return Array.from(byId.values());
-  }, [messages]);
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [channelFilesKey]);
 
   // A different channel means a different session set — drop any prior target.
   // Also drop any buffered stream deltas + cancel a pending flush so a stale frame
@@ -577,6 +589,20 @@ export function ChannelView({ channel, onBack, sidebarOpen, onToggleSidebar }: P
     setShowConnBanner(rtStatus === "offline");
   }, [rtStatus]);
 
+  // If the realtime connection drops mid-stream, a bot bubble can be left with
+  // `_streaming: true` indefinitely (no done frame arrives), which the streaming
+  // fast-path renders as plain pre-wrap text. Finalize any lingering stream when we
+  // go offline so it falls back to full Markdown; reconnect catch-up then reconciles
+  // it with the persisted server state.
+  useEffect(() => {
+    if (rtStatus !== "offline") return;
+    setMessages((prev) =>
+      prev.some((m) => m._streaming)
+        ? prev.map((m) => (m._streaming ? { ...m, _streaming: false } : m))
+        : prev
+    );
+  }, [rtStatus]);
+
   // Re-flatten the palette when bot labels resolve after the initial fetch.
   useEffect(() => {
     void loadCommands();
diff --git a/frontend/src/features/chat/MessageItem.tsx b/frontend/src/features/chat/MessageItem.tsx
index 6b27c68a..a9dba1ca 100644
--- a/frontend/src/features/chat/MessageItem.tsx
+++ b/frontend/src/features/chat/MessageItem.tsx
@@ -472,14 +472,20 @@ function MessageBody({
     return <p className="text-sm text-red-400 italic">{message.error}</p>;
   }
 
+  // While a bubble is streaming, re-parsing the whole accumulated Markdown +
+  // re-highlighting the growing code block every animation frame is ~O(len^2)
+  // main-thread work. Render the in-flight text as plain whitespace-pre-wrap and
+  // only switch to full Markdown + highlighting once the turn finalizes
+  // (_streaming clears), which leaves completed messages rendered exactly as before.
   const hasMarkdown =
-    content.includes("```") ||
-    content.includes("**") ||
-    content.includes("*") ||
-    content.includes("#") ||
-    content.includes("[") ||
-    content.includes("\n") ||
-    content.includes("`");
+    !message._streaming &&
+    (content.includes("```") ||
+      content.includes("**") ||
+      content.includes("*") ||
+      content.includes("#") ||
+      content.includes("[") ||
+      content.includes("\n") ||
+      content.includes("`"));
 
   return (
     <div className="relative">
diff --git a/packages/cheers-acp-connector-rs/src/acp_adapter.rs b/packages/cheers-acp-connector-rs/src/acp_adapter.rs
index bac2e561..023d691e 100644
--- a/packages/cheers-acp-connector-rs/src/acp_adapter.rs
+++ b/packages/cheers-acp-connector-rs/src/acp_adapter.rs
@@ -979,23 +979,31 @@ async fn handle_peer_message(
 async fn handle_peer_notification(
     account_id: &str,
     event_tx: &mpsc::Sender<RuntimeEvent>,
-    value: Value,
+    mut value: Value,
 ) -> anyhow::Result<()> {
     let method = value.get("method").and_then(Value::as_str).unwrap_or("");
     if method != "session/update" {
         tracing::debug!(account = %account_id, method, "ACP peer notification (not session/update; ignored)");
         return Ok(());
     }
-    let params = value.get("params").cloned().unwrap_or(Value::Null);
-    let acp_session_id = params
-        .get("sessionId")
+    // Read only the small sessionId out of the borrowed value...
+    let acp_session_id = value
+        .get("params")
+        .and_then(|params| params.get("sessionId"))
         .and_then(Value::as_str)
         .unwrap_or_default()
         .to_string();
     if acp_session_id.is_empty() {
         return Ok(());
     }
-    let update = params.get("update").cloned().unwrap_or(Value::Null);
+    // ...then MOVE the (potentially large) update subtree out of the owned value
+    // instead of deep-cloning it — session/update is the highest-frequency event
+    // and tool_call_update carries the bulky tool output.
+    let update = value
+        .get_mut("params")
+        .and_then(|params| params.get_mut("update"))
+        .map(Value::take)
+        .unwrap_or(Value::Null);
     tracing::debug!(
         account = %account_id,
         session = %acp_session_id,
diff --git a/packages/cheers-acp-connector-rs/src/bridge_runtime/mod.rs b/packages/cheers-acp-connector-rs/src/bridge_runtime/mod.rs
index 423b3c16..d6a7f429 100644
--- a/packages/cheers-acp-connector-rs/src/bridge_runtime/mod.rs
+++ b/packages/cheers-acp-connector-rs/src/bridge_runtime/mod.rs
@@ -1966,8 +1966,14 @@ impl RuntimeContext {
                 )
                 .await?;
                 let (final_text, file_ids) = {
-                    let guard = run.lock().await;
-                    (guard.text.clone(), guard.created_file_ids.clone())
+                    let mut guard = run.lock().await;
+                    // Move the accumulated text out (turn is over; the run is about
+                    // to be dropped from the shared maps below) so the whole streamed
+                    // response isn't deep-cloned just to hand it to the Done frame.
+                    (
+                        std::mem::take(&mut guard.text),
+                        guard.created_file_ids.clone(),
+                    )
                 };
                 let terminal_ack = self
                     .io
diff --git a/packages/cheers-acp-connector-rs/src/loopback.rs b/packages/cheers-acp-connector-rs/src/loopback.rs
index f5545c7d..5a46a17d 100644
--- a/packages/cheers-acp-connector-rs/src/loopback.rs
+++ b/packages/cheers-acp-connector-rs/src/loopback.rs
@@ -16,6 +16,13 @@ use uuid::Uuid;
 /// a pooled-but-idle MCP socket can't leak a task/fd for the life of the session.
 const LOOPBACK_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
 
+/// Hard ceiling on a single loopback request body. The only legitimate client is
+/// the local MCP bridge, whose resource calls are small JSON payloads, so a few
+/// MiB is generous. Without this cap a local process could set an arbitrary
+/// `Content-Length` and force the connection task to buffer that many bytes on the
+/// heap before the token is even checked.
+const MAX_LOOPBACK_BODY_BYTES: usize = 8 * 1024 * 1024;
+
 #[derive(Debug, Clone)]
 pub struct LoopbackHandle {
     pub url: String,
@@ -233,6 +240,21 @@ async fn read_http_request(
         .get("content-length")
         .and_then(|value| value.parse::<usize>().ok())
         .unwrap_or(0);
+    // Reject an oversized body from the declared Content-Length BEFORE draining it,
+    // so a local process can't force us to buffer an arbitrary amount of heap. We
+    // still write a proper 413 and then close (returning Err drops the connection).
+    if content_length > MAX_LOOPBACK_BODY_BYTES {
+        write_http_json(
+            stream,
+            413,
+            false,
+            &json!({ "ok": false, "code": "PAYLOAD_TOO_LARGE", "error": "loopback request body too large" }),
+        )
+        .await?;
+        return Err(anyhow!(
+            "loopback request body too large: {content_length} bytes (max {MAX_LOOPBACK_BODY_BYTES})"
+        ));
+    }
     let body_start = header_end + 4;
     let body_end = body_start + content_length;
     while buf.len() < body_end {
@@ -280,6 +302,7 @@ async fn write_http_json(
         200 => "OK",
         401 => "Unauthorized",
         405 => "Method Not Allowed",
+        413 => "Payload Too Large",
         _ => "Error",
     };
     // Content-Length is always present, so the client can frame the body and reuse
diff --git a/packages/cheers-acp-connector-rs/src/self_update.rs b/packages/cheers-acp-connector-rs/src/self_update.rs
index 08d1e7e3..425a007a 100644
--- a/packages/cheers-acp-connector-rs/src/self_update.rs
+++ b/packages/cheers-acp-connector-rs/src/self_update.rs
@@ -183,6 +183,12 @@ struct Marker {
     /// can't put the connector into a swap/rollback loop.
     #[serde(default)]
     blocked_versions: Vec<String>,
+    /// Highest version this connector has ever applied. Persisted so a hostile
+    /// gateway can't force a downgrade to a superseded — but still correctly
+    /// signed — historical build that merely happens to be newer than whatever
+    /// is running right now. Monotonic: only ever moves forward.
+    #[serde(default)]
+    highest_applied: Option<String>,
 }
 
 fn marker_path(state_dir: &Path) -> PathBuf {
@@ -485,9 +491,19 @@ impl SelfUpdater {
             .connect_timeout(Duration::from_secs(10))
             .timeout(Duration::from_secs(300))
             .build()?;
-        let manifest_bytes = fetch(&client, &format!("{base}/connector-manifest.json")).await?;
+        let manifest_bytes = fetch(
+            &client,
+            &format!("{base}/connector-manifest.json"),
+            MAX_METADATA_BYTES,
+        )
+        .await?;
         let sig_b64 = String::from_utf8(
-            fetch(&client, &format!("{base}/connector-manifest.json.sig")).await?,
+            fetch(
+                &client,
+                &format!("{base}/connector-manifest.json.sig"),
+                MAX_METADATA_BYTES,
+            )
+            .await?,
         )
         .context("manifest signature is not UTF-8")?;
         let manifest = verify_and_parse_manifest(&manifest_bytes, &sig_b64, self.pubkey_pem())?;
@@ -502,6 +518,22 @@ impl SelfUpdater {
             ));
         }
 
+        // Client-side monotonic anti-downgrade: a manifest may be older than the
+        // highest version we've ever applied yet still newer than what's running
+        // (e.g. a bad manual downgrade, or a rollback that restored an older
+        // binary). The signed manifest can't carry a per-client floor, so we keep
+        // one on disk and refuse anything at or below it. This blocks a hostile
+        // gateway from pinning us to a superseded, correctly-signed build.
+        if let Some(highest) = read_marker(&self.state_dir).and_then(|m| m.highest_applied) {
+            if !version_is_newer(&manifest.version, &highest) {
+                return Err(anyhow!(
+                    "signed manifest is for {} which is not newer than the highest \
+                     version ever applied ({highest}) — refusing (anti-downgrade)",
+                    manifest.version
+                ));
+            }
+        }
+
         let suffix = platform_asset_suffix().expect("checked in disabled_reason");
         let exe = std::env::current_exe()?;
         let exe_dir = exe
@@ -532,7 +564,7 @@ impl SelfUpdater {
                 .get(asset)
                 .ok_or_else(|| anyhow!("manifest has no asset {asset}"))?
                 .sha256;
-            let bytes = fetch(&client, &format!("{base}/{asset}")).await?;
+            let bytes = fetch(&client, &format!("{base}/{asset}"), MAX_BINARY_BYTES).await?;
             let actual = sha256_hex(&bytes);
             if &actual != expected {
                 return Err(anyhow!(
@@ -601,6 +633,17 @@ impl SelfUpdater {
                 previous_exe = Some(old);
             }
         }
+        let prev_marker = read_marker(&self.state_dir);
+        let blocked_versions = prev_marker
+            .as_ref()
+            .map(|m| m.blocked_versions.clone())
+            .unwrap_or_default();
+        // Advance the monotonic floor to the version we're applying (never let it
+        // move backward — read_marker may carry a higher one).
+        let highest_applied = match prev_marker.and_then(|m| m.highest_applied) {
+            Some(prev) if !version_is_newer(version, &prev) => Some(prev),
+            _ => Some(version.to_string()),
+        };
         write_marker(
             &self.state_dir,
             &Marker {
@@ -608,9 +651,8 @@ impl SelfUpdater {
                 previous_exe,
                 attempts: 0,
                 confirmed: false,
-                blocked_versions: read_marker(&self.state_dir)
-                    .map(|m| m.blocked_versions)
-                    .unwrap_or_default(),
+                blocked_versions,
+                highest_applied,
             },
         )?;
         tracing::info!(version = %version, "self-update: binaries swapped, restarting");
@@ -618,8 +660,20 @@ impl SelfUpdater {
     }
 }
 
-async fn fetch(client: &reqwest::Client, url: &str) -> anyhow::Result<Vec<u8>> {
-    let resp = client
+/// Download cap for the manifest and its signature — both are tiny JSON/sig.
+const MAX_METADATA_BYTES: usize = 64 * 1024;
+/// Download cap for a single binary asset.
+const MAX_BINARY_BYTES: usize = 128 * 1024 * 1024;
+
+/// GET `url` into memory, refusing to buffer more than `max_bytes`. The cap is
+/// enforced twice: against an advertised `Content-Length` before reading (fast
+/// reject) and against the running byte count while streaming (defends against a
+/// lying or absent `Content-Length` — localhost gateways are plain http, so a
+/// MITM/hostile proxy could otherwise stream a multi-GB body and OOM us). All
+/// signature/sha256 verification still happens on the caller side, after this
+/// bounded download returns.
+async fn fetch(client: &reqwest::Client, url: &str, max_bytes: usize) -> anyhow::Result<Vec<u8>> {
+    let mut resp = client
         .get(url)
         .send()
         .await
@@ -627,7 +681,27 @@ async fn fetch(client: &reqwest::Client, url: &str) -> anyhow::Result<Vec<u8>> {
     if !resp.status().is_success() {
         return Err(anyhow!("GET {url} returned HTTP {}", resp.status()));
     }
-    Ok(resp.bytes().await?.to_vec())
+    if let Some(len) = resp.content_length() {
+        if len > max_bytes as u64 {
+            return Err(anyhow!(
+                "GET {url}: Content-Length {len} exceeds cap of {max_bytes} bytes"
+            ));
+        }
+    }
+    let mut buf: Vec<u8> = Vec::new();
+    while let Some(chunk) = resp
+        .chunk()
+        .await
+        .with_context(|| format!("read body of {url}"))?
+    {
+        if buf.len() + chunk.len() > max_bytes {
+            return Err(anyhow!(
+                "GET {url}: response body exceeds cap of {max_bytes} bytes"
+            ));
+        }
+        buf.extend_from_slice(&chunk);
+    }
+    Ok(buf)
 }
 
 /// Replace this process with `exe`, keeping argv (daemon metadata matches on
@@ -785,6 +859,7 @@ mod tests {
                 attempts: 0,
                 confirmed: false,
                 blocked_versions: vec![],
+                highest_applied: None,
             },
         )
         .expect("write marker");
@@ -807,6 +882,7 @@ mod tests {
                 attempts: 1,
                 confirmed: false,
                 blocked_versions: vec![],
+                highest_applied: None,
             },
         )
         .expect("write marker");
diff --git a/server/migrations/0049_messages_pending_permission_index.sql b/server/migrations/0049_messages_pending_permission_index.sql
new file mode 100644
index 00000000..1fc72e4d
--- /dev/null
+++ b/server/migrations/0049_messages_pending_permission_index.sql
@@ -0,0 +1,9 @@
+-- Fleet inbox pending scan (fleet::find_pending_for_user / find_pending_for_user_all)
+-- selects unresolved `permission` messages `ORDER BY created_at DESC LIMIT 100`. With no
+-- index on `msg_type`, that scan reads the hot `messages` table by created_at and filters
+-- rows post-hoc. A partial index on the small `permission` subset makes the ordered scan
+-- index-driven (permission rows are a tiny fraction of all messages), so the badge/fleet
+-- queries touch only pending cards instead of walking the whole timeline.
+CREATE INDEX IF NOT EXISTS ix_messages_pending_permission
+    ON messages (created_at DESC)
+    WHERE msg_type = 'permission';
diff --git a/server/src/api/approval.rs b/server/src/api/approval.rs
index fcb4a6af..707af8b2 100644
--- a/server/src/api/approval.rs
+++ b/server/src/api/approval.rs
@@ -173,6 +173,22 @@ pub async fn resolve_permission(
     let pending = approval::find_pending(&state.db, channel_id, &request_id)
         .await?
         .ok_or(AppError::NotFound)?;
+    // Authorization gate: a current channel member (or a platform admin) may proceed
+    // to the fine-grained approver/RESPOND check below; additionally the bot OWNER may
+    // always resolve their own bot's request even without joining the channel ("the
+    // owner may always resolve", see below). A REMOVED member is neither a member nor
+    // the owner, so this — together with the delegation/RESPOND purge on member removal
+    // (remove_channel_member / leave_channel) — stops a stale grant from letting an
+    // ex-member resolve.
+    if ensure_member(&state, channel_id, uid, &claims.role)
+        .await
+        .is_err()
+        && approval::bot_owner(&state.db, pending.bot_id).await? != Some(uid)
+    {
+        return Err(AppError::Forbidden(
+            "not authorized to resolve this bot's permission".into(),
+        ));
+    }
 
     if pending
         .content_data
diff --git a/server/src/api/auth.rs b/server/src/api/auth.rs
index c5d5e34f..4e2e1a1a 100644
--- a/server/src/api/auth.rs
+++ b/server/src/api/auth.rs
@@ -262,7 +262,8 @@ pub async fn register(
         .map(str::trim)
         .filter(|s| !s.is_empty())
         .map(str::to_string);
-    let hash = bcrypt::hash(&body.password, bcrypt::DEFAULT_COST)
+    let hash = crate::infra::crypto::hash_password(body.password.clone())
+        .await
         .map_err(|e| AppError::Internal(format!("hash: {e}")))?;
     let user_id = Uuid::new_v4();
 
@@ -336,12 +337,16 @@ pub async fn change_password(
     .ok_or(AppError::NotFound)?;
 
     let hashed: String = row.try_get("password_hash").map_err(AppError::Db)?;
-    if !bcrypt::verify(&body.current_password, &hashed).unwrap_or(false) {
+    if !crate::infra::crypto::verify_password(body.current_password.clone(), hashed)
+        .await
+        .unwrap_or(false)
+    {
         return Err(AppError::Unauthorized(
             "current password is incorrect".into(),
         ));
     }
-    let new_hash = bcrypt::hash(&body.new_password, bcrypt::DEFAULT_COST)
+    let new_hash = crate::infra::crypto::hash_password(body.new_password.clone())
+        .await
         .map_err(|e| AppError::Internal(format!("hash: {e}")))?;
 
     // Revoke all existing tokens (other devices) by bumping the version.
@@ -496,7 +501,8 @@ pub async fn reset_password(
     .ok_or(AppError::NotFound)?;
     let user_id: String = user.try_get("user_id").map_err(AppError::Db)?;
 
-    let hash = bcrypt::hash(&body.new_password, bcrypt::DEFAULT_COST)
+    let hash = crate::infra::crypto::hash_password(body.new_password.clone())
+        .await
         .map_err(|e| AppError::Internal(format!("hash: {e}")))?;
     sqlx::query(
         "UPDATE users SET password_hash = $2, token_version = token_version + 1 WHERE user_id = $1",
diff --git a/server/src/api/channels.rs b/server/src/api/channels.rs
index 461175d7..a7ea716a 100644
--- a/server/src/api/channels.rs
+++ b/server/src/api/channels.rs
@@ -204,17 +204,28 @@ pub async fn list_channels(
              LEFT JOIN workspace_memberships wm ON wm.workspace_id = c.workspace_id
                     AND wm.user_id = $1 AND wm.status = 'active'
              CROSS JOIN LATERAL (
+                 -- Bounded unread scan: cap at the newest 100 unread messages so a
+                 -- huge backlog can't force an unbounded count. `unread_count` is
+                 -- therefore min(actual, 100) (99+ semantics); `mention_count`
+                 -- is computed over the same capped window via a single LEFT JOIN
+                 -- (PK (msg_id, member_id) ⇒ at most one match per message) rather
+                 -- than a per-row correlated EXISTS.
                  SELECT count(*) AS unread_count,
-                        count(*) FILTER (WHERE EXISTS (
-                            SELECT 1 FROM message_mentions mm
-                            WHERE mm.msg_id = m.msg_id
-                              AND mm.member_id = $1 AND mm.member_type = 'user'
-                        )) AS mention_count
+                        count(mm.msg_id) AS mention_count
                  FROM messages m
-                 WHERE m.channel_id = c.channel_id
-                   AND m.is_partial = FALSE
-                   AND m.sender_id <> $1
-                   AND m.created_at > COALESCE(cm.last_read_at, 'epoch'::timestamptz)
+                 LEFT JOIN message_mentions mm
+                        ON mm.msg_id = m.msg_id
+                       AND mm.member_id = $1 AND mm.member_type = 'user'
+                 WHERE m.msg_id IN (
+                     SELECT m2.msg_id
+                     FROM messages m2
+                     WHERE m2.channel_id = c.channel_id
+                       AND m2.is_partial = FALSE
+                       AND m2.sender_id <> $1
+                       AND m2.created_at > COALESCE(cm.last_read_at, 'epoch'::timestamptz)
+                     ORDER BY m2.created_at DESC
+                     LIMIT 100
+                 )
              ) counts
              WHERE c.type != 'dm'
                AND wm.user_id IS NULL
@@ -249,18 +260,30 @@ pub async fn list_channels(
          LEFT JOIN workspace_memberships wm ON wm.workspace_id = c.workspace_id AND wm.user_id = $1
                 AND wm.status = 'active'
          CROSS JOIN LATERAL (
+             -- Bounded unread scan: cap at the newest 100 unread messages so a huge
+             -- backlog can't force an unbounded count. `unread_count` is min(actual,
+             -- 100) (99+ semantics); `mention_count` is computed over the same
+             -- capped window via a single LEFT JOIN (PK (msg_id, member_id) ⇒ at most
+             -- one match per message) rather than a per-row correlated EXISTS. The
+             -- non-member guard (`cm.member_id IS NOT NULL`) stays inside the window:
+             -- a non-member's window is empty, so both counts are 0 as before.
              SELECT count(*) AS unread_count,
-                    count(*) FILTER (WHERE EXISTS (
-                        SELECT 1 FROM message_mentions mm
-                        WHERE mm.msg_id = m.msg_id
-                          AND mm.member_id = $1 AND mm.member_type = 'user'
-                    )) AS mention_count
+                    count(mm.msg_id) AS mention_count
              FROM messages m
-             WHERE cm.member_id IS NOT NULL
-               AND m.channel_id = c.channel_id
-               AND m.is_partial = FALSE
-               AND m.sender_id <> $1
-               AND m.created_at > COALESCE(cm.last_read_at, 'epoch'::timestamptz)
+             LEFT JOIN message_mentions mm
+                    ON mm.msg_id = m.msg_id
+                   AND mm.member_id = $1 AND mm.member_type = 'user'
+             WHERE m.msg_id IN (
+                 SELECT m2.msg_id
+                 FROM messages m2
+                 WHERE cm.member_id IS NOT NULL
+                   AND m2.channel_id = c.channel_id
+                   AND m2.is_partial = FALSE
+                   AND m2.sender_id <> $1
+                   AND m2.created_at > COALESCE(cm.last_read_at, 'epoch'::timestamptz)
+                 ORDER BY m2.created_at DESC
+                 LIMIT 100
+             )
          ) counts
          WHERE c.type != 'dm'
            AND (cm.member_id IS NOT NULL
@@ -349,11 +372,19 @@ pub async fn list_dms(
         "SELECT c.channel_id, c.workspace_id, c.name, c.type, c.purpose, c.auto_assist,
                 c.allow_member_invites, c.allow_bot_adds,
                 COALESCE((
+                    -- Bounded unread scan: cap at the newest 100 unread messages so a
+                    -- huge backlog can't force an unbounded count (min(actual, 100),
+                    -- i.e. 99+ semantics).
                     SELECT count(*) FROM messages msg
-                    WHERE msg.channel_id = c.channel_id
-                      AND msg.is_partial = FALSE
-                      AND msg.sender_id <> $1
-                      AND msg.created_at > COALESCE(cm.last_read_at, 'epoch'::timestamptz)
+                    WHERE msg.msg_id IN (
+                        SELECT m2.msg_id FROM messages m2
+                        WHERE m2.channel_id = c.channel_id
+                          AND m2.is_partial = FALSE
+                          AND m2.sender_id <> $1
+                          AND m2.created_at > COALESCE(cm.last_read_at, 'epoch'::timestamptz)
+                        ORDER BY m2.created_at DESC
+                        LIMIT 100
+                    )
                 ), 0) AS unread_count,
                 COALESCE(
                   (SELECT COALESCE(u.display_name, u.username) FROM channel_memberships m
@@ -1116,6 +1147,32 @@ pub async fn decline_channel_invite(
     Ok(Json(json!({"declined": true})))
 }
 
+/// Revoke a user's channel-scoped ACP approval authority when they leave / are
+/// removed: their approver delegations (`approval_delegations`) and any per-user
+/// RESPOND overrides (`bot_event_access`) for this channel. Parameterized;
+/// harmless no-op when `user_id` is a bot (neither table holds bot user rows).
+async fn purge_channel_approval_authority(
+    state: &AppState,
+    channel_id: &str,
+    user_id: &str,
+) -> Result<(), AppError> {
+    sqlx::query("DELETE FROM approval_delegations WHERE channel_id = $1 AND user_id = $2")
+        .bind(channel_id)
+        .bind(user_id)
+        .execute(&state.db)
+        .await?;
+    sqlx::query(
+        "DELETE FROM bot_event_access
+         WHERE channel_id = $1 AND subject_kind = 'user' AND subject_id = $2
+           AND capability = 'respond'",
+    )
+    .bind(channel_id)
+    .bind(user_id)
+    .execute(&state.db)
+    .await?;
+    Ok(())
+}
+
 pub async fn remove_channel_member(
     State(state): State<AppState>,
     Extension(claims): Extension<Claims>,
@@ -1134,6 +1191,11 @@ pub async fn remove_channel_member(
         .bind(&member_id)
         .execute(&state.db)
         .await?;
+    // A removed member must not retain approval authority in this channel: drop any
+    // approver delegations they held (approval_delegations) and any per-user RESPOND
+    // overrides (bot_event_access) — otherwise a stale row would still let them
+    // resolve this channel's ACP permission requests. No-op for bots.
+    purge_channel_approval_authority(&state, &channel_id, &member_id).await?;
     if let Ok(cid) = Uuid::parse_str(&channel_id) {
         // Membership checks happen at subscribe time, so removal must cut any
         // LIVE subscriptions too — otherwise the removed member keeps receiving
@@ -1252,6 +1314,9 @@ pub async fn leave_channel(
         .execute(&state.db)
         .await?;
     }
+    // Same authority purge as remove_channel_member: a member who leaves must not
+    // keep approver delegations or per-user RESPOND overrides in this channel.
+    purge_channel_approval_authority(&state, &channel_id, &claims.sub).await?;
     // Same reverse edge as remove_channel_member: cut the leaver's live
     // subscriptions so they stop receiving new frames immediately.
     if let (Ok(cid), Ok(uid)) = (Uuid::parse_str(&channel_id), Uuid::parse_str(&claims.sub)) {
diff --git a/server/src/api/files.rs b/server/src/api/files.rs
index efcc2bd1..5324c262 100644
--- a/server/src/api/files.rs
+++ b/server/src/api/files.rs
@@ -55,6 +55,72 @@ fn safe_filename(raw: &str) -> Result<String, AppError> {
     Ok(name.to_string())
 }
 
+/// Defense-in-depth: normalize a caller-supplied `content_type` to a known-safe
+/// MIME type before persisting it. The serve side (`infra::http::file_response`)
+/// already forces active types to download, so this only hardens the stored value
+/// — an unrecognized type collapses to `application/octet-stream` rather than
+/// being trusted verbatim. Never rejects: legit files just get a generic type.
+fn normalize_content_type(raw: &str) -> String {
+    // Drop parameters (e.g. `; charset=utf-8`) and match the bare MIME base.
+    let base = raw
+        .split(';')
+        .next()
+        .unwrap_or("")
+        .trim()
+        .to_ascii_lowercase();
+    const ALLOWED: &[&str] = &[
+        // Images (svg+xml is downloaded, not rendered, by the serve side).
+        "image/png",
+        "image/jpeg",
+        "image/gif",
+        "image/webp",
+        "image/svg+xml",
+        "image/bmp",
+        "image/tiff",
+        "image/avif",
+        "image/heic",
+        // Documents / text.
+        "application/pdf",
+        "application/json",
+        "application/rtf",
+        "application/zip",
+        "text/plain",
+        "text/csv",
+        "text/markdown",
+        // Office (legacy + OOXML + OpenDocument).
+        "application/msword",
+        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+        "application/vnd.ms-excel",
+        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        "application/vnd.ms-powerpoint",
+        "application/vnd.openxmlformats-officedocument.presentationml.presentation",
+        "application/vnd.oasis.opendocument.text",
+        "application/vnd.oasis.opendocument.spreadsheet",
+        "application/vnd.oasis.opendocument.presentation",
+        // Audio.
+        "audio/mpeg",
+        "audio/mp4",
+        "audio/wav",
+        "audio/x-wav",
+        "audio/webm",
+        "audio/ogg",
+        "audio/flac",
+        "audio/aac",
+        "audio/x-m4a",
+        // Video.
+        "video/mp4",
+        "video/webm",
+        "video/quicktime",
+        "video/ogg",
+        "video/x-msvideo",
+    ];
+    if ALLOWED.contains(&base.as_str()) {
+        base
+    } else {
+        "application/octet-stream".to_string()
+    }
+}
+
 fn resolve_expires_in(seconds: Option<i64>) -> i64 {
     const MIN_SECONDS: i64 = 60;
     const MAX_SECONDS: i64 = 7 * 24 * 60 * 60;
@@ -331,11 +397,13 @@ pub async fn upload_file(
         return Err(AppError::BadRequest("empty file".into()));
     }
     let filename = safe_filename(&q.filename)?;
-    let content_type = q
-        .content_type
-        .clone()
-        .filter(|s| !s.trim().is_empty())
-        .unwrap_or_else(|| "application/octet-stream".to_string());
+    let content_type = normalize_content_type(
+        q.content_type
+            .as_deref()
+            .map(str::trim)
+            .filter(|s| !s.is_empty())
+            .unwrap_or("application/octet-stream"),
+    );
     let size_bytes =
         i32::try_from(body.len()).map_err(|_| AppError::BadRequest("file too large".into()))?;
     let file_id = Uuid::new_v4().to_string();
@@ -417,6 +485,7 @@ pub async fn request_presign(
     .ok_or_else(|| AppError::Forbidden("not a channel member".into()))?;
 
     let filename = safe_filename(&body.filename)?;
+    let content_type = normalize_content_type(&body.content_type);
     let file_id = Uuid::new_v4().to_string();
     let object_key = format!("uploads/{}/{}", file_id, filename);
     let requested_size = body.size.unwrap_or(body.size_bytes);
@@ -438,7 +507,7 @@ pub async fn request_presign(
     .bind(&object_key)
     .bind(&state.config.s3_bucket)
     .bind(&filename)
-    .bind(&body.content_type)
+    .bind(&content_type)
     .bind(size_bytes)
     .bind(expires_at)
     .execute(&state.db)
@@ -449,7 +518,7 @@ pub async fn request_presign(
     Ok(Json(json!({
         "file_id": file_id,
         "upload_url": upload_url,
-        "headers": { "Content-Type": body.content_type },
+        "headers": { "Content-Type": content_type },
         "expires_in": expires_in_seconds,
         "object_key": object_key,
     })))
diff --git a/server/src/api/fleet.rs b/server/src/api/fleet.rs
index 39acdde4..d4db7403 100644
--- a/server/src/api/fleet.rs
+++ b/server/src/api/fleet.rs
@@ -23,7 +23,11 @@ use std::collections::HashMap;
 use crate::{
     api::middleware::Claims,
     app_state::AppState,
-    domain::{approval, bot_event_policy::Capability, fleet},
+    domain::{
+        acp_policy, approval,
+        bot_event_policy::{self, Capability},
+        fleet,
+    },
     errors::AppError,
 };
 
@@ -52,23 +56,30 @@ async fn channel_role(state: &AppState, channel_id: Uuid, uid: Uuid) -> String {
 
 /// Resolve (may_see, may_answer) for one pending card — the same SEE gate and
 /// 3-way answer compose as `resolve_permission`. Fail-closed on errors.
+///
+/// `rules`/`groups` are the bot's access rules and the caller's matched groups,
+/// batch-loaded ONCE per bot by the caller (see [`get_fleet`]/[`get_fleet_badge`]);
+/// both the SEE and RESPOND gates resolve in memory against them, so this does at
+/// most one DB round-trip per row (the `is_approver` delegation lookup, which is
+/// per-`op_kind` and can't be memoized here). Decision is identical to resolving
+/// each gate independently against the DB.
 async fn see_and_answer(
     state: &AppState,
     p: &crate::domain::fleet::FleetPending,
     uid: Uuid,
     role: &str,
+    rules: &[bot_event_policy::Rule],
+    groups: &[String],
 ) -> (bool, bool) {
-    let may_see = crate::domain::acp_policy::allows(
-        &state.db,
-        &p.bot_id.to_string(),
+    let may_see = acp_policy::allows_with_rules(
+        rules,
         &p.channel_id.to_string(),
         &uid.to_string(),
         role,
+        groups,
         "session/request_permission",
         Capability::See,
-    )
-    .await
-    .unwrap_or(false);
+    );
     if !may_see {
         return (false, false);
     }
@@ -81,20 +92,55 @@ async fn see_and_answer(
     let actionable = approval::is_approver(&state.db, p.bot_id, p.channel_id, uid, op_kind)
         .await
         .unwrap_or(false)
-        || crate::domain::acp_policy::allows(
-            &state.db,
-            &p.bot_id.to_string(),
+        || acp_policy::allows_with_rules(
+            rules,
             &p.channel_id.to_string(),
             &uid.to_string(),
             role,
+            groups,
             "session/request_permission",
             Capability::Respond,
-        )
-        .await
-        .unwrap_or(false);
+        );
     (true, actionable)
 }
 
+/// Batch-load each distinct bot's access rules and the caller's matched groups
+/// ONCE (mirrors `api::approval::filter_traces_by_see`). Keyed by `bot_id` so the
+/// per-row SEE/RESPOND gates resolve in memory instead of re-loading the same
+/// bot's rules for every pending row (fixes the N+1 policy fanout).
+///
+/// A bot whose rule load FAILS is deliberately left out of the caches so the
+/// caller drops its rows — preserving this surface's fail-closed contract (the
+/// previous per-row `acp_policy::allows(...).unwrap_or(false)` did the same on a
+/// DB error). A tracked-but-missing bot therefore means "policy unavailable".
+async fn load_policy_caches(
+    state: &AppState,
+    pending: &[crate::domain::fleet::FleetPending],
+    uid: Uuid,
+) -> (
+    HashMap<Uuid, Vec<bot_event_policy::Rule>>,
+    HashMap<Uuid, Vec<String>>,
+) {
+    let uid_s = uid.to_string();
+    let mut rules_by_bot: HashMap<Uuid, Vec<bot_event_policy::Rule>> = HashMap::new();
+    let mut groups_by_bot: HashMap<Uuid, Vec<String>> = HashMap::new();
+    let mut tried: std::collections::HashSet<Uuid> = std::collections::HashSet::new();
+    for p in pending {
+        if !tried.insert(p.bot_id) {
+            continue; // already attempted this bot
+        }
+        let bid = p.bot_id.to_string();
+        // Fail-closed: on a rule-load error, leave this bot out so its rows drop.
+        let Ok(rules) = bot_event_policy::load_rules(&state.db, &bid).await else {
+            continue;
+        };
+        let groups = bot_event_policy::matched_groups(&state.db, &bid, &uid_s, &rules).await;
+        rules_by_bot.insert(p.bot_id, rules);
+        groups_by_bot.insert(p.bot_id, groups);
+    }
+    (rules_by_bot, groups_by_bot)
+}
+
 // ── GET /fleet/badge ─────────────────────────────────────────────────────────
 
 /// Workspace-agnostic count of pending approvals the caller may answer —
@@ -105,6 +151,8 @@ pub async fn get_fleet_badge(
 ) -> Result<Json<Value>, AppError> {
     let uid = user_id(&claims)?;
     let pending = fleet::find_pending_for_user_all(&state.db, uid).await?;
+    // Batch-load policy rules/groups once per bot, then resolve each row in memory.
+    let (rules_by_bot, groups_by_bot) = load_policy_caches(&state, &pending, uid).await;
     let mut roles: HashMap<Uuid, String> = HashMap::new();
     let mut count: i64 = 0;
     for p in pending {
@@ -116,7 +164,15 @@ pub async fn get_fleet_badge(
                 r
             }
         };
-        let (_, actionable) = see_and_answer(&state, &p, uid, &role).await;
+        // Fail-closed: a bot absent from the cache had a policy-load error → skip.
+        let Some(rules) = rules_by_bot.get(&p.bot_id) else {
+            continue;
+        };
+        let groups = groups_by_bot
+            .get(&p.bot_id)
+            .map(Vec::as_slice)
+            .unwrap_or(&[]);
+        let (_, actionable) = see_and_answer(&state, &p, uid, &role, rules, groups).await;
         if actionable {
             count += 1;
         }
@@ -138,6 +194,8 @@ pub async fn get_fleet(
 
     // ── Zone A: pending approvals (SEE-gated, flagged with may-answer) ──────
     let pending = fleet::find_pending_for_user(&state.db, workspace_id, uid).await?;
+    // Batch-load policy rules/groups once per bot, then resolve each row in memory.
+    let (rules_by_bot, groups_by_bot) = load_policy_caches(&state, &pending, uid).await;
     // Channel roles are shared by both policy checks below; resolve each once.
     let mut roles: HashMap<Uuid, String> = HashMap::new();
     let mut approvals: Vec<Value> = Vec::with_capacity(pending.len());
@@ -151,8 +209,16 @@ pub async fn get_fleet(
                 r
             }
         };
+        // Fail-closed: a bot absent from the cache had a policy-load error → drop.
+        let Some(rules) = rules_by_bot.get(&p.bot_id) else {
+            continue;
+        };
+        let groups = groups_by_bot
+            .get(&p.bot_id)
+            .map(Vec::as_slice)
+            .unwrap_or(&[]);
         // SEE gate — fail-closed: on error, drop the row (see module docs).
-        let (may_see, actionable) = see_and_answer(&state, &p, uid, &role).await;
+        let (may_see, actionable) = see_and_answer(&state, &p, uid, &role, rules, groups).await;
         if !may_see {
             continue;
         }
diff --git a/server/src/api/users.rs b/server/src/api/users.rs
index 245f9399..80bdfd1b 100644
--- a/server/src/api/users.rs
+++ b/server/src/api/users.rs
@@ -339,7 +339,8 @@ pub async fn create_user(
         .map(str::trim)
         .filter(|s| !s.is_empty())
         .map(str::to_string);
-    let hash = bcrypt::hash(&body.password, bcrypt::DEFAULT_COST)
+    let hash = crate::infra::crypto::hash_password(body.password.clone())
+        .await
         .map_err(|e| AppError::Internal(format!("hash: {e}")))?;
     let user_id = Uuid::new_v4().to_string();
 
diff --git a/server/src/api/workspace.rs b/server/src/api/workspace.rs
index e1aa3208..67ff6614 100644
--- a/server/src/api/workspace.rs
+++ b/server/src/api/workspace.rs
@@ -286,16 +286,21 @@ pub async fn resolve_ref(
 
     // 2. Desk (context_files): prefer an exact path, else fall back to basename
     //    (a Desk file the bot mentioned by name, e.g. `todo.md` for `notes/todo.md`).
+    // `base` is a literal filename here; escape LIKE metacharacters so a name
+    // containing `%`/`_`/`\` matches by value instead of acting as a wildcard.
+    // The `path = $3` exact-match branch keeps the raw `base` (parameter $3).
+    let base_like = crate::domain::messages::escape_like_pattern(base);
     let desk = sqlx::query_as::<_, (String, Option<String>)>(
         "SELECT path, content FROM context_files
          WHERE channel_id = $1 AND is_dir = FALSE
-           AND (path = $2 OR path = $3 OR path LIKE '%/' || $3)
+           AND (path = $2 OR path = $3 OR path LIKE '%/' || $4 ESCAPE '\\')
          ORDER BY (path = $2) DESC, length(path) ASC
          LIMIT 1",
     )
     .bind(channel_id.to_string())
     .bind(r)
     .bind(base)
+    .bind(&base_like)
     .fetch_optional(&state.db)
     .await
     .map_err(AppError::Db)?;
diff --git a/server/src/domain/acp_policy.rs b/server/src/domain/acp_policy.rs
index da22a80d..23046bcb 100644
--- a/server/src/domain/acp_policy.rs
+++ b/server/src/domain/acp_policy.rs
@@ -9,7 +9,10 @@
 
 use sqlx::PgPool;
 
-use crate::domain::{acp_events, bot_event_policy::Capability};
+use crate::domain::{
+    acp_events,
+    bot_event_policy::{self, Capability},
+};
 use crate::errors::AppError;
 
 /// May `(user_id, role)` perform ACP event `acp_event_name` for `(bot, channel)`
@@ -31,6 +34,37 @@ pub async fn allows(
         .await
 }
 
+/// In-memory twin of [`allows`]: same name→class mapping and gate, but resolves
+/// against already-loaded `rules` + the requester's precomputed `matched_groups`
+/// instead of hitting the DB. Lets a caller batch-load a bot's rules/groups ONCE
+/// (via [`bot_event_policy::load_rules`] + [`bot_event_policy::matched_groups`])
+/// and then decide many rows for that bot with zero further round-trips. The
+/// decision is identical to `allows` for the same inputs — unmodeled events are
+/// not gated here either (`true`).
+#[allow(clippy::too_many_arguments)]
+pub fn allows_with_rules(
+    rules: &[bot_event_policy::Rule],
+    channel_id: &str,
+    user_id: &str,
+    role: &str,
+    matched_groups: &[String],
+    acp_event_name: &str,
+    cap: Capability,
+) -> bool {
+    let Some(class) = acp_events::classify(acp_event_name).and_then(|e| e.event_class) else {
+        return true;
+    };
+    bot_event_policy::resolve_access(
+        rules,
+        channel_id,
+        user_id,
+        role,
+        matched_groups,
+        class,
+        cap,
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/server/src/domain/auth.rs b/server/src/domain/auth.rs
index f5a388e2..10a4f7c0 100644
--- a/server/src/domain/auth.rs
+++ b/server/src/domain/auth.rs
@@ -62,8 +62,10 @@ pub async fn authenticate(
 
     let hashed: String = row.try_get("password_hash").map_err(AppError::Db)?;
 
-    // Python passlib 使用 bcrypt（$2b$ 前缀）
-    let ok = bcrypt::verify(password, &hashed)
+    // Python passlib 使用 bcrypt（$2b$ 前缀）。bcrypt 是 CPU 密集（~200-300ms），
+    // 放到 spawn_blocking 线程池执行，避免阻塞 tokio worker。
+    let ok = crate::infra::crypto::verify_password(password.to_string(), hashed)
+        .await
         .map_err(|_| AppError::Unauthorized("invalid credentials".into()))?;
 
     if !ok {
diff --git a/server/src/domain/seed.rs b/server/src/domain/seed.rs
index 42f648da..b4e6cfb9 100644
--- a/server/src/domain/seed.rs
+++ b/server/src/domain/seed.rs
@@ -11,7 +11,6 @@
 
 use std::env;
 
-use bcrypt::{hash, DEFAULT_COST};
 use sqlx::{PgPool, Row};
 use tracing::{info, warn};
 use uuid::Uuid;
@@ -37,7 +36,9 @@ pub async fn ensure_admin_user(db: &PgPool) -> anyhow::Result<()> {
     let username = env::var("ADMIN_USERNAME").unwrap_or_else(|_| "admin".to_string());
     let display_name =
         env::var("ADMIN_DISPLAY_NAME").unwrap_or_else(|_| "Administrator".to_string());
-    let password_hash = hash(&password, DEFAULT_COST)?;
+    // Startup-only (runs once against an empty users table), but route through the
+    // spawn_blocking helper anyway so no bcrypt call runs on an async worker thread.
+    let password_hash = crate::infra::crypto::hash_password(password.clone()).await?;
     let user_id = Uuid::new_v4().to_string();
 
     sqlx::query(
diff --git a/server/src/gateway/presence.rs b/server/src/gateway/presence.rs
index 8e0c47f7..7a8ae096 100644
--- a/server/src/gateway/presence.rs
+++ b/server/src/gateway/presence.rs
@@ -10,7 +10,9 @@
 
 use serde_json::json;
 use sqlx::PgPool;
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::{Duration, Instant};
 use uuid::Uuid;
 
 use crate::{
@@ -18,6 +20,57 @@ use crate::{
     gateway::{realtime::frame::WireFrame, registry::BotLocator},
 };
 
+/// 每频道 bot 成员名单（DB channel_memberships 查询结果）的进程内短 TTL 缓存。
+///
+/// 成员名单是慢变量，而在线与否由内存 registry（BotLocator）实时判定。连接抖动风暴下
+/// broadcast_presence 会被高频调用，缓存把 DB 查询从 O(事件) 降到 O(去重频道/TTL 窗口)，
+/// 同时 online/offline 信号仍逐次实时计算，presence 不会因缓存而变陈旧。
+const BOT_ROSTER_TTL: Duration = Duration::from_secs(2);
+
+struct RosterEntry {
+    members: Vec<String>,
+    fetched_at: Instant,
+}
+
+fn roster_cache() -> &'static Mutex<HashMap<Uuid, RosterEntry>> {
+    static CACHE: OnceLock<Mutex<HashMap<Uuid, RosterEntry>>> = OnceLock::new();
+    CACHE.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+/// 取频道的 bot 成员名单：命中未过期缓存直接返回，否则查 DB 并刷新缓存。
+/// 查询失败时与原行为一致——返回空且不写入缓存（下次调用重试）。
+async fn channel_bot_members(db: &PgPool, channel_id: Uuid) -> Vec<String> {
+    {
+        let cache = roster_cache().lock().unwrap();
+        if let Some(entry) = cache.get(&channel_id) {
+            if entry.fetched_at.elapsed() < BOT_ROSTER_TTL {
+                return entry.members.clone();
+            }
+        }
+    }
+    let fetched: Result<Vec<String>, _> = sqlx::query_scalar(
+        "SELECT member_id FROM channel_memberships
+         WHERE channel_id = $1 AND member_type = 'bot'",
+    )
+    .bind(channel_id.to_string())
+    .fetch_all(db)
+    .await;
+    match fetched {
+        Ok(member_ids) => {
+            let mut cache = roster_cache().lock().unwrap();
+            cache.insert(
+                channel_id,
+                RosterEntry {
+                    members: member_ids.clone(),
+                    fetched_at: Instant::now(),
+                },
+            );
+            member_ids
+        }
+        Err(_) => Vec::new(),
+    }
+}
+
 /// 计算并广播一个频道的全量在线名单（用户 + bot）。
 pub async fn broadcast_presence(state: &AppState, channel_id: Uuid) {
     let online_user_ids = state.fanout.online_users(channel_id);
@@ -52,14 +105,8 @@ pub async fn channel_online_bots(
     bot_locator: &Arc<dyn BotLocator>,
     channel_id: Uuid,
 ) -> Vec<String> {
-    let member_ids: Vec<String> = sqlx::query_scalar(
-        "SELECT member_id FROM channel_memberships
-         WHERE channel_id = $1 AND member_type = 'bot'",
-    )
-    .bind(channel_id.to_string())
-    .fetch_all(db)
-    .await
-    .unwrap_or_default();
+    // 成员名单走短 TTL 缓存（慢变量）；在线状态每次实时判定（快变量）。
+    let member_ids = channel_bot_members(db, channel_id).await;
 
     let mut online = Vec::new();
     for id in member_ids {
diff --git a/server/src/gateway/ws/agent_bridge.rs b/server/src/gateway/ws/agent_bridge.rs
index 6efe892a..d48e0d17 100644
--- a/server/src/gateway/ws/agent_bridge.rs
+++ b/server/src/gateway/ws/agent_bridge.rs
@@ -905,6 +905,30 @@ async fn handle_data_frame(frame: &Value, state: &AppState, bot: &BotInfo, socke
 /// a log-write failure must never disrupt the live agent. Streaming text chunks
 /// are dropped here (the bot's message already carries the text).
 async fn handle_acp_event_frame(frame: &Value, state: &AppState, bot: &BotInfo) {
+    // The channel_id is bot-supplied and must not be trusted: only record,
+    // promote (plan/usage/commands stores) and broadcast into channels the bot is
+    // actually a member of (mirrors handle_trace_frame / handle_permission_request_frame).
+    // Otherwise a self-registered bot could spoof artifact updates and board signals
+    // into arbitrary channels. A present channel_id must pass the membership check
+    // before any store write / acp_event_log insert / broadcast below.
+    if let Some(channel_id) = frame
+        .get("channel_id")
+        .and_then(Value::as_str)
+        .and_then(|s| s.parse::<Uuid>().ok())
+    {
+        if ensure_bot_channel_member(&state.db, bot.bot_id, channel_id)
+            .await
+            .is_err()
+        {
+            tracing::debug!(
+                bot_id = %bot.bot_id,
+                %channel_id,
+                "acp_event dropped: bot not a member of target channel"
+            );
+            return;
+        }
+    }
+
     let name = frame
         .get("name")
         .or_else(|| frame.get("kind"))
@@ -1100,6 +1124,59 @@ async fn handle_trace_frame(frame: &Value, state: &AppState, bot: &BotInfo) {
         .await;
 }
 
+/// Short TTL for the in-process [`bot_event_policy::load_rules`] cache. `allowed_seers`
+/// runs on the hot connector read loop (once per `bot_trace` frame), so an uncached
+/// `load_rules` would issue a DB SELECT per frame at ~1 RTT each. Expiry-only: entries
+/// are not invalidated on rule writes — a few seconds of staleness for a live-visibility
+/// decision is acceptable, and the durable `*.read` verbs remain authoritative.
+const POLICY_RULES_TTL: std::time::Duration = std::time::Duration::from_secs(3);
+
+#[allow(clippy::type_complexity)]
+fn policy_rules_cache() -> &'static std::sync::Mutex<
+    std::collections::HashMap<
+        Uuid,
+        (
+            std::time::Instant,
+            Vec<crate::domain::bot_event_policy::Rule>,
+        ),
+    >,
+> {
+    static CACHE: std::sync::OnceLock<
+        std::sync::Mutex<
+            std::collections::HashMap<
+                Uuid,
+                (
+                    std::time::Instant,
+                    Vec<crate::domain::bot_event_policy::Rule>,
+                ),
+            >,
+        >,
+    > = std::sync::OnceLock::new();
+    CACHE.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
+}
+
+/// TTL-cached wrapper over [`bot_event_policy::load_rules`]. Serves a cloned copy of the
+/// bot's rules on a cache hit within [`POLICY_RULES_TTL`]; otherwise refreshes from the
+/// DB and repopulates. Preserves `load_rules`' error contract so the caller can fail-open.
+async fn cached_load_rules(
+    state: &AppState,
+    bot_id: Uuid,
+) -> Result<Vec<crate::domain::bot_event_policy::Rule>, crate::errors::AppError> {
+    let now = std::time::Instant::now();
+    if let Ok(guard) = policy_rules_cache().lock() {
+        if let Some((at, rules)) = guard.get(&bot_id) {
+            if now.duration_since(*at) < POLICY_RULES_TTL {
+                return Ok(rules.clone());
+            }
+        }
+    }
+    let rules = crate::domain::bot_event_policy::load_rules(&state.db, &bot_id.to_string()).await?;
+    if let Ok(mut guard) = policy_rules_cache().lock() {
+        guard.insert(bot_id, (now, rules.clone()));
+    }
+    Ok(rules)
+}
+
 /// Live per-subscriber SEE (docs/arch/ACP_EVENT_TAXONOMY.md): the online channel
 /// users allowed to SEE an agent event of `event_class` for `bot`. Platform admins
 /// bypass; absent rules → members allowed (the default). Fail-open on a rules error
@@ -1114,11 +1191,10 @@ async fn allowed_seers(
     if online.is_empty() {
         return Vec::new();
     }
-    let rules =
-        match crate::domain::bot_event_policy::load_rules(&state.db, &bot_id.to_string()).await {
-            Ok(r) => r,
-            Err(_) => return online,
-        };
+    let rules = match cached_load_rules(state, bot_id).await {
+        Ok(r) => r,
+        Err(_) => return online,
+    };
     // Fast path: `resolve_access` for `See` only ever consults `see`-capability rules
     // for this event_class, and the membership default for `See` is always allow. So
     // with no matching `see` rule, everyone online is a seer — skip the channel-role
diff --git a/server/src/infra/crypto.rs b/server/src/infra/crypto.rs
index 9993e212..e8e2cae0 100644
--- a/server/src/infra/crypto.rs
+++ b/server/src/infra/crypto.rs
@@ -106,6 +106,30 @@ pub fn decrypt_secret(master_key: &[u8; 32], blob_b64: &str) -> anyhow::Result<S
     Ok(String::from_utf8(plaintext)?)
 }
 
+// ── Password hashing (bcrypt, off the async reactor) ─────────────────────────
+//
+// bcrypt at DEFAULT_COST burns ~200-300ms of CPU per call. Running it directly
+// on a tokio worker blocks that thread from making progress on other tasks, so
+// a burst of logins/registrations can starve the whole runtime. These helpers
+// move the CPU work onto the blocking-thread pool via `spawn_blocking`, which
+// wants owned data — callers hand over owned `String`s.
+
+/// Hash a password with bcrypt at `DEFAULT_COST`, off the async reactor.
+/// Returns the same `bcrypt::BcryptError` the underlying call would.
+pub async fn hash_password(plain: String) -> Result<String, bcrypt::BcryptError> {
+    tokio::task::spawn_blocking(move || bcrypt::hash(&plain, bcrypt::DEFAULT_COST))
+        .await
+        .map_err(|e| bcrypt::BcryptError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?
+}
+
+/// Verify a password against a bcrypt hash, off the async reactor.
+/// Returns the same `bcrypt::BcryptError` the underlying call would.
+pub async fn verify_password(plain: String, hash: String) -> Result<bool, bcrypt::BcryptError> {
+    tokio::task::spawn_blocking(move || bcrypt::verify(&plain, &hash))
+        .await
+        .map_err(|e| bcrypt::BcryptError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?
+}
+
 /// A short, unambiguous one-time code for email flows (e.g. password reset). 8 chars
 /// from a 31-symbol alphabet (no 0/O/1/I/L) → ~31^8 ≈ 8.5e11 combos, fits the
 /// `email_codes.code` column (VARCHAR(10)). Pair with a short TTL + rate limiting.
diff --git a/server/src/infra/email.rs b/server/src/infra/email.rs
index 9a703b96..6789f302 100644
--- a/server/src/infra/email.rs
+++ b/server/src/infra/email.rs
@@ -56,10 +56,12 @@ async fn deliver(
         "{intro}\n\n    {code}\n\n{CODE_TTL_TEXT}\nIf you didn't request this, you can ignore this email."
     );
     if config.brevo_api_key.is_some() && config.email_from_email.is_some() {
+        // A provider IS configured — never write the code to the logs on this path,
+        // even on a send failure (a transient 5xx/429/timeout must not leak the live
+        // code into logs in prod). Log success/failure without the code and return.
         match send_brevo(config, to_email, subject, intro, code, &text).await {
             Ok(()) => {
                 tracing::info!(target: "cheers::email", to = %to_email, kind, "email sent via Brevo");
-                return;
             }
             Err(e) => {
                 tracing::error!(
@@ -67,14 +69,17 @@ async fn deliver(
                     to = %to_email,
                     kind,
                     error = %e,
-                    "Brevo delivery failed — falling back to log delivery"
+                    "Brevo delivery failed — code NOT delivered (no log fallback when a provider is configured)"
                 );
             }
         }
+        return;
     }
-    // Dev/staging delivery: the code is visible in the gateway logs
-    // (`docker logs cheers-gateway-1 | grep cheers::email`). Only taken when Brevo
-    // is unconfigured or the send above failed.
+    // Local-dev delivery: NO email provider is configured at all, so the code is
+    // written to the gateway logs to keep the flow exercisable
+    // (`docker logs cheers-gateway-1 | grep cheers::email`). This branch is never
+    // reached once BREVO_API_KEY + EMAIL_FROM_EMAIL are set, so the code is never
+    // logged in a provider-configured (prod/staging) environment.
     tracing::info!(
         target: "cheers::email",
         to = %to_email,
diff --git a/server/src/infra/ratelimit.rs b/server/src/infra/ratelimit.rs
index 242c7ebe..ae124ef6 100644
--- a/server/src/infra/ratelimit.rs
+++ b/server/src/infra/ratelimit.rs
@@ -51,9 +51,16 @@ impl FixedWindowLimiter {
     /// Record one failed attempt for `key`, opening or extending its window.
     pub fn record_failure(&self, key: &str) {
         let now = Instant::now();
-        // Bound memory under a spoofed-key flood: hard-clear if the map blows up.
+        // Bound memory under a spoofed-key flood: evict only already-expired windows
+        // (never live ones), so a flood of throwaway keys can't reset the counters of
+        // clients with in-progress windows. If the map is STILL over the cap afterwards
+        // — a sustained flood of >100k *live* distinct keys — fall back to a hard clear
+        // so memory stays bounded (retain-only would free nothing and grow unbounded).
         if self.hits.len() > 100_000 {
-            self.hits.clear();
+            self.hits.retain(|_, v| v.reset_at > now);
+            if self.hits.len() > 100_000 {
+                self.hits.clear();
+            }
         }
         let mut e = self.hits.entry(key.to_string()).or_insert_with(|| Window {
             count: 0,
@@ -79,9 +86,16 @@ impl FixedWindowLimiter {
     /// dispatch fan-out) where concurrent callers race on the same key.
     pub fn try_hit(&self, key: &str) -> bool {
         let now = Instant::now();
-        // Bound memory under a spoofed-key flood: hard-clear if the map blows up.
+        // Bound memory under a spoofed-key flood: evict only already-expired windows
+        // (never live ones), so a flood of throwaway keys can't reset the counters of
+        // clients with in-progress windows. If the map is STILL over the cap afterwards
+        // — a sustained flood of >100k *live* distinct keys — fall back to a hard clear
+        // so memory stays bounded (retain-only would free nothing and grow unbounded).
         if self.hits.len() > 100_000 {
-            self.hits.clear();
+            self.hits.retain(|_, v| v.reset_at > now);
+            if self.hits.len() > 100_000 {
+                self.hits.clear();
+            }
         }
         let mut e = self.hits.entry(key.to_string()).or_insert_with(|| Window {
             count: 0,
@@ -139,11 +153,19 @@ impl MinIntervalLimiter {
     /// Record that `key` just acted (set its "last" to now). Call this only after
     /// the guarded action succeeds — see [`peek`](Self::peek).
     pub fn record(&self, key: &str) {
-        // Bound memory under a spoofed-key flood: hard-clear if the map blows up.
+        let now = Instant::now();
+        // Bound memory under a spoofed-key flood: evict only entries whose interval has
+        // already elapsed (never live ones), so a flood of throwaway keys can't clear a
+        // client's in-flight interval and let it act again early. If still over the cap
+        // after that (a flood of >100k live keys), hard-clear so memory stays bounded.
         if self.last.len() > 100_000 {
-            self.last.clear();
+            self.last
+                .retain(|_, v| now.saturating_duration_since(*v) < self.interval);
+            if self.last.len() > 100_000 {
+                self.last.clear();
+            }
         }
-        self.last.insert(key.to_string(), Instant::now());
+        self.last.insert(key.to_string(), now);
     }
 
     /// Convenience for callers gating a cheap, always-succeeding action: peek and,