From e1e2f676380b9a938de3b7a18f4022fc060b4048 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=86=A0=E8=BE=B0?= Date: Wed, 3 Jun 2026 14:44:10 +0800 Subject: [PATCH 1/4] feat(artifacts): publish ArtifactPending on create + bridge to artifact_pending socket event (#3162) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Today the chat surface only learns an artifact exists when `artifacts::store::finalize_artifact` (or `fail_artifact`) publishes `DomainEvent::ArtifactReady` / `ArtifactFailed`. For tools that take real time to produce a file (e.g. `generate_presentation`) the user sees nothing between hitting send and the card appearing — no "Generating..." surface, no proof the dispatch even happened. Sub-task AC#1 of #3162. This PR wires the missing pending-state event end-to-end on the Rust side. The frontend (ArtifactCard render path) is a focused follow-up. - **`DomainEvent::ArtifactPending`** (new variant in `src/core/event_bus/events.rs`): same shape as `ArtifactReady` minus `size_bytes` (no file yet), with the same `thread_id` / `client_id` routing fields. Lives under the existing `"artifact"` domain so the existing web-channel subscriber's filter picks it up without re-registration. Stable-name table extended for observability/tracing. - **`artifacts::store::create_artifact`** now publishes `ArtifactPending` immediately after the meta row is reserved. Reads the routing pair from `APPROVAL_CHAT_CONTEXT` the same way `finalize` / `fail` already do, so CLI / cron / sub-agent paths surface `None` and the web bridge silently drops the event — same degradation contract as the existing Ready/Failed events. - **`channels::providers::web::ArtifactSurfaceSubscriber`** now bridges `ArtifactPending` → `artifact_pending` socket event with the same `artifact_id` the future Ready/Failed will carry, so the frontend can swap the card in place when the matching terminal event lands (no flicker). Tests: - `artifacts::store::tests::create_artifact_publishes_artifact_pending_event` — spins up the event bus, subscribes a typed collector, calls `create_artifact`, asserts exactly one `ArtifactPending` lands with the right kind/title/workspace_dir/path and that `thread_id` / `client_id` are `None` for the unbound test task (the expected no-chat-context degradation). - Existing `artifacts::` (45/45) + `channels::providers::web` (63/63) pass — no regressions. Run: `GGML_NATIVE=OFF cargo test -p openhuman --lib artifacts:: channels::providers::web` — all green. `cargo fmt --all --check` passes. --- src/core/event_bus/events.rs | 39 ++++++++- src/openhuman/artifacts/store.rs | 28 +++++++ src/openhuman/artifacts/store_tests.rs | 106 ++++++++++++++++++++++++ src/openhuman/channels/providers/web.rs | 57 +++++++++++-- 4 files changed, 221 insertions(+), 9 deletions(-) diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index f0e3a402d7..3767dcf70d 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -408,6 +408,40 @@ pub enum DomainEvent { thread_id: Option, client_id: Option, }, + /// An artifact record has been **created** (`ArtifactStatus::Pending`) + /// but no bytes are on disk yet — the producing tool has only just + /// reserved the row. Published by + /// [`crate::openhuman::artifacts::store::create_artifact`]. + /// Bridged to the web channel as an `artifact_pending` socket event + /// so the frontend can render an in-progress / "Generating…" card the + /// moment the tool dispatches, instead of waiting until the file + /// arrives via [`Self::ArtifactReady`]. The pending card is replaced + /// in place when the matching `ArtifactReady` / `ArtifactFailed` + /// event with the same `artifact_id` arrives. Sub-task #3162 of #1535. + ArtifactPending { + /// UUID of the freshly-created artifact record. + artifact_id: String, + /// Lowercase variant of `ArtifactKind` (`presentation`, + /// `document`, `image`, `other`). + kind: String, + /// Human-readable title (also the on-disk filename stem). + title: String, + /// Absolute workspace root the artifact belongs to — see + /// [`Self::ArtifactReady::workspace_dir`] for rationale. + workspace_dir: String, + /// Relative path under `/artifacts/` where the file + /// *will* land. The frontend uses it to render a stable card key + /// so subsequent `ArtifactReady` can swap the same surface in + /// place without flicker. + path: String, + /// Chat thread the artifact belongs to, when the producing turn + /// carried an `APPROVAL_CHAT_CONTEXT`. `None` for CLI / cron / + /// sub-agent paths — no client to fan out to. + thread_id: Option, + /// Socket.IO client id (room) to surface the card to, when known. + /// `None` for non-chat callers. + client_id: Option, + }, // ── Webhooks ──────────────────────────────────────────────────────── /// An incoming webhook request from the transport layer, ready for routing. @@ -956,7 +990,9 @@ impl DomainEvent { Self::ApprovalRequested { .. } | Self::ApprovalDecided { .. } => "approval", - Self::ArtifactReady { .. } | Self::ArtifactFailed { .. } => "artifact", + Self::ArtifactReady { .. } + | Self::ArtifactFailed { .. } + | Self::ArtifactPending { .. } => "artifact", Self::McpServerInstalled { .. } | Self::McpServerConnected { .. } @@ -1056,6 +1092,7 @@ impl DomainEvent { Self::ApprovalDecided { .. } => "ApprovalDecided", Self::ArtifactReady { .. } => "ArtifactReady", Self::ArtifactFailed { .. } => "ArtifactFailed", + Self::ArtifactPending { .. } => "ArtifactPending", Self::McpServerInstalled { .. } => "McpServerInstalled", Self::McpServerConnected { .. } => "McpServerConnected", Self::McpServerDisconnected { .. } => "McpServerDisconnected", diff --git a/src/openhuman/artifacts/store.rs b/src/openhuman/artifacts/store.rs index 96afc3ae2e..2c549c6c98 100644 --- a/src/openhuman/artifacts/store.rs +++ b/src/openhuman/artifacts/store.rs @@ -288,6 +288,17 @@ fn sanitize_filename_stem(title: &str) -> String { /// `extension` is the file extension WITHOUT the leading dot /// (e.g. `"pptx"`, `"pdf"`). Used to build the rendered filename /// under the artifact directory. +/// +/// Publishes [`DomainEvent::ArtifactPending`] on the global bus the +/// moment the row is reserved so the chat surface can render an +/// in-progress / "Generating…" card immediately (#3162). When the +/// matching [`finalize_artifact`] / [`fail_artifact`] later fires it +/// reuses the same `artifact_id`, so the card swaps in place without +/// flicker. Same chat-context routing rules as the Ready/Failed pair — +/// `thread_id` / `client_id` come from the +/// [`crate::openhuman::approval::ApprovalChatContext`] task-local and +/// are `None` for CLI / cron / sub-agent paths, in which case the web +/// bridge silently drops the event for lack of a routing target. pub async fn create_artifact( workspace_dir: &Path, kind: super::types::ArtifactKind, @@ -342,6 +353,23 @@ pub async fn create_artifact( meta.kind.as_str(), absolute_path ); + + // Surface the "Generating…" card the moment the row is reserved so + // the user doesn't stare at an empty composer until the tool finishes + // (#3162). When `finalize_artifact` / `fail_artifact` later fires the + // matching Ready/Failed event with the same `artifact_id`, the + // frontend can swap the card in place. + let (thread_id, client_id) = current_chat_context(); + crate::core::event_bus::publish_global(crate::core::event_bus::DomainEvent::ArtifactPending { + artifact_id: meta.id.clone(), + kind: meta.kind.as_str().to_string(), + title: meta.title.clone(), + workspace_dir: workspace_dir.to_string_lossy().into_owned(), + path: meta.path.clone(), + thread_id, + client_id, + }); + Ok((meta, absolute_path)) } diff --git a/src/openhuman/artifacts/store_tests.rs b/src/openhuman/artifacts/store_tests.rs index 48452acc56..166677c92c 100644 --- a/src/openhuman/artifacts/store_tests.rs +++ b/src/openhuman/artifacts/store_tests.rs @@ -190,3 +190,109 @@ async fn validate_artifact_id_rejects_slashes() { "unexpected error: {err}" ); } + +// ── create_artifact event publication (#3162) ───────────────────────────── + +use crate::core::event_bus::{ + init_global, subscribe_global, DomainEvent, EventHandler, SubscriptionHandle, +}; +use async_trait::async_trait; +use std::sync::{Arc, Mutex as StdMutex}; + +#[derive(Clone)] +struct PendingCollector { + events: Arc>>, +} + +impl PendingCollector { + fn new() -> Self { + Self { + events: Arc::new(StdMutex::new(Vec::new())), + } + } + + fn subscribe(&self) -> Option { + subscribe_global(Arc::new(self.clone())) + } + + fn snapshot(&self) -> Vec { + self.events.lock().unwrap().clone() + } +} + +#[async_trait] +impl EventHandler for PendingCollector { + fn name(&self) -> &str { + "test::pending_collector" + } + + async fn handle(&self, event: &DomainEvent) { + // Only capture artifact-domain events so the receive buffer can't + // overflow with unrelated traffic from neighbouring tests. + if matches!( + event, + DomainEvent::ArtifactPending { .. } + | DomainEvent::ArtifactReady { .. } + | DomainEvent::ArtifactFailed { .. } + ) { + self.events.lock().unwrap().push(event.clone()); + } + } +} + +/// #3162: `create_artifact` publishes `DomainEvent::ArtifactPending` +/// the moment the row is reserved, so the chat surface can render an +/// in-progress "Generating…" card before the file lands on disk. +#[tokio::test] +async fn create_artifact_publishes_artifact_pending_event() { + init_global(256); + let collector = PendingCollector::new(); + let _handle = collector.subscribe(); + + let tmp = TempDir::new().unwrap(); + let (meta, _path) = create_artifact(tmp.path(), ArtifactKind::Presentation, "Q3 Deck", "pptx") + .await + .expect("create_artifact succeeds"); + + // The bus is broadcast-based and processed off-task — yield until the + // subscriber's tokio task drains the message. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + if !collector.snapshot().is_empty() { + break; + } + if std::time::Instant::now() >= deadline { + panic!("ArtifactPending event was not observed within 2s"); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + + let events = collector.snapshot(); + assert_eq!( + events.len(), + 1, + "exactly one event expected, got {events:?}" + ); + let DomainEvent::ArtifactPending { + artifact_id, + kind, + title, + workspace_dir, + path, + thread_id, + client_id, + } = &events[0] + else { + panic!("expected ArtifactPending, got {:?}", events[0]); + }; + assert_eq!(*artifact_id, meta.id); + assert_eq!(kind, "presentation"); + assert_eq!(title, "Q3 Deck"); + assert_eq!(*workspace_dir, tmp.path().to_string_lossy()); + assert_eq!(*path, meta.path); + // No chat context bound on this test task, so the routing fields are + // None — the web bridge drops the event in that case, which is the + // intended degradation path for CLI / cron / sub-agent callers. + assert!(thread_id.is_none(), "thread_id leaked, got {thread_id:?}"); + assert!(client_id.is_none(), "client_id leaked, got {client_id:?}"); +} diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index 434e8f98b8..d8f365dc9b 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -67,13 +67,22 @@ pub fn register_approval_surface_subscriber() { static ARTIFACT_SURFACE_HANDLE: OnceLock = OnceLock::new(); /// Bridge artifact lifecycle events onto the web channel. -/// `DomainEvent::ArtifactReady` / `ArtifactFailed` (published by -/// `artifacts::store::{finalize,fail}_artifact` from #2778's producer -/// surface) carry the thread_id + client_id when the producing turn -/// ran under an `APPROVAL_CHAT_CONTEXT`. When present, fan out as an -/// `artifact_ready` / `artifact_failed` socket event so the frontend -/// `chatRuntimeSlice` can upsert the snapshot and the `ArtifactCard` -/// can render in the message timeline. Sub-task #2779 of #1535. +/// `DomainEvent::ArtifactPending` / `ArtifactReady` / `ArtifactFailed` +/// (published by `artifacts::store::{create,finalize,fail}_artifact`) +/// carry the thread_id + client_id when the producing turn ran under an +/// `APPROVAL_CHAT_CONTEXT`. When present, fan out as an +/// `artifact_pending` / `artifact_ready` / `artifact_failed` socket +/// event so the frontend `chatRuntimeSlice` can upsert the snapshot and +/// the `ArtifactCard` can render in the message timeline: +/// +/// - `artifact_pending` → render an in-progress "Generating…" card the +/// moment the producing tool dispatches (#3162). +/// - `artifact_ready` → swap the same card to a download surface when +/// the file lands (#2779). +/// - `artifact_failed` → swap to a retry-hint card on producer error. +/// +/// The card is keyed on `artifact_id`, so the Pending → Ready/Failed +/// transition reuses the same surface instead of flickering a new one. /// Idempotent. No-op for non-chat events (thread/client id absent). pub fn register_artifact_surface_subscriber() { if ARTIFACT_SURFACE_HANDLE.get().is_some() { @@ -83,7 +92,7 @@ pub fn register_artifact_surface_subscriber() { Some(handle) => { let _ = ARTIFACT_SURFACE_HANDLE.set(handle); log::info!( - "[web-channel] artifact-surface subscriber registered (domain=artifact) — will bridge ArtifactReady/Failed → artifact_ready/artifact_failed socket events" + "[web-channel] artifact-surface subscriber registered (domain=artifact) — will bridge ArtifactPending/Ready/Failed → artifact_pending/artifact_ready/artifact_failed socket events" ); } None => { @@ -175,6 +184,38 @@ impl EventHandler for ArtifactSurfaceSubscriber { ..Default::default() }); } + DomainEvent::ArtifactPending { + artifact_id, + kind, + title, + workspace_dir, + path, + thread_id, + client_id, + } => { + let (Some(thread_id), Some(client_id)) = (thread_id, client_id) else { + log::debug!( + "[web-channel] artifact-surface skip ArtifactPending id={artifact_id}: no chat context" + ); + return; + }; + log::info!( + "[web-channel] artifact-surface emitting artifact_pending id={artifact_id} kind={kind} thread_id={thread_id} client_id={client_id}" + ); + publish_web_channel_event(WebChannelEvent { + event: "artifact_pending".to_string(), + client_id: client_id.clone(), + thread_id: thread_id.clone(), + args: Some(serde_json::json!({ + "artifact_id": artifact_id, + "kind": kind, + "title": title, + "workspace_dir": workspace_dir, + "path": path, + })), + ..Default::default() + }); + } _ => {} } } From 7e86adf0eea69263b249fa7fbb8d1adf9e4946d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=86=A0=E8=BE=B0?= Date: Wed, 3 Jun 2026 15:01:45 +0800 Subject: [PATCH 2/4] test(artifacts): scope PendingCollector to this test's artifact_id (#3162) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address CodeRabbit feedback on #3277: the previous PendingCollector subscribed to the process-wide event stream and asserted on the full snapshot, so a parallel `cargo test --lib` run that also exercised create_artifact / finalize_artifact / fail_artifact could land unrelated events in the same buffer and fail this test's `events.len() == 1` check. Two changes, both hardening: - `PendingCollector::domains() -> Some(&["artifact"])` so the bus filters before delivery — non-artifact traffic never reaches this subscriber's receive buffer at all. - Wait loop + assertion now filter the snapshot by `artifact_id` == `meta.id` && `workspace_dir` == our tempdir before counting, so a concurrent test publishing a sibling artifact event can no longer pollute the assertion. Local: `GGML_NATIVE=OFF cargo test -p openhuman --lib artifacts::store::tests::create_artifact_publishes_artifact_pending_event` — passes. --- src/openhuman/artifacts/store_tests.rs | 56 ++++++++++++++++++++------ 1 file changed, 43 insertions(+), 13 deletions(-) diff --git a/src/openhuman/artifacts/store_tests.rs b/src/openhuman/artifacts/store_tests.rs index 166677c92c..30ccb26c78 100644 --- a/src/openhuman/artifacts/store_tests.rs +++ b/src/openhuman/artifacts/store_tests.rs @@ -226,9 +226,17 @@ impl EventHandler for PendingCollector { "test::pending_collector" } + /// Filter at the bus boundary so the broadcast channel never delivers + /// non-artifact traffic to this subscriber — keeps the per-test + /// receive buffer small even when other tests pump unrelated events + /// in parallel. + fn domains(&self) -> Option<&[&str]> { + Some(&["artifact"]) + } + async fn handle(&self, event: &DomainEvent) { - // Only capture artifact-domain events so the receive buffer can't - // overflow with unrelated traffic from neighbouring tests. + // domains() filter guarantees only artifact-domain variants + // arrive, but match defensively in case the enum grows. if matches!( event, DomainEvent::ArtifactPending { .. } @@ -253,25 +261,47 @@ async fn create_artifact_publishes_artifact_pending_event() { let (meta, _path) = create_artifact(tmp.path(), ArtifactKind::Presentation, "Q3 Deck", "pptx") .await .expect("create_artifact succeeds"); - - // The bus is broadcast-based and processed off-task — yield until the - // subscriber's tokio task drains the message. + let expected_workspace = tmp.path().to_string_lossy().into_owned(); + + // The bus is broadcast-based and processed off-task — wait until the + // subscriber's tokio task delivers OUR event. Filtering by + // artifact_id + workspace_dir keeps the test robust against parallel + // `cargo test` runs that may publish unrelated artifact lifecycle + // events into the same process-wide bus. + let matches_this_artifact = |event: &DomainEvent| { + matches!( + event, + DomainEvent::ArtifactPending { + artifact_id, + workspace_dir, + .. + } if artifact_id == &meta.id && workspace_dir == &expected_workspace + ) + }; let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); loop { - if !collector.snapshot().is_empty() { + if collector.snapshot().iter().any(matches_this_artifact) { break; } if std::time::Instant::now() >= deadline { - panic!("ArtifactPending event was not observed within 2s"); + panic!( + "ArtifactPending event for id={} was not observed within 2s", + meta.id + ); } tokio::time::sleep(std::time::Duration::from_millis(10)).await; } - let events = collector.snapshot(); + let mine: Vec = collector + .snapshot() + .into_iter() + .filter(matches_this_artifact) + .collect(); assert_eq!( - events.len(), + mine.len(), 1, - "exactly one event expected, got {events:?}" + "exactly one ArtifactPending for {} expected, got {mine:?}", + meta.id ); let DomainEvent::ArtifactPending { artifact_id, @@ -281,14 +311,14 @@ async fn create_artifact_publishes_artifact_pending_event() { path, thread_id, client_id, - } = &events[0] + } = &mine[0] else { - panic!("expected ArtifactPending, got {:?}", events[0]); + unreachable!("filter pinned us to ArtifactPending"); }; assert_eq!(*artifact_id, meta.id); assert_eq!(kind, "presentation"); assert_eq!(title, "Q3 Deck"); - assert_eq!(*workspace_dir, tmp.path().to_string_lossy()); + assert_eq!(*workspace_dir, expected_workspace); assert_eq!(*path, meta.path); // No chat context bound on this test task, so the routing fields are // None — the web bridge drops the event in that case, which is the From fe8fd294f7c1e050f7d237189d62b8b671a8e1fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=86=A0=E8=BE=B0?= Date: Wed, 3 Jun 2026 15:47:44 +0800 Subject: [PATCH 3/4] ci: retrigger after flaky chat-management E2E lane 1 (#3162) From 82d3b1035d476e88155d8e00e91448cfa1b0e0c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=86=A0=E8=BE=B0?= Date: Wed, 3 Jun 2026 16:25:20 +0800 Subject: [PATCH 4/4] ci: retrigger after process-global rpc_mutations flake in Rust Core Coverage (#3162)