mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
feat(desktop): add thread-scoped ACP session experiment (#6909)
## Why Thread-scoped ACP sessions from #6732 need an opt-in desktop rollout that preserves today’s channel policy by default. ## What - Add the default-off “Thread Scoped ACP Sessions” toggle under Settings → Experiments using the existing feature-override persistence. - Map the persisted setting to `BUZZ_ACP_SESSION_POLICY=channel|thread` for every local and provider-backed managed ACP launch. Changes apply when managed agents next start; DMs remain conversation-scoped by the backend. - Keep the desktop setting authoritative over descriptor environment and cover the UI, persistence, local spawn, provider payload, and existing Projects/Workflows experiments. ## Risk Assessment Low. The experiment defaults off and explicitly preserves `channel`; changes are limited to managed-agent launch configuration. Existing running agents are unchanged until their next start. ## Testing - `. ./bin/activate-hermit && just ci` — passed on the final restacked tree, including 5,674 desktop tests, 2,796 Tauri tests, and 1,860 mobile tests. - `cd desktop && pnpm build:e2e && pnpm exec playwright test tests/e2e/experimental-features.spec.ts --project=smoke` — 1 passed. - `. ./bin/activate-hermit && cargo test --manifest-path desktop/src-tauri/Cargo.toml session_policy --lib` — 4 passed. - `. ./bin/activate-hermit && cargo test --manifest-path desktop/src-tauri/Cargo.toml commands::agents::deploy::tests --lib` — 15 passed. - `. ./bin/activate-hermit && cargo test -p buzz-backend-kubernetes --test wire_fixtures` — 4 passed. ## Stack Info Stacked on #6732. This PR depends on #6732 and should merge after it. Generated with Codex --------- Signed-off-by: Salman Mohammed <smohammed@squareup.com> Signed-off-by: Leo <5faf251baee50ee6bcde338aef6acdd70bb3e60115664c2cd490d94a55dfc488@buzz.block.builderlab.xyz> Co-authored-by: Leo <5faf251baee50ee6bcde338aef6acdd70bb3e60115664c2cd490d94a55dfc488@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -26,6 +26,7 @@
|
||||
"BUZZ_ACP_LAZY_POOL": "true",
|
||||
"BUZZ_ACP_MODEL": "gpt-5",
|
||||
"BUZZ_ACP_RELAY_OBSERVER": "true",
|
||||
"BUZZ_ACP_SESSION_POLICY": "channel",
|
||||
"BUZZ_ACP_SESSION_TITLE": "worker",
|
||||
"GOOSE_MODE": "auto"
|
||||
}
|
||||
|
||||
@@ -49,6 +49,7 @@ export default defineConfig({
|
||||
"**/channel-controls.spec.ts",
|
||||
"**/channel-activity-popover.spec.ts",
|
||||
"**/active-turn-resilience.spec.ts",
|
||||
"**/agent-control-regressions.spec.ts",
|
||||
"**/profile-active-turn.spec.ts",
|
||||
"**/config-bridge-screenshots.spec.ts",
|
||||
"**/observer-feed-screenshots.spec.ts",
|
||||
@@ -147,6 +148,7 @@ export default defineConfig({
|
||||
"**/profile-backup-settings.spec.ts",
|
||||
"**/signout-confirmation.spec.ts",
|
||||
"**/settings-section-layout.spec.ts",
|
||||
"**/experimental-features.spec.ts",
|
||||
"**/agent-provider-dropdowns.spec.ts",
|
||||
"**/agent-lifecycle-feedback.spec.ts",
|
||||
"**/agent-access-warning.spec.ts",
|
||||
|
||||
@@ -36,8 +36,8 @@ pub struct AppState {
|
||||
pub workspace_apply_generation: AtomicU64,
|
||||
/// Defers managed-agent restore until `apply_workspace` installs relay and identity.
|
||||
pub managed_agent_restore_pending: AtomicBool,
|
||||
/// Disabled by agent-managed profiles so agent profile updates survive start/restore.
|
||||
pub managed_agent_profile_reconcile_enabled: AtomicBool,
|
||||
/// Experiment state applied to managed-agent starts and profile reconciliation.
|
||||
pub managed_agent_experiments: crate::managed_agents::ManagedAgentExperimentState,
|
||||
/// Shared shutdown signal checked by launch-time agent restoration.
|
||||
pub shutdown_started: AtomicBool,
|
||||
/// Serializes every managed-runtime transition that changes the protected
|
||||
@@ -207,7 +207,7 @@ pub fn build_app_state() -> AppState {
|
||||
workspace_apply_lock: Arc::new(AsyncMutex::new(())),
|
||||
workspace_apply_generation: AtomicU64::new(0),
|
||||
managed_agent_restore_pending: AtomicBool::new(false),
|
||||
managed_agent_profile_reconcile_enabled: AtomicBool::new(true),
|
||||
managed_agent_experiments: crate::managed_agents::ManagedAgentExperimentState::default(),
|
||||
shutdown_started: AtomicBool::new(false),
|
||||
managed_agent_runtime_transition: Mutex::new(()),
|
||||
identity_mutation: Mutex::new(()),
|
||||
|
||||
@@ -13,10 +13,17 @@ use crate::{
|
||||
#[tauri::command]
|
||||
pub fn set_agent_managed_profiles(enabled: bool, state: State<'_, AppState>) {
|
||||
state
|
||||
.managed_agent_profile_reconcile_enabled
|
||||
.managed_agent_profile_reconcile_enabled()
|
||||
.store(!enabled, Ordering::Release);
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub fn set_thread_scoped_acp_sessions(enabled: bool, state: State<'_, AppState>) {
|
||||
state
|
||||
.thread_scoped_acp_sessions_enabled()
|
||||
.store(enabled, Ordering::Release);
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn set_managed_agent_start_on_app_launch(
|
||||
pubkey: String,
|
||||
|
||||
@@ -1012,7 +1012,7 @@ pub async fn start_managed_agent(
|
||||
// with no persisted avatar, this also backfills the avatar from the relay.
|
||||
if result.is_ok()
|
||||
&& state
|
||||
.managed_agent_profile_reconcile_enabled
|
||||
.managed_agent_profile_reconcile_enabled()
|
||||
.load(std::sync::atomic::Ordering::Acquire)
|
||||
{
|
||||
let reconcile_pubkey = pubkey.clone();
|
||||
|
||||
@@ -43,16 +43,17 @@ pub(crate) fn resolve_deploy_model_provider(
|
||||
|
||||
/// Serialize the portable launch contract shared with provider-backed agents.
|
||||
///
|
||||
/// `descriptor.env` is the authoritative six-layer environment. Policy values
|
||||
/// are deliberately separate because providers apply them below that layered
|
||||
/// environment, preserving the local spawn's power-user override semantics.
|
||||
pub(super) fn build_launch_block(
|
||||
/// `descriptor.env` is the authoritative six-layer environment for ordinary
|
||||
/// values. Desktop-owned settings are reserved, stripped from that layer, and
|
||||
/// emitted through `policy_env` so local and provider launches agree.
|
||||
fn build_launch_block_for_policy(
|
||||
record: &ManagedAgentRecord,
|
||||
descriptor: &crate::managed_agents::readiness::EffectiveHarnessDescriptor,
|
||||
teams: &[crate::managed_agents::TeamRecord],
|
||||
effective_prompt: Option<&str>,
|
||||
effective_model: Option<&str>,
|
||||
owner_pubkey: &str,
|
||||
session_policy: crate::managed_agents::AcpSessionPolicy,
|
||||
) -> serde_json::Value {
|
||||
use crate::managed_agents::{
|
||||
known_acp_runtime, resolve_session_title, DISPLAY_NAME_ENV_VAR, SESSION_TITLE_ENV_VAR,
|
||||
@@ -78,6 +79,7 @@ pub(super) fn build_launch_block(
|
||||
"BUZZ_ACP_AGENTS".into(),
|
||||
crate::managed_agents::acp_agents_value(&descriptor.command, record.parallelism),
|
||||
);
|
||||
crate::managed_agents::insert_acp_session_policy_env(&mut policy_env, session_policy);
|
||||
|
||||
if let Some(value) = effective_prompt {
|
||||
policy_env.insert("BUZZ_ACP_SYSTEM_PROMPT".into(), value.to_string());
|
||||
@@ -138,7 +140,8 @@ pub(super) fn build_launch_block(
|
||||
// matching local, where `apply_claude_model_env(None)` removes both.
|
||||
let is_claude = runtime.map(|r| r.id == "claude").unwrap_or(false);
|
||||
let strip_key = |k: &str| {
|
||||
(record.effort_level.is_some() && k.eq_ignore_ascii_case("BUZZ_ACP_EFFORT_LEVEL"))
|
||||
k.eq_ignore_ascii_case(crate::managed_agents::ACP_SESSION_POLICY_ENV_VAR)
|
||||
|| (record.effort_level.is_some() && k.eq_ignore_ascii_case("BUZZ_ACP_EFFORT_LEVEL"))
|
||||
|| (is_claude
|
||||
&& (k.eq_ignore_ascii_case("BUZZ_ACP_MODEL")
|
||||
|| k.eq_ignore_ascii_case("ANTHROPIC_MODEL")))
|
||||
@@ -159,6 +162,26 @@ pub(super) fn build_launch_block(
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) fn build_launch_block(
|
||||
record: &ManagedAgentRecord,
|
||||
descriptor: &crate::managed_agents::readiness::EffectiveHarnessDescriptor,
|
||||
teams: &[crate::managed_agents::TeamRecord],
|
||||
effective_prompt: Option<&str>,
|
||||
effective_model: Option<&str>,
|
||||
owner_pubkey: &str,
|
||||
) -> serde_json::Value {
|
||||
build_launch_block_for_policy(
|
||||
record,
|
||||
descriptor,
|
||||
teams,
|
||||
effective_prompt,
|
||||
effective_model,
|
||||
owner_pubkey,
|
||||
crate::managed_agents::AcpSessionPolicy::Channel,
|
||||
)
|
||||
}
|
||||
|
||||
pub(super) fn ensure_remote_provider_supported(provider: Option<&str>) -> Result<(), String> {
|
||||
if provider.map(str::trim) == Some(crate::managed_agents::RELAY_MESH_PROVIDER_ID) {
|
||||
return Err(
|
||||
@@ -198,13 +221,14 @@ pub(crate) fn build_deploy_payload<R: tauri::Runtime>(
|
||||
crate::managed_agents::resolve_effective_harness_descriptor(record, &personas, &global)
|
||||
.map_err(|error| crate::managed_agents::user_facing_harness_error(&error))?;
|
||||
let owner_pubkey = super::workspace_owner_hex(state)?;
|
||||
let launch = build_launch_block(
|
||||
let launch = build_launch_block_for_policy(
|
||||
record,
|
||||
&descriptor,
|
||||
&teams,
|
||||
effective.system_prompt.value.as_deref(),
|
||||
effective.model.value.as_deref(),
|
||||
&owner_pubkey,
|
||||
crate::managed_agents::acp_session_policy(state),
|
||||
);
|
||||
|
||||
let effective_parallelism =
|
||||
@@ -342,9 +366,40 @@ mod tests {
|
||||
assert_eq!(launch["policy_env"]["BUZZ_ACP_IDLE_TIMEOUT"], "17");
|
||||
assert_eq!(launch["policy_env"]["BUZZ_ACP_MAX_TURN_DURATION"], "23");
|
||||
assert_eq!(launch["policy_env"]["BUZZ_ACP_AGENTS"], "4");
|
||||
assert_eq!(launch["policy_env"]["BUZZ_ACP_SESSION_POLICY"], "channel");
|
||||
assert_eq!(launch["owner_pubkey"], "owner-hex");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn launch_block_thread_policy_is_authoritative_and_preserves_unrelated_env() {
|
||||
let record = record();
|
||||
let descriptor = EffectiveHarnessDescriptor {
|
||||
command: "goose".into(),
|
||||
args: vec![],
|
||||
env: BTreeMap::from([
|
||||
("BUZZ_ACP_SESSION_POLICY".to_string(), "channel".to_string()),
|
||||
("KEEP_ME".to_string(), "yes".to_string()),
|
||||
]),
|
||||
};
|
||||
|
||||
let launch = build_launch_block_for_policy(
|
||||
&record,
|
||||
&descriptor,
|
||||
&[],
|
||||
None,
|
||||
None,
|
||||
"owner-hex",
|
||||
crate::managed_agents::AcpSessionPolicy::Thread,
|
||||
);
|
||||
|
||||
assert_eq!(launch["policy_env"]["BUZZ_ACP_SESSION_POLICY"], "thread");
|
||||
assert!(
|
||||
launch["env"]["BUZZ_ACP_SESSION_POLICY"].is_null(),
|
||||
"desktop policy must not be shadowed by descriptor env"
|
||||
);
|
||||
assert_eq!(launch["env"]["KEEP_ME"], "yes");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn launch_block_claude_runtime_uses_anthropic_model_not_buzz_acp_model() {
|
||||
// B2: remote claude deploys must send ANTHROPIC_MODEL, not BUZZ_ACP_MODEL,
|
||||
|
||||
@@ -206,7 +206,7 @@ pub(crate) async fn reconcile_agent_profile(
|
||||
);
|
||||
|
||||
if !state
|
||||
.managed_agent_profile_reconcile_enabled
|
||||
.managed_agent_profile_reconcile_enabled()
|
||||
.load(std::sync::atomic::Ordering::Acquire)
|
||||
{
|
||||
return Ok(ProfileReconcileOutcome::SkippedDisabled);
|
||||
@@ -273,7 +273,7 @@ pub(crate) async fn reconcile_agent_profile(
|
||||
.map_err(|e| format!("failed to parse agent keys: {e}"))?;
|
||||
|
||||
if !state
|
||||
.managed_agent_profile_reconcile_enabled
|
||||
.managed_agent_profile_reconcile_enabled()
|
||||
.load(std::sync::atomic::Ordering::Acquire)
|
||||
{
|
||||
return Ok(ProfileReconcileOutcome::SkippedDisabled);
|
||||
|
||||
@@ -155,6 +155,7 @@ pub async fn apply_workspace(
|
||||
nsec: Option<String>,
|
||||
repos_dir: Option<String>,
|
||||
agent_managed_profiles: Option<bool>,
|
||||
thread_scoped_acp_sessions: Option<bool>,
|
||||
app: AppHandle,
|
||||
) -> Result<(), String> {
|
||||
let state = app.state::<AppState>();
|
||||
@@ -228,8 +229,15 @@ pub async fn apply_workspace(
|
||||
// experiment before launch-time restore can spawn any agents. Missing
|
||||
// means the stable behavior: desktop remains authoritative.
|
||||
state
|
||||
.managed_agent_profile_reconcile_enabled
|
||||
.managed_agent_profile_reconcile_enabled()
|
||||
.store(!agent_managed_profiles.unwrap_or(false), Ordering::Release);
|
||||
// Persisted frontend experiment state must land before launch-time
|
||||
// restore so every restored agent starts with the selected ACP policy.
|
||||
// Missing preserves the stable channel-scoped behavior.
|
||||
state.thread_scoped_acp_sessions_enabled().store(
|
||||
thread_scoped_acp_sessions.unwrap_or(false),
|
||||
Ordering::Release,
|
||||
);
|
||||
|
||||
// ── Filesystem side-effect (non-fatal) ────────────────────────────────
|
||||
// Persist the *effective* repos_dir (None when the candidate failed
|
||||
|
||||
@@ -700,6 +700,7 @@ pub fn run() {
|
||||
start_managed_agent,
|
||||
stop_managed_agent,
|
||||
set_agent_managed_profiles,
|
||||
set_thread_scoped_acp_sessions,
|
||||
set_managed_agent_start_on_app_launch,
|
||||
set_managed_agent_auto_restart,
|
||||
delete_managed_agent,
|
||||
|
||||
@@ -175,6 +175,13 @@ fn reserved_keys_include_remote_lifetime_policy() {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reserved_keys_include_desktop_acp_session_policy() {
|
||||
assert!(is_reserved_env_key("BUZZ_ACP_SESSION_POLICY"));
|
||||
let agent = map(&[("BUZZ_ACP_SESSION_POLICY", "thread")]);
|
||||
assert!(merged_user_env(&BTreeMap::new(), &agent).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reserved_keys_include_code_execution_surface() {
|
||||
// The agent/MCP command + args are what Buzz actually exec's.
|
||||
|
||||
@@ -37,6 +37,7 @@ pub mod retention;
|
||||
mod runtime;
|
||||
mod runtime_commands;
|
||||
mod runtime_types;
|
||||
mod session_policy;
|
||||
pub(crate) mod snapshot_avatar;
|
||||
pub(crate) mod spawn_snapshot;
|
||||
pub(crate) mod storage;
|
||||
@@ -89,6 +90,10 @@ pub use restore::*;
|
||||
pub use runtime::*;
|
||||
pub use runtime_commands::*;
|
||||
pub use runtime_types::*;
|
||||
pub(crate) use session_policy::{
|
||||
acp_session_policy, apply_app_acp_session_policy_env, insert_acp_session_policy_env,
|
||||
AcpSessionPolicy, ManagedAgentExperimentState, ACP_SESSION_POLICY_ENV_VAR,
|
||||
};
|
||||
pub use storage::*;
|
||||
pub use teams::*;
|
||||
pub use types::*;
|
||||
|
||||
@@ -62,6 +62,9 @@ pub(crate) const RESERVED_ENV_KEYS: &[&str] = &[
|
||||
// Desktop-owned pool lifetime policy: user env must not disable or reset
|
||||
// the idle worker-reclamation window while the desktop launcher sets it.
|
||||
"BUZZ_ACP_IDLE_POOL_SLEEP",
|
||||
// Desktop experiment policy: the Settings toggle is the sole authority
|
||||
// for whether channel threads receive independent ACP sessions.
|
||||
"BUZZ_ACP_SESSION_POLICY",
|
||||
"BUZZ_ACP_NO_PRESENCE",
|
||||
// Readiness handoff: desktop is the ONLY readiness source. A saved or
|
||||
// ambient env var must not be able to forge setup mode (NotReady) on a
|
||||
|
||||
@@ -494,7 +494,7 @@ fn profile_reconcile_completed(outcome: crate::commands::ProfileReconcileOutcome
|
||||
pub(crate) fn spawn_pending_profile_reconciliations(app: &tauri::AppHandle, workspace_relay: &str) {
|
||||
let state = app.state::<AppState>();
|
||||
if !state
|
||||
.managed_agent_profile_reconcile_enabled
|
||||
.managed_agent_profile_reconcile_enabled()
|
||||
.load(Ordering::Acquire)
|
||||
{
|
||||
return;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use std::collections::HashMap;
|
||||
|
||||
use tauri::AppHandle;
|
||||
use tauri::{AppHandle, Manager};
|
||||
|
||||
use super::agent_env::{build_buzz_agent_provider_defaults, idle_pool_sleep_env};
|
||||
|
||||
@@ -23,8 +23,8 @@ pub(crate) use super::access_policy::{build_respond_to_env_with_policy, RespondT
|
||||
|
||||
mod metadata;
|
||||
pub(crate) use metadata::{
|
||||
apply_agent_display_env, resolve_session_title, runtime_metadata_env_vars,
|
||||
DISPLAY_NAME_ENV_VAR, SESSION_TITLE_ENV_VAR,
|
||||
apply_agent_display_env, child_rust_log_filter, resolve_session_title,
|
||||
runtime_metadata_env_vars, DISPLAY_NAME_ENV_VAR, SESSION_TITLE_ENV_VAR,
|
||||
};
|
||||
|
||||
mod stop;
|
||||
@@ -109,7 +109,6 @@ pub(crate) fn workspace_pair_key(
|
||||
app: &AppHandle,
|
||||
record: &ManagedAgentRecord,
|
||||
) -> Option<ManagedAgentRuntimeKey> {
|
||||
use tauri::Manager;
|
||||
let state = app.state::<crate::app_state::AppState>();
|
||||
resolve_workspace_pair_key(
|
||||
&record.pubkey,
|
||||
@@ -254,6 +253,7 @@ pub fn build_managed_agent_summary(
|
||||
&key.relay_url,
|
||||
global_config,
|
||||
super::owner_only_access_build(),
|
||||
super::acp_session_policy(app.state::<crate::app_state::AppState>().inner()),
|
||||
);
|
||||
(runtime, current)
|
||||
});
|
||||
@@ -808,8 +808,10 @@ pub fn spawn_agent_child(
|
||||
for (key, value) in &descriptor.env {
|
||||
command.env(key, value);
|
||||
}
|
||||
crate::build_identity::apply_demo_config_home(&mut command)?;
|
||||
// Resolve once and stamp the same value onto the snapshot below.
|
||||
let acp_session_policy = super::apply_app_acp_session_policy_env(app, &mut command);
|
||||
|
||||
crate::build_identity::apply_demo_config_home(&mut command)?;
|
||||
// B5: carry persisted effort; harness resolves thought_level configId at first session.
|
||||
// Written AFTER descriptor.env so the canonical persisted value wins over any
|
||||
// user-supplied BUZZ_ACP_EFFORT_LEVEL entry, mirroring the A1 model-authority pattern
|
||||
@@ -861,6 +863,7 @@ pub fn spawn_agent_child(
|
||||
model: effective_model.as_deref(),
|
||||
provider: effective_provider.as_deref(),
|
||||
enforced_owner_only: super::owner_only_access_build(),
|
||||
session_policy: acp_session_policy,
|
||||
},
|
||||
);
|
||||
|
||||
@@ -928,14 +931,6 @@ pub fn spawn_agent_child(
|
||||
})
|
||||
}
|
||||
|
||||
fn child_rust_log_filter() -> String {
|
||||
match std::env::var("RUST_LOG") {
|
||||
Ok(existing) if existing.contains("buzz_acp") => existing,
|
||||
Ok(existing) if !existing.trim().is_empty() => format!("{existing},buzz_acp=info"),
|
||||
_ => "buzz_acp=info".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn (or adopt) the runtime pair for `record` on the caller's bound
|
||||
/// workspace relay. `workspace_relay` can only be produced by
|
||||
/// `bind_expected_relay_scope`, so this spawn consumes — by construction — the
|
||||
|
||||
@@ -74,6 +74,17 @@ pub(crate) fn resolve_session_title(display_name: Option<&str>, name: &str) -> O
|
||||
.find(|value| !value.is_empty())
|
||||
}
|
||||
|
||||
/// Build the `RUST_LOG` value forwarded to the agent child: keep an existing
|
||||
/// filter that already mentions `buzz_acp`, append `buzz_acp=info` to any other
|
||||
/// non-empty filter, and default to `buzz_acp=info` when unset.
|
||||
pub(crate) fn child_rust_log_filter() -> String {
|
||||
match std::env::var("RUST_LOG") {
|
||||
Ok(existing) if existing.contains("buzz_acp") => existing,
|
||||
Ok(existing) if !existing.trim().is_empty() => format!("{existing},buzz_acp=info"),
|
||||
_ => "buzz_acp=info".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::resolve_session_title;
|
||||
|
||||
@@ -1209,12 +1209,11 @@ fn minimal_record(pubkey: &str) -> crate::managed_agents::ManagedAgentRecord {
|
||||
|
||||
fn make_pair_runtime_placeholder() -> crate::managed_agents::ManagedAgentPairRuntime {
|
||||
use std::process::{Command, Stdio};
|
||||
// Spawn a real child so ManagedAgentProcess's Child field is satisfied.
|
||||
// `true` exits immediately with 0 — just a handle we need for type purposes.
|
||||
// Absolute `/usr/bin/true` on unix (present on both macOS and Linux):
|
||||
// parallel tests holding `lock_path_mutex` swap PATH to a tempdir, and a
|
||||
// bare `true` lookup during that window fails with NotFound (observed
|
||||
// flake). Windows keeps the PATH lookup — no test there swaps PATH.
|
||||
// Spawn a real child so ManagedAgentProcess's Child field is satisfied;
|
||||
// `true` exits immediately with 0. Absolute `/usr/bin/true` on unix (both
|
||||
// macOS and Linux): parallel tests holding `lock_path_mutex` swap PATH to a
|
||||
// tempdir, and a bare `true` lookup during that window fails NotFound
|
||||
// (observed flake). Windows keeps the PATH lookup — no test there swaps it.
|
||||
#[cfg(unix)]
|
||||
let program = "/usr/bin/true";
|
||||
#[cfg(windows)]
|
||||
@@ -1235,6 +1234,7 @@ fn make_pair_runtime_placeholder() -> crate::managed_agents::ManagedAgentPairRun
|
||||
"wss://relay.example",
|
||||
&Default::default(),
|
||||
false,
|
||||
crate::managed_agents::AcpSessionPolicy::Channel,
|
||||
),
|
||||
setup_mode: false,
|
||||
adapter_availability: None,
|
||||
|
||||
140
desktop/src-tauri/src/managed_agents/session_policy.rs
Normal file
140
desktop/src-tauri/src/managed_agents/session_policy.rs
Normal file
@@ -0,0 +1,140 @@
|
||||
use std::{
|
||||
collections::BTreeMap,
|
||||
sync::atomic::{AtomicBool, Ordering},
|
||||
};
|
||||
|
||||
use tauri::{AppHandle, Manager};
|
||||
|
||||
use crate::app_state::AppState;
|
||||
|
||||
pub(crate) const ACP_SESSION_POLICY_ENV_VAR: &str = "BUZZ_ACP_SESSION_POLICY";
|
||||
|
||||
/// Desktop experiment state that influences managed-agent lifecycle behavior.
|
||||
pub struct ManagedAgentExperimentState {
|
||||
pub(crate) profile_reconcile_enabled: AtomicBool,
|
||||
pub(crate) thread_scoped_acp_sessions_enabled: AtomicBool,
|
||||
}
|
||||
|
||||
impl Default for ManagedAgentExperimentState {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
profile_reconcile_enabled: AtomicBool::new(true),
|
||||
thread_scoped_acp_sessions_enabled: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
pub(crate) fn managed_agent_profile_reconcile_enabled(&self) -> &AtomicBool {
|
||||
&self.managed_agent_experiments.profile_reconcile_enabled
|
||||
}
|
||||
|
||||
pub(crate) fn thread_scoped_acp_sessions_enabled(&self) -> &AtomicBool {
|
||||
&self
|
||||
.managed_agent_experiments
|
||||
.thread_scoped_acp_sessions_enabled
|
||||
}
|
||||
}
|
||||
|
||||
/// Desktop-owned ACP session policy applied to every managed-agent launch.
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub(crate) enum AcpSessionPolicy {
|
||||
Channel,
|
||||
Thread,
|
||||
}
|
||||
|
||||
impl AcpSessionPolicy {
|
||||
pub(crate) fn from_thread_scoped_enabled(enabled: bool) -> Self {
|
||||
if enabled {
|
||||
Self::Thread
|
||||
} else {
|
||||
Self::Channel
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) const fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Channel => "channel",
|
||||
Self::Thread => "thread",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve the persisted experiment state at the shared launch boundary.
|
||||
pub(crate) fn acp_session_policy(state: &AppState) -> AcpSessionPolicy {
|
||||
AcpSessionPolicy::from_thread_scoped_enabled(
|
||||
state
|
||||
.thread_scoped_acp_sessions_enabled()
|
||||
.load(Ordering::Acquire),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn apply_acp_session_policy_env(
|
||||
command: &mut std::process::Command,
|
||||
policy: AcpSessionPolicy,
|
||||
) {
|
||||
command.env(ACP_SESSION_POLICY_ENV_VAR, policy.as_str());
|
||||
}
|
||||
|
||||
/// Resolve the effective policy, apply it to `command`, and return it so the
|
||||
/// caller can stamp the same value onto the spawn snapshot (env and badge can
|
||||
/// never disagree about what the child launched with).
|
||||
pub(crate) fn apply_app_acp_session_policy_env(
|
||||
app: &AppHandle,
|
||||
command: &mut std::process::Command,
|
||||
) -> AcpSessionPolicy {
|
||||
let policy = acp_session_policy(app.state::<AppState>().inner());
|
||||
apply_acp_session_policy_env(command, policy);
|
||||
policy
|
||||
}
|
||||
|
||||
pub(crate) fn insert_acp_session_policy_env(
|
||||
policy_env: &mut BTreeMap<String, String>,
|
||||
policy: AcpSessionPolicy,
|
||||
) {
|
||||
policy_env.insert(
|
||||
ACP_SESSION_POLICY_ENV_VAR.to_string(),
|
||||
policy.as_str().to_string(),
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn command_policy(command: &std::process::Command) -> Option<&str> {
|
||||
command
|
||||
.get_envs()
|
||||
.find(|(key, _)| *key == ACP_SESSION_POLICY_ENV_VAR)
|
||||
.and_then(|(_, value)| value)
|
||||
.and_then(std::ffi::OsStr::to_str)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn absent_or_disabled_experiment_selects_channel_policy() {
|
||||
assert_eq!(
|
||||
AcpSessionPolicy::from_thread_scoped_enabled(false),
|
||||
AcpSessionPolicy::Channel
|
||||
);
|
||||
assert_eq!(AcpSessionPolicy::Channel.as_str(), "channel");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn enabled_experiment_selects_thread_policy() {
|
||||
assert_eq!(
|
||||
AcpSessionPolicy::from_thread_scoped_enabled(true),
|
||||
AcpSessionPolicy::Thread
|
||||
);
|
||||
assert_eq!(AcpSessionPolicy::Thread.as_str(), "thread");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn local_launch_env_receives_the_selected_policy() {
|
||||
let mut command = std::process::Command::new("true");
|
||||
command.env(ACP_SESSION_POLICY_ENV_VAR, "ambient");
|
||||
|
||||
apply_acp_session_policy_env(&mut command, AcpSessionPolicy::Thread);
|
||||
|
||||
assert_eq!(command_policy(&command), Some("thread"));
|
||||
}
|
||||
}
|
||||
@@ -38,7 +38,7 @@ use super::{
|
||||
readiness::EffectiveHarnessDescriptor,
|
||||
runtime::{resolve_session_title, SESSION_TITLE_ENV_VAR},
|
||||
types::{AgentDefinition, ManagedAgentRecord, TeamRecord},
|
||||
GlobalAgentConfig,
|
||||
AcpSessionPolicy, GlobalAgentConfig,
|
||||
};
|
||||
|
||||
pub(crate) mod diff;
|
||||
@@ -76,6 +76,12 @@ pub(crate) struct SpawnConfigInputs<'a> {
|
||||
/// Compile-time distribution capability projected at this runtime boundary.
|
||||
/// The stored record remains portable; only effective spawned access is stamped.
|
||||
pub enforced_owner_only: bool,
|
||||
/// The effective ACP session policy (`channel`/`thread`) the launch applies.
|
||||
/// Resolved from the desktop experiment toggle at the shared launch
|
||||
/// boundary; captured here so flipping the experiment while an agent runs
|
||||
/// drives the existing restart-required path (the harness only reads
|
||||
/// `BUZZ_ACP_SESSION_POLICY` at launch).
|
||||
pub session_policy: AcpSessionPolicy,
|
||||
}
|
||||
|
||||
/// The effective spawn configuration of one managed-agent process.
|
||||
@@ -136,6 +142,14 @@ pub(crate) struct SpawnConfigSnapshot {
|
||||
/// user env `low`, or the reverse) produces no spurious drift entry, and an
|
||||
/// env-only edit still surfaces as exactly one `effort_level` entry.
|
||||
pub effort_level: Option<String>,
|
||||
/// The effective ACP session policy this launch applies (`channel` or
|
||||
/// `thread`). The harness reads `BUZZ_ACP_SESSION_POLICY` only at launch, so
|
||||
/// capturing the resolved policy here lets a toggle flip while an agent runs
|
||||
/// raise the restart-required badge instead of silently leaving the running
|
||||
/// process on the old policy. Written directly on the spawn `Command` (not
|
||||
/// via layered env), so it must be captured explicitly rather than read back
|
||||
/// out of `env`.
|
||||
pub session_policy: String,
|
||||
}
|
||||
|
||||
/// The startup effort a spawn would actually apply, mirroring `apply_effort_env`
|
||||
@@ -166,6 +180,7 @@ impl SpawnConfigSnapshot {
|
||||
model,
|
||||
provider,
|
||||
enforced_owner_only,
|
||||
session_policy,
|
||||
} = inputs;
|
||||
let (respond_to, respond_to_allowlist) =
|
||||
super::projected_access_with_policy(record, enforced_owner_only);
|
||||
@@ -219,6 +234,7 @@ impl SpawnConfigSnapshot {
|
||||
// raw descriptor env (before the strip), so a user-seeded env value
|
||||
// is preserved as the effective effort when no canonical is set.
|
||||
effort_level: effective_effort(record, &descriptor.env),
|
||||
session_policy: session_policy.as_str().to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -259,6 +275,7 @@ pub(crate) fn prospective_spawn_config_snapshot(
|
||||
workspace_relay: &str,
|
||||
global: &GlobalAgentConfig,
|
||||
enforced_owner_only: bool,
|
||||
session_policy: AcpSessionPolicy,
|
||||
) -> SpawnConfigSnapshot {
|
||||
// Prospective re-snapshot: apply the same `apply_persona_snapshot` the
|
||||
// start/restore paths run right before spawning, so this describes what a
|
||||
@@ -309,6 +326,7 @@ pub(crate) fn prospective_spawn_config_snapshot(
|
||||
model: model.as_deref(),
|
||||
provider: provider.as_deref(),
|
||||
enforced_owner_only,
|
||||
session_policy,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ fn base() -> SpawnConfigSnapshot {
|
||||
max_turn_duration_seconds: Some(7200),
|
||||
parallelism: 1,
|
||||
effort_level: Some("high".into()),
|
||||
session_policy: "channel".into(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -72,6 +73,7 @@ fn mutations() -> Vec<Mutation> {
|
||||
}),
|
||||
("parallelism", |s| s.parallelism = 8),
|
||||
("effort_level", |s| s.effort_level = None),
|
||||
("session_policy", |s| s.session_policy = "thread".into()),
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ fn snapshot_with_policy(
|
||||
workspace_relay,
|
||||
global,
|
||||
enforced_owner_only,
|
||||
AcpSessionPolicy::Channel,
|
||||
)
|
||||
.canonical()
|
||||
}
|
||||
|
||||
@@ -187,3 +187,77 @@ fn canonical_effort_edit_changes_snapshot() {
|
||||
"a canonical effort edit must trip the restart badge"
|
||||
);
|
||||
}
|
||||
|
||||
use crate::managed_agents::spawn_snapshot::{
|
||||
eligible_restart_diff, prospective_spawn_config_snapshot, RestartDiffEntry,
|
||||
SpawnConfigSnapshot, TrackedSpawnState,
|
||||
};
|
||||
use crate::managed_agents::AcpSessionPolicy;
|
||||
|
||||
/// Build the prospective snapshot for a bare record under one session policy.
|
||||
fn snapshot_under(policy: AcpSessionPolicy) -> SpawnConfigSnapshot {
|
||||
prospective_spawn_config_snapshot(
|
||||
&record(),
|
||||
&[],
|
||||
&[],
|
||||
"wss://ws.example",
|
||||
&Default::default(),
|
||||
false,
|
||||
policy,
|
||||
)
|
||||
}
|
||||
|
||||
/// Restart-badge entries for a stamped→current session-policy transition,
|
||||
/// exercising the real badge path (`eligible_restart_diff`).
|
||||
fn policy_transition_diff(
|
||||
stamped: &SpawnConfigSnapshot,
|
||||
current: &SpawnConfigSnapshot,
|
||||
) -> Vec<RestartDiffEntry> {
|
||||
eligible_restart_diff(
|
||||
false,
|
||||
Some(TrackedSpawnState {
|
||||
stamped,
|
||||
current,
|
||||
stamped_availability: None,
|
||||
current_availability: None,
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn toggling_session_policy_while_running_requires_restart() {
|
||||
// Regression: flipping the desktop experiment must reach the config-drift
|
||||
// path so a running agent restarts onto the new policy. The harness reads
|
||||
// BUZZ_ACP_SESSION_POLICY only at launch, so without the snapshot field the
|
||||
// badge stayed dark and the process silently kept the old policy.
|
||||
let channel = snapshot_under(AcpSessionPolicy::Channel);
|
||||
let thread = snapshot_under(AcpSessionPolicy::Thread);
|
||||
|
||||
// channel -> thread lights exactly the session_policy entry.
|
||||
let forward = policy_transition_diff(&channel, &thread);
|
||||
assert_eq!(
|
||||
forward.iter().map(|e| e.field.as_str()).collect::<Vec<_>>(),
|
||||
vec!["session_policy"],
|
||||
);
|
||||
|
||||
// thread -> channel is equally visible (rollback also restarts).
|
||||
let reverse = policy_transition_diff(&thread, &channel);
|
||||
assert_eq!(
|
||||
reverse.iter().map(|e| e.field.as_str()).collect::<Vec<_>>(),
|
||||
vec!["session_policy"],
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unchanged_session_policy_does_not_require_restart() {
|
||||
// An unchanged policy must not badge — the default (channel) case must stay
|
||||
// byte-for-byte inert so existing running agents don't flash a spurious
|
||||
// restart badge after this change ships.
|
||||
let channel = snapshot_under(AcpSessionPolicy::Channel);
|
||||
assert!(
|
||||
policy_transition_diff(&channel, &snapshot_under(AcpSessionPolicy::Channel)).is_empty()
|
||||
);
|
||||
|
||||
let thread = snapshot_under(AcpSessionPolicy::Thread);
|
||||
assert!(policy_transition_diff(&thread, &snapshot_under(AcpSessionPolicy::Thread)).is_empty());
|
||||
}
|
||||
|
||||
@@ -138,6 +138,7 @@ fn backfill_of_promptless_record_keeps_spawn_snapshot_stable() {
|
||||
"wss://ws.example",
|
||||
&Default::default(),
|
||||
false,
|
||||
crate::managed_agents::AcpSessionPolicy::Channel,
|
||||
);
|
||||
|
||||
backfill_standalone_agents_in_dir(&base(dir.path())).unwrap();
|
||||
@@ -155,6 +156,7 @@ fn backfill_of_promptless_record_keeps_spawn_snapshot_stable() {
|
||||
"wss://ws.example",
|
||||
&Default::default(),
|
||||
false,
|
||||
crate::managed_agents::AcpSessionPolicy::Channel,
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
@@ -190,6 +192,7 @@ fn backfill_of_prompted_record_keeps_spawn_snapshot_stable() {
|
||||
"wss://ws.example",
|
||||
&Default::default(),
|
||||
false,
|
||||
crate::managed_agents::AcpSessionPolicy::Channel,
|
||||
);
|
||||
|
||||
backfill_standalone_agents_in_dir(&base(dir.path())).unwrap();
|
||||
@@ -207,6 +210,7 @@ fn backfill_of_prompted_record_keeps_spawn_snapshot_stable() {
|
||||
"wss://ws.example",
|
||||
&Default::default(),
|
||||
false,
|
||||
crate::managed_agents::AcpSessionPolicy::Channel,
|
||||
);
|
||||
|
||||
assert_eq!(before.canonical(), after.canonical());
|
||||
|
||||
@@ -283,6 +283,31 @@ with a TypeScript lookup table or an id comparison in a component.
|
||||
|
||||
17. **Databricks model discovery has one shared catalog authority.** Desktop and ACP call the shared `buzz-agent` discovery library; Desktop passes the effective merged `DATABRICKS_MODEL_FILTER` explicitly, and the library applies it to raw workspace endpoint IDs and Unity Catalog model-service FQNs after the additive union. A successful filtered-empty catalog is authoritative: it stays empty, disables switching, and never falls through to configured or known-model fallback. UC FQNs are catalog data and always use the MLflow Chat Completions route, regardless of family-looking text in their components. Global Defaults preserves the discovered model ID as the selected value while its closed trigger renders the provider-scoped display label; do not force the raw persisted ID over that label.
|
||||
|
||||
## Channel-only runtime controls
|
||||
|
||||
Desktop observer controls identify a channel, not a thread session. The harness
|
||||
rejects `cancel_turn` and `switch_model` with `ambiguous_target` when that channel
|
||||
has multiple known session scopes, including retained idle scopes. Do not treat
|
||||
that result as success or a deferred model switch. Stop feedback waits for the
|
||||
harness result matching the control type, channel, and request ID; relay delivery
|
||||
alone does not prove that a turn was signalled. A missing result is unconfirmed,
|
||||
not success. The activity pane must use its resolved `sessionChannelId` for
|
||||
both the outgoing control and result correlation, even without a loaded
|
||||
`Channel` object. Stop is unavailable in an unscoped all-channel pane.
|
||||
|
||||
Per-thread observer controls remain a separate protocol/UI change. Do not tell
|
||||
users to type `!cancel` beside an inline mention: the owner command requires
|
||||
kind 9, body exactly `!cancel` after trimming, and the agent's separate `p` tag.
|
||||
The automatic-mention picker also inserts literal `@Name` into the body, so it
|
||||
does not provide an exact-command workaround. The UI must state this limitation
|
||||
rather than offer an ineffective command. An authorized owner can instead use
|
||||
the CLI with the channel and target thread root:
|
||||
|
||||
```sh
|
||||
buzz messages send --channel <channel-id> --reply-to <thread-root-id> \
|
||||
--mention <agent-pubkey> --content '!cancel'
|
||||
```
|
||||
|
||||
## The tests that enforce this
|
||||
|
||||
- `lib/agentConfigCore.test.mjs` — field model per harness × scope, clearing
|
||||
|
||||
102
desktop/src/features/agents/lib/cancelTurnOutcome.test.mjs
Normal file
102
desktop/src/features/agents/lib/cancelTurnOutcome.test.mjs
Normal file
@@ -0,0 +1,102 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import { awaitCancelTurnOutcome } from "./cancelTurnOutcome.ts";
|
||||
|
||||
function harness(sendCancel = async () => {}) {
|
||||
let listener;
|
||||
let timeout;
|
||||
let unsubscribed = false;
|
||||
let timeoutCancelled = false;
|
||||
const outcome = awaitCancelTurnOutcome({
|
||||
requestId: "request-a",
|
||||
channelId: "channel-a",
|
||||
subscribe: (fn) => {
|
||||
listener = fn;
|
||||
return () => {
|
||||
unsubscribed = true;
|
||||
};
|
||||
},
|
||||
sendCancel,
|
||||
scheduleTimeout: (fn) => {
|
||||
timeout = fn;
|
||||
return () => {
|
||||
timeoutCancelled = true;
|
||||
};
|
||||
},
|
||||
});
|
||||
return {
|
||||
outcome,
|
||||
push: (status, overrides = {}) =>
|
||||
listener({
|
||||
type: "cancel_turn",
|
||||
requestId: "request-a",
|
||||
channelId: "channel-a",
|
||||
status,
|
||||
...overrides,
|
||||
}),
|
||||
timeout: () => timeout(),
|
||||
assertCleaned: () => {
|
||||
assert.equal(unsubscribed, true);
|
||||
assert.equal(timeoutCancelled, true);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
for (const status of ["sent", "no_active_turn", "ambiguous_target"]) {
|
||||
test(`stop returns the correlated harness result: ${status}`, async () => {
|
||||
const h = harness();
|
||||
h.push(status);
|
||||
assert.equal(await h.outcome, status);
|
||||
h.assertCleaned();
|
||||
});
|
||||
}
|
||||
|
||||
test("relay delivery, old harnesses, replay, other channels and model acks cannot confirm a stop", async () => {
|
||||
const h = harness();
|
||||
h.push("sent", { requestId: undefined });
|
||||
h.push("sent", { requestId: "old-request" });
|
||||
h.push("sent", { channelId: "channel-b" });
|
||||
h.push("sent", { type: "switch_model" });
|
||||
h.push("future_status");
|
||||
h.timeout();
|
||||
assert.equal(await h.outcome, "unconfirmed");
|
||||
h.assertCleaned();
|
||||
});
|
||||
|
||||
test("stop unsubscribes and clears timeout after a transport error", async () => {
|
||||
const h = harness(async () => {
|
||||
throw new Error("transport");
|
||||
});
|
||||
await assert.rejects(h.outcome, /transport/);
|
||||
h.assertCleaned();
|
||||
});
|
||||
|
||||
test("a hung transport cannot block the unconfirmed timeout", async () => {
|
||||
const h = harness(() => new Promise(() => {}));
|
||||
h.timeout();
|
||||
assert.equal(await h.outcome, "unconfirmed");
|
||||
h.assertCleaned();
|
||||
});
|
||||
|
||||
test("a harness result can settle before the send promise resolves", async () => {
|
||||
const h = harness(() => new Promise(() => {}));
|
||||
h.push("sent");
|
||||
assert.equal(await h.outcome, "sent");
|
||||
h.assertCleaned();
|
||||
});
|
||||
|
||||
test("a late transport rejection does not replace the settled result", async () => {
|
||||
let rejectSend;
|
||||
const h = harness(
|
||||
() =>
|
||||
new Promise((_resolve, reject) => {
|
||||
rejectSend = reject;
|
||||
}),
|
||||
);
|
||||
h.timeout();
|
||||
assert.equal(await h.outcome, "unconfirmed");
|
||||
rejectSend(new Error("late transport error"));
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
h.assertCleaned();
|
||||
});
|
||||
76
desktop/src/features/agents/lib/cancelTurnOutcome.ts
Normal file
76
desktop/src/features/agents/lib/cancelTurnOutcome.ts
Normal file
@@ -0,0 +1,76 @@
|
||||
import type { ControlResultFrame } from "@/shared/api/types";
|
||||
|
||||
/** Stop feedback must describe the harness result, not relay delivery alone. */
|
||||
export async function awaitCancelTurnOutcome({
|
||||
requestId,
|
||||
channelId,
|
||||
subscribe,
|
||||
sendCancel,
|
||||
scheduleTimeout,
|
||||
}: {
|
||||
requestId: string;
|
||||
channelId: string;
|
||||
subscribe: (listener: (frame: ControlResultFrame) => void) => () => void;
|
||||
sendCancel: () => Promise<void>;
|
||||
scheduleTimeout: (onTimeout: () => void) => () => void;
|
||||
}): Promise<"sent" | "no_active_turn" | "ambiguous_target" | "unconfirmed"> {
|
||||
type Outcome = "sent" | "no_active_turn" | "ambiguous_target" | "unconfirmed";
|
||||
|
||||
let settled = false;
|
||||
let unsubscribe = () => {};
|
||||
let cancelTimeout = () => {};
|
||||
let resolveResult: (outcome: Outcome) => void = () => {};
|
||||
let rejectResult: (error: unknown) => void = () => {};
|
||||
const result = new Promise<Outcome>((resolve, reject) => {
|
||||
resolveResult = resolve;
|
||||
rejectResult = reject;
|
||||
});
|
||||
const cleanup = () => {
|
||||
unsubscribe();
|
||||
cancelTimeout();
|
||||
};
|
||||
const settle = (outcome: Outcome) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
resolveResult(outcome);
|
||||
};
|
||||
const fail = (error: unknown) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
rejectResult(error);
|
||||
};
|
||||
|
||||
unsubscribe = subscribe((frame) => {
|
||||
if (
|
||||
frame.type !== "cancel_turn" ||
|
||||
frame.requestId !== requestId ||
|
||||
frame.channelId !== channelId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (
|
||||
frame.status === "sent" ||
|
||||
frame.status === "no_active_turn" ||
|
||||
frame.status === "ambiguous_target"
|
||||
) {
|
||||
settle(frame.status);
|
||||
}
|
||||
});
|
||||
// Start the timeout before sending. A hung relay transport must not keep the
|
||||
// caller pending forever; timeout truthfully reports that the harness result
|
||||
// was not confirmed. The send promise is still observed below so a later
|
||||
// rejection cannot become an unhandled rejection.
|
||||
cancelTimeout = scheduleTimeout(() => settle("unconfirmed"));
|
||||
try {
|
||||
// Race transport failure against the harness result. A correlated result
|
||||
// may arrive before publish resolves, and a transport that hangs must not
|
||||
// block the timeout from settling the outer operation.
|
||||
void Promise.resolve(sendCancel()).catch(fail);
|
||||
} catch (error) {
|
||||
fail(error);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
@@ -75,6 +75,23 @@ const drainMicrotasks = async () => {
|
||||
}
|
||||
};
|
||||
|
||||
test("ambiguous sibling sessions reject the pick even after another channel switched", async () => {
|
||||
const h = harness([CH_A, CH_B]);
|
||||
h.push(frame("switched"));
|
||||
h.push(frame("ambiguous_target", { channelId: CH_B }));
|
||||
assert.equal(await h.outcome, "ambiguous");
|
||||
assert.equal(h.cancelTimeoutCalls, 1);
|
||||
assert.equal(h.unsubscribeCalls, 1);
|
||||
});
|
||||
|
||||
test("stale or foreign ambiguity does not reject a live model pick", async () => {
|
||||
const h = harness([CH_A]);
|
||||
h.push(frame("ambiguous_target", { requestId: "old-pick" }));
|
||||
h.push(frame("ambiguous_target", { channelId: CH_B }));
|
||||
h.push(frame("switched"));
|
||||
assert.equal(await h.outcome, "ok");
|
||||
});
|
||||
|
||||
test("awaitLiveSwitchOutcome fast sent on one channel does not mask a later unsupported on another", async () => {
|
||||
const h = harness([CH_A, CH_B]);
|
||||
// Channel A acks fast as `sent`; a first-ack-resolves impl would settle "ok"
|
||||
|
||||
@@ -41,6 +41,9 @@ import type { ControlResultFrame } from "@/shared/api/types";
|
||||
* Both fail-fast to `"not_delivered"`, distinct from `"pending"` (which DID
|
||||
* ride the requeued session): here the switch never landed at all.
|
||||
*
|
||||
* `ambiguous_target` rejects a channel-only pick when the harness knows more
|
||||
* than one session scope in that channel. No sibling session was changed.
|
||||
*
|
||||
* Any other status — a `sent` provisional ack, or an unknown future status — is
|
||||
* inert: it is never counted as success. A new producer status that should
|
||||
* settle the pick must add its own explicit branch.
|
||||
@@ -79,16 +82,24 @@ export async function awaitLiveSwitchOutcome({
|
||||
sendSwitches: () => Promise<void>;
|
||||
/** Schedule the no-reply fallback; returns a cancel function. */
|
||||
scheduleTimeout: (onTimeout: () => void) => () => void;
|
||||
}): Promise<"ok" | "unsupported" | "failed" | "not_delivered" | "pending"> {
|
||||
}): Promise<
|
||||
"ok" | "unsupported" | "failed" | "not_delivered" | "pending" | "ambiguous"
|
||||
> {
|
||||
const expected = new Set(channelIds);
|
||||
const settled = new Promise<
|
||||
"ok" | "unsupported" | "failed" | "not_delivered" | "pending"
|
||||
"ok" | "unsupported" | "failed" | "not_delivered" | "pending" | "ambiguous"
|
||||
>((resolve) => {
|
||||
let unsubscribe = () => {};
|
||||
let cancelTimeout = () => {};
|
||||
const succeeded = new Set<string>();
|
||||
const finish = (
|
||||
outcome: "ok" | "unsupported" | "failed" | "not_delivered" | "pending",
|
||||
outcome:
|
||||
| "ok"
|
||||
| "unsupported"
|
||||
| "failed"
|
||||
| "not_delivered"
|
||||
| "pending"
|
||||
| "ambiguous",
|
||||
) => {
|
||||
cancelTimeout();
|
||||
unsubscribe();
|
||||
@@ -112,6 +123,10 @@ export async function awaitLiveSwitchOutcome({
|
||||
if (!frame.channelId || !expected.has(frame.channelId)) {
|
||||
return;
|
||||
}
|
||||
if (frame.status === "ambiguous_target") {
|
||||
finish("ambiguous");
|
||||
return;
|
||||
}
|
||||
if (frame.status === "unsupported_model") {
|
||||
// Model unavailable — reject the whole pick immediately.
|
||||
finish("unsupported");
|
||||
|
||||
@@ -157,6 +157,12 @@ export function ModelPicker({
|
||||
try {
|
||||
if (isLiveSwitch) {
|
||||
const outcome = await sendLiveSwitch(modelId);
|
||||
if (outcome === "ambiguous") {
|
||||
toast.error(
|
||||
"Couldn't switch all sessions — a channel has multiple agent sessions. Stop and restart the agent with the new model.",
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (outcome === "unsupported") {
|
||||
toast.error("That model isn't available for this agent.");
|
||||
return;
|
||||
|
||||
@@ -25,6 +25,8 @@ import {
|
||||
import { useAnchoredScroll } from "@/features/messages/ui/useAnchoredScroll";
|
||||
import { useStableArrayShallow } from "@/shared/hooks/useStableReference";
|
||||
import { cancelManagedAgentTurn } from "@/shared/api/agentControl";
|
||||
import { awaitCancelTurnOutcome } from "@/features/agents/lib/cancelTurnOutcome";
|
||||
import { subscribeControlResults } from "@/features/agents/observerRelayStore";
|
||||
import type { Channel } from "@/shared/api/types";
|
||||
import { useEscapeKey } from "@/shared/hooks/useEscapeKey";
|
||||
import { useIsThreadPanelOverlay } from "@/shared/hooks/use-mobile";
|
||||
@@ -105,7 +107,8 @@ export function AgentSessionThreadPanel({
|
||||
agent.pubkey,
|
||||
sessionChannelId,
|
||||
);
|
||||
const canStopCurrentTurn = isWorking && canInterruptTurn;
|
||||
const canStopCurrentTurn =
|
||||
Boolean(sessionChannelId) && isWorking && canInterruptTurn;
|
||||
useEscapeKey(onClose, isOverlay || isSinglePanelView);
|
||||
|
||||
const scrollRef = React.useRef<HTMLDivElement>(null);
|
||||
@@ -241,12 +244,38 @@ export function AgentSessionThreadPanel({
|
||||
const animateActivity = useTranscriptAnimationEnabled();
|
||||
const showTimestamps = useTranscriptTimestampsEnabled();
|
||||
async function handleInterruptTurn() {
|
||||
if (!channel) {
|
||||
if (!sessionChannelId) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await cancelManagedAgentTurn(agent.pubkey, channel.id);
|
||||
const requestId = crypto.randomUUID();
|
||||
const outcome = await awaitCancelTurnOutcome({
|
||||
requestId,
|
||||
channelId: sessionChannelId,
|
||||
subscribe: (listener) =>
|
||||
subscribeControlResults(agent.pubkey, listener),
|
||||
sendCancel: () =>
|
||||
cancelManagedAgentTurn(agent.pubkey, sessionChannelId, requestId),
|
||||
scheduleTimeout: (onTimeout) => {
|
||||
const timeout = window.setTimeout(onTimeout, 8_000);
|
||||
return () => window.clearTimeout(timeout);
|
||||
},
|
||||
});
|
||||
if (outcome === "ambiguous_target") {
|
||||
toast.error(
|
||||
"This channel has multiple agent sessions. Stopping a specific thread isn't available here yet.",
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (outcome === "no_active_turn") {
|
||||
toast.info("No active turn to stop.");
|
||||
return;
|
||||
}
|
||||
if (outcome === "unconfirmed") {
|
||||
toast.info("Stop requested, but the agent hasn't confirmed it.");
|
||||
return;
|
||||
}
|
||||
toast.success(
|
||||
`Stop signal sent to ${agent.name}. It may take a moment to respond.`,
|
||||
);
|
||||
@@ -385,9 +414,11 @@ export function AgentSessionThreadPanel({
|
||||
title={
|
||||
canStopCurrentTurn
|
||||
? "Interrupt the current ACP turn without stopping the agent process."
|
||||
: isWorking
|
||||
? "Only locally managed agents can be interrupted from this community."
|
||||
: "Available while the agent is working."
|
||||
: !sessionChannelId
|
||||
? "Open activity for a channel to stop its current turn."
|
||||
: isWorking
|
||||
? "Only locally managed agents can be interrupted from this community."
|
||||
: "Available while the agent is working."
|
||||
}
|
||||
>
|
||||
<Octagon className="mt-0.5 h-4 w-4 text-muted-foreground" />
|
||||
@@ -397,9 +428,11 @@ export function AgentSessionThreadPanel({
|
||||
</span>
|
||||
{!canStopCurrentTurn ? (
|
||||
<span className="mt-0.5 block text-xs text-muted-foreground">
|
||||
{isWorking
|
||||
? "Only available for locally managed agents."
|
||||
: "Available while the agent is working."}
|
||||
{!sessionChannelId
|
||||
? "Open activity for a channel to stop its current turn."
|
||||
: isWorking
|
||||
? "Only available for locally managed agents."
|
||||
: "Available while the agent is working."}
|
||||
</span>
|
||||
) : null}
|
||||
</span>
|
||||
|
||||
@@ -5,10 +5,10 @@ import { isMacPlatform } from "@/shared/lib/platform";
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { resetRateLimitGate } from "@/shared/api/relayRateLimitGate";
|
||||
import {
|
||||
applyCommunity,
|
||||
autoConnectDefaultRelayEnabled,
|
||||
getDefaultRelayUrl,
|
||||
} from "@/shared/api/tauri";
|
||||
import { applyCommunity } from "@/shared/api/tauriWorkspace";
|
||||
import { getIdentity } from "@/shared/api/tauriIdentity";
|
||||
import { clearTrayAgentActivity } from "@/shared/api/trayMenu";
|
||||
import { getOverrides } from "@/shared/features";
|
||||
@@ -300,6 +300,7 @@ export function useCommunityInit(
|
||||
activeCommunity.token,
|
||||
activeCommunity.reposDir,
|
||||
getOverrides().agentManagedProfiles === true,
|
||||
getOverrides().threadScopedAcpSessions === true,
|
||||
);
|
||||
} catch (error) {
|
||||
// A bad `repos_dir` no longer reaches here — `apply_workspace` treats
|
||||
|
||||
@@ -1,4 +1,7 @@
|
||||
import { setAgentManagedProfiles } from "@/shared/api/tauri";
|
||||
import {
|
||||
setAgentManagedProfiles,
|
||||
setThreadScopedAcpSessions,
|
||||
} from "@/shared/api/tauriWorkspace";
|
||||
import { desktopFeatures, useFeatureToggle } from "@/shared/features";
|
||||
import type { FeatureDefinition } from "@/shared/features";
|
||||
import { Switch } from "@/shared/ui/switch";
|
||||
@@ -33,6 +36,14 @@ function FeatureRow({ feature }: { feature: FeatureDefinition }) {
|
||||
);
|
||||
});
|
||||
}
|
||||
if (feature.id === "threadScopedAcpSessions") {
|
||||
void setThreadScopedAcpSessions(value).catch((error) => {
|
||||
console.error(
|
||||
"Failed to apply thread-scoped ACP sessions setting:",
|
||||
error,
|
||||
);
|
||||
});
|
||||
}
|
||||
}}
|
||||
/>
|
||||
</SettingsOptionRow>
|
||||
|
||||
@@ -1,15 +1,16 @@
|
||||
import { sendAgentObserverControl } from "@/shared/api/observerRelay";
|
||||
import type { CancelManagedAgentTurnResult } from "@/shared/api/types";
|
||||
|
||||
/** Send a stop request; the harness acknowledges it via control_result. */
|
||||
export async function cancelManagedAgentTurn(
|
||||
pubkey: string,
|
||||
channelId: string,
|
||||
): Promise<CancelManagedAgentTurnResult> {
|
||||
requestId: string,
|
||||
): Promise<void> {
|
||||
await sendAgentObserverControl(pubkey, {
|
||||
type: "cancel_turn",
|
||||
channelId,
|
||||
requestId,
|
||||
});
|
||||
return { status: "sent" };
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1071,22 +1071,6 @@ export async function cancelPairing(): Promise<void> {
|
||||
await invokeTauri("cancel_pairing");
|
||||
}
|
||||
|
||||
export async function applyCommunity(
|
||||
relayUrl: string,
|
||||
nsec?: string,
|
||||
token?: string,
|
||||
reposDir?: string,
|
||||
agentManagedProfiles?: boolean,
|
||||
): Promise<void> {
|
||||
await invokeTauri("apply_workspace", {
|
||||
relayUrl,
|
||||
nsec: nsec ?? null,
|
||||
token: token ?? null,
|
||||
reposDir: reposDir ?? null,
|
||||
agentManagedProfiles: agentManagedProfiles ?? false,
|
||||
});
|
||||
}
|
||||
|
||||
// Validate a candidate repos dir without mutating the filesystem. Rejects
|
||||
// with a human-readable reason; resolves for a valid or empty path.
|
||||
export async function validateReposDir(dir: string): Promise<void> {
|
||||
@@ -1096,9 +1080,6 @@ export async function validateReposDir(dir: string): Promise<void> {
|
||||
export const setPreventSleepActive = (active: boolean) =>
|
||||
invokeTauri("set_prevent_sleep_active", { active });
|
||||
|
||||
export const setAgentManagedProfiles = (enabled: boolean) =>
|
||||
invokeTauri("set_agent_managed_profiles", { enabled });
|
||||
|
||||
/** Returns true on macOS, Windows, and Linux AppImage installs.
|
||||
* Returns false on Linux non-AppImage packages (e.g. .deb) where
|
||||
* Tauri's updater cannot swap the binary. */
|
||||
|
||||
25
desktop/src/shared/api/tauriWorkspace.ts
Normal file
25
desktop/src/shared/api/tauriWorkspace.ts
Normal file
@@ -0,0 +1,25 @@
|
||||
import { invokeTauri } from "@/shared/api/tauri";
|
||||
|
||||
export async function applyCommunity(
|
||||
relayUrl: string,
|
||||
nsec?: string,
|
||||
token?: string,
|
||||
reposDir?: string,
|
||||
agentManagedProfiles?: boolean,
|
||||
threadScopedAcpSessions?: boolean,
|
||||
): Promise<void> {
|
||||
await invokeTauri("apply_workspace", {
|
||||
relayUrl,
|
||||
nsec: nsec ?? null,
|
||||
token: token ?? null,
|
||||
reposDir: reposDir ?? null,
|
||||
agentManagedProfiles: agentManagedProfiles ?? false,
|
||||
threadScopedAcpSessions: threadScopedAcpSessions ?? false,
|
||||
});
|
||||
}
|
||||
|
||||
export const setAgentManagedProfiles = (enabled: boolean) =>
|
||||
invokeTauri("set_agent_managed_profiles", { enabled });
|
||||
|
||||
export const setThreadScopedAcpSessions = (enabled: boolean) =>
|
||||
invokeTauri("set_thread_scoped_acp_sessions", { enabled });
|
||||
@@ -444,14 +444,11 @@ export type ManagedAgentLog = {
|
||||
logPath: string;
|
||||
};
|
||||
|
||||
export type CancelManagedAgentTurnResult = {
|
||||
status: "sent" | "no_active_turn";
|
||||
};
|
||||
|
||||
/** Outcome of a live `switch_model` control frame; `failure` lands late. */
|
||||
export type SwitchManagedAgentModelStatus =
|
||||
| "sent"
|
||||
| "turn_ending"
|
||||
| "ambiguous_target"
|
||||
| "switched"
|
||||
| "unsupported_model"
|
||||
| "no_active_turn"
|
||||
|
||||
48
desktop/src/shared/features/manifest.test.mjs
Normal file
48
desktop/src/shared/features/manifest.test.mjs
Normal file
@@ -0,0 +1,48 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { readFileSync } from "node:fs";
|
||||
import test from "node:test";
|
||||
|
||||
const manifest = JSON.parse(
|
||||
readFileSync(
|
||||
new URL("../../../../preview-features.json", import.meta.url),
|
||||
"utf8",
|
||||
),
|
||||
);
|
||||
|
||||
test("thread-scoped ACP sessions is a default-off desktop experiment", () => {
|
||||
const feature = manifest.features.find(
|
||||
({ id }) => id === "threadScopedAcpSessions",
|
||||
);
|
||||
|
||||
assert.deepEqual(feature, {
|
||||
id: "threadScopedAcpSessions",
|
||||
name: "Thread Scoped ACP Sessions",
|
||||
description:
|
||||
"Give each channel thread isolated agent context. Applies when managed agents next start; DMs stay conversation-scoped.",
|
||||
platforms: ["desktop"],
|
||||
});
|
||||
assert.equal(feature.defaultEnabled, undefined);
|
||||
});
|
||||
|
||||
test("existing Projects and Workflows experiments remain unchanged", () => {
|
||||
const existing = Object.fromEntries(
|
||||
manifest.features
|
||||
.filter(({ id }) => id === "projects" || id === "workflows")
|
||||
.map((feature) => [feature.id, feature]),
|
||||
);
|
||||
|
||||
assert.deepEqual(existing, {
|
||||
projects: {
|
||||
id: "projects",
|
||||
name: "Projects",
|
||||
description: "Git repository browser and collaboration",
|
||||
platforms: ["desktop"],
|
||||
},
|
||||
workflows: {
|
||||
id: "workflows",
|
||||
name: "Workflows",
|
||||
description: "YAML-defined automations with approval gates",
|
||||
platforms: ["desktop"],
|
||||
},
|
||||
});
|
||||
});
|
||||
@@ -46,6 +46,21 @@ test("setOverride persists filtered overrides", () => {
|
||||
);
|
||||
});
|
||||
|
||||
test("thread-scoped ACP session preference persists without changing existing experiments", () => {
|
||||
const { values } = installStorage({ workflows: true, projects: false });
|
||||
|
||||
setOverride("threadScopedAcpSessions", true);
|
||||
|
||||
assert.equal(
|
||||
values.get(OVERRIDES_KEY),
|
||||
JSON.stringify({
|
||||
workflows: true,
|
||||
projects: false,
|
||||
threadScopedAcpSessions: true,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
test("getOverrides drops non-boolean values", () => {
|
||||
installStorage({ workflows: "yes", projects: false });
|
||||
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
import * as React from "react";
|
||||
import { createRoot, type Root } from "react-dom/client";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { AgentSessionThreadPanel } from "@/features/channels/ui/AgentSessionThreadPanel";
|
||||
import { CommunitiesProvider } from "@/features/communities/useCommunities";
|
||||
import { bytesToHex, hexToBytes } from "@noble/hashes/utils.js";
|
||||
import { emit, listen } from "@tauri-apps/api/event";
|
||||
import { mockIPC, mockWindows } from "@tauri-apps/api/mocks";
|
||||
@@ -18,6 +23,15 @@ import type { UnreadCatchUpChannelResult } from "@/shared/api/tauriUnreadCatchUp
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { activateRateLimit } from "@/shared/api/relayRateLimitGate";
|
||||
import { resolveAgentParallelism } from "@/features/agents/lib/agentParallelism";
|
||||
import { awaitLiveSwitchOutcome } from "@/features/agents/lib/liveSwitchOutcome";
|
||||
import {
|
||||
_testRegisterKnownAgents,
|
||||
ensureRelayObserverSubscription,
|
||||
subscribeControlResults,
|
||||
} from "@/features/agents/observerRelayStore";
|
||||
import { switchManagedAgentModel } from "@/shared/api/agentControl";
|
||||
import { mockSearchHitMatches } from "./e2eBridgeSearch.ts";
|
||||
export { mockSearchHitMatches };
|
||||
import type { ConnectionState } from "@/shared/api/relayClientShared";
|
||||
import type {
|
||||
ChannelTemplate,
|
||||
@@ -370,6 +384,14 @@ type E2eConfig = {
|
||||
closeChannelLiveSubscriptionOnce?: boolean;
|
||||
/** Reject successive kind-9 sends with these messages, then resume. */
|
||||
sendMessageErrors?: string[];
|
||||
/** Test-only observer control requests captured after mock publish. */
|
||||
observerControlResults?: Array<{
|
||||
type: "cancel_turn" | "switch_model";
|
||||
status: string;
|
||||
channelId?: string | null;
|
||||
requestId?: string;
|
||||
modelId?: string;
|
||||
}>;
|
||||
/** Reject successive managed-agent starts, then resume. */
|
||||
startManagedAgentErrors?: string[];
|
||||
/** Delay (ms) after snapshotting a thread-replies page so E2E tests can
|
||||
@@ -1385,6 +1407,25 @@ declare global {
|
||||
peers: number;
|
||||
}>;
|
||||
}) => void;
|
||||
/** Run the real live-switch outcome and control transport through mock relay. */
|
||||
__BUZZ_E2E_RUN_MODEL_SWITCH__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelIds: string[];
|
||||
modelId: string;
|
||||
requestId: string;
|
||||
timeoutMs?: number;
|
||||
}) => Promise<string>;
|
||||
/** Mount AgentSessionThreadPanel directly for channelId-only UI coverage. */
|
||||
__BUZZ_E2E_MOUNT_AGENT_SESSION_PANEL__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelId: string;
|
||||
canInterruptTurn?: boolean;
|
||||
}) => Promise<void>;
|
||||
/** Resolve decoded test-only observer controls back through the mock relay. */
|
||||
__BUZZ_E2E_OBSERVER_CONTROLS__?: Array<{
|
||||
agentPubkey: string;
|
||||
payload: unknown;
|
||||
}>;
|
||||
__BUZZ_E2E_SEED_ACTIVE_TURNS__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelId: string;
|
||||
@@ -4075,6 +4116,9 @@ const mockFeedOverrides: RawHomeFeedResponse["feed"] = {
|
||||
};
|
||||
|
||||
let installed = false;
|
||||
let directPanelRoot: Root | null = null;
|
||||
let directPanelContainer: HTMLDivElement | null = null;
|
||||
let directPanelQueryClient: QueryClient | null = null;
|
||||
let nextSocketId = 1;
|
||||
|
||||
function syncMockRelayAgentsFromManagedAgents() {
|
||||
@@ -4795,6 +4839,62 @@ function emitMockLiveEvent(channelId: string, event: RelayEvent) {
|
||||
}
|
||||
}
|
||||
|
||||
let mockObserverControlSeq = 0;
|
||||
|
||||
function emitMockObserverControlResult(
|
||||
agentPubkey: string,
|
||||
request: {
|
||||
type: "cancel_turn" | "switch_model";
|
||||
channelId?: string | null;
|
||||
requestId?: string;
|
||||
modelId?: string;
|
||||
},
|
||||
result: {
|
||||
type: "cancel_turn" | "switch_model";
|
||||
status: string;
|
||||
channelId?: string | null;
|
||||
requestId?: string;
|
||||
modelId?: string;
|
||||
},
|
||||
) {
|
||||
const channelId = result.channelId ?? request.channelId ?? null;
|
||||
const payload = {
|
||||
type: result.type,
|
||||
status: result.status,
|
||||
...(result.requestId !== undefined
|
||||
? { requestId: result.requestId }
|
||||
: request.requestId !== undefined
|
||||
? { requestId: request.requestId }
|
||||
: {}),
|
||||
...(result.modelId !== undefined
|
||||
? { modelId: result.modelId }
|
||||
: request.modelId !== undefined
|
||||
? { modelId: request.modelId }
|
||||
: {}),
|
||||
};
|
||||
mockObserverControlSeq += 1;
|
||||
const observerEvent = createMockEvent(
|
||||
KIND_AGENT_OBSERVER_FRAME,
|
||||
JSON.stringify({
|
||||
seq: mockObserverControlSeq,
|
||||
timestamp: new Date().toISOString(),
|
||||
kind: "control_result",
|
||||
agentIndex: null,
|
||||
channelId,
|
||||
sessionId: null,
|
||||
turnId: null,
|
||||
payload,
|
||||
}),
|
||||
[
|
||||
["p", MOCK_IDENTITY_PUBKEY],
|
||||
["agent", agentPubkey],
|
||||
["frame", "telemetry"],
|
||||
],
|
||||
agentPubkey,
|
||||
);
|
||||
emitMockLiveEvent(GLOBAL_MOCK_SUBSCRIPTION, observerEvent);
|
||||
}
|
||||
|
||||
function emitOrDeferMockSendMessageLiveEcho(
|
||||
channelId: string,
|
||||
event: RelayEvent,
|
||||
@@ -9655,52 +9755,6 @@ async function handleUpdateManagedAgent(args: {
|
||||
return { agent: cloneManagedAgent(agent), profile_sync_error: null };
|
||||
}
|
||||
|
||||
/**
|
||||
* Mock-mode `search_messages` predicate, mirroring the relay's filter contract.
|
||||
*
|
||||
* `since`/`until` are NIP-01 bounds and both inclusive — the relay keeps events
|
||||
* where `since <= created_at <= until` (`crates/buzz-core/src/filter.rs`). The
|
||||
* `before:` operator's exclusivity is encoded upstream in
|
||||
* `parseSearchOperators`, which subtracts a second; the mock must not subtract
|
||||
* it a second time.
|
||||
*
|
||||
* Exported so the boundary behavior is unit-testable without a browser.
|
||||
*/
|
||||
export function mockSearchHitMatches(
|
||||
hit: Pick<
|
||||
RawSearchHit,
|
||||
"channel_id" | "pubkey" | "created_at" | "content" | "channel_name"
|
||||
>,
|
||||
filters: {
|
||||
/** Lowercased FTS query; empty matches everything. */
|
||||
query: string;
|
||||
channelId?: string;
|
||||
authorSet: Set<string> | null;
|
||||
since?: number;
|
||||
until?: number;
|
||||
},
|
||||
): boolean {
|
||||
if (filters.channelId && hit.channel_id !== filters.channelId) {
|
||||
return false;
|
||||
}
|
||||
if (filters.authorSet && !filters.authorSet.has(hit.pubkey.toLowerCase())) {
|
||||
return false;
|
||||
}
|
||||
if (filters.since != null && hit.created_at < filters.since) {
|
||||
return false;
|
||||
}
|
||||
if (filters.until != null && hit.created_at > filters.until) {
|
||||
return false;
|
||||
}
|
||||
if (!filters.query) {
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
hit.content.toLowerCase().includes(filters.query) ||
|
||||
(hit.channel_name?.toLowerCase().includes(filters.query) ?? false)
|
||||
);
|
||||
}
|
||||
|
||||
async function handleSearchMessages(
|
||||
args: {
|
||||
q: string;
|
||||
@@ -10868,6 +10922,51 @@ function sendToMockSocket(args: {
|
||||
if (type === "EVENT") {
|
||||
const event = rest[0] as RelayEvent;
|
||||
|
||||
if (event.kind === KIND_AGENT_OBSERVER_FRAME) {
|
||||
const frame = event.tags.find((tag) => tag[0] === "frame")?.[1];
|
||||
if (frame === "control") {
|
||||
let payload: {
|
||||
type: "cancel_turn" | "switch_model";
|
||||
channelId?: string | null;
|
||||
requestId?: string;
|
||||
modelId?: string;
|
||||
};
|
||||
try {
|
||||
payload = JSON.parse(event.content);
|
||||
} catch {
|
||||
sendWsText(socket.handler, [
|
||||
"OK",
|
||||
event.id,
|
||||
false,
|
||||
"mock observer control payload is not JSON",
|
||||
]);
|
||||
return;
|
||||
}
|
||||
window.__BUZZ_E2E_OBSERVER_CONTROLS__?.push({
|
||||
agentPubkey: event.tags.find((tag) => tag[0] === "agent")?.[1] ?? "",
|
||||
payload,
|
||||
});
|
||||
const configured = getConfig()?.mock?.observerControlResults ?? [];
|
||||
const resultIndex = configured.findIndex(
|
||||
(candidate) => candidate.type === payload.type,
|
||||
);
|
||||
const result =
|
||||
resultIndex >= 0
|
||||
? configured.splice(resultIndex, 1)[0]
|
||||
: {
|
||||
type: payload.type,
|
||||
status: payload.type === "cancel_turn" ? "sent" : "switched",
|
||||
};
|
||||
emitMockObserverControlResult(
|
||||
event.tags.find((tag) => tag[0] === "agent")?.[1] ?? "",
|
||||
payload,
|
||||
result,
|
||||
);
|
||||
sendWsText(socket.handler, ["OK", event.id, true, ""]);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (event.kind === 28936) {
|
||||
sendWsText(socket.handler, ["OK", event.id, true, ""]);
|
||||
return;
|
||||
@@ -11138,6 +11237,92 @@ export function maybeInstallE2eTauriMocks() {
|
||||
window.__BUZZ_E2E_COMMANDS__ = [];
|
||||
window.__BUZZ_E2E_COMMAND_PAYLOADS__ = [];
|
||||
window.__BUZZ_E2E_COMMAND_LOG__ = [];
|
||||
window.__BUZZ_E2E_OBSERVER_CONTROLS__ = [];
|
||||
window.__BUZZ_E2E_RUN_MODEL_SWITCH__ = async ({
|
||||
agentPubkey,
|
||||
channelIds,
|
||||
modelId,
|
||||
requestId,
|
||||
timeoutMs = 8_000,
|
||||
}) => {
|
||||
_testRegisterKnownAgents("e2e-model-switch", [agentPubkey]);
|
||||
await ensureRelayObserverSubscription();
|
||||
// The mock relay has already been exercised by the app's normal observer
|
||||
// subscription. The outcome helper below is the direct browser seam for the
|
||||
// control-specific behavior, while the existing request capture proves the
|
||||
// outgoing payload shape.
|
||||
return awaitLiveSwitchOutcome({
|
||||
requestId,
|
||||
channelIds,
|
||||
subscribe: (listener) => subscribeControlResults(agentPubkey, listener),
|
||||
sendSwitches: async () => {
|
||||
await Promise.all(
|
||||
channelIds.map((channelId) =>
|
||||
switchManagedAgentModel(agentPubkey, channelId, modelId, requestId),
|
||||
),
|
||||
);
|
||||
},
|
||||
scheduleTimeout: (onTimeout) => {
|
||||
const timeout = window.setTimeout(onTimeout, timeoutMs);
|
||||
return () => window.clearTimeout(timeout);
|
||||
},
|
||||
});
|
||||
};
|
||||
window.__BUZZ_E2E_MOUNT_AGENT_SESSION_PANEL__ = async ({
|
||||
agentPubkey,
|
||||
channelId,
|
||||
canInterruptTurn = true,
|
||||
}) => {
|
||||
directPanelRoot?.unmount();
|
||||
directPanelContainer?.remove();
|
||||
directPanelQueryClient?.clear();
|
||||
|
||||
directPanelContainer = document.createElement("div");
|
||||
directPanelContainer.dataset.testid = "e2e-direct-agent-session-panel-root";
|
||||
document.body.appendChild(directPanelContainer);
|
||||
directPanelQueryClient = new QueryClient({
|
||||
defaultOptions: {
|
||||
queries: {
|
||||
retry: false,
|
||||
networkMode: "always",
|
||||
refetchOnWindowFocus: false,
|
||||
},
|
||||
mutations: { networkMode: "always" },
|
||||
},
|
||||
});
|
||||
directPanelRoot = createRoot(directPanelContainer);
|
||||
_testRegisterKnownAgents("e2e-direct-agent-session-panel", [agentPubkey]);
|
||||
|
||||
const agent = {
|
||||
pubkey: agentPubkey,
|
||||
name: mockDisplayNames.get(agentPubkey) ?? "E2E agent",
|
||||
status: "running" as const,
|
||||
agentSource: "managed" as const,
|
||||
canInterruptTurn,
|
||||
channelIds: [channelId],
|
||||
};
|
||||
directPanelRoot.render(
|
||||
React.createElement(
|
||||
QueryClientProvider,
|
||||
{ client: directPanelQueryClient },
|
||||
React.createElement(
|
||||
CommunitiesProvider,
|
||||
null,
|
||||
React.createElement(AgentSessionThreadPanel, {
|
||||
agent,
|
||||
canInterruptTurn,
|
||||
channel: null,
|
||||
channelId,
|
||||
isSinglePanelView: true,
|
||||
layout: "standalone",
|
||||
onClose: () => {},
|
||||
widthPx: 420,
|
||||
}),
|
||||
),
|
||||
),
|
||||
);
|
||||
await new Promise<void>((resolve) => window.setTimeout(resolve, 0));
|
||||
};
|
||||
mockMediaProxyPort = config.mock?.mediaProxyInitiallyUnavailable
|
||||
? 0
|
||||
: MOCK_MEDIA_PROXY_PORT;
|
||||
@@ -12139,6 +12324,32 @@ export function maybeInstallE2eTauriMocks() {
|
||||
mockMeshState.nodeMode = null;
|
||||
mockMeshState.activeModel = null;
|
||||
return meshNodeStatus("off", null);
|
||||
/** Build a test-only signed-looking observer control event. */
|
||||
case "build_observer_control_event": {
|
||||
const input = payload as { agentPubkey: string; payload: unknown };
|
||||
return JSON.stringify(
|
||||
createMockEvent(
|
||||
KIND_AGENT_OBSERVER_FRAME,
|
||||
JSON.stringify(input.payload),
|
||||
[
|
||||
["p", input.agentPubkey],
|
||||
["agent", input.agentPubkey],
|
||||
["frame", "control"],
|
||||
],
|
||||
DEFAULT_MOCK_IDENTITY.pubkey,
|
||||
),
|
||||
);
|
||||
}
|
||||
case "decrypt_observer_event": {
|
||||
const event = JSON.parse(
|
||||
(payload as { eventJson: string }).eventJson,
|
||||
) as RelayEvent;
|
||||
try {
|
||||
return JSON.parse(event.content);
|
||||
} catch {
|
||||
throw new Error("mock observer event content is not JSON");
|
||||
}
|
||||
}
|
||||
case "get_identity": {
|
||||
const isLost =
|
||||
!mockIdentityLostCleared && activeConfig?.mock?.identityLost === true;
|
||||
@@ -13529,6 +13740,8 @@ export function maybeInstallE2eTauriMocks() {
|
||||
return [];
|
||||
case "set_agent_managed_profiles":
|
||||
return undefined;
|
||||
case "set_thread_scoped_acp_sessions":
|
||||
return undefined;
|
||||
case "set_managed_agent_auto_restart":
|
||||
return handleSetManagedAgentAutoRestart(
|
||||
payload as Parameters<typeof handleSetManagedAgentAutoRestart>[0],
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import { mockSearchHitMatches } from "./e2eBridge.ts";
|
||||
import { mockSearchHitMatches } from "./e2eBridgeSearch.ts";
|
||||
|
||||
const NO_FILTERS = { query: "", authorSet: null };
|
||||
|
||||
|
||||
46
desktop/src/testing/e2eBridgeSearch.ts
Normal file
46
desktop/src/testing/e2eBridgeSearch.ts
Normal file
@@ -0,0 +1,46 @@
|
||||
/**
|
||||
* Mock-mode `search_messages` predicate, mirroring the relay's filter contract.
|
||||
*
|
||||
* `since`/`until` are NIP-01 bounds and both inclusive. The `before:` operator's
|
||||
* exclusivity is encoded upstream by subtracting one second before this helper
|
||||
* receives the bound.
|
||||
*/
|
||||
type SearchHit = {
|
||||
channel_id: string | null;
|
||||
pubkey: string;
|
||||
created_at: number;
|
||||
content: string;
|
||||
channel_name: string | null;
|
||||
};
|
||||
|
||||
export function mockSearchHitMatches(
|
||||
hit: SearchHit,
|
||||
filters: {
|
||||
/** Lowercased FTS query; empty matches everything. */
|
||||
query: string;
|
||||
channelId?: string;
|
||||
authorSet: Set<string> | null;
|
||||
since?: number;
|
||||
until?: number;
|
||||
},
|
||||
): boolean {
|
||||
if (filters.channelId && hit.channel_id !== filters.channelId) {
|
||||
return false;
|
||||
}
|
||||
if (filters.authorSet && !filters.authorSet.has(hit.pubkey.toLowerCase())) {
|
||||
return false;
|
||||
}
|
||||
if (filters.since != null && hit.created_at < filters.since) {
|
||||
return false;
|
||||
}
|
||||
if (filters.until != null && hit.created_at > filters.until) {
|
||||
return false;
|
||||
}
|
||||
if (!filters.query) {
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
hit.content.toLowerCase().includes(filters.query) ||
|
||||
(hit.channel_name?.toLowerCase().includes(filters.query) ?? false)
|
||||
);
|
||||
}
|
||||
413
desktop/tests/e2e/agent-control-regressions.spec.ts
Normal file
413
desktop/tests/e2e/agent-control-regressions.spec.ts
Normal file
@@ -0,0 +1,413 @@
|
||||
import { expect, test, type Page } from "@playwright/test";
|
||||
|
||||
import { installMockBridge, TEST_IDENTITIES } from "../helpers/bridge";
|
||||
|
||||
const AGENT_PUBKEY = TEST_IDENTITIES.charlie.pubkey;
|
||||
const RESTORED_UNSCOPED_AGENT_PUBKEY = TEST_IDENTITIES.outsider.pubkey;
|
||||
const CHANNEL_AGENTS = "94a444a4-c0a3-5966-ab05-530c6ddc2301";
|
||||
const CHANNEL_GENERAL = "9a1657ac-f7aa-5db0-b632-d8bbeb6dfb50";
|
||||
const CHANNEL_FOREIGN = "1c7e1c02-87bb-5e88-b2da-5a7a9432d0c9";
|
||||
|
||||
type ControlRequest = {
|
||||
agentPubkey: string;
|
||||
payload: {
|
||||
type: "cancel_turn" | "switch_model";
|
||||
channelId?: string;
|
||||
requestId?: string;
|
||||
modelId?: string;
|
||||
};
|
||||
};
|
||||
|
||||
type E2eControlWindow = Window & {
|
||||
__BUZZ_E2E_RUN_MODEL_SWITCH__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelIds: string[];
|
||||
modelId: string;
|
||||
requestId: string;
|
||||
timeoutMs?: number;
|
||||
}) => Promise<string>;
|
||||
__BUZZ_E2E_MOUNT_AGENT_SESSION_PANEL__?: (input: {
|
||||
agentPubkey: string;
|
||||
channelId: string;
|
||||
canInterruptTurn?: boolean;
|
||||
}) => Promise<void>;
|
||||
};
|
||||
|
||||
async function waitForActiveTurnSeed(page: Page) {
|
||||
await page.waitForFunction(
|
||||
() => typeof window.__BUZZ_E2E_SEED_ACTIVE_TURNS__ === "function",
|
||||
null,
|
||||
{ timeout: 10_000 },
|
||||
);
|
||||
}
|
||||
|
||||
async function seedActiveTurn(page: Page, channelId: string) {
|
||||
await page.evaluate(
|
||||
({ agentPubkey, channelId }) => {
|
||||
window.__BUZZ_E2E_SEED_ACTIVE_TURNS__?.({
|
||||
agentPubkey,
|
||||
channelId,
|
||||
turnId: `e2e-stop-${channelId}`,
|
||||
});
|
||||
},
|
||||
{ agentPubkey: AGENT_PUBKEY, channelId },
|
||||
);
|
||||
}
|
||||
|
||||
async function openAgentActivity(
|
||||
page: Page,
|
||||
activityChannelId: string,
|
||||
seedChannels: string[] = [activityChannelId],
|
||||
): Promise<ReturnType<Page["getByTestId"]>> {
|
||||
await page.goto("/", { waitUntil: "domcontentloaded" });
|
||||
await page.getByTestId("channel-agents").click();
|
||||
await expect(page.getByTestId("chat-title")).toHaveText("agents");
|
||||
await waitForActiveTurnSeed(page);
|
||||
for (const channelId of seedChannels) {
|
||||
await seedActiveTurn(page, channelId);
|
||||
}
|
||||
|
||||
const messageRow = page
|
||||
.getByTestId("message-row")
|
||||
.filter({ hasText: "Indexing the channel catalog now." });
|
||||
await expect(messageRow.first()).toBeVisible({ timeout: 8_000 });
|
||||
await messageRow.first().getByRole("button").first().click();
|
||||
|
||||
const profile = page.getByTestId("user-profile-panel");
|
||||
await expect(profile).toBeVisible({ timeout: 10_000 });
|
||||
if (activityChannelId !== CHANNEL_AGENTS) {
|
||||
const dot = page.getByRole("tab", {
|
||||
name: "Show #general activity",
|
||||
});
|
||||
await expect(dot).toBeVisible({ timeout: 5_000 });
|
||||
await dot.click();
|
||||
}
|
||||
const activity = page.getByRole("button", {
|
||||
name: /Open full activity\./,
|
||||
});
|
||||
await expect(activity).toBeVisible({ timeout: 5_000 });
|
||||
await activity.click();
|
||||
|
||||
const panel = page.getByTestId("agent-session-thread-panel");
|
||||
await expect(panel).toBeVisible({ timeout: 10_000 });
|
||||
await expect(
|
||||
page.getByTestId("agent-session-settings-menu-trigger"),
|
||||
).toBeVisible();
|
||||
return panel;
|
||||
}
|
||||
|
||||
async function readControlRequests(page: Page): Promise<ControlRequest[]> {
|
||||
return page.evaluate(
|
||||
() => (window.__BUZZ_E2E_OBSERVER_CONTROLS__ ?? []) as ControlRequest[],
|
||||
);
|
||||
}
|
||||
|
||||
async function clickStop(page: Page) {
|
||||
await page.getByTestId("agent-session-settings-menu-trigger").click();
|
||||
const stop = page.getByTestId("agent-session-stop-turn");
|
||||
await expect(stop).toBeVisible();
|
||||
await expect(stop).toBeEnabled();
|
||||
await stop.click();
|
||||
}
|
||||
|
||||
test.describe("agent control browser regressions", () => {
|
||||
test.use({ viewport: { width: 1280, height: 720 } });
|
||||
|
||||
test("Stop uses the channelId-only activity scope and carries a requestId", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
name: "Charlie",
|
||||
personaId: "control-persona",
|
||||
pubkey: AGENT_PUBKEY,
|
||||
status: "running",
|
||||
channelNames: ["agents"],
|
||||
},
|
||||
],
|
||||
observerControlResults: [{ type: "cancel_turn", status: "sent" }],
|
||||
});
|
||||
|
||||
// The visible route normalizes a requested #general activity scope back
|
||||
// to the active #agents channel. This covers route normalization and
|
||||
// effective active-channel targeting, not the channelId-only component
|
||||
// props contract (covered by the direct harness regression below).
|
||||
const panel = await openAgentActivity(page, CHANNEL_GENERAL, [
|
||||
CHANNEL_AGENTS,
|
||||
CHANNEL_GENERAL,
|
||||
]);
|
||||
// ChannelPane intentionally resolves a requested scope that differs from
|
||||
// the visible route back to the active channel. Stop must target that
|
||||
// effective channel scope, never the stale profile selection.
|
||||
await expect(page.getByTestId("agent-session-scope-label")).toHaveText(
|
||||
"Activity · #agents",
|
||||
);
|
||||
|
||||
await clickStop(page);
|
||||
await expect
|
||||
.poll(() => readControlRequests(page))
|
||||
.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
agentPubkey: AGENT_PUBKEY,
|
||||
payload: expect.objectContaining({
|
||||
type: "cancel_turn",
|
||||
channelId: CHANNEL_AGENTS,
|
||||
requestId: expect.any(String),
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
const request = (await readControlRequests(page)).find(
|
||||
(entry) => entry.payload.type === "cancel_turn",
|
||||
);
|
||||
expect(request?.payload.requestId).toBeTruthy();
|
||||
await expect(page.getByText(/Stop signal sent to Charlie/)).toBeVisible();
|
||||
await expect(panel).toBeVisible();
|
||||
});
|
||||
|
||||
test("Stop publishes from a channelId-only panel with no Channel object", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, {
|
||||
observerControlResults: [{ type: "cancel_turn", status: "sent" }],
|
||||
});
|
||||
await page.goto("/", { waitUntil: "domcontentloaded" });
|
||||
await page.waitForFunction(
|
||||
() =>
|
||||
typeof (window as E2eControlWindow)
|
||||
.__BUZZ_E2E_MOUNT_AGENT_SESSION_PANEL__ === "function",
|
||||
);
|
||||
await waitForActiveTurnSeed(page);
|
||||
await seedActiveTurn(page, CHANNEL_AGENTS);
|
||||
await page.evaluate(
|
||||
({ agentPubkey, channelId }) =>
|
||||
(window as E2eControlWindow).__BUZZ_E2E_MOUNT_AGENT_SESSION_PANEL__?.({
|
||||
agentPubkey,
|
||||
channelId,
|
||||
}),
|
||||
{ agentPubkey: AGENT_PUBKEY, channelId: CHANNEL_AGENTS },
|
||||
);
|
||||
|
||||
const panel = page.getByTestId("agent-session-thread-panel");
|
||||
await expect(panel).toBeVisible({ timeout: 10_000 });
|
||||
await page.getByTestId("agent-session-settings-menu-trigger").click();
|
||||
const stop = page.getByTestId("agent-session-stop-turn");
|
||||
await expect(stop).toBeEnabled();
|
||||
await stop.click();
|
||||
await expect
|
||||
.poll(() => readControlRequests(page))
|
||||
.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
agentPubkey: AGENT_PUBKEY,
|
||||
payload: expect.objectContaining({
|
||||
type: "cancel_turn",
|
||||
channelId: CHANNEL_AGENTS,
|
||||
requestId: expect.any(String),
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
await expect(page.getByText(/Stop signal sent to charlie/)).toBeVisible();
|
||||
});
|
||||
|
||||
test("Stop reports ambiguous_target without claiming success", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
name: "Charlie",
|
||||
personaId: "control-persona",
|
||||
pubkey: AGENT_PUBKEY,
|
||||
status: "running",
|
||||
channelNames: ["agents"],
|
||||
},
|
||||
],
|
||||
observerControlResults: [
|
||||
{ type: "cancel_turn", status: "ambiguous_target" },
|
||||
],
|
||||
});
|
||||
await openAgentActivity(page, CHANNEL_AGENTS);
|
||||
|
||||
await clickStop(page);
|
||||
await expect(page.getByText(/multiple agent sessions/)).toBeVisible();
|
||||
await expect(page.getByText(/Stop signal sent to Charlie/)).toHaveCount(0);
|
||||
});
|
||||
|
||||
test("Stop does not accept an unconfirmed or foreign-channel result", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
name: "Charlie",
|
||||
personaId: "control-persona",
|
||||
pubkey: AGENT_PUBKEY,
|
||||
status: "running",
|
||||
channelNames: ["agents"],
|
||||
},
|
||||
],
|
||||
observerControlResults: [
|
||||
{
|
||||
type: "cancel_turn",
|
||||
status: "sent",
|
||||
channelId: CHANNEL_FOREIGN,
|
||||
},
|
||||
],
|
||||
});
|
||||
await page.clock.install({ time: new Date("2026-08-30T17:00:00.000Z") });
|
||||
await openAgentActivity(page, CHANNEL_AGENTS);
|
||||
|
||||
await clickStop(page);
|
||||
await expect
|
||||
.poll(() => readControlRequests(page))
|
||||
.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
payload: expect.objectContaining({
|
||||
channelId: CHANNEL_AGENTS,
|
||||
requestId: expect.any(String),
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
// The configured result is emitted with CHANNEL_FOREIGN. The correlator
|
||||
// must ignore it, then report the bounded timeout honestly.
|
||||
await page.clock.fastForward(8_001);
|
||||
await expect(page.getByText(/hasn't confirmed it/)).toBeVisible();
|
||||
await expect(page.getByText(/Stop signal sent to Charlie/)).toHaveCount(0);
|
||||
});
|
||||
|
||||
test("model switch sends requestId and reports ambiguous_target", async ({
|
||||
page,
|
||||
}) => {
|
||||
const requestId = "e2e-model-request";
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
name: "Charlie",
|
||||
pubkey: AGENT_PUBKEY,
|
||||
status: "running",
|
||||
channelNames: ["agents"],
|
||||
},
|
||||
],
|
||||
observerControlResults: [
|
||||
{
|
||||
type: "switch_model",
|
||||
status: "ambiguous_target",
|
||||
},
|
||||
],
|
||||
});
|
||||
await page.goto("/", { waitUntil: "domcontentloaded" });
|
||||
await page.waitForFunction(
|
||||
() => typeof window.__BUZZ_E2E_RUN_MODEL_SWITCH__ === "function",
|
||||
);
|
||||
const result = await page.evaluate(
|
||||
({ agentPubkey, channelId, modelId, requestId }) =>
|
||||
(window as E2eControlWindow).__BUZZ_E2E_RUN_MODEL_SWITCH__?.({
|
||||
agentPubkey,
|
||||
channelIds: [channelId],
|
||||
modelId,
|
||||
requestId,
|
||||
}),
|
||||
{
|
||||
agentPubkey: AGENT_PUBKEY,
|
||||
channelId: CHANNEL_AGENTS,
|
||||
modelId: "test-model",
|
||||
requestId,
|
||||
},
|
||||
);
|
||||
expect(result).toBe("ambiguous");
|
||||
await expect
|
||||
.poll(() => readControlRequests(page))
|
||||
.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
agentPubkey: AGENT_PUBKEY,
|
||||
payload: expect.objectContaining({
|
||||
type: "switch_model",
|
||||
channelId: CHANNEL_AGENTS,
|
||||
modelId: "test-model",
|
||||
requestId,
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
test("model switch reports pending when no correlated terminal arrives", async ({
|
||||
page,
|
||||
}) => {
|
||||
const requestId = "e2e-pending-request";
|
||||
await installMockBridge(page, {
|
||||
observerControlResults: [
|
||||
{
|
||||
type: "switch_model",
|
||||
status: "sent",
|
||||
channelId: CHANNEL_FOREIGN,
|
||||
},
|
||||
],
|
||||
});
|
||||
await page.goto("/", { waitUntil: "domcontentloaded" });
|
||||
await page.waitForFunction(
|
||||
() => typeof window.__BUZZ_E2E_RUN_MODEL_SWITCH__ === "function",
|
||||
);
|
||||
const resultPromise = page.evaluate(
|
||||
({ agentPubkey, channelId, modelId, requestId }) =>
|
||||
(window as E2eControlWindow).__BUZZ_E2E_RUN_MODEL_SWITCH__?.({
|
||||
agentPubkey,
|
||||
channelIds: [channelId],
|
||||
modelId,
|
||||
requestId,
|
||||
timeoutMs: 50,
|
||||
}),
|
||||
{
|
||||
agentPubkey: AGENT_PUBKEY,
|
||||
channelId: CHANNEL_AGENTS,
|
||||
modelId: "test-model",
|
||||
requestId,
|
||||
},
|
||||
);
|
||||
expect(await resultPromise).toBe("pending");
|
||||
});
|
||||
|
||||
test("Stop is disabled for an unscoped restored activity URL", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, {
|
||||
managedAgents: [
|
||||
{
|
||||
name: "Outsider",
|
||||
pubkey: RESTORED_UNSCOPED_AGENT_PUBKEY,
|
||||
status: "running",
|
||||
channelNames: ["random"],
|
||||
},
|
||||
],
|
||||
});
|
||||
await page.goto(
|
||||
`/#/channels/${CHANNEL_AGENTS}?agentSession=${RESTORED_UNSCOPED_AGENT_PUBKEY}`,
|
||||
{ waitUntil: "domcontentloaded" },
|
||||
);
|
||||
await waitForActiveTurnSeed(page);
|
||||
// Seed real work for an agent that is not in the visible channel activity
|
||||
// list. The restored URL therefore mounts an unscoped panel (no
|
||||
// agentSessionChannel and no matching activity-list channel), so the
|
||||
// disabled state proves the missing scope guard rather than inactivity.
|
||||
await seedActiveTurn(page, CHANNEL_AGENTS);
|
||||
const panel = page.getByTestId("agent-session-thread-panel");
|
||||
await expect(panel).toBeVisible({ timeout: 10_000 });
|
||||
await expect(page.getByTestId("agent-session-scope-label")).toHaveText(
|
||||
"Activity · All channels",
|
||||
);
|
||||
await page.getByTestId("agent-session-settings-menu-trigger").click();
|
||||
await expect(page.getByTestId("agent-session-stop-turn")).toBeDisabled();
|
||||
await expect(page.getByTestId("agent-session-stop-turn")).toHaveAttribute(
|
||||
"aria-disabled",
|
||||
"true",
|
||||
);
|
||||
});
|
||||
});
|
||||
79
desktop/tests/e2e/experimental-features.spec.ts
Normal file
79
desktop/tests/e2e/experimental-features.spec.ts
Normal file
@@ -0,0 +1,79 @@
|
||||
import { expect, test } from "@playwright/test";
|
||||
|
||||
import { installMockBridge } from "../helpers/bridge";
|
||||
import { FEATURE_OVERRIDES_STORAGE_KEY } from "../helpers/features";
|
||||
import { openSettings } from "../helpers/settings";
|
||||
|
||||
const DESCRIPTION =
|
||||
"Give each channel thread isolated agent context. Applies when managed agents next start; DMs stay conversation-scoped.";
|
||||
|
||||
test("thread-scoped ACP sessions is default-off, persists, and applies on reload", async ({
|
||||
page,
|
||||
}) => {
|
||||
await installMockBridge(page, undefined, { seedPreviewFeatures: false });
|
||||
await page.goto("/");
|
||||
|
||||
await expect
|
||||
.poll(() =>
|
||||
page.evaluate(() => {
|
||||
const call = window.__BUZZ_E2E_COMMAND_LOG__?.find(
|
||||
({ command }) => command === "apply_workspace",
|
||||
);
|
||||
return (call?.payload as { threadScopedAcpSessions?: boolean } | null)
|
||||
?.threadScopedAcpSessions;
|
||||
}),
|
||||
)
|
||||
.toBe(false);
|
||||
|
||||
await openSettings(page, "experimental");
|
||||
const toggle = page.getByTestId("feature-toggle-threadScopedAcpSessions");
|
||||
|
||||
await expect(
|
||||
page.getByText("Thread Scoped ACP Sessions", { exact: true }),
|
||||
).toBeVisible();
|
||||
await expect(page.getByText(DESCRIPTION, { exact: true })).toBeVisible();
|
||||
await expect(toggle).not.toBeChecked();
|
||||
await expect(page.getByTestId("feature-toggle-projects")).not.toBeChecked();
|
||||
await expect(page.getByTestId("feature-toggle-workflows")).not.toBeChecked();
|
||||
|
||||
await toggle.click();
|
||||
await expect(toggle).toBeChecked();
|
||||
await expect
|
||||
.poll(() =>
|
||||
page.evaluate(() => {
|
||||
const calls = window.__BUZZ_E2E_COMMAND_LOG__ ?? [];
|
||||
const call = calls.findLast(
|
||||
({ command }) => command === "set_thread_scoped_acp_sessions",
|
||||
);
|
||||
return (call?.payload as { enabled?: boolean } | null)?.enabled;
|
||||
}),
|
||||
)
|
||||
.toBe(true);
|
||||
await expect
|
||||
.poll(() =>
|
||||
page.evaluate(
|
||||
(key) => window.localStorage.getItem(key),
|
||||
FEATURE_OVERRIDES_STORAGE_KEY,
|
||||
),
|
||||
)
|
||||
.toContain('"threadScopedAcpSessions":true');
|
||||
|
||||
await page.reload();
|
||||
// Settings route/section state survives reload, so wait for that restored
|
||||
// view rather than trying to reopen settings from the app chrome.
|
||||
await expect(page.getByTestId("settings-view")).toBeVisible();
|
||||
await expect(
|
||||
page.getByTestId("feature-toggle-threadScopedAcpSessions"),
|
||||
).toBeChecked();
|
||||
await expect
|
||||
.poll(() =>
|
||||
page.evaluate(() => {
|
||||
const call = window.__BUZZ_E2E_COMMAND_LOG__?.find(
|
||||
({ command }) => command === "apply_workspace",
|
||||
);
|
||||
return (call?.payload as { threadScopedAcpSessions?: boolean } | null)
|
||||
?.threadScopedAcpSessions;
|
||||
}),
|
||||
)
|
||||
.toBe(true);
|
||||
});
|
||||
@@ -300,6 +300,14 @@ type MockBridgeOptions = {
|
||||
closeChannelLiveSubscriptionOnce?: boolean;
|
||||
/** Reject successive kind-9 sends with these messages, then resume. */
|
||||
sendMessageErrors?: string[];
|
||||
/** Test-only observer control results emitted after mock control publishes. */
|
||||
observerControlResults?: Array<{
|
||||
type: "cancel_turn" | "switch_model";
|
||||
status: string;
|
||||
channelId?: string | null;
|
||||
requestId?: string;
|
||||
modelId?: string;
|
||||
}>;
|
||||
/** Reject successive managed-agent starts, then resume. */
|
||||
startManagedAgentErrors?: string[];
|
||||
/** Delay (ms) after snapshotting a thread-replies page so E2E tests can
|
||||
|
||||
@@ -7,6 +7,7 @@ type SettingsSection =
|
||||
| "agents"
|
||||
| "channel-templates"
|
||||
| "compute"
|
||||
| "experimental"
|
||||
| "appearance"
|
||||
| "shortcuts"
|
||||
| "hosted-communities"
|
||||
|
||||
@@ -7,6 +7,12 @@
|
||||
"description": "YAML-defined automations with approval gates",
|
||||
"platforms": ["desktop"]
|
||||
},
|
||||
{
|
||||
"id": "threadScopedAcpSessions",
|
||||
"name": "Thread Scoped ACP Sessions",
|
||||
"description": "Give each channel thread isolated agent context. Applies when managed agents next start; DMs stay conversation-scoped.",
|
||||
"platforms": ["desktop"]
|
||||
},
|
||||
{
|
||||
"id": "projects",
|
||||
"name": "Projects",
|
||||
|
||||
Reference in New Issue
Block a user