mirror of
https://github.com/block/buzz.git
synced 2026-10-29 13:31:44 +00:00
fix(desktop): preserve launch community separately from runtime aliases
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
@@ -176,14 +176,19 @@ async fn start_local_agent_with_preflight(
|
||||
replay_floor_unix: Option<u64>,
|
||||
) -> Result<ManagedAgentSummary, String> {
|
||||
let launch_owner = workspace_owner_hex(state)?;
|
||||
let launch_key = crate::managed_agents::ManagedAgentRuntimeKey::new(
|
||||
pubkey,
|
||||
&relay_ws_url_with_override(state),
|
||||
// Runtime keys fold loopback aliases for process bookkeeping, not tenant
|
||||
// identity. Preserve the workspace authority across the preflight await.
|
||||
let launch_relay = crate::relay::bind_expected_relay_scope(
|
||||
expected_relay_url,
|
||||
relay_ws_url_with_override(state),
|
||||
)?;
|
||||
let launch_key =
|
||||
crate::managed_agents::ManagedAgentRuntimeKey::new(pubkey, launch_relay.as_str())?;
|
||||
let resume = if matches!(intent, LocalStartIntent::Explicit) {
|
||||
Some(crate::managed_agents::remote_stop::capture_resume(
|
||||
app,
|
||||
&launch_key,
|
||||
launch_relay.as_str(),
|
||||
&launch_owner,
|
||||
)?)
|
||||
} else {
|
||||
@@ -236,10 +241,8 @@ async fn start_local_agent_with_preflight(
|
||||
// below — the check is tied to its use, so a switch landing after this
|
||||
// point can no longer retarget the spawn (it only changes state this
|
||||
// call no longer consults).
|
||||
let workspace_relay_url = crate::relay::bind_expected_relay_scope(
|
||||
expected_relay_url.or(Some(launch_key.relay_url.as_str())),
|
||||
crate::relay::relay_ws_url_with_override(state),
|
||||
)?;
|
||||
let workspace_relay_url =
|
||||
launch_relay.revalidate(crate::relay::relay_ws_url_with_override(state))?;
|
||||
// Bind the active owner after the same final await as the relay. A
|
||||
// same-relay identity replacement during mesh preflight must not release
|
||||
// the stale preflight owner to spawn.
|
||||
|
||||
@@ -122,11 +122,12 @@ pub(crate) struct ResumeTicket {
|
||||
}
|
||||
|
||||
pub(crate) fn capture_resume(
|
||||
app: &AppHandle,
|
||||
app: &AppHandle<impl tauri::Runtime>,
|
||||
key: &ManagedAgentRuntimeKey,
|
||||
community: &str,
|
||||
owner: &str,
|
||||
) -> Result<ResumeTicket, String> {
|
||||
let conn = connection(app, key, owner)?;
|
||||
let conn = connection(app, community, owner)?;
|
||||
schema(&conn)?;
|
||||
let previous = conn
|
||||
.query_row(
|
||||
@@ -140,12 +141,11 @@ pub(crate) fn capture_resume(
|
||||
}
|
||||
|
||||
fn connection(
|
||||
app: &AppHandle,
|
||||
key: &ManagedAgentRuntimeKey,
|
||||
app: &AppHandle<impl tauri::Runtime>,
|
||||
community: &str,
|
||||
owner: &str,
|
||||
) -> Result<Connection, String> {
|
||||
let path =
|
||||
scoped_retention_db_path(&super::managed_agents_base_dir(app)?, &key.relay_url, owner);
|
||||
let path = scoped_retention_db_path(&super::managed_agents_base_dir(app)?, community, owner);
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(|e| e.to_string())?;
|
||||
}
|
||||
@@ -155,8 +155,9 @@ fn connection(
|
||||
/// Every ordinary spawn passes here, including restore/config/reconcile.
|
||||
/// Caller holds the existing transition lock through child registration.
|
||||
pub(crate) fn check_launch(
|
||||
app: &AppHandle,
|
||||
app: &AppHandle<impl tauri::Runtime>,
|
||||
key: &ManagedAgentRuntimeKey,
|
||||
community: &str,
|
||||
owner: Option<&str>,
|
||||
resume: Option<&ResumeTicket>,
|
||||
) -> Result<(), String> {
|
||||
@@ -165,18 +166,18 @@ pub(crate) fn check_launch(
|
||||
if owner != Some(current_owner.as_str()) {
|
||||
return Err("Desktop launch owner changed".into());
|
||||
}
|
||||
let conn = connection(app, key, ¤t_owner)?;
|
||||
let conn = connection(app, community, ¤t_owner)?;
|
||||
schema(&conn)?;
|
||||
let scope = super::retention::RetentionScope {
|
||||
db_path: scoped_retention_db_path(
|
||||
&super::managed_agents_base_dir(app)?,
|
||||
&key.relay_url,
|
||||
community,
|
||||
¤t_owner,
|
||||
),
|
||||
relay_url: key.relay_url.clone(),
|
||||
relay_url: community.to_owned(),
|
||||
owner_keys: state.signing_keys()?,
|
||||
};
|
||||
let mut local_conn = connection(app, key, ¤t_owner)?;
|
||||
let mut local_conn = connection(app, community, ¤t_owner)?;
|
||||
let host = crate::commands::desktop_stop::local_id(&mut local_conn, &scope)?;
|
||||
if super::placement::blocked(&conn, &key.pubkey, &host)?
|
||||
&& (resume.is_none() || super::placement::has_start(&conn, &key.pubkey)?)
|
||||
@@ -210,8 +211,9 @@ fn allow_launch(row: Option<&(String, bool)>, resume: Option<&ResumeTicket>) ->
|
||||
/// A failed spawn must not unblock config/restore. Commit only after the child
|
||||
/// has its ordinary receipt and tracked handle, still under the transition lock.
|
||||
pub(crate) fn finish_resume(
|
||||
app: &AppHandle,
|
||||
app: &AppHandle<impl tauri::Runtime>,
|
||||
key: &ManagedAgentRuntimeKey,
|
||||
community: &str,
|
||||
owner: Option<&str>,
|
||||
ticket: Option<&ResumeTicket>,
|
||||
) -> Result<(), String> {
|
||||
@@ -219,8 +221,8 @@ pub(crate) fn finish_resume(
|
||||
return Ok(());
|
||||
}
|
||||
let owner = owner.ok_or("Desktop launch owner unavailable")?;
|
||||
check_launch(app, key, Some(owner), ticket)?;
|
||||
connection(app, key, owner)?
|
||||
check_launch(app, key, community, Some(owner), ticket)?;
|
||||
connection(app, community, owner)?
|
||||
.execute(
|
||||
"UPDATE desktop_stop_fence SET blocked=0 WHERE agent=?1",
|
||||
[&key.pubkey],
|
||||
@@ -250,6 +252,69 @@ mod tests {
|
||||
.sign_with_keys(keys)
|
||||
.unwrap()
|
||||
}
|
||||
#[test]
|
||||
fn launch_reads_receiver_fence_in_original_community_not_runtime_alias() {
|
||||
let mut context = tauri::test::mock_context(tauri::test::noop_assets());
|
||||
context.config_mut().identifier = format!("buzz-test-{}", uuid::Uuid::new_v4());
|
||||
let state = crate::app_state::build_app_state();
|
||||
let keys = state.signing_keys().unwrap();
|
||||
let owner = keys.public_key().to_hex();
|
||||
let app = tauri::test::mock_builder()
|
||||
.manage(state)
|
||||
.build(context)
|
||||
.unwrap();
|
||||
let root = app.path().app_data_dir().unwrap();
|
||||
let community = "ws://localhost:3037";
|
||||
let agent = Keys::generate().public_key().to_hex();
|
||||
let key = ManagedAgentRuntimeKey::new(&agent, community).unwrap();
|
||||
assert_ne!(key.relay_url, community);
|
||||
let scope = super::super::retention::RetentionScope {
|
||||
db_path: scoped_retention_db_path(&root.join("agents"), community, &owner),
|
||||
relay_url: community.into(),
|
||||
owner_keys: keys.clone(),
|
||||
};
|
||||
let mut conn = connection(app.handle(), community, &owner).unwrap();
|
||||
let host = crate::commands::desktop_stop::local_id(&mut conn, &scope).unwrap();
|
||||
let target = StopTarget {
|
||||
v: 1,
|
||||
community: community.into(),
|
||||
desktop: host,
|
||||
agent,
|
||||
};
|
||||
let stop = request(&keys, &target, 100);
|
||||
receive(
|
||||
&mut conn,
|
||||
&stop,
|
||||
&keys,
|
||||
community,
|
||||
&target.desktop,
|
||||
true,
|
||||
|_| Ok(()),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(check_launch(app.handle(), &key, community, Some(&owner), None).is_err());
|
||||
// The numeric community is a different authority, even though the
|
||||
// process bookkeeping key historically folds the two spellings.
|
||||
assert!(check_launch(app.handle(), &key, &key.relay_url, Some(&owner), None).is_ok());
|
||||
let resume = capture_resume(app.handle(), &key, community, &owner).unwrap();
|
||||
assert!(check_launch(app.handle(), &key, community, Some(&owner), Some(&resume)).is_ok());
|
||||
let newer = request(&keys, &target, 101);
|
||||
receive(
|
||||
&mut conn,
|
||||
&newer,
|
||||
&keys,
|
||||
community,
|
||||
&target.desktop,
|
||||
true,
|
||||
|_| Ok(()),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(check_launch(app.handle(), &key, community, Some(&owner), Some(&resume)).is_err());
|
||||
drop(conn);
|
||||
drop(app);
|
||||
std::fs::remove_dir_all(root).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn launch_fence_requires_explicit_start_and_rejects_delayed_preflight() {
|
||||
let stopped = ("stop-a".to_owned(), true);
|
||||
|
||||
@@ -21,7 +21,11 @@ use tauri::Manager;
|
||||
enum SpawnOutcome {
|
||||
/// Boxed: the spawned process carries its full spawn-config snapshot, so an
|
||||
/// inline variant would make every `Skipped`/`Failed` outcome pay for it.
|
||||
Spawned(super::ManagedAgentRuntimeKey, Box<ManagedAgentProcess>),
|
||||
Spawned(
|
||||
super::ManagedAgentRuntimeKey,
|
||||
String,
|
||||
Box<ManagedAgentProcess>,
|
||||
),
|
||||
Skipped,
|
||||
Failed(String),
|
||||
}
|
||||
@@ -341,7 +345,7 @@ pub async fn restore_managed_agents_on_launch(
|
||||
spawn_agent_child(
|
||||
app,
|
||||
record,
|
||||
&key.relay_url,
|
||||
&relay_url,
|
||||
true,
|
||||
owner_hex_ref,
|
||||
None,
|
||||
@@ -349,7 +353,7 @@ pub async fn restore_managed_agents_on_launch(
|
||||
)
|
||||
}) {
|
||||
Ok(process) => {
|
||||
SpawnOutcome::Spawned(key, Box::new(process))
|
||||
SpawnOutcome::Spawned(key, relay_url, Box::new(process))
|
||||
}
|
||||
Err(error) => SpawnOutcome::Failed(error),
|
||||
}
|
||||
@@ -388,7 +392,7 @@ pub async fn restore_managed_agents_on_launch(
|
||||
// Skipped means a concurrent reconcile already owns a live child for
|
||||
// this pair; leave its runtime and record state untouched.
|
||||
SpawnOutcome::Skipped => continue,
|
||||
SpawnOutcome::Spawned(key, mut process) => {
|
||||
SpawnOutcome::Spawned(key, relay_url, mut process) => {
|
||||
let Ok(record) = find_managed_agent_mut(&mut records, &pubkey) else {
|
||||
continue;
|
||||
};
|
||||
@@ -416,11 +420,11 @@ pub async fn restore_managed_agents_on_launch(
|
||||
key.clone(),
|
||||
super::ManagedAgentPairRuntime::starting(*process),
|
||||
);
|
||||
// Carry the spawn key's relay into profile reconciliation so
|
||||
// Carry the original launch community into profile reconciliation so
|
||||
// the background task queries/publishes on the relay this
|
||||
// spawn was actually keyed to — not whatever workspace is
|
||||
// active when the task eventually executes.
|
||||
successfully_spawned.push((pubkey, key.relay_url.clone()));
|
||||
successfully_spawned.push((pubkey, relay_url));
|
||||
}
|
||||
SpawnOutcome::Failed(error) => {
|
||||
let Ok(record) = find_managed_agent_mut(&mut records, &pubkey) else {
|
||||
|
||||
@@ -477,7 +477,7 @@ pub(crate) fn spawn_agent_child_with_broker(
|
||||
broker: Option<&super::broker_launch::BrokerSession>,
|
||||
) -> Result<crate::managed_agents::ManagedAgentProcess, String> {
|
||||
let key = ManagedAgentRuntimeKey::new(record.pubkey.clone(), relay_url)?;
|
||||
super::remote_stop::check_launch(app, &key, owner_hex, resume)?;
|
||||
super::remote_stop::check_launch(app, &key, relay_url, owner_hex, resume)?;
|
||||
if let Some(session) = broker {
|
||||
session.validate(super::broker_launch::LaunchScope {
|
||||
owner: owner_hex.ok_or("Desktop owner unavailable")?,
|
||||
@@ -574,7 +574,8 @@ pub(crate) fn spawn_agent_child_with_broker(
|
||||
|
||||
// The caller supplies the explicit canonical pair relay. This is the only
|
||||
// relay this child may connect to, regardless of the record/workspace default.
|
||||
let effective_relay_url = runtime_key.relay_url.clone();
|
||||
// Process identity normalization must not select a different relay tenant.
|
||||
let effective_relay_url = relay_url.to_owned();
|
||||
// Augment PATH for DMG launches so child processes can find:
|
||||
// - bundled CLI via ~/.local/bin symlink
|
||||
// - nvm-managed node/npm (nvm initializes only in interactive shells)
|
||||
@@ -946,7 +947,7 @@ pub fn start_managed_agent_process(
|
||||
let mut process = spawn_agent_child(
|
||||
app,
|
||||
record,
|
||||
&key.relay_url,
|
||||
workspace_relay.as_str(),
|
||||
false,
|
||||
owner_hex,
|
||||
replay_floor_unix,
|
||||
@@ -973,7 +974,7 @@ pub fn start_managed_agent_process(
|
||||
record.last_error_code = None;
|
||||
|
||||
runtimes.insert(key.clone(), ManagedAgentPairRuntime::starting(process));
|
||||
super::remote_stop::finish_resume(app, &key, owner_hex, resume)?;
|
||||
super::remote_stop::finish_resume(app, &key, workspace_relay.as_str(), owner_hex, resume)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -322,6 +322,7 @@ pub(crate) fn start_pair_locked(
|
||||
Some(super::remote_stop::capture_resume(
|
||||
&app,
|
||||
&key,
|
||||
&relay_url,
|
||||
owner.as_deref().ok_or("Desktop owner unavailable")?,
|
||||
)?)
|
||||
} else {
|
||||
@@ -330,7 +331,7 @@ pub(crate) fn start_pair_locked(
|
||||
let mut process = super::spawn_agent_child_with_broker(
|
||||
&app,
|
||||
record,
|
||||
&key.relay_url,
|
||||
&relay_url,
|
||||
lazy,
|
||||
owner.as_deref(),
|
||||
None,
|
||||
@@ -355,7 +356,7 @@ pub(crate) fn start_pair_locked(
|
||||
record.last_stopped_at = None;
|
||||
record.last_error = None;
|
||||
runtimes.insert(key.clone(), ManagedAgentPairRuntime::starting(process));
|
||||
super::remote_stop::finish_resume(&app, &key, owner.as_deref(), resume.as_ref())?;
|
||||
super::remote_stop::finish_resume(&app, &key, &relay_url, owner.as_deref(), resume.as_ref())?;
|
||||
let status = status_for(&app, record, &key, runtimes.get(&key), None);
|
||||
drop(runtimes);
|
||||
save_managed_agents(&app, &records)?;
|
||||
@@ -462,7 +463,7 @@ async fn probe_agent_relay_access(
|
||||
let key = ManagedAgentRuntimeKey::new(record.pubkey.clone(), &requested_relay_url)?;
|
||||
let keys = nostr::Keys::parse(record.private_key_nsec.trim())
|
||||
.map_err(|error| format!("invalid managed-agent key: {error}"))?;
|
||||
let api_base = crate::relay::relay_http_base_url(&key.relay_url);
|
||||
let api_base = crate::relay::relay_http_base_url(&requested_relay_url);
|
||||
tokio::time::timeout(
|
||||
std::time::Duration::from_secs(10),
|
||||
crate::relay::query_relay_at_with_keys(
|
||||
@@ -563,7 +564,7 @@ pub async fn reconcile_managed_agent_runtimes(
|
||||
Ok((record, key, requested)) => {
|
||||
match start_pair(
|
||||
record.pubkey.clone(),
|
||||
key.relay_url.clone(),
|
||||
requested.clone(),
|
||||
true,
|
||||
Some(&record.updated_at),
|
||||
false,
|
||||
|
||||
@@ -41,6 +41,12 @@ impl ScopedWorkspaceRelay {
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
|
||||
/// Recheck a captured workspace after preflight without substituting a
|
||||
/// runtime-normalized URL (which may alias distinct tenant authorities).
|
||||
pub fn revalidate(self, workspace_relay_url: String) -> Result<Self, String> {
|
||||
bind_expected_relay_scope(Some(self.as_str()), workspace_relay_url)
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate a caller-captured relay scope against one workspace-relay read
|
||||
@@ -119,6 +125,30 @@ mod tests {
|
||||
bind_expected_signer,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn preflight_revalidation_preserves_localhost_authority_not_runtime_alias() {
|
||||
let relay = "ws://localhost:3037";
|
||||
let captured = bind_expected_relay_scope(None, relay.into()).unwrap();
|
||||
let runtime =
|
||||
crate::managed_agents::ManagedAgentRuntimeKey::new("a".repeat(64), captured.as_str())
|
||||
.unwrap();
|
||||
assert_eq!(runtime.relay_url, "ws://127.0.0.1:3037");
|
||||
assert_eq!(captured.revalidate(relay.into()).unwrap().as_str(), relay);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preflight_revalidation_rejects_switch_even_to_same_runtime_alias() {
|
||||
for changed in ["ws://127.0.0.1:3037", "wss://other.example"] {
|
||||
let captured = bind_expected_relay_scope(None, "ws://localhost:3037".into()).unwrap();
|
||||
assert!(captured.revalidate(changed.into()).is_err());
|
||||
}
|
||||
assert!(bind_expected_relay_scope(
|
||||
Some("ws://localhost:3037"),
|
||||
"ws://127.0.0.1:3037".into(),
|
||||
)
|
||||
.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn matching_scope_passes_across_ws_http_normalization() {
|
||||
assert_expected_relay_scope(Some("wss://tenant-a.example"), "https://tenant-a.example")
|
||||
|
||||
Reference in New Issue
Block a user