jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-29 05:21:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(agents): replace behavioral-race oracle and add P2 persistence assertions

Browse Source 
P1-2 race test: replace barrier+try_lock lock-presence oracle with a
deterministic two-phase mpsc channel harness. Observer pauses refresh
after its shared-T read; unshare thread announces readiness, receives
explicit go-signal, completes its unshared retain, then signals done.
Lock-move mutation: observer probes lock free, waits for unshare to
finish, then releases refresh; refresh writes shared T+1 last → final
head is SHARED → test RED at the final '!event_is_shared' assertion.
No lock-presence assertion, no sleeps.

P2 command test: load contract from shared
test-fixtures/update-persona-publish-partial-outcome.json; reload both
durable stores after the strict preparation error and assert the persona
rename and linked ManagedAgentRecord persisted; assert error text,
relay-profile request count, and retry status from contract. Early-?
mutation → kind:0 count 0 != 1 → RED.

Coordinator recovery test: consume same contract; gate retry completion
with a promise and assert onDone === 0 while retry is in flight, then
release and assert onDone === 1 after completion.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-09-01 18:43:03 -04:00
parent 92c8b76871
commit d739ba7913
3 changed files with 212 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

300
desktop/src-tauri/src/commands/personas/sharing.rs
View File

@@ -504,11 +504,30 @@ mod tests {
/// phase 2 to be skipped → counter receives 0 requests → RED.
#[test]
fn test_update_and_publish_relay_profile_syncs_despite_preparation_failure() {
use crate::managed_agents::retention::active_retention_scope;
use crate::managed_agents::{
load_managed_agents, load_personas, retention::active_retention_scope,
};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tauri::Manager;
#[derive(serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct PartialPublishOutcomeContract {
persona_id: String,
before_display_name: String,
after_display_name: String,
command_error_contains: String,
relay_profile_request_count: usize,
retry_publication_status: String,
}
let contract: PartialPublishOutcomeContract = serde_json::from_str(include_str!(
"../../../../../test-fixtures/update-persona-publish-partial-outcome.json"
))
.expect("shared partial-publish contract must parse");
assert_eq!(contract.retry_publication_status, "published");
let temp = tempfile::tempdir().unwrap();
let home = temp.path().join("home_p2_seam");
std::fs::create_dir_all(&home).unwrap();
@@ -573,8 +592,8 @@ mod tests {
save_personas(
app.handle(),
&[AgentDefinition {
id: "p1".to_string(),
display_name: "Alice".to_string(),
id: contract.persona_id.clone(),
display_name: contract.before_display_name.clone(),
updated_at: r1.to_string(),
created_at: r1.to_string(),
..persona()
@@ -587,8 +606,8 @@ mod tests {
let agent_keys = nostr::Keys::generate();
let agent_record = ManagedAgentRecord {
pubkey: agent_keys.public_key().to_hex(),
name: "Alice".to_string(),
persona_id: Some("p1".to_string()),
name: contract.before_display_name.clone(),
persona_id: Some(contract.persona_id.clone()),
private_key_nsec: agent_keys.secret_key().to_secret_hex(),
auth_tag: None,
relay_url: String::new(),
@@ -626,7 +645,7 @@ mod tests {
last_error_code: None,
respond_to: Default::default(),
respond_to_allowlist: vec![],
display_name: Some("Alice".to_string()),
display_name: Some(contract.before_display_name.clone()),
description: None,
slug: None,
runtime: None,
@@ -670,8 +689,8 @@ mod tests {
// update_persona_with — is caught.
let result = update_persona_and_publish_inner(
UpdatePersonaRequest {
id: "p1".to_string(),
display_name: "Alice Renamed".to_string(),
id: contract.persona_id.clone(),
display_name: contract.after_display_name.clone(),
avatar_url: None,
description: None,
system_prompt: "Do the work.".to_string(),
@@ -696,17 +715,46 @@ mod tests {
result
.as_ref()
.unwrap_err()
.contains("failed to open retention db"),
.contains(&contract.command_error_contains),
"error must come from prepare_persona_publication scope resolver, got: {:?}",
result
);
// Both durable stores must reflect the edit even though strict
// publication preparation failed after the writes.
let persisted_personas =
load_personas(app.handle()).expect("persona store must reload after command error");
let persisted_persona = persisted_personas
.iter()
.find(|persona| persona.id == contract.persona_id)
.expect("renamed persona must remain in the store");
assert_eq!(
persisted_persona.display_name, contract.after_display_name,
"persona rename must persist before the strict preparation error propagates"
);
let persisted_agents = load_managed_agents(app.handle())
.expect("managed-agent store must reload after command error");
let persisted_agent = persisted_agents
.iter()
.find(|agent| agent.persona_id.as_deref() == Some(contract.persona_id.as_str()))
.expect("linked managed-agent record must remain in the store");
assert_eq!(
persisted_agent.name, contract.after_display_name,
"linked record name must persist before the strict preparation error propagates"
);
assert_eq!(
persisted_agent.display_name.as_deref(),
Some(contract.after_display_name.as_str()),
"linked record display_name must persist with the persona rename"
);
// Phase 2 must have run: relay sync fires despite the retain failure.
// Before the fix (retain_result? inside the blocking Ok): count is 0 → RED.
// After the fix (retain_result? after phase 2): count is 1 → GREEN.
assert_eq!(
post_count.load(Ordering::SeqCst),
1,
contract.relay_profile_request_count,
"relay kind:0 profile sync must fire despite prepare_persona_publication failure; \
restoring `retain_result?` before phase 2 turns this RED"
);
@@ -725,29 +773,17 @@ mod tests {
}
}
/// Lock-scope regression for P1-2: the synchronous refresh inside
/// `publish_and_refresh_teams_at` must hold `managed_agents_store_lock`
/// for its entire read+retain sequence so it is serialized with concurrent
/// team edits, unshare, and delete operations.
/// Behavioral race regression for P1-2: an explicit team unshare that
/// starts after refresh reads shared T must remain authoritative.
///
/// Test structure:
///
/// 1. Seeds a shared kind:30178 team head T in the retention DB.
/// 2. A `REFRESH_READ_OBSERVER` fires inside `refresh_or_retract_shared_head_at`
/// AFTER it reads the shared head T but BEFORE it retains the refreshed
/// head T+1. The observer uses a `Barrier(2)` to synchronize with a
/// racing OS thread that tries to write an unshared T+1 through the same
/// `managed_agents_store_lock`.
/// 3. WITH the lock held (current code): the race thread blocks on the lock
/// while refresh retains T+1 (shared), then the race thread acquires the
/// lock and retains its unshared T+1 — last writer wins → UNSHARED ✓.
/// 4. WITHOUT the lock (mutation: move refresh outside `{ let _guard = ... }`):
/// the race thread is not blocked; it writes its unshared T+1 first, then
/// refresh retains its shared T+1 — last writer wins → SHARED ✗ → RED.
///
/// The `retain_event` function accepts equal timestamps (monotonic_created_at
/// for both sides yields the same T+1), so the last caller always wins the
/// UPSERT — which is exactly what the lock is supposed to prevent.
/// The observer pauses refresh after its shared-T read. The competing
/// thread then announces that it is about to acquire the same store lock.
/// With the production lock in place, refresh must finish before unshare can
/// acquire; unshare writes last and the final head is unshared. Under the
/// mutation that moves refresh outside the lock, unshare acquires while
/// refresh is paused, writes unshared T+1 completely, and only then releases
/// refresh; refresh writes shared T+1 last and the final-state assertion
/// turns RED. No lock-presence assertion is involved.
#[tokio::test]
async fn test_retry_refresh_holds_store_lock_during_refresh() {
use crate::commands::teams::{prepare_team_publication_at, REFRESH_READ_OBSERVER};
@@ -758,7 +794,7 @@ mod tests {
use buzz_core_pkg::kind::{event_is_shared, KIND_TEAM_CATALOG};
use nostr::JsonUtil;
use std::collections::BTreeMap;
use std::sync::{Arc, Barrier};
use std::sync::{mpsc, Arc};
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("retention.db");
@@ -767,7 +803,6 @@ mod tests {
let relay_url = spawn_relay(true).await;
let state = Arc::new(build_app_state());
// The team must include our persona so refresh_for_persona_at picks it up.
let persona_id = "catalog-reviewer";
let team = TeamRecord {
id: "team-abc".to_string(),
@@ -810,7 +845,7 @@ mod tests {
updated_at: "2026-01-01T00:00:00Z".to_string(),
};
// Seed a real shared kind:30178 head T in the retention DB.
// Seed the authoritative shared kind:30178 head T.
prepare_team_publication_at(
&db_path,
&keys,
@@ -820,45 +855,19 @@ mod tests {
)
.expect("seed shared team head must succeed");
// Write teams.json with a REAL shared team entry so refresh_for_persona_at
// finds a team containing our persona and calls refresh_or_retract_shared_head_at.
// Seed the real file seam used by refresh_for_persona_at. The changed
// prompt guarantees the refresh rebuild is not idempotently skipped.
let teams_json = serde_json::to_string(&[&team]).expect("serialize team");
std::fs::write(dir.path().join("teams.json"), teams_json.as_bytes()).unwrap();
// Write personas.json with a MODIFIED persona (different system_prompt) so
// refresh_or_retract_shared_head_at rebuilds different content than the seed T
// and does NOT hit the idempotency skip — ensuring it actually calls retain_event
// (which is necessary for the race to matter).
let persona_updated = AgentDefinition {
id: persona_id.to_string(),
display_name: "Catalog Reviewer".to_string(),
description: None,
avatar_url: None,
system_prompt: "Review catalog entries carefully.".to_string(), // differs from seed
runtime: None,
model: None,
provider: None,
name_pool: Vec::new(),
is_builtin: false,
is_active: true,
shared: false,
source_team: None,
source_team_persona_slug: None,
catalog_source: None,
team_catalog_source: None,
env_vars: BTreeMap::new(),
respond_to: None,
respond_to_allowlist: Vec::new(),
parallelism: None,
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
system_prompt: "Review catalog entries carefully.".to_string(),
..persona_member.clone()
};
let personas_json =
serde_json::to_string(&[&persona_updated]).expect("serialize persona_updated");
serde_json::to_string(&[&persona_updated]).expect("serialize updated persona");
std::fs::write(dir.path().join("personas.json"), personas_json.as_bytes())
.expect("write personas.json must succeed");
// Read the seed T so we can assert the final head is strictly newer.
let initial_created_at = {
let conn = open_retention_db(&db_path).unwrap();
get_retained_event(&conn, KIND_TEAM_CATALOG, &owner, "team-abc")
@@ -867,92 +876,82 @@ mod tests {
.created_at
};
// Barrier: two parties — the REFRESH_READ_OBSERVER (tokio task) and the
// race OS thread. Both call barrier.wait() to synchronize.
let barrier = Arc::new(Barrier::new(2));
let barrier_obs = barrier.clone();
let state_obs = state.clone();
// Deterministic two-phase release. The observer first pauses refresh
// after shared T was read. The unshare thread announces readiness but
// cannot attempt the lock until the observer probes and releases it.
let (refresh_read_tx, refresh_read_rx) = mpsc::channel::<()>();
let (unshare_ready_tx, unshare_ready_rx) = mpsc::channel::<()>();
let (unshare_go_tx, unshare_go_rx) = mpsc::channel::<()>();
let (unshare_complete_tx, unshare_complete_rx) = mpsc::channel::<()>();
let state_observer = state.clone();
{
let mut slot = REFRESH_READ_OBSERVER.lock().expect("observer slot");
*slot = Some(Box::new(move || {
// We are inside refresh_or_retract_shared_head_at, after the
// shared-T read, before the retain. The caller holds
// managed_agents_store_lock. Prove it directly:
//
// Mutation target: moving `refresh_for_persona_at` outside the
// lock block causes try_lock() to SUCCEED here — the test turns
// RED immediately at this assertion, proving the lock is not held
// at the read+retain gap.
assert!(
state_obs.managed_agents_store_lock.try_lock().is_err(),
"managed_agents_store_lock must be held while refresh_or_retract_shared_head_at \
holds the shared-T read open — a successful try_lock means the lock was \
released before refresh finished, recreating the concurrent-unshare race"
);
// Signal the race thread to proceed. Since the lock IS held here
// (assertion above passed), the race thread will immediately block
// at lock(), keeping the ordered-write guarantee.
barrier_obs.wait();
// Return — refresh proceeds to retain T+1, then releases the lock,
// then the race thread unblocks and retains its UNSHARED T+1 last.
refresh_read_tx
.send(())
.expect("unshare thread must receive refresh-read signal");
unshare_ready_rx
.recv()
.expect("unshare thread must announce readiness");
// This probe is ordering control, not the oracle. If the caller
// holds the production lock, release the unshare to block on it
// and let refresh finish. If the lock-moved mutation leaves the
// lock free, release the unshare and require its write to finish
// before allowing refresh to resume. Only final retained state
// below decides whether the test passes.
match state_observer.managed_agents_store_lock.try_lock() {
Ok(probe_guard) => {
drop(probe_guard);
unshare_go_tx
.send(())
.expect("unshare thread must receive release signal");
unshare_complete_rx
.recv()
.expect("unshare must complete before unlocked refresh resumes");
}
Err(_) => {
unshare_go_tx
.send(())
.expect("unshare thread must receive release signal");
}
}
}));
}
// Race thread: waits at the barrier (released by the observer WHILE the
// lock is held), then acquires the same lock and retains an unshared T+1.
// With the lock: blocked until refresh retains T+1, then it retains its
// own unshared T+1 — final = UNSHARED ✓.
// Without the lock (mutation): races freely, retains unshared first, then
// refresh retains shared T+1 over it — final = SHARED ✗.
let state_race = state.clone();
let db_path_race = db_path.clone();
let keys_race = keys.clone();
let team_race = team.clone();
let member_race = AgentDefinition {
id: persona_id.to_string(),
display_name: "Catalog Reviewer".to_string(),
description: None,
avatar_url: None,
system_prompt: "Review.".to_string(),
runtime: None,
model: None,
provider: None,
name_pool: Vec::new(),
is_builtin: false,
is_active: true,
shared: false,
source_team: None,
source_team_persona_slug: None,
catalog_source: None,
team_catalog_source: None,
env_vars: BTreeMap::new(),
respond_to: None,
respond_to_allowlist: Vec::new(),
parallelism: None,
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
};
let race_thread = std::thread::spawn(move || {
barrier.wait(); // released by observer while lock is held
// Acquire the same store lock — this blocks until publish_and_refresh_teams_at
// releases it, ensuring the race write comes AFTER refresh's retain.
let _guard = state_race
let state_unshare = state.clone();
let db_path_unshare = db_path.clone();
let keys_unshare = keys.clone();
let team_unshare = team.clone();
let member_unshare = persona_member.clone();
let unshare_thread = std::thread::spawn(move || {
refresh_read_rx
.recv()
.expect("refresh must announce its shared-T read");
unshare_ready_tx
.send(())
.expect("observer must receive readiness signal");
unshare_go_rx
.recv()
.expect("observer must release the unshare lock attempt");
let _guard = state_unshare
.managed_agents_store_lock
.lock()
.expect("race lock must not be poisoned");
// Retain an unshared team head T+1 — the write that must win.
.expect("unshare lock must not be poisoned");
prepare_team_publication_at(
&db_path_race,
&keys_race,
&team_race,
&[member_race],
Some(false), // unshare
&db_path_unshare,
&keys_unshare,
&team_unshare,
&[member_unshare],
Some(false),
)
.expect("race unshare must succeed");
.expect("unshare must retain its unshared head");
// Under correct locking the observer has already returned and
// drops this receiver; under the lock-moved mutation it is waiting
// for this completion signal. Both outcomes are intentional.
let _ = unshare_complete_tx.send(());
});
// Run publish_and_refresh_teams_at using a prepared shared persona head.
let prepared = prepared(&db_path, relay_url, keys.clone(), Some(true));
let persona_id_str = prepared.persona.id.clone();
let result = publish_and_refresh_teams_at(
@@ -965,26 +964,19 @@ mod tests {
)
.await;
// Clear observer before any assertion that could panic.
{
let mut slot = REFRESH_READ_OBSERVER.lock().expect("observer slot");
*slot = None;
}
// Wait for the race thread to complete its unshare write.
race_thread.join().expect("race thread must not panic");
unshare_thread
.join()
.expect("unshare thread must not panic");
result.expect("publish_and_refresh_teams_at must succeed");
// The race thread always runs AFTER refresh (because the lock serializes
// them) and retains an unshared head at T+1. Since retain_event accepts
// equal timestamps, the last writer wins — and the race thread is last.
// Mutation (move refresh outside lock): race thread writes first, refresh
// writes over it → SHARED → test turns RED.
let conn = open_retention_db(&db_path).unwrap();
let retained = get_retained_event(&conn, KIND_TEAM_CATALOG, &owner, "team-abc")
.expect("db query must not fail")
.expect("retained team catalog head must exist after refresh + race");
.expect("retained team catalog head must exist after refresh + unshare");
assert!(
retained.created_at >= initial_created_at,
"retained head must be at or after the seed timestamp"
@@ -993,10 +985,8 @@ mod tests {
nostr::Event::from_json(&retained.raw_event).expect("must parse retained event");
assert!(
!event_is_shared(&event),
"final retained kind:30178 head must be UNSHARED — \
race thread ran last (lock held during refresh) and its write must not \
have been overwritten by the refreshed shared head; \
moving refresh outside the lock causes the shared head to win → RED"
"final retained kind:30178 head must be UNSHARED; moving refresh outside \
managed_agents_store_lock lets the paused refresh overwrite the completed unshare"
);
}
}

70
desktop/src/features/agents/ui/agentSaveCoordinator.test.mjs
View File

@@ -1,4 +1,5 @@
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import test from "node:test";
import { toast } from "sonner";
@@ -32,6 +33,16 @@ function captureToasts() {
// ── Shared fixtures ────────────────────────────────────────────────────────────
const partialPublishOutcomeContract = JSON.parse(
readFileSync(
new URL(
"../../../../../test-fixtures/update-persona-publish-partial-outcome.json",
import.meta.url,
),
"utf8",
),
);
function makeDefinition(overrides = {}) {
return {
id: "def-1",
@@ -1712,31 +1723,68 @@ test("test_publish_throws_post_persist_returns_false_not_partial_failure_toast",
// tests cover the retry layer on top.
test("test_publish_retry_succeeds_settles_as_full_success", async () => {
// Initial updatePersonaAndPublish throws (retain step failed), but the
// persona IS observed as persisted. publishRetry succeeds → coordinator
// should call onDone and return true.
// Consume the same observed partial-outcome contract as the Rust command
// test: the rename persisted, the command returned the strict preparation
// error, and recovery is therefore a publish-only retry.
const cap = captureToasts();
try {
const persisted = makeDefinition({ displayName: "Alice" });
const contract = partialPublishOutcomeContract;
const persisted = makeDefinition({
id: contract.personaId,
displayName: contract.afterDisplayName,
});
let retryCalls = 0;
let announceRetryStarted;
const retryStarted = new Promise((resolve) => {
announceRetryStarted = resolve;
});
let allowRetryCompletion;
const retryCompletionGate = new Promise((resolve) => {
allowRetryCompletion = resolve;
});
const opts = makeOpts({
ctx: {
kind: "definition-only",
definition: makeDefinition({ displayName: "Alice" }),
definition: makeDefinition({
id: contract.personaId,
displayName: contract.beforeDisplayName,
}),
},
personaInput: makePersonaInput({ displayName: "Alice" }),
personaInput: makePersonaInput({
id: contract.personaId,
displayName: contract.afterDisplayName,
}),
publishCatalogUpdates: true,
updatePersonaAndPublish: async () => {
throw new Error("failed to open retention db");
throw new Error(contract.commandErrorContains);
},
refetchStores: async () => ({ persona: persisted, agent: null }),
publishRetry: async () => {
publishRetry: async (personaId) => {
assert.equal(
personaId,
contract.personaId,
"retry must target the definition from the observed command outcome",
);
retryCalls++;
return { persona: persisted, publicationStatus: "published" };
announceRetryStarted();
await retryCompletionGate;
return {
persona: persisted,
publicationStatus: contract.retryPublicationStatus,
};
},
});
const result = await runAgentSaveCoordinator(opts);
const coordinatorResult = runAgentSaveCoordinator(opts);
await retryStarted;
assert.equal(
opts._calls.onDone,
0,
"dialog must remain open while the publish retry is still in flight",
);
allowRetryCompletion();
const result = await coordinatorResult;
assert.equal(
result,
@@ -1746,7 +1794,7 @@ test("test_publish_retry_succeeds_settles_as_full_success", async () => {
assert.equal(
opts._calls.onDone,
1,
"onDone must be called on retry success",
"dialog closes exactly once after retry completion",
);
assert.equal(retryCalls, 1, "publishRetry must be called exactly once");
const warnings = cap.captured.filter((c) => c.kind === "warning");

8
test-fixtures/update-persona-publish-partial-outcome.json Normal file
View File

@@ -0,0 +1,8 @@
{
"personaId": "p1",
"beforeDisplayName": "Alice",
"afterDisplayName": "Alice Renamed",
"commandErrorContains": "failed to open retention db",
"relayProfileRequestCount": 1,
"retryPublicationStatus": "published"
}