test(integration): combine PR #30-#38 for session dev validation

PR_INTEGRATION_REPORT.md

@@ -0,0 +1,76 @@
+# OpenAB Plus：开放 PR 整合与验证报告
+
+**报告日期：** 2026-08-12  
+**改造分支：** `agent/pr-integration-20260812`  
+**基线：** `upstream/main` 的 `a03d9de5`  
+**整合头提交：** `ba7925a5`  
+**范围：** 当前开放的六个 PR（#30–#35）。[1]
+
+> 本报告记录的是一条**隔离的本地改造分支**。未向 `upstream/main` 推送、合并或创建远程 PR，因此上游仓库状态未被改变。
+
+## 1. 整合结论
+
+六个开放 PR 均已抓取并按功能依赖顺序尝试合并至隔离分支。由于 #30 与 #31 是同一会话工作台方向的两套重叠实现，且 #32、#33、#35 都会更新详情页、状态映射或事件流，原始顺序合并会产生多处内容冲突。改造分支保留了较完整的 #30 工作台主实现，并吸收 #31 的兼容模块、#32 的活动流代码、#33 的历史与来源链接修复、#34 的 transcript 存储及只读流，以及 #35 的冷启动回放与集中式状态映射。
+
+在首次合并后，前端出现状态映射 API 不一致，后端出现来源链接回填 API 的参数与返回值不一致。两类问题均已在整合分支修复，并通过前端类型检查、前端单测、Rust 工作区静态检查及核心/网关定向测试验证。
+
+| PR | 标题 | 处理结果 | 整合提交 | 关键处理 |
+|---|---|---|---|---|
+| #30 | 三栏只读会话工作台骨架 | 已整合 | `eec52862` | 保留 main 的详情页和已构建产物，接入工作台页面、面板与会话列表能力。 |
+| #31 | 用工作台替代 sessions table | 已整合（兼容部分） | `ae12f3d9` | 与 #30 为重叠实现；保留 #30 的活动工作台，吸收 #31 非冲突组件、列表及数据逻辑。 |
+| #32 | AionUi 风格会话活动流 | 已整合 | `0e692192` | 接入活动流组件、工具调用规范化、文件差异和终端输出展示能力。 |
+| #33 | 修复详情历史与来源链接 | 已整合 | `dd4144b7` | 接入服务端历史/来源修复、状态与详情页面能力，并保留工作台列表。 |
+| #34 | transcript 存储和只读流 | 已整合 | `65496780` | 接入 transcript 存储、快照接口、带 generation 的 SSE 流和管理端测试。 |
+| #35 | 冷启动回放、来源链接与状态映射 | 已整合 | `885e0bcf` | 无 `Last-Event-ID` 时从序列零回放；统一状态展示与来源跳转。 |
+
+## 2. 关键冲突解决
+
+### 2.1 工作台 UI 的双实现
+
+#30 与 #31 都从较早的共同基线引入会话工作台，涉及同名路由、会话表组件、布局、样式、依赖锁文件及生成的 `web/app.js`。本次整合将 #30 作为运行时的主工作台实现，以避免删除较新的主线详情页；#31 的无冲突列表、Inspector、MainPanel、Sidebar 与会话数据逻辑仍被吸收。生成产物在最终前端构建后重新生成，而不是直接采用任一过期 PR 的构建文件。[2] [3]
+
+### 2.2 Transcript 流与冷启动历史回放
+
+#34 将管理端事件流从原有事件总线演进为带 generation 的 `SessionStreamBus`；#35 在同一区域加入无游标冷启动时从序列零回放历史的行为。最终实现使用 `SessionStreamBus` 和 generation 校验，且在无 `Last-Event-ID` 时订阅并回放序列零之后的保留历史。对于旧 generation 的游标，仍返回显式 reset 事件而非错误重放。[5] [6]
+
+### 2.3 来源链接回填 API
+
+#34 需要在创建快照时支持可选来源链接，#35 需要让适配器可以幂等回填链接并据此判断是否发布变化。整合后，`SessionSnapshot::set_source_permalink` 接受 `Option<&str>`、只在来源值确有变化时返回 `true`，且不修改 `updated_at`，从而同时满足快照创建、延迟回填和幂等行为。
+
+## 3. 验证结果
+
+| 验证项目 | 结果 | 说明 |
+|---|---|---|
+| Git 冲突标记扫描 | 通过 | 未发现 `<<<<<<<`、`=======` 或 `>>>>>>>` 标记。 |
+| `git diff --check upstream/main...HEAD` | 通过 | 无空白错误。 |
+| Web 依赖安装 | 通过 | 使用锁文件完成 `pnpm --dir web install --frozen-lockfile`。 |
+| Web 类型检查 | 通过 | `pnpm --dir web lint` 成功。 |
+| Web 单元测试 | 通过 | Vitest：**8 个测试文件、39 个用例全部通过**。 |
+| Web 生产构建 | 通过 | `pnpm --dir web build` 成功；Vite 提示部分 bundle 大于 500 kB，属于性能告警而非构建失败。 |
+| Rust 工作区静态检查 | 通过 | `cargo check --workspace` 成功。 |
+| Rust 核心定向测试 | 通过 | `cargo test -p openab-core --no-default-features --lib --jobs 1`：**544 通过、0 失败**。 |
+| Rust 网关定向测试 | 通过 | `cargo test -p openab-gateway --no-default-features --lib --jobs 1`：**30 通过、0 失败**。 |
+| 核心模块严格 Clippy | 通过 | `cargo clippy -p openab-core --no-default-features --lib -- -D warnings` 成功。 |
+| 修改文件格式检查 | 通过 | `session_snapshot.rs` 和 `session_admin.rs` 的定向 `rustfmt --check` 成功。 |
+
+## 4. 已知边界与后续工作
+
+严格的 `cargo clippy -p openab-gateway --no-default-features --lib -- -D warnings` 未通过，但诊断集中在 `crates/openab-gateway/src/lib.rs` 与 `media.rs` 中原有的未使用导入、未使用变量和 dead-code 项，并非本次修改的 transcript、SSE 或来源链接文件。应将这些 22 项作为单独的网关清理任务处理，避免把无关重构混入本次 PR 整合。
+
+完整 `cargo test --workspace` 在首次全量链接/测试构建期间，`aws-sdk-s3` 的编译进程收到 `SIGTERM` 而中止，并非测试断言失败。为避免云依赖的重型构建掩盖本次代码验证，已完成并通过核心与网关的定向测试。建议在资源更充足的 CI runner 上补跑完整工作区测试，以及平台特性组合测试。
+
+Vite 的生产构建成功，但生成的 `app.js` 约 2.3 MB（gzip 后约 744 kB）并触发大 chunk 告警。后续可通过路由级动态导入或手工分包，降低首屏加载体积。
+
+## 5. 建议的下一步
+
+建议将本地分支推送到团队可写的远程 fork 或集成仓库后，创建一条仅指向该改造分支的审查 PR。审查时应重点验证：工作台中 #30/#31 重叠 UI 的最终视觉选择、#34/#35 的冷启动 SSE 回放、Slack/Discord 来源链接回填，以及在 CI 中补跑全量 Rust 工作区测试。
+
+## References
+
+[1]: https://github.com/ZeroPointSix/openab-plus/pulls "ZeroPointSix/openab-plus Pull Requests"
+[2]: https://github.com/ZeroPointSix/openab-plus/pull/30 "PR #30 — ZER-404 W1: 三栏只读会话工作台骨架"
+[3]: https://github.com/ZeroPointSix/openab-plus/pull/31 "PR #31 — feat(web): replace sessions table with workbench"
+[4]: https://github.com/ZeroPointSix/openab-plus/pull/32 "PR #32 — feat(web): add AionUi-inspired session activity feed"
+[5]: https://github.com/ZeroPointSix/openab-plus/pull/34 "PR #34 — feat(session): add transcript store and read-only stream"
+[6]: https://github.com/ZeroPointSix/openab-plus/pull/35 "PR #35 — fix: ZER-669 P1 遗留修复"
+[7]: https://github.com/ZeroPointSix/openab-plus/pull/33 "PR #33 — fix: repair session detail history and source links"

crates/openab-core/src/acp/profile_pool.rs

@@ -3,8 +3,9 @@ use super::pool;
 use super::protocol::ConfigOption;
 use crate::agent_profile::{AgentProfileService, ProfileSessionOverrides, RecoveryStrategy};
 use crate::config::AgentConfig;
-use crate::session_event::{SessionEventBus, SessionEventKind};
+use crate::session_event::{SessionEventBus, SessionEventKind, SessionStreamBus};
 use crate::session_snapshot::{SessionRuntimeMetadata, SessionSnapshot, SessionStatus};
+use crate::transcript::SessionTranscriptStore;
 use anyhow::{anyhow, Result};
 use std::collections::HashMap;
 use std::env;
@@ -32,6 +33,8 @@ pub struct SessionPool {
     thread_policies: RwLock<HashMap<String, ThreadProfilePolicy>>,
     thread_gates: RwLock<HashMap<String, Arc<Mutex<()>>>>,
     session_events: SessionEventBus,
+    session_stream: SessionStreamBus,
+    transcripts: SessionTranscriptStore,
     snapshots: RwLock<HashMap<String, SessionSnapshot>>,
     external_base_url: Option<String>,
     #[cfg(any(test, feature = "test-support"))]
@@ -54,6 +57,13 @@ impl SessionPool {
         let mut pools = HashMap::new();
         pools.insert("system".to_string(), system_pool);
         let external_base_url = session_external_base_url_from_env();
+        let transcript_capacity = SessionTranscriptStore::capacity_from_env();
+        // Transcript retention is configurable independently from the existing
+        // lifecycle event history. Keep the lifecycle/SSE replay buffer at its
+        // established default rather than changing it with transcript tuning.
+        let session_stream = SessionStreamBus::default();
+        let session_events = SessionEventBus::new_with_stream(session_stream.clone());
+        let transcripts = SessionTranscriptStore::new(transcript_capacity, session_stream.clone());
         Self {
             base_config: config,
             max_sessions,
@@ -64,7 +74,9 @@ impl SessionPool {
             thread_pools: RwLock::new(HashMap::new()),
             thread_policies: RwLock::new(HashMap::new()),
             thread_gates: RwLock::new(HashMap::new()),
-            session_events: SessionEventBus::default(),
+            session_events,
+            session_stream,
+            transcripts,
             snapshots: RwLock::new(HashMap::new()),
             external_base_url,
             #[cfg(any(test, feature = "test-support"))]
@@ -85,6 +97,16 @@ impl SessionPool {
         self.session_events.clone()
     }
 
+    /// Unified, read-only cursor source for status and transcript SSE events.
+    pub fn session_stream_bus(&self) -> SessionStreamBus {
+        self.session_stream.clone()
+    }
+
+    /// Independent per-session ring buffers for ACP transcript data.
+    pub fn transcript_store(&self) -> SessionTranscriptStore {
+        self.transcripts.clone()
+    }
+
     /// Seed a session snapshot and emit `session.created` for integration tests.
     #[cfg(any(test, feature = "test-support"))]
     pub async fn seed_session_snapshot_for_test(&self, snapshot: SessionSnapshot) {
@@ -124,14 +146,36 @@ impl SessionPool {
         working_dir_override: Option<&str>,
         profile_id: Option<&str>,
         overrides: Option<&ProfileSessionOverrides>,
+    ) -> Result<bool> {
+        self.get_or_create_with_profile_and_source(
+            thread_id,
+            working_dir_override,
+            profile_id,
+            overrides,
+            None,
+        )
+        .await
+    }
+
+    pub async fn get_or_create_with_profile_and_source(
+        &self,
+        thread_id: &str,
+        working_dir_override: Option<&str>,
+        profile_id: Option<&str>,
+        overrides: Option<&ProfileSessionOverrides>,
+        source_permalink: Option<&str>,
     ) -> Result<bool> {
         let gate = self.thread_gate(thread_id).await;
         let _guard = gate.lock().await;
 
         if let Some(pool) = self.existing_pool(thread_id).await {
             let result = pool.get_or_create(thread_id, working_dir_override).await;
             return match result {
-                Ok(outcome) => Ok(self.apply_ensure_outcome(thread_id, outcome).await),
+                Ok(outcome) => {
+                    self.backfill_source_permalink(thread_id, source_permalink)
+                        .await;
+                    Ok(self.apply_ensure_outcome(thread_id, outcome).await)
+                }
                 Err(err) => {
                     self.mark_session_error(thread_id, err.to_string()).await;
                     Err(err)
@@ -210,6 +254,7 @@ impl SessionPool {
                         self.external_base_url.as_deref(),
                     );
                     snapshot.replace_runtime_metadata(runtime_metadata);
+                    snapshot.set_source_permalink(source_permalink);
                     if !profile_config_errors.is_empty() {
                         snapshot.set_profile_config_errors(profile_config_errors);
                     }
@@ -308,7 +353,8 @@ impl SessionPool {
         } else {
             Some(
                 crate::agent_profile::AgentCapabilityResolver::config_schema_from_options(
-                    agent_type, &options,
+                    agent_type,
+                    &options,
                 ),
             )
         }
@@ -453,6 +499,22 @@ impl SessionPool {
         .await;
     }
 
+    /// Backfill the session source permalink (e.g. a Slack/Discord thread URL)
+    /// resolved lazily by the originating adapter.
+    ///
+    /// The permalink is immutable source metadata, so unlike status changes
+    /// this only updates the stored snapshot — no timeline event is emitted.
+    /// REST reads and every subsequent event snapshot carry the value.
+    /// Idempotent: an unchanged permalink skips the write entirely.
+    pub async fn record_session_source_permalink(&self, thread_id: &str, permalink: String) {
+        let mut snapshots = self.snapshots.write().await;
+        if let Some(snapshot) = snapshots.get_mut(thread_id) {
+            if snapshot.source.permalink.as_deref() != Some(permalink.as_str()) {
+                snapshot.set_source_permalink(Some(&permalink));
+            }
+        }
+    }
+
     pub async fn mark_profile_deleted(&self, profile_id: &str) {
         let updated: Vec<SessionSnapshot> = {
             let mut snapshots = self.snapshots.write().await;
@@ -483,6 +545,21 @@ impl SessionPool {
             .publish(SessionEventKind::SessionCreated, snapshot);
     }
 
+    async fn backfill_source_permalink(&self, thread_id: &str, permalink: Option<&str>) {
+        let snapshot = {
+            let mut snapshots = self.snapshots.write().await;
+            let Some(snapshot) = snapshots.get_mut(thread_id) else {
+                return;
+            };
+            if !snapshot.set_source_permalink(permalink) {
+                return;
+            }
+            snapshot.clone()
+        };
+        self.session_events
+            .publish(SessionEventKind::SourceChanged, snapshot);
+    }
+
     async fn update_snapshot<F>(&self, thread_id: &str, kind: SessionEventKind, apply: F)
     where
         F: FnOnce(&mut SessionSnapshot),
@@ -992,7 +1069,9 @@ mod tests {
         ));
         outer.seed_session_snapshot_for_test(snapshot).await;
 
-        outer.record_session_config_update("slack:thread", &[]).await;
+        outer
+            .record_session_config_update("slack:thread", &[])
+            .await;
 
         let snapshot = outer
             .session_snapshot("slack:thread")
@@ -1113,6 +1192,57 @@ mod tests {
         assert_eq!(snapshot.profile_status, Some(ProfileStatus::Deleted));
     }
 
+    #[tokio::test]
+    async fn record_session_source_permalink_backfills_snapshot_source() {
+        let outer = SessionPool::new(AgentConfig::default(), 2, 120, HashMap::new());
+        outer
+            .seed_session_snapshot_for_test(SessionSnapshot::new(
+                "slack:thread".into(),
+                "codex".into(),
+                "/workspace".into(),
+                None,
+                None,
+                None,
+                None,
+            ))
+            .await;
+
+        outer
+            .record_session_source_permalink(
+                "slack:thread",
+                "https://acme.slack.com/archives/C1/p1700000000000100".into(),
+            )
+            .await;
+
+        let snapshot = outer
+            .session_snapshot("slack:thread")
+            .await
+            .expect("snapshot");
+        assert_eq!(
+            snapshot.source.permalink.as_deref(),
+            Some("https://acme.slack.com/archives/C1/p1700000000000100")
+        );
+
+        // Unknown sessions are ignored; an unchanged permalink skips the write.
+        outer
+            .record_session_source_permalink("slack:missing", "https://x".into())
+            .await;
+        outer
+            .record_session_source_permalink(
+                "slack:thread",
+                "https://acme.slack.com/archives/C1/p1700000000000100".into(),
+            )
+            .await;
+        let snapshot = outer
+            .session_snapshot("slack:thread")
+            .await
+            .expect("snapshot");
+        assert_eq!(
+            snapshot.source.permalink.as_deref(),
+            Some("https://acme.slack.com/archives/C1/p1700000000000100")
+        );
+    }
+
     #[tokio::test]
     async fn config_schema_for_agent_uses_live_options_from_matching_session() {
         let outer = SessionPool::new(AgentConfig::default(), 2, 120, HashMap::new());