新增: 无人值守安全系列(H1-H3) + P5 审计日志与部署运维 + #14 异步委托

- H3 外部干预感知: run_agentic_loop 增 h3_watch_dir 参数, loop 启动取 git HEAD 基线, 每 5 轮探测, HEAD 被外部改写(切分支/reset/他话提交)即 emit_task_parked(stop_kind=external_change) + teardown 停靠; 4 调用方(发送/编辑/强制发送/审批续跑)从 allowed_dirs.persistent 排序取首目录, 修复初版 current_dir 恒空缺陷 - P5 审计日志: df-im-gateway audit.rs append-only SQLite(tunnel_audit 表, payload 4KB 截断, recent/by_request 查询), main.rs 8 打点点位(inbound/command/outbound), 写失败不阻断主链路, 17 tests passed - P5 部署运维: publish-im-gateway.ps1(musl 交叉编译→scp .new→systemd 原子切换→9191 验证→自动回滚) + systemd unit + nginx im.1216.top 反代 + 运维手册 - #14 异步委托: 受理/完成两段式 + 桌面任务卡(后台进行中徽章/轮次/切换) + 飞书双端同构, 关联缺陷 #8-#12 速清验收: cargo check workspace EXIT 0, cargo test workspace 全绿(711+560+33+12...), vue-tsc EXIT 0
2026-10-06 15:59:10 +08:00
parent 7c252cff96
commit 438d47bc9f
17 changed files with 987 additions and 12 deletions
@@ -0,0 +1,22 @@
+[Unit]
+Description=DevFlow IM Gateway (飞书/企微/钉钉/Web 四端远程接入)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=/opt/df-im-gateway
+ExecStart=/opt/df-im-gateway/df-im-gateway
+Restart=on-failure
+RestartSec=3
+# ── 运行时 env(与本地 config.rs 对齐;token 与 u-relay 同源)──
+Environment=GATEWAY_RELAY_URL=ws://127.0.0.1:9180/ws/commander
+Environment=GATEWAY_RELAY_TOKEN=devflow-relay-default-token
+Environment=GATEWAY_DEFAULT_CHANNEL=chan-work
+Environment=GATEWAY_DATA_DIR=/opt/df-im-gateway/data
+Environment=RUST_LOG=info,df_im_gateway=debug
+StandardOutput=append:/opt/df-im-gateway/gateway.log
+StandardError=append:/opt/df-im-gateway/gateway.err.log
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,23 @@
+# df-im-gateway nginx 反代片段(与 u-relay 同机,复用 *.1216.top 通配符证书)
+# 放入 /etc/nginx/conf.d/ 后 nginx -t && systemctl reload nginx
+# Web Chat 页面 + WS 同端口 9191(网关 axum 直接托管)
+server {
+    listen 443 ssl;
+    server_name im.1216.top;
+
+    ssl_certificate     /etc/nginx/sslkey/_.1216.top.pem;
+    ssl_certificate_key /etc/nginx/sslkey/_.1216.top.key;
+
+    location / {
+        proxy_pass http://127.0.0.1:9191;
+        proxy_http_version 1.1;
+        # WS 升级头(与 u-relay 反代同款)
+        proxy_set_header Upgrade $http_upgrade;
+        proxy_set_header Connection "upgrade";
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        # 流式响应不缓冲(SSE/流式文本)
+        proxy_buffering off;
+        proxy_read_timeout 300s;
+    }
+}
@@ -0,0 +1,86 @@
+# df-im-gateway 运维手册
+
+> 四端 IM 网关（飞书/企微/钉钉/Web）远程接入 DevFlow AI Chat。
+> 部署拓扑：与 u-relay 同机（39.99.243.191），WS Commander 经 u-relay 连桌面端 df-tunnel。
+
+## 一、拓扑与环境
+
+```
+飞书/企微/钉钉/Web ──► df-im-gateway(9191, 同机)
+                          │ ws://127.0.0.1:9180/ws/commander (Commander 身份)
+                          ▼
+                      u-relay(9180, nginx 前置 wss://u-relay.1216.top)
+                          │ /ws/agent
+                          ▼
+                      桌面端 DevFlow(df-tunnel Agent)
+```
+
+| 项 | 值 |
+|---|---|
+| 服务器 | 39.99.243.191:22 (root 免密, Ubuntu 22.04) |
+| 部署路径 | /opt/df-im-gateway/（二进制 + data/ + 日志） |
+| 监听端口 | 9191（本机明文，nginx 前置 im.1216.top → wss） |
+| systemd | df-im-gateway.service（Restart=on-failure, 开机自启） |
+| token | devflow-relay-default-token（与 u-relay/桌面端三端同源） |
+| 数据库 | /opt/df-im-gateway/data/{sessions.db, audit.db}（SQLite WAL） |
+
+## 二、日常运维
+
+```bash
+# 状态 / 日志 / 重启
+systemctl status df-im-gateway
+journalctl -u df-im-gateway -n 50 --no-pager
+systemctl restart df-im-gateway
+
+# 端口验证
+ss -tlnp | grep 9191
+
+# 审计查询(最近 20 条)
+sqlite3 /opt/df-im-gateway/data/audit.db \
+  "SELECT datetime(ts/1000,'unixepoch','+8 hours'), direction, action, chat_id, outcome FROM tunnel_audit ORDER BY id DESC LIMIT 20;"
+
+# 会话映射表
+sqlite3 /opt/df-im-gateway/data/sessions.db "SELECT * FROM session_map;"
+```
+
+⚠️ systemd 托管，勿 kill Main PID、勿 nohup 起野进程；重启一律 `systemctl restart`。
+
+## 三、发布流程
+
+```powershell
+# 本机(Windows)一键发布：单测 → musl 交叉编译 → scp → 原子切换 → 9191 验证 → 失败自动回滚
+powershell -File E:\wk-lab\devflow\crates\df-im-gateway\scripts\publish-im-gateway.ps1
+# 仅构建不部署: 加 -SkipDeploy；仅部署不构建: 加 -SkipBuild
+```
+
+首次部署需手动铺底：
+```bash
+scp crates/df-im-gateway/deploy/df-im-gateway.service root@39.99.243.191:/etc/systemd/system/
+ssh root@39.99.243.191 'systemctl daemon-reload && systemctl enable df-im-gateway'
+# nginx(可选,Web 端公网访问时): scp deploy/nginx-im.1216.top.conf → /etc/nginx/conf.d/ && nginx -t && systemctl reload nginx
+```
+
+## 四、回滚
+
+```bash
+cd /opt/df-im-gateway
+systemctl stop df-im-gateway
+LATEST=$(ls -t df-im-gateway.bak.* | head -1)
+cp "$LATEST" df-im-gateway && chmod +x df-im-gateway
+systemctl start df-im-gateway
+```
+
+## 五、配置变更
+
+| env | 说明 | 改法 |
+|---|---|---|
+| GATEWAY_RELAY_TOKEN | 与 u-relay 三端同源，**改必须同步桌面端** | 改 unit Environment= → daemon-reload → restart |
+| GATEWAY_DEFAULT_CHANNEL | 默认路由 channel | 同上 |
+| GATEWAY_WHITELIST | 用户白名单（空=任何人可用，生产必设！） | 同上 |
+| FEISHU_APP_ID/SECRET | 飞书长连接 | 同上 |
+
+## 六、已知项
+
+- token 仍为 MVP 单共享值（per-user 是 S2 计划）
+- 日志 gateway.log 追加无轮转，可配 logrotate 或定期 truncate
+- 四端 E2E 联调依赖各平台凭证到位（飞书/钉钉/企微）
@@ -0,0 +1,70 @@
+# df-im-gateway 一键发布脚本（复用 publish-u-relay 模式）
+# 用法: powershell -File scripts\publish-im-gateway.ps1 [-SkipBuild] [-SkipDeploy]
+# 流程: cargo test → musl 交叉编译 → scp 上传(.new 中转) → systemd 原子切换 → 9191 验证
+param(
+    [switch]$SkipBuild,
+    [switch]$SkipDeploy
+)
+
+$ErrorActionPreference = "Stop"
+$Repo = "E:/wk-lab/devflow"
+$Target = "x86_64-unknown-linux-musl"
+$Server = "root@39.99.243.191"
+$RemoteDir = "/opt/df-im-gateway"
+$ServiceName = "df-im-gateway"
+$LocalBin = "$Repo/target-musl/$Target/release/df-im-gateway"
+
+function Step($msg) { Write-Host "`n=== $msg ===" -ForegroundColor Cyan }
+
+if (-not $SkipBuild) {
+    Step "Step 0: 单测基线(df-im-gateway)"
+    cargo test -p df-im-gateway --manifest-path "$Repo/Cargo.toml"
+    if ($LASTEXITCODE -ne 0) { throw "单测失败,中止发布" }
+
+    Step "Step 1: musl 交叉编译(rust-lld self-contained)"
+    $env:CARGO_TARGET_DIR = "$Repo/target-musl"
+    cargo build --release -p df-im-gateway --manifest-path "$Repo/Cargo.toml" --target $Target
+    if ($LASTEXITCODE -ne 0) { throw "编译失败,中止发布" }
+    Remove-Item Env:\CARGO_TARGET_DIR -ErrorAction SilentlyContinue
+    if (-not (Test-Path $LocalBin)) { throw "产物不存在: $LocalBin" }
+    $size = [math]::Round((Get-Item $LocalBin).Length / 1MB, 1)
+    Write-Host "产物: $LocalBin (${size}MB)"
+}
+
+if ($SkipDeploy) { Write-Host "`nSkipDeploy: 仅构建完成"; exit 0 }
+
+Step "Step 2: 服务器预检 + 备份"
+ssh $Server "mkdir -p $RemoteDir && test -f $RemoteDir/df-im-gateway && cp $RemoteDir/df-im-gateway $RemoteDir/df-im-gateway.bak.`$(date +%Y%m%d%H%M%S) || echo '首次部署,无备份'"
+
+Step "Step 3: 上传(.new 中转防 Text file busy)"
+scp $LocalBin "${Server}:$RemoteDir/df-im-gateway.new"
+if ($LASTEXITCODE -ne 0) { throw "scp 失败" }
+
+Step "Step 4: 原子切换 + systemd 重启"
+ssh $Server "bash -s" @'
+set -e
+cd /opt/df-im-gateway
+systemctl stop df-im-gateway 2>/dev/null || true
+mv -f df-im-gateway.new df-im-gateway
+chmod +x df-im-gateway
+systemctl start df-im-gateway
+sleep 1
+systemctl is-active df-im-gateway
+'@
+if ($LASTEXITCODE -ne 0) { throw "切换/启动失败,检查 journalctl -u df-im-gateway"; }
+
+Step "Step 5: 健康验证"
+ssh $Server "ss -tlnp | grep 9191 && journalctl -u df-im-gateway -n 5 --no-pager | sed 's/\x1b\[[0-9;]*m//g'"
+if ($LASTEXITCODE -ne 0) {
+    Write-Host "❌ 9191 未监听,执行回滚..." -ForegroundColor Red
+    ssh $Server "bash -s" @'
+set -e
+cd /opt/df-im-gateway
+systemctl stop df-im-gateway
+LATEST=$(ls -t df-im-gateway.bak.* 2>/dev/null | head -1)
+if [ -n "$LATEST" ]; then cp "$LATEST" df-im-gateway; chmod +x df-im-gateway; systemctl start df-im-gateway; echo "回滚完成: $LATEST"; else echo "无备份可回滚(首次部署失败,保持停止)"; fi
+'@
+    throw "健康检查失败,已回滚"
+}
+
+Write-Host "`n✅ df-im-gateway 发布完成 (9191)" -ForegroundColor Green
@@ -0,0 +1,236 @@
+//! 审计日志(P5):IM 消息 ↔ TunnelCommand 全量记录(SQLite + WAL,与 SessionStore 同机库文件)
+//!
+//! 设计口径:
+//! - **append-only**:只写不改,无 UPDATE/DELETE——审计的公信力来自不可篡改。
+//! - **双向打点**:inbound(IM 用户 → 网关)与 outbound(网关/桌面端 → IM)各记一行,
+//!   同一 request_id 关联请求与响应(排障时按 id 串链)。
+//! - **失败不阻断**:审计写入失败仅 tracing::warn,绝不影响主链路(审计是增益非闸门)。
+//! - 表:tunnel_audit(id 自增 / ts 毫秒 / direction / platform / chat_id / user_id /
+//!   conv_id / action / request_id / payload 截断 4KB / outcome ok|error|denied)。
+//!   payload 截断防长消息撑爆库;敏感凭证不经过此层(config 层已隔离)。
+//!
+//! 查询接口(运维用):recent 按时间倒序限量;by_request 串同请求双向链。
+
+use std::path::Path;
+use std::sync::Arc;
+
+use rusqlite::Connection;
+
+const MAX_PAYLOAD: usize = 4096;
+
+/// 审计方向
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Direction {
+    /// IM 用户 → 网关(入站消息/审批点击/新会话)
+    Inbound,
+    /// 网关/桌面端 → IM(流式输出/审批卡片/通知)
+    Outbound,
+    /// 网关 → 桌面端(TunnelCommand 下发)
+    Command,
+}
+
+impl Direction {
+    fn as_str(self) -> &'static str {
+        match self {
+            Direction::Inbound => "inbound",
+            Direction::Outbound => "outbound",
+            Direction::Command => "command",
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct AuditLog {
+    conn: Arc<std::sync::Mutex<Connection>>,
+}
+
+/// 打点参数(字段可选,None 写空串,保持列 NOT NULL 简化查询)
+pub struct AuditEntry {
+    pub direction: Direction,
+    pub platform: String,
+    pub chat_id: String,
+    pub user_id: String,
+    pub conv_id: String,
+    /// 动作名:message/new_session/approval_click/approval_card/stream/completed/notice/send_cmd/approve_cmd
+    pub action: String,
+    /// 关联 ID:入站↔出站串链用(消息场景可用 chat+ts 派生,审批场景用 approval_id)
+    pub request_id: String,
+    pub payload: String,
+    /// ok / error / denied(白名单拒/无 channel)
+    pub outcome: String,
+}
+
+impl AuditLog {
+    /// 打开(或创建)审计库。默认与 sessions.db 同目录独立文件(隔离写放大)。
+    pub fn open(path: &Path) -> anyhow::Result<Self> {
+        let conn = Connection::open(path)?;
+        conn.pragma_update(None, "journal_mode", "WAL")?;
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS tunnel_audit (
+                id         INTEGER PRIMARY KEY AUTOINCREMENT,
+                ts         INTEGER NOT NULL,
+                direction  TEXT NOT NULL,
+                platform   TEXT NOT NULL DEFAULT '',
+                chat_id    TEXT NOT NULL DEFAULT '',
+                user_id    TEXT NOT NULL DEFAULT '',
+                conv_id    TEXT NOT NULL DEFAULT '',
+                action     TEXT NOT NULL,
+                request_id TEXT NOT NULL DEFAULT '',
+                payload    TEXT NOT NULL DEFAULT '',
+                outcome    TEXT NOT NULL DEFAULT 'ok'
+            );
+            CREATE INDEX IF NOT EXISTS idx_audit_ts ON tunnel_audit(ts);
+            CREATE INDEX IF NOT EXISTS idx_audit_request ON tunnel_audit(request_id);",
+        )?;
+        Ok(Self { conn: Arc::new(std::sync::Mutex::new(conn)) })
+    }
+
+    /// 内存库(测试)
+    #[allow(dead_code)]
+    pub fn open_in_memory() -> anyhow::Result<Self> {
+        let conn = Connection::open_in_memory()?;
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS tunnel_audit (
+                id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL,
+                direction TEXT NOT NULL, platform TEXT NOT NULL DEFAULT '',
+                chat_id TEXT NOT NULL DEFAULT '', user_id TEXT NOT NULL DEFAULT '',
+                conv_id TEXT NOT NULL DEFAULT '', action TEXT NOT NULL,
+                request_id TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL DEFAULT '',
+                outcome TEXT NOT NULL DEFAULT 'ok');",
+        )?;
+        Ok(Self { conn: Arc::new(std::sync::Mutex::new(conn)) })
+    }
+
+    fn now_ms() -> i64 {
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as i64)
+            .unwrap_or(0)
+    }
+
+    /// 追加一条审计(失败仅告警不冒泡)
+    pub fn record(&self, e: AuditEntry) {
+        let payload = if e.payload.len() > MAX_PAYLOAD {
+            format!("{}…[截断]", &e.payload[..MAX_PAYLOAD])
+        } else {
+            e.payload
+        };
+        let result = (|| -> anyhow::Result<()> {
+            let conn = self.conn.lock().map_err(|err| anyhow::anyhow!("audit 锁中毒: {err}"))?;
+            conn.execute(
+                "INSERT INTO tunnel_audit (ts, direction, platform, chat_id, user_id, conv_id, action, request_id, payload, outcome) \
+                 VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",
+                rusqlite::params![
+                    Self::now_ms(), e.direction.as_str(), e.platform, e.chat_id,
+                    e.user_id, e.conv_id, e.action, e.request_id, payload, e.outcome,
+                ],
+            )?;
+            Ok(())
+        })();
+        if let Err(err) = result {
+            tracing::warn!("[audit] 写入失败(不影响主链路): {err}");
+        }
+    }
+
+    /// 最近 N 条(时间倒序,运维排障)
+    #[allow(dead_code)]
+    pub fn recent(&self, limit: usize) -> anyhow::Result<Vec<String>> {
+        let conn = self.conn.lock().map_err(|e| anyhow::anyhow!("audit 锁中毒: {e}"))?;
+        let mut stmt = conn.prepare(
+            "SELECT ts, direction, platform, chat_id, user_id, conv_id, action, request_id, outcome \
+             FROM tunnel_audit ORDER BY id DESC LIMIT ?1",
+        )?;
+        let rows = stmt.query_map([limit as i64], |row| {
+            Ok(format!(
+                "{} [{}] {} plat={} chat={} user={} conv={} req={} outcome={}",
+                row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(6)?,
+                row.get::<_, String>(2)?, row.get::<_, String>(3)?, row.get::<_, String>(4)?,
+                row.get::<_, String>(5)?, row.get::<_, String>(7)?, row.get::<_, String>(8)?,
+            ))
+        })?;
+        Ok(rows.filter_map(|r| r.ok()).collect())
+    }
+
+    /// 按 request_id 串双向链(入站→命令→出站全链)
+    #[allow(dead_code)]
+    pub fn by_request(&self, request_id: &str) -> anyhow::Result<Vec<String>> {
+        let conn = self.conn.lock().map_err(|e| anyhow::anyhow!("audit 锁中毒: {e}"))?;
+        let mut stmt = conn.prepare(
+            "SELECT ts, direction, action, outcome FROM tunnel_audit WHERE request_id = ?1 ORDER BY id ASC",
+        )?;
+        let rows = stmt.query_map([request_id], |row| {
+            Ok(format!(
+                "{} [{}] {} outcome={}",
+                row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?,
+            ))
+        })?;
+        Ok(rows.filter_map(|r| r.ok()).collect())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn entry(direction: Direction, action: &str, req: &str, outcome: &str) -> AuditEntry {
+        AuditEntry {
+            direction,
+            platform: "feishu".into(),
+            chat_id: "oc_123".into(),
+            user_id: "u_1".into(),
+            conv_id: "feishu:oc_123".into(),
+            action: action.into(),
+            request_id: req.into(),
+            payload: "hello".into(),
+            outcome: outcome.into(),
+        }
+    }
+
+    #[test]
+    fn record_and_recent_roundtrip() {
+        let log = AuditLog::open_in_memory().unwrap();
+        log.record(entry(Direction::Inbound, "message", "req-1", "ok"));
+        log.record(entry(Direction::Command, "send_cmd", "req-1", "ok"));
+        log.record(entry(Direction::Outbound, "stream", "req-1", "ok"));
+        log.record(entry(Direction::Inbound, "message", "req-2", "denied"));
+        let rows = log.recent(10).unwrap();
+        assert_eq!(rows.len(), 4);
+        // 倒序:最新在前
+        assert!(rows[0].contains("req-2"));
+        assert!(rows[0].contains("denied"));
+        let chain = log.by_request("req-1").unwrap();
+        assert_eq!(chain.len(), 3);
+    }
+
+    #[test]
+    fn payload_truncated() {
+        let log = AuditLog::open_in_memory().unwrap();
+        let mut e = entry(Direction::Inbound, "message", "req-x", "ok");
+        e.payload = "x".repeat(10_000);
+        log.record(e);
+        let conn_count = |log: &AuditLog| -> i64 {
+            let conn = log.conn.lock().unwrap();
+            conn.query_row("SELECT COUNT(*) FROM tunnel_audit", [], |r| r.get(0)).unwrap()
+        };
+        assert_eq!(conn_count(&log), 1);
+        let conn = log.conn.lock().unwrap();
+        let stored: String = conn
+            .query_row("SELECT payload FROM tunnel_audit", [], |r| r.get(0))
+            .unwrap();
+        assert!(stored.len() < 5_000);
+        assert!(stored.ends_with("…[截断]"));
+    }
+
+    #[test]
+    fn record_never_panics_on_poisoned_lock() {
+        let log = AuditLog::open_in_memory().unwrap();
+        // 毒锁:持锁 panic 使 Mutex poisoned,record 应 warn 而非 panic
+        let guard = log.conn.lock().unwrap();
+        std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            std::panic::panic_any("poison");
+        }))
+        .err();
+        drop(guard);
+        log.record(entry(Direction::Inbound, "message", "req-p", "ok")); // 不 panic 即通过
+    }
+}
@@ -9,6 +9,7 @@
 //! ```

 mod adapters;
+mod audit;
 mod commander;
 mod config;
 mod feishu;
@@ -43,6 +44,8 @@ async fn main() -> anyhow::Result<()> {
        .unwrap_or_else(|_| PathBuf::from("data"));
    std::fs::create_dir_all(&data_dir)?;
    let store = SessionStore::open(&data_dir.join("sessions.db"))?;
+    // 审计日志(P5):IM 消息 ↔ TunnelCommand 全量记录,写失败不阻断主链路
+    let audit = audit::AuditLog::open(&data_dir.join("audit.db"))?;

    info_banner(&cfg);

@@ -153,11 +156,25 @@ async fn main() -> anyhow::Result<()> {
                        let tail = throttle.finish(&conv_id);
                        let full = tail.unwrap_or_default();
                        let idx = platform_of(&conv_id);
+                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Outbound,
+                            platform: String::new(), chat_id: conv_id.clone(),
+                            user_id: String::new(), conv_id: conv_id.clone(),
+                            action: "completed".into(), request_id: String::new(),
+                            payload: full.clone(), outcome: "ok".into(),
+                        });
                        let _ = adapters[idx]
                            .dispatch(OutboundAction::StreamText { chat_id: conv_id, text: full, done: true })
                            .await;
                    }
                    df_tunnel::TunnelEvent::Approval { conv_id, approval_id, description } => {
+                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Outbound,
+                            platform: String::new(), chat_id: conv_id.clone(),
+                            user_id: String::new(), conv_id: conv_id.clone(),
+                            action: "approval_card".into(), request_id: approval_id.clone(),
+                            payload: description.clone(), outcome: "ok".into(),
+                        });
                        let idx = platform_of(&conv_id);
                        let _ = adapters[idx]
                            .dispatch(OutboundAction::ApprovalCard {
@@ -182,12 +199,26 @@ async fn main() -> anyhow::Result<()> {
                        // 回复适配器:按来源平台回路由
                        let reply_idx = platform_of(&platform);
                        if !cfg_loop.whitelist.is_empty() && !cfg_loop.whitelist.contains(&user_id) {
+                            audit.record(audit::AuditEntry {
+                                direction: audit::Direction::Inbound,
+                                platform: platform.clone(), chat_id: chat_id.clone(),
+                                user_id: user_id.clone(), conv_id: String::new(),
+                                action: "message".into(), request_id: String::new(),
+                                payload: text.clone(), outcome: "denied".into(),
+                            });
                            let _ = adapters[reply_idx]
                                .dispatch(OutboundAction::Notice { chat_id, text: "⛔ 你不在白名单内".into() })
                                .await;
                            continue;
                        }
                        let Some(channel_id) = cfg_loop.route(&text) else {
+                            audit.record(audit::AuditEntry {
+                                direction: audit::Direction::Inbound,
+                                platform: platform.clone(), chat_id: chat_id.clone(),
+                                user_id: user_id.clone(), conv_id: String::new(),
+                                action: "message".into(), request_id: String::new(),
+                                payload: text.clone(), outcome: "error".into(),
+                            });
                            let _ = adapters[reply_idx]
                                .dispatch(OutboundAction::Notice { chat_id, text: "⚠️ 无可用 channel".into() })
                                .await;
@@ -202,14 +233,45 @@ async fn main() -> anyhow::Result<()> {
                                // 走查 #3:新建会话时清理流式残留(上轮异常中止无
                                // Completed 事件,streams entry 永驻会错覆写旧消息)
                                adapters[reply_idx].reset_stream(&chat_id).await;
+                                audit.record(audit::AuditEntry {
+                                    direction: audit::Direction::Inbound,
+                                    platform: platform.clone(), chat_id: chat_id.clone(),
+                                    user_id: user_id.clone(), conv_id: conv.clone(),
+                                    action: "new_session".into(), request_id: String::new(),
+                                    payload: String::new(), outcome: "ok".into(),
+                                });
                                conv
                            }
                        };
+                        // 入站消息审计:request_id 用 conv+毫秒派生,出站 Completed 事件无法
+                        // 回带,此处以 (direction=inbound,message) 为链头,command/outbound
+                        // 侧用同一 conv_id 关联(recent 按时间倒序即还原会话时间线)。
+                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Inbound,
+                            platform: platform.clone(), chat_id: chat_id.clone(),
+                            user_id: user_id.clone(), conv_id: conv_id.clone(),
+                            action: "message".into(), request_id: String::new(),
+                            payload: text.clone(), outcome: "ok".into(),
+                        });
+                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Command,
+                            platform: platform.clone(), chat_id: chat_id.clone(),
+                            user_id: user_id.clone(), conv_id: conv_id.clone(),
+                            action: "send_cmd".into(), request_id: String::new(),
+                            payload: text.clone(), outcome: "ok".into(),
+                        });
                        let _ = cmd_tx
                            .send(df_tunnel::TunnelCommand::Send { conv_id, content: text })
                            .await;
                    }
                    InboundEvent::NewSession { platform, chat_id } => {
+                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Inbound,
+                            platform: platform.clone(), chat_id: chat_id.clone(),
+                            user_id: String::new(), conv_id: String::new(),
+                            action: "new_session_reset".into(), request_id: String::new(),
+                            payload: String::new(), outcome: "ok".into(),
+                        });
                        let _ = store.remove(&platform, &chat_id);
                        let reply_idx = platform_of(&platform);
                        let _ = adapters[reply_idx]
@@ -227,6 +289,19 @@ async fn main() -> anyhow::Result<()> {
                            .get_conv(&platform, &chat_id)
                            .map(|m| m.conv_id)
                            .unwrap_or_else(|| format!("{platform}:{chat_id}"));
+                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Inbound,
+                            platform: platform.clone(), chat_id: chat_id.clone(),
+                            user_id: String::new(), conv_id: conv_id.clone(),
+                            action: "approval_click".into(), request_id: approval_id.clone(),
+                            payload: if approved { "approved".into() } else { "rejected".into() },
+                            outcome: "ok".into(),
+                        });                        audit.record(audit::AuditEntry {
+                            direction: audit::Direction::Command,
+                            platform, chat_id, user_id: String::new(), conv_id: conv_id.clone(),
+                            action: "approve_cmd".into(), request_id: approval_id.clone(),                            payload: if approved { "approved".into() } else { "rejected".into() },
+                            outcome: "ok".into(),
+                        });
                        let _ = cmd_tx
                            .send(df_tunnel::TunnelCommand::Approve { conv_id, approval_id, accepted: approved })
                            .await;