From a3f95e7205f1390e85bbfee10d011229fb241c43 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 08:15:46 -0500 Subject: [PATCH] feat(loadgen): model agent turns and sustained holds Agents in an agent harness are idle between prompts and work in bursts; the loadgen agent polled HTTP /query on a fixed interval instead. Add an [agent.turn] model of what the harness sends around a prompt: context reads, two reactions, typing and observer frames while the turn runs, tool reads, a threaded reply and the reactions' removal. Agents can also open one live mentions REQ per channel and send presence heartbeats, and polling can be switched off. Ramp steps may now repeat a level, so a long sustained hold is reported and judged in windows. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-test-client/src/loadgen/run.rs | 577 +++++++++++++++--- .../buzz-test-client/src/loadgen/scenario.rs | 96 ++- crates/buzz-test-client/src/loadgen/stats.rs | 6 +- perf/LOADGEN.md | 33 +- perf/scenarios/local-agent-turns.toml | 50 ++ 5 files changed, 677 insertions(+), 85 deletions(-) create mode 100644 perf/scenarios/local-agent-turns.toml diff --git a/crates/buzz-test-client/src/loadgen/run.rs b/crates/buzz-test-client/src/loadgen/run.rs index d1e68ebe58..d6edde7e72 100644 --- a/crates/buzz-test-client/src/loadgen/run.rs +++ b/crates/buzz-test-client/src/loadgen/run.rs @@ -5,21 +5,26 @@ use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use buzz_core::kind::{ - KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, - KIND_PRESENCE_UPDATE, KIND_REACTION, KIND_STREAM_MESSAGE, KIND_TYPING_INDICATOR, + KIND_AGENT_OBSERVER_FRAME, KIND_DELETION, KIND_GIT_REPO_ANNOUNCEMENT, KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, KIND_NIP29_GROUP_METADATA, KIND_PRESENCE_UPDATE, + KIND_PROFILE, KIND_PROJECT, KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_EDIT, + KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TYPING_INDICATOR, + KIND_WORKFLOW_APPROVAL_REQUESTED, }; use buzz_core::observer::{ encrypt_observer_payload, OBSERVER_AGENT_TAG, OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY, }; use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; -use nostr::{Event, EventBuilder, Filter, Keys, Kind, PublicKey, Tag, Timestamp}; +use nostr::{ + Alphabet, Event, EventBuilder, Filter, Keys, Kind, PublicKey, SingleLetterTag, Tag, Timestamp, +}; use serde_json::{json, Value}; use tokio::sync::{watch, Mutex}; use super::desktop; use super::identity::{self, Role}; use super::net::{http_reason, Net}; -use super::scenario::{Community, HumanProfile, Members, Scenario, WideSplit}; +use super::scenario::{AgentTurn, Community, HumanProfile, Members, Scenario, WideSplit}; use super::seed::{find_roots, hot_root_keys, reply_tags, HOT_ROOT_PREFIX, ROOT_PREFIX}; use super::stats::{abort_reason, ops_json, Op, Stats}; @@ -1035,8 +1040,76 @@ fn subscription_of(message: &RelayMessage) -> Option<&str> { } } -/// An agent client: NIP-OA authenticated socket, HTTP `/query` polling, -/// HTTP `/events` posts and encrypted observer frames to its owner. +/// Kinds an agent harness subscribes to for its mentions in each channel. +const AGENT_MENTION_KINDS: [u32; 4] = [ + KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_EDIT, + KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_STREAM_REMINDER, +]; +/// Kinds a turn reads as channel context and tool reads. +const AGENT_READ_KINDS: [u32; 2] = [KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2]; +/// Reactions a turn adds to the event it answers and removes at its end: +/// "seen" and "working". +const TURN_REACTIONS: [&str; 2] = ["👀", "💬"]; + +/// An agent's turn under way: the channel it works in, the event it +/// answers, and when its next typing indicator, observer frame and tool +/// reads are due. +struct Turn { + channel: String, + target: Option, + ends: Instant, + next_typing: Option, + next_frame: Option, + /// Due times of the tool reads still to make, latest first. + reads: Vec, +} + +/// What a turn does next. +enum TurnStep { + Typing(String), + Frame, + Read(String), + End, +} + +impl Turn { + fn due(&self) -> Option { + [self.next_typing, self.next_frame, self.reads.last().copied()] + .into_iter() + .flatten() + .chain([self.ends]) + .min() + } + + /// The step due at `now`, if any, with its next one scheduled. Typing, + /// frames and reads run before the end, so none is lost to a late wakeup. + fn step(&mut self, now: Instant, profile: &AgentTurn) -> Option { + let due = |t: Option| t.is_some_and(|t| t <= now); + if due(self.next_typing) { + self.next_typing = every(profile.typing_secs, now).filter(|t| *t < self.ends); + Some(TurnStep::Typing(self.channel.clone())) + } else if due(self.next_frame) { + self.next_frame = every(profile.frame_secs, now).filter(|t| *t < self.ends); + Some(TurnStep::Frame) + } else if due(self.reads.last().copied()) { + self.reads.pop(); + Some(TurnStep::Read(self.channel.clone())) + } else { + (self.ends <= now).then_some(TurnStep::End) + } + } +} + +/// `Some(now + secs)`, or `None` for an interval of 0 (off). +fn every(secs: f64, from: Instant) -> Option { + (secs > 0.0).then(|| from + Duration::from_secs_f64(secs)) +} + +/// An agent client: NIP-OA authenticated socket with its mentions, then +/// either HTTP polling, posts and observer frames on Poisson schedules, or +/// turns of work (see [`AgentTurn`]), or both. async fn agent(shared: Arc, global: usize) { let scenario = &shared.scenario; let profile = &scenario.agent; @@ -1053,6 +1126,18 @@ async fn agent(shared: Arc, global: usize) { let mut rng = Rng::new(u64::MAX - global as u64); let mut backoff = Backoff::new(); let mut last_message = None; + let query_per_hour = if profile.query_interval_secs > 0.0 { + 3600.0 / profile.query_interval_secs + } else { + 0.0 + }; + let mut next_query = rng.after(query_per_hour); + let mut next_post = rng.after(profile.events_per_hour); + let mut next_frame = rng.after(profile.observer_frames_per_hour); + let mut next_turn = rng.after(profile.turn.per_hour); + // A turn outlives a dropped socket, as a harness keeps working while it + // reconnects. + let mut turn: Option = None; while let Some(mut conn) = connect( &shared, @@ -1066,31 +1151,49 @@ async fn agent(shared: Arc, global: usize) { { let since = Timestamp::now(); let subscribed = Instant::now(); - let mut ok = match send_req( - &mut conn, - "mentions", + let mut reqs = vec![( + "mentions".to_string(), mentions_filter(keys.public_key(), since), - ) - .await - { - Ok(()) => true, - Err(e) => { + )]; + if profile.channel_subscriptions { + reqs.extend(channels.iter().enumerate().map(|(i, channel)| { + let filter = Filter::new() + .kinds(AGENT_MENTION_KINDS.map(kind)) + .custom_tags(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]) + .pubkey(keys.public_key()) + .since(since); + (format!("ch-{i}"), filter) + })); + } + let mut ok = true; + for (id, filter) in reqs { + if let Err(e) = send_req(&mut conn, &id, filter).await { shared .stats .err(Op::WsConnect, &format!("disconnected: {e}")); - false + ok = false; + break; } - }; - let query_per_hour = 3600.0 / profile.query_interval_secs; - let mut next_query = rng.after(query_per_hour); - let mut next_post = rng.after(profile.events_per_hour); - let mut next_frame = rng.after(profile.observer_frames_per_hour); + } + // Presence starts at a random point of its interval, so a fleet that + // connects together does not heartbeat together. + let mut next_presence = every( + profile.presence_interval_secs * rng.unit(), + Instant::now(), + ) + .or_else(|| every(profile.presence_interval_secs, Instant::now())); while ok && !stopped(&shared) { - let due = [next_query, next_post, next_frame] - .into_iter() - .flatten() - .min(); + let due = [ + next_query, + next_post, + next_frame, + next_presence, + turn.as_ref().map_or(next_turn, Turn::due), + ] + .into_iter() + .flatten() + .min(); let wait = due .map(|d| d.saturating_duration_since(Instant::now())) .unwrap_or(MAX_IDLE) @@ -1112,9 +1215,49 @@ async fn agent(shared: Arc, global: usize) { } } let now = Instant::now(); - let channel = &channels[rng.below(channels.len())]; - if next_query.is_some_and(|t| t <= now) { + // One action per wakeup; anything else due runs on the next pass. + let sent = if next_presence.is_some_and(|t| t <= now) { + next_presence = every(profile.presence_interval_secs, now); + let event = + EventBuilder::new(kind(KIND_PRESENCE_UPDATE), "online").sign_with_keys(&keys); + send(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + } else if let Some(step) = turn.as_mut().and_then(|t| t.step(now, &profile.turn)) { + match step { + TurnStep::Typing(channel) => { + let event = EventBuilder::new(kind(KIND_TYPING_INDICATOR), "") + .tags([tag(["h", &channel])]) + .sign_with_keys(&keys); + send(&shared.stats, &mut conn, Op::PublishEphemeral, event).await + } + TurnStep::Frame => { + let frame = observer_frame(&keys, &owner, profile.turn.frame_bytes); + publish_frame(&shared.stats, &mut conn, frame).await + } + TurnStep::Read(channel) => { + let filters = json!([{ + "kinds": AGENT_READ_KINDS, + "#h": [channel], + "limit": 50, + }]); + query(&shared, Op::HttpQuery, &keys, community, &filters, &auth).await; + Ok(true) + } + TurnStep::End => { + if let Some(t) = turn.take() { + end_turn(&shared, community, &keys, &auth, t).await; + } + next_turn = rng.after(profile.turn.per_hour); + Ok(true) + } + } + } else if turn.is_none() && next_turn.is_some_and(|t| t <= now) { + next_turn = None; + let channel = channels[rng.below(channels.len())].clone(); + turn = Some(start_turn(&shared, community, &keys, &auth, channel, &mut rng).await); + Ok(true) + } else if next_query.is_some_and(|t| t <= now) { next_query = rng.after(query_per_hour); + let channel = &channels[rng.below(channels.len())]; let window = Timestamp::now().as_secs().saturating_sub(3600); let filters = json!([ { "kinds": [KIND_STREAM_MESSAGE], "#h": [channel], "since": window, "limit": 20 }, @@ -1130,60 +1273,29 @@ async fn agent(shared: Arc, global: usize) { Some(&auth), ) .await; + Ok(true) } else if next_post.is_some_and(|t| t <= now) { next_post = rng.after(profile.events_per_hour); + let channel = &channels[rng.below(channels.len())]; let event = EventBuilder::new( kind(KIND_STREAM_MESSAGE), format!("{TIMED_PREFIX}{} {me}", now_micros()), ) - .tags([tag(["h", channel])]) - .sign_with_keys(&keys); - match event { - Ok(event) => { - let pending = shared.stats.begin(Op::HttpEvents); - match shared - .net - .submit(&keys, community, &event, Some(&auth)) - .await - { - Ok(()) => pending.ok(), - Err(e) => pending.err(&e.to_string()), - } - } - Err(e) => shared.stats.err(Op::HttpEvents, &format!("sign: {e}")), - } + .tags([tag(["h", channel])]); + submit(&shared, community, &keys, &auth, event, Op::HttpEvents).await; + Ok(true) } else if next_frame.is_some_and(|t| t <= now) { next_frame = rng.after(profile.observer_frames_per_hour); - let payload = json!({ "type": "loadgen", "sent_micros": now_micros().to_string() }); - let event = encrypt_observer_payload(&keys, &owner.public_key(), &payload) - .map_err(|e| e.to_string()) - .and_then(|content| { - EventBuilder::new(kind(KIND_AGENT_OBSERVER_FRAME), content) - .tags([ - tag(["p", &owner.public_key().to_hex()]), - tag([OBSERVER_AGENT_TAG, &me]), - tag([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]), - ]) - .sign_with_keys(&keys) - .map_err(|e| e.to_string()) - }); - let result = match event { - Ok(event) => { - publish(&shared.stats, &mut conn, Op::PublishEphemeral, event).await - } - Err(e) => { - shared - .stats - .err(Op::PublishEphemeral, &format!("observer: {e}")); - Ok(false) - } - }; - if let Err(e) = result { - shared - .stats - .err(Op::WsConnect, &format!("disconnected: {e}")); - ok = false; - } + let frame = observer_frame(&keys, &owner, 0); + publish_frame(&shared.stats, &mut conn, frame).await + } else { + Ok(true) + }; + if let Err(e) = sent { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; } } let _ = conn.disconnect().await; @@ -1198,6 +1310,167 @@ async fn agent(shared: Arc, global: usize) { } } +/// Starts a turn in `channel` the way an agent harness starts a prompt: +/// HTTP `/query` reads of the channel's newest messages, its metadata, its +/// project home (two reads) and the newest author's profile, then two +/// reactions on the newest message. The reads are `turn_context`. +async fn start_turn( + shared: &Shared, + community: &Community, + keys: &Keys, + auth: &Tag, + channel: String, + rng: &mut Rng, +) -> Turn { + let profile = &shared.scenario.agent.turn; + let context = json!([{ "kinds": AGENT_READ_KINDS, "#h": [channel], "limit": 13 }]); + let newest: Option = query(shared, Op::TurnContext, keys, community, &context, auth) + .await + .and_then(|body| serde_json::from_str::>(&body).ok()) + .and_then(|events| events.into_iter().max_by_key(|e| e.created_at)); + let author = newest.as_ref().map(|e| e.pubkey.to_hex()); + let reads = [ + json!([{ "kinds": [KIND_NIP29_GROUP_METADATA], "#d": [channel], "limit": 1 }]), + json!([{ "kinds": [KIND_PROJECT], "#buzz-channel": [channel] }]), + json!([{ "kinds": [KIND_GIT_REPO_ANNOUNCEMENT], "#buzz-channel": [channel] }]), + json!([{ "kinds": [KIND_PROFILE], "authors": [author.unwrap_or_else(|| keys.public_key().to_hex())] }]), + ]; + // Each request checks the stop, so a turn starting at the stop sends + // nothing more than the request under way. + for filters in reads { + if stopped(shared) { + break; + } + query(shared, Op::TurnContext, keys, community, &filters, auth).await; + } + if let Some(target) = newest.as_ref().filter(|_| !stopped(shared)) { + for emoji in TURN_REACTIONS { + let event = EventBuilder::new(kind(KIND_REACTION), emoji) + .tags([tag(["h", &channel]), tag(["e", &target.id.to_hex()])]); + submit(shared, community, keys, auth, event, Op::HttpEvents).await; + } + } + let now = Instant::now(); + let length = Duration::from_secs_f64(profile.mean_secs * (0.5 + rng.unit())); + let mut reads: Vec = (0..profile.reads) + .map(|_| now + length.mul_f64(rng.unit())) + .collect(); + reads.sort_unstable_by(|a, b| b.cmp(a)); + Turn { + channel, + target: newest, + ends: now + length, + next_typing: every(profile.typing_secs, now).map(|_| now), + next_frame: every(profile.frame_secs, now).map(|_| now), + reads, + } +} + +/// Ends a turn: a timed reply in the answered message's thread (or the +/// channel), then each reaction's removal, which reads the agent's +/// reactions on the target and deletes them, as the harness does. +async fn end_turn(shared: &Shared, community: &Community, keys: &Keys, auth: &Tag, turn: Turn) { + let me = keys.public_key().to_hex(); + let mut tags = vec![tag(["h", &turn.channel])]; + if let Some(target) = &turn.target { + let id = target.id.to_hex(); + let root = target + .tags + .iter() + .find(|t| t.as_slice().get(3).is_some_and(|m| m == "root")) + .and_then(|t| t.content()) + .map_or_else(|| id.clone(), str::to_string); + tags.extend(reply_tags(&root, &id)); + } + let reply = EventBuilder::new( + kind(KIND_STREAM_MESSAGE), + format!("{TIMED_PREFIX}{} {me}", now_micros()), + ) + .tags(tags); + submit(shared, community, keys, auth, reply, Op::HttpEvents).await; + let Some(target) = turn.target.filter(|_| !stopped(shared)) else { + return; + }; + let mine = json!([{ + "kinds": [KIND_REACTION], + "authors": [me], + "#e": [target.id.to_hex()], + }]); + for _ in TURN_REACTIONS { + if stopped(shared) { + return; + } + let found = query(shared, Op::HttpQuery, keys, community, &mine, auth) + .await + .and_then(|body| serde_json::from_str::>(&body).ok()) + .unwrap_or_default(); + // The first pass deletes every reaction found, so the second only + // reads, as the harness's second removal finds nothing left. + for reaction in found { + let event = EventBuilder::new(kind(KIND_DELETION), "") + .tags([tag(["e", &reaction.id.to_hex()])]); + submit(shared, community, keys, auth, event, Op::HttpEvents).await; + } + } +} + +/// An encrypted kind:24200 telemetry frame to `owner`, its plaintext padded +/// by `pad` bytes. +fn observer_frame(keys: &Keys, owner: &Keys, pad: usize) -> Result { + let payload = json!({ + "type": "loadgen", + "sent_micros": now_micros().to_string(), + "pad": "x".repeat(pad), + }); + let content = + encrypt_observer_payload(keys, &owner.public_key(), &payload).map_err(|e| e.to_string())?; + EventBuilder::new(kind(KIND_AGENT_OBSERVER_FRAME), content) + .tags([ + tag(["p", &owner.public_key().to_hex()]), + tag([OBSERVER_AGENT_TAG, &keys.public_key().to_hex()]), + tag([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]), + ]) + .sign_with_keys(keys) + .map_err(|e| e.to_string()) +} + +async fn publish_frame( + stats: &Stats, + conn: &mut NostrWsConnection, + frame: Result, +) -> Result { + match frame { + Ok(event) => publish(stats, conn, Op::PublishEphemeral, event).await, + Err(e) => { + stats.err(Op::PublishEphemeral, &format!("observer: {e}")); + Ok(false) + } + } +} + +/// Signs and submits `event` through HTTP `/events` as `op`. A reaction the +/// relay already holds counts as answered: an earlier turn cut short by the +/// stop left it, and the next removal deletes it. +async fn submit( + shared: &Shared, + community: &Community, + keys: &Keys, + auth: &Tag, + event: EventBuilder, + op: Op, +) { + let event = match event.sign_with_keys(keys) { + Ok(event) => event, + Err(e) => return shared.stats.err(op, &format!("sign: {e}")), + }; + let pending = shared.stats.begin(op); + match shared.net.submit(keys, community, &event, Some(auth)).await { + Ok(()) => pending.ok(), + Err(e) if e.to_string().contains("reaction already exists") => pending.ok(), + Err(e) => pending.err(&e.to_string()), + } +} + /// A hot-thread author: replies to the hot root through HTTP `/events` and /// reads the thread's newest replies through HTTP `/query`. Replies carry /// their send time, so humans in the hot channel record their delivery. @@ -1303,7 +1576,7 @@ async fn http( path: &str, body: &Value, auth: Option<&Tag>, -) { +) -> Option { let pending = shared.stats.begin(op); let body = serde_json::to_vec(body).expect("serialize JSON body"); match shared @@ -1311,12 +1584,33 @@ async fn http( .post(keys, &community.http_url, path, body, auth) .await { - Ok((status, _)) if status.is_success() => pending.ok(), - Ok((status, text)) => pending.err(&http_reason(status, &text)), - Err(e) => pending.err(&e.to_string()), + Ok((status, text)) if status.is_success() => { + pending.ok(); + Some(text) + } + Ok((status, text)) => { + pending.err(&http_reason(status, &text)); + None + } + Err(e) => { + pending.err(&e.to_string()); + None + } } } +/// An agent's HTTP `/query` as `op`; the response body when it succeeded. +async fn query( + shared: &Shared, + op: Op, + keys: &Keys, + community: &Community, + filters: &Value, + auth: &Tag, +) -> Option { + http(shared, op, keys, community, "/query", filters, Some(auth)).await +} + #[cfg(test)] mod tests { use super::*; @@ -1671,6 +1965,9 @@ hold_secs = 1 json!(["EOSE", frame[1]]) } Some("EVENT") => { + if let Some(k) = frame[1]["kind"].as_u64() { + seen.kinds.lock().unwrap().push(k); + } let tags = frame[1]["tags"].as_array().cloned().unwrap_or_default(); if let Some(h) = tags.iter().find(|t| t[0] == "h") { let h = h[1].as_str().unwrap_or_default().to_string(); @@ -1710,6 +2007,8 @@ hold_secs = 1 closed: AtomicUsize, /// The `h` tag of every EVENT that had one. channels: std::sync::Mutex>, + /// The kind of every EVENT. + kinds: std::sync::Mutex>, } impl Seen { @@ -2098,6 +2397,132 @@ hold_secs = 2 } /// Holds `/events` responses until opened, counting the requests held. + /// One agent running turns over a fake socket and a fake HTTP relay + /// that holds one channel message and the agent's live reactions. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn agent_turns_read_react_stream_reply_and_remove_their_reactions() { + let (ws, seen) = fake_ws_relay(FakeRelay::PROMPT).await; + let channel = identity::channel_id("t", 0, 0); + let author = Keys::generate(); + let target = EventBuilder::new(kind(KIND_STREAM_MESSAGE), "question") + .tags([tag(["h", &channel])]) + .sign_with_keys(&author) + .unwrap(); + let posted: Log = Default::default(); + let queries: Log = Default::default(); + let (post_log, query_log, held) = (posted.clone(), queries.clone(), target.clone()); + let http = super::super::seed::tests::fake_http(move |_, path, body| { + if path == "/events" { + let event: Event = serde_json::from_slice(body).unwrap(); + post_log.lock().unwrap().push(event); + return (200, json!({"accepted": true, "message": ""}).to_string()); + } + let filters: Value = serde_json::from_slice(body).unwrap(); + query_log.lock().unwrap().push(filters.clone()); + let found = match filters[0]["kinds"][0].as_u64() { + Some(k) if k == u64::from(KIND_STREAM_MESSAGE) => json!([held]), + // Reactions not deleted since they were posted. Two turns in + // one second post the same reaction id again. + Some(k) if k == u64::from(KIND_REACTION) => { + let mut live: Vec = Vec::new(); + for e in post_log.lock().unwrap().iter() { + if e.kind == kind(KIND_REACTION) { + live.push(e.clone()); + } else if e.kind == kind(KIND_DELETION) { + let gone: Vec<&str> = e.tags.iter().filter_map(|t| t.content()).collect(); + live.retain(|r| !gone.contains(&r.id.to_hex().as_str())); + } + } + json!(live) + } + _ => json!([]), + }; + (200, found.to_string()) + }) + .await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{ws}" +http_url = "http://{http}" +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 0 +agents = 1 +channels_per_member = 1 +[agent] +query_interval_secs = 0.0 +events_per_hour = 0.0 +observer_frames_per_hour = 0.0 +presence_interval_secs = 1.0 +channel_subscriptions = true +[agent.turn] +per_hour = 36000.0 +mean_secs = 0.6 +frame_secs = 0.1 +typing_secs = 0.1 +reads = 2 +[[ramp]] +users = 1.0 +hold_secs = 3 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let shard = Shard::parse("0/1").unwrap(); + let report = run_with(scenario, net, shard, Duration::from_secs(5)) + .await + .expect("agent turn run passes"); + let ops = &report["steps"][0]["ops"]; + assert!(ops["turn_context"]["count"].as_u64() > Some(0), "{report}"); + + // Mentions plus one channel subscription on the one session. + assert_eq!(seen.reqs_after(1).await, 2); + let kinds = seen.kinds.lock().unwrap().clone(); + for k in [KIND_TYPING_INDICATOR, KIND_AGENT_OBSERVER_FRAME, KIND_PRESENCE_UPDATE] { + assert!(kinds.contains(&u64::from(k)), "no kind {k} over the socket: {kinds:?}"); + } + + let posted = posted.lock().unwrap().clone(); + let of = |k: u32| posted.iter().filter(move |e| e.kind == kind(k)); + let target_id = target.id.to_hex(); + let reactions: Vec<&Event> = of(KIND_REACTION).collect(); + assert!(reactions.len() >= 4, "two reactions a turn: {posted:?}"); + for reaction in &reactions { + let e = reaction.tags.iter().find(|t| t.kind().as_str() == "e"); + assert_eq!(e.and_then(|t| t.content()), Some(target_id.as_str())); + } + // Every finished turn replied in the target's thread and deleted + // both of its reactions; only a turn cut by the stop leaves any. + let replies: Vec<&Event> = of(KIND_STREAM_MESSAGE).collect(); + assert!(!replies.is_empty(), "{posted:?}"); + for reply in &replies { + let markers = buzz_core::nip10::parse_thread_markers(&reply.tags); + assert_eq!(markers.root.as_deref(), Some(target_id.as_str())); + assert_eq!(markers.reply.as_deref(), Some(target_id.as_str())); + } + let deleted: Vec = of(KIND_DELETION) + .flat_map(|e| e.tags.iter().filter_map(|t| t.content()).map(str::to_string)) + .collect(); + assert_eq!(deleted.len(), 2 * replies.len(), "{posted:?}"); + let ids: Vec = reactions.iter().map(|e| e.id.to_hex()).collect(); + assert!(deleted.iter().all(|id| ids.contains(id)), "{deleted:?}"); + + // Each turn reads its context: metadata, project home and profile. + let queries = queries.lock().unwrap().clone(); + let has = |k: u32| queries.iter().any(|q| q[0]["kinds"][0] == json!(k)); + for k in [KIND_NIP29_GROUP_METADATA, KIND_PROJECT, KIND_GIT_REPO_ANNOUNCEMENT] { + assert!(has(k), "no kind {k} read: {queries:?}"); + } + let profile = json!([{ "kinds": [KIND_PROFILE], "authors": [author.public_key().to_hex()] }]); + assert!(queries.contains(&profile), "{queries:?}"); + // Tool reads fetch the channel's newest 50 messages. + assert!(queries.iter().any(|q| q[0]["limit"] == json!(50)), "{queries:?}"); + } + #[derive(Default)] struct Gate { arrived: AtomicUsize, diff --git a/crates/buzz-test-client/src/loadgen/scenario.rs b/crates/buzz-test-client/src/loadgen/scenario.rs index aabfb899e5..291d578386 100644 --- a/crates/buzz-test-client/src/loadgen/scenario.rs +++ b/crates/buzz-test-client/src/loadgen/scenario.rs @@ -184,12 +184,19 @@ impl Default for HumanProfile { #[derive(Clone, Debug, Deserialize)] #[serde(default, deny_unknown_fields)] pub struct AgentProfile { - /// Mean interval between HTTP `/query` polls. + /// Mean interval between HTTP `/query` polls (0 disables polling). pub query_interval_secs: f64, /// Messages posted through HTTP `/events` per hour. pub events_per_hour: f64, - /// Encrypted kind:24200 observer frames sent to the owner per hour. + /// Encrypted kind:24200 observer frames sent to the owner per hour, + /// outside turns. pub observer_frames_per_hour: f64, + /// Seconds between presence heartbeats (0 sends none). + pub presence_interval_secs: f64, + /// Open one live REQ per channel for the agent's mentions there, as the + /// agent harness does, instead of relying on HTTP polling. + pub channel_subscriptions: bool, + pub turn: AgentTurn, } impl Default for AgentProfile { @@ -198,10 +205,52 @@ impl Default for AgentProfile { query_interval_secs: 5.0, events_per_hour: 60.0, observer_frames_per_hour: 60.0, + presence_interval_secs: 0.0, + channel_subscriptions: false, + turn: AgentTurn::default(), } } } +/// Agent turns: bursts of work in one channel, as an agent harness runs a +/// prompt. A turn starts with context reads and two reactions, streams +/// typing and observer frames while it runs, makes `reads` tool reads, and +/// ends with a reply and the reactions' removal. Disabled at 0 `per_hour`. +#[derive(Clone, Debug, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct AgentTurn { + /// Turns started per hour while the agent is idle. An agent runs one + /// turn at a time. + pub per_hour: f64, + /// Mean turn length. Each turn lasts 0.5–1.5 times this. + pub mean_secs: f64, + /// Seconds between observer frames during a turn (0 sends none). + pub frame_secs: f64, + /// Plaintext size of each observer frame, before encryption. + pub frame_bytes: usize, + /// Seconds between typing indicators during a turn (0 sends none). + pub typing_secs: f64, + /// Channel reads over HTTP `/query` during a turn, as tool calls. + pub reads: usize, +} + +impl Default for AgentTurn { + fn default() -> Self { + Self { + per_hour: 0.0, + mean_secs: 60.0, + frame_secs: 1.0, + frame_bytes: 1024, + typing_secs: 3.0, + reads: 2, + } + } +} + +/// Largest observer frame plaintext: NIP-44 encrypts at most 65,535 bytes, +/// and the frame's JSON envelope needs some of them. +pub const MAX_FRAME_BYTES: usize = 60_000; + /// One thread in community 0, channel 0 that keeps growing: seeded with /// many direct replies, then replied to and read throughout the run. /// Disabled while `seed_replies` and `authors` are both 0. @@ -416,11 +465,13 @@ impl Scenario { if self.ramp.is_empty() { return Err("ramp needs at least one step".into()); } - // Clients are only ever added, so a step can never shrink the population. + // Clients are only ever added, so a step can never shrink the + // population. Repeating a level splits a long hold into reported, + // judged windows. let mut previous = 0.0; for step in &self.ramp { - if !step.users.is_finite() || step.users <= previous { - return Err("ramp users must be finite and increase every step".into()); + if !step.users.is_finite() || step.users <= 0.0 || step.users < previous { + return Err("ramp users must be finite, above 0 and never decrease".into()); } previous = step.users; } @@ -434,7 +485,7 @@ impl Scenario { "population.connect_rate_per_sec", self.population.connect_rate_per_sec, ), - ("agent.query_interval_secs", a.query_interval_secs), + ("agent.turn.mean_secs", a.turn.mean_secs), ("abort.max_p99_ms", self.abort.max_p99_ms), ("desktop.backstop_secs", self.desktop.backstop_secs), ]; @@ -455,6 +506,20 @@ impl Scenario { if !d.renewal_secs.is_finite() || !(d.renewal_secs == 0.0 || d.renewal_secs >= 1.0) { return Err("desktop.renewal_secs must be 0 or a finite number >= 1".into()); } + // Agent intervals are 0 (off) or slow enough to schedule. + let intervals = [ + ("agent.presence_interval_secs", a.presence_interval_secs, 1.0), + ("agent.turn.frame_secs", a.turn.frame_secs, 0.1), + ("agent.turn.typing_secs", a.turn.typing_secs, 0.1), + ]; + for (name, value, min) in intervals { + if !value.is_finite() || !(value == 0.0 || value >= min) { + return Err(format!("{name} must be 0 or a finite number >= {min}")); + } + } + if a.turn.frame_bytes > MAX_FRAME_BYTES { + return Err(format!("agent.turn.frame_bytes must be at most {MAX_FRAME_BYTES}")); + } let non_negative = [ ("human.typing_per_hour", h.typing_per_hour), ("human.messages_per_hour", h.messages_per_hour), @@ -462,6 +527,8 @@ impl Scenario { ("human.mean_session_secs", h.mean_session_secs), ("agent.events_per_hour", a.events_per_hour), ("agent.observer_frames_per_hour", a.observer_frames_per_hour), + ("agent.query_interval_secs", a.query_interval_secs), + ("agent.turn.per_hour", a.turn.per_hour), ( "hot_thread.replies_per_hour", self.hot_thread.replies_per_hour, @@ -700,12 +767,18 @@ hold_secs = 10 "[human]\nthread_reply_fraction = 1.5\n", "[agent]\nquery_interval_secs = nan\n", "[agent]\nobserver_frames_per_hour = inf\n", + "[agent]\nquery_interval_secs = -1.0\n", + "[agent]\npresence_interval_secs = 0.5\n", + "[agent.turn]\nper_hour = -1.0\n", + "[agent.turn]\nmean_secs = 0.0\n", + "[agent.turn]\nframe_secs = 0.05\n", + "[agent.turn]\ntyping_secs = nan\n", + "[agent.turn]\nframe_bytes = 60001\n", "[abort]\nmax_error_rate = nan\n", "[abort]\nmax_p99_ms = inf\n", "[[ramp]]\nusers = inf\nhold_secs = 10\n", // A down-ramp would report a smaller population than is running. "[[ramp]]\nusers = 0.5\nhold_secs = 10\n", - "[[ramp]]\nusers = 1.0\nhold_secs = 10\n", ]; for case in cases { let text = format!("{MINIMAL}{case}"); @@ -718,6 +791,15 @@ hold_secs = 10 assert!(Scenario::parse(&text).is_err()); let up = format!("{MINIMAL}[[ramp]]\nusers = 2.0\nhold_secs = 10\n"); assert!(Scenario::parse(&up).is_ok()); + // A repeated level splits a long hold into judged windows. + let held = format!("{MINIMAL}[[ramp]]\nusers = 1.0\nhold_secs = 10\n"); + assert!(Scenario::parse(&held).is_ok()); + // Polling, presence, frames and typing can each be switched off. + let off = format!( + "{MINIMAL}[agent]\nquery_interval_secs = 0.0\npresence_interval_secs = 0.0\n\ + [agent.turn]\nper_hour = 6.0\nframe_secs = 0.0\ntyping_secs = 0.0\n" + ); + assert!(Scenario::parse(&off).is_ok()); } #[test] diff --git a/crates/buzz-test-client/src/loadgen/stats.rs b/crates/buzz-test-client/src/loadgen/stats.rs index 6da0c1e46f..1f469a437b 100644 --- a/crates/buzz-test-client/src/loadgen/stats.rs +++ b/crates/buzz-test-client/src/loadgen/stats.rs @@ -21,6 +21,8 @@ pub enum Op { Subscribe, HotReply, ThreadFetch, + /// Agent turn: an HTTP `/query` context read at the start of a turn. + TurnContext, /// Desktop client: NIP-11 `GET /` on connect. DesktopNip11, /// Desktop client: EOSE of a live subscription (global or channel batch). @@ -46,7 +48,7 @@ pub enum Op { } impl Op { - pub const ALL: [Op; 28] = [ + pub const ALL: [Op; 29] = [ Op::WsConnect, Op::History, Op::PublishEphemeral, @@ -58,6 +60,7 @@ impl Op { Op::Subscribe, Op::HotReply, Op::ThreadFetch, + Op::TurnContext, Op::DesktopNip11, Op::DesktopLive, Op::DesktopReplay, @@ -90,6 +93,7 @@ impl Op { Op::Subscribe => "subscribe_eose", Op::HotReply => "hot_reply", Op::ThreadFetch => "thread_fetch", + Op::TurnContext => "turn_context", Op::DesktopNip11 => "desktop_nip11", Op::DesktopLive => "desktop_live_eose", Op::DesktopReplay => "desktop_replay_eose", diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index 5aaf7d2bee..e9821480a7 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -77,6 +77,10 @@ Members are assigned deterministically by global index, so every shard and - **Agent**: authenticates with a NIP-OA tag from its owner, polls HTTP `/query` (channel window plus mentions), posts messages through HTTP `/events`, and sends encrypted kind:24200 observer frames to its owner. + With `agent.channel_subscriptions` it also opens one live REQ per channel + for its mentions there, and with `agent.presence_interval_secs` it sends + presence heartbeats. `agent.turn` adds turns of work, described under + [Agent turns](#agent-turns). - **Desktop human**: the `desktop.fraction` of humans that model the desktop app instead of the basic human above. They keep the basic human's presence, typing, messages and reactions; the subscriptions and HTTP reads are the @@ -97,6 +101,31 @@ results as the scenario's rates. Then a ramp step's `users` reads as a multiple of that deployment's population. Keep such scenarios with that deployment's own configuration. +## Agent turns + +An agent harness is idle between prompts and then works in bursts. Set +`agent.turn.per_hour` above 0 to model that. An idle agent starts a turn in +one of its channels on a Poisson schedule and runs one turn at a time. A +turn lasts 0.5–1.5 times `agent.turn.mean_secs` and does what the harness +does around a prompt: + +1. **Start** (`turn_context`): HTTP `/query` reads of the channel's newest + 13 messages, its kind:39000 metadata, its project home (kind:30621 and + kind:30617, one read each) and the newest message's author profile. + Then two kind:7 reactions on that message through HTTP `/events`. +2. **Work**: a typing indicator every `typing_secs` and an encrypted + observer frame of `frame_bytes` plaintext every `frame_secs` over the + socket, plus `reads` HTTP `/query` reads of the channel's newest 50 + messages at random points in the turn, as tool calls. +3. **End**: a timed reply in the message's thread through HTTP `/events`, + then each reaction's removal: a `/query` for the agent's reactions on + the message and a kind:5 deletion of each one found. + +For a harness-like agent, also set `agent.query_interval_secs = 0` (no +polling) and `events_per_hour = observer_frames_per_hour = 0`, so turns are +its only work. A turn under way when the socket drops continues after the +reconnect. + ## Targeting and membership `target.connect_addr` (`host:port`) applies only together with @@ -258,7 +287,9 @@ session. Each connect plus authentication must finish within `rate-limited: quota exceeded` text, are reported as `throttled`, not errors. Every other rejection, including `rate-limited: shared admission unavailable` (the relay's admission store is down), counts as an error. -Ramp steps must increase `users`, since clients are only ever added. +Ramp steps must never decrease `users`, since clients are only ever added. +Repeat a level to split a long hold into windows that are each reported and +judged, for example six 600-second steps at the same `users` for an hour. When the run stops, clients get 15 seconds to settle what they have in flight and disconnect. An operation already under way runs to its own deadline diff --git a/perf/scenarios/local-agent-turns.toml b/perf/scenarios/local-agent-turns.toml new file mode 100644 index 0000000000..d828bcb9c8 --- /dev/null +++ b/perf/scenarios/local-agent-turns.toml @@ -0,0 +1,50 @@ +# Agents that work in turns, as an agent harness does, against a local +# relay (`just relay`). Each human owns two agents; agents do not poll. +# cargo run -p buzz-test-client --bin buzz-loadgen -- seed perf/scenarios/local-agent-turns.toml +# cargo run -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios/local-agent-turns.toml +run_id = "local-agent-turns" + +[target] +ws_url = "ws://localhost:3000" +http_url = "http://localhost:3000" + +[shape] +communities = 1 +channels_per_community = 10 +history_messages_per_channel = 20 +thread_roots_per_channel = 1 +thread_depth = 5 + +[population] +humans = 20 +agents = 40 +channels_per_member = 3 +connect_rate_per_sec = 50.0 + +[human] +messages_per_hour = 30.0 + +[agent] +query_interval_secs = 0.0 +events_per_hour = 0.0 +observer_frames_per_hour = 0.0 +presence_interval_secs = 60.0 +channel_subscriptions = true + +# Short turns, often, so a short run finishes many of them. +[agent.turn] +per_hour = 120.0 +mean_secs = 15.0 +frame_secs = 1.0 +frame_bytes = 1024 +typing_secs = 3.0 +reads = 2 + +# The same level twice: a sustained hold reported in two windows. +[[ramp]] +users = 1.0 +hold_secs = 60 + +[[ramp]] +users = 1.0 +hold_secs = 60