From 3fdd8c6ff1ca42fe0be9f8ae78a8f8f7b8075711 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 11:33:28 -0500 Subject: [PATCH] feat(loadgen): model agent turns and sustained holds (#9182) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Why The loadgen agent polled HTTP `/query` every few seconds. A real agent harness (`buzz-acp`) never polls when idle. It works in bursts around each prompt, and those bursts are the load a fleet of active agents puts on the relay. To answer "how well does the relay handle N active users plus their agents over a sustained period", the loadgen needs to model those bursts and hold them for an hour or more. ## What - **`[agent.turn]`**: an idle agent starts a turn on a Poisson schedule and runs one turn at a time. A turn lasts 0.5–1.5 × `mean_secs`. The steps follow `buzz-acp`: - **Start**: a `limit:1` `/query` finds the message to answer (`http_query`), which stands in for the harness's subscription. Then the harness's own reads (`turn_context` op): for a threaded message, its thread context in one `/query` (root by id, newest 13 replies, the agent's own newest reply) plus a `/count` when the window is full. Then kind:39000 metadata, project home (kind:30621, kind:30617) and the author's profile. Then two kind:7 reactions on the message. - **Work**: typing every `typing_secs` and observer frames of `frame_bytes` every `frame_secs` over the socket, plus `reads` tool reads (`/query`, limit 50). - **End**: a threaded reply, under the root that NIP-10 `resolve()` gives. Then reaction removal: query own reactions, then a kind:5 deletion for each. - One HTTP action at a time runs next to the socket. Socket publishes (presence, typing, frames) don't wait for their OK: each OK is matched as it arrives, one missing after 30 s is an error, and at most 128 wait at once. So the socket and HTTP sides stay on schedule while either is slow, as in the harness. A 2xx response that doesn't decode counts as an error. No agent request starts after the stop. - The request mix approximates sustained turns: every turn reads like a first prompt. A harness already holding a thread reads 25 replies (not 13), batches profile lookups and caches project-home lookups for 30 s, so this reads profiles and project homes more often than a busy harness. - `agent.channel_subscriptions`: one live mentions REQ per channel, like the harness. `agent.presence_interval_secs`: presence heartbeats. - `agent.query_interval_secs = 0` turns polling off. - **Sustained holds**: ramp steps may now repeat a level. A 60-minute hold can be six 600 s steps, each one reported and judged, so drift shows up as a trend. - New op `turn_context`. New scenario `perf/scenarios/local-agent-turns.toml`. `perf/LOADGEN.md` documents the model. All new fields are off by default (`turn.per_hour = 0`, presence 0, no channel subscriptions), so existing scenarios run unchanged. ## What got simpler - The agent loop's per-action branches now go through one `sent` result. The observer frame build, the HTTP submit and the `/query` body read are shared helpers. The old inline frame code is gone. - "Hold for an hour, judge every 10 minutes" is now a ramp of repeated levels. No new abort-window concept was needed. - Identified debt, not touched here: abort is still judged only at step end, so a collapsed relay is not detected until the step closes. Repeated short steps reduce that gap. ## Testing - On a Blox workstation at 6bb151d1: `cargo fmt --check`, `cargo clippy -p buzz-test-client --all-targets -D warnings`, and `cargo test -p buzz-test-client --lib` (103 passed). The five turn tests passed 10/10 repeated runs. Three mutants (awaiting the OK again, dropping the stop guard in `submit`, dropping it in `query`) each fail a test. - `local-agent-turns` against a local relay at 6bb151d1: both steps 0 errors, 0 clients cancelled. - `agent_turns_read_react_stream_reply_and_remove_their_reactions` runs an agent against a fake socket and a fake HTTP relay. It asserts the exact thread-context filters and count, the metadata, project and profile reads, the reactions, typing, frames and presence, and that typing or frames arrive while a tool read is held. It also asserts the threaded reply under the right root and the deletion of exactly the reactions the agent added. - `agent_turns_keep_moving_while_socket_oks_lag`: the same journey with every socket OK 1.5 s late, past the 1 s heartbeat. - `a_turn_stopped_while_adding_a_reaction_sends_nothing_more` and `a_turn_stopped_while_reading_its_reactions_deletes_none`: the stop lands while a reaction POST or removal query is held, and no request follows it. - `agent_turn_reads_that_do_not_decode_are_errors`: 200 responses that are not events count as errors, and no turn reacts. - `a_timed_out_route_is_recovered_again_after_a_renewal` (desktop) fails 2 times in 10 on unmodified main as well. That flake is pre-existing and unrelated to this change. - Live: a single-relay run with 10k humans and 20k turn-taking agents (held at 1x for 60 min, then 1.5x and 2x). Results will be added here. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-test-client/src/loadgen/run.rs | 1441 +++++++++++++++-- .../buzz-test-client/src/loadgen/scenario.rs | 102 +- crates/buzz-test-client/src/loadgen/stats.rs | 6 +- perf/LOADGEN.md | 57 +- perf/scenarios/local-agent-turns.toml | 50 + 5 files changed, 1515 insertions(+), 141 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..436ecd99e2 100644 --- a/crates/buzz-test-client/src/loadgen/run.rs +++ b/crates/buzz-test-client/src/loadgen/run.rs @@ -1,27 +1,35 @@ //! Drives the simulated population through the scenario's ramp. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet, VecDeque}; +use std::future::Future; +use std::pin::Pin; 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 buzz_ws_client::connection::PUBLISH_OK_TIMEOUT_SECS; +use buzz_ws_client::{NostrWsConnection, OkResponse, RelayMessage, WsClientError}; +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}; +use super::stats::{abort_reason, ops_json, Op, Pending, Stats}; /// Content prefix of timed messages: `lg:`. const TIMED_PREFIX: &str = "lg:"; @@ -1035,8 +1043,259 @@ 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] = ["👀", "💬"]; +/// Replies a harness puts in a thread's context, as its default context +/// message limit. It fetches one more, and counts the thread when it gets it. +const THREAD_CONTEXT_REPLIES: usize = 12; + +/// 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 { + /// When the turn next has something to do. Without `http`, only its + /// socket steps count: typing and frames. + fn due(&self, http: bool) -> Option { + let http_due = http.then(|| self.reads.last().copied().unwrap_or(self.ends)); + [self.next_typing, self.next_frame, http_due] + .into_iter() + .flatten() + .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. + /// Without `http`, reads and the end wait. + fn step(&mut self, now: Instant, profile: &AgentTurn, http: bool) -> 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 !http { + None + } else if due(self.reads.last().copied()) { + self.reads.pop(); + Some(TurnStep::Read(self.channel.clone())) + } else { + (self.ends <= now).then_some(TurnStep::End) + } + } +} + +/// A turn's random draws, made before its requests start. +struct TurnPlan { + channel: String, + length: Duration, + /// Each tool read's offset into the turn, as a fraction of its length. + reads: Vec, +} + +impl TurnPlan { + fn draw(channels: &[String], profile: &AgentTurn, rng: &mut Rng) -> Self { + let channel = channels[rng.below(channels.len())].clone(); + let length = Duration::from_secs_f64(profile.mean_secs * (0.5 + rng.unit())); + let reads = (0..profile.reads).map(|_| rng.unit()).collect(); + Self { + channel, + length, + reads, + } + } +} + +/// How an agent's HTTP action finished. +enum Outcome { + Started(Box), + Ended, + Done, +} + +/// An agent's HTTP action in flight. One runs at a time, beside the socket. +type Busy<'a> = Option + Send + 'a>>>; + +/// The in-flight action's outcome; never resolves when there is none. +async fn settle(busy: &mut Busy<'_>) -> Outcome { + match busy { + Some(action) => action.await, + None => std::future::pending().await, + } +} + +/// Applies a finished HTTP action to the agent's turn state. +fn land( + outcome: Outcome, + turn: &mut Option, + next_turn: &mut Option, + profile: &AgentTurn, + rng: &mut Rng, +) { + match outcome { + Outcome::Started(started) => *turn = Some(*started), + Outcome::Ended => *next_turn = rng.after(profile.per_hour), + Outcome::Done => {} + } +} + +/// `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)) +} + +/// The most socket publishes an agent leaves awaiting their `OK`; past it, a +/// publish is an error instead of sent. A relay that answers within the +/// publish timeout keeps an agent far below it. +const MAX_UNACKED: usize = 128; + +/// An agent's socket publishes awaiting their `OK`, oldest first. An agent +/// sends a frame and moves on, as a harness does, so a slow `OK` holds up +/// neither its HTTP action nor its next frame. Each `OK` is matched as it +/// arrives. A publish gets `timeout` from the start of its send: one not +/// written by then fails and drops the socket, and one not answered by then +/// is an error. +struct Unacked<'a> { + stats: &'a Stats, + timeout: Duration, + /// Event id, the publish's timing and its deadline. Identical frames sent + /// within a second share an id; each `OK` settles the oldest. + waiting: VecDeque<(String, Pending<'a>, Instant)>, +} + +impl<'a> Unacked<'a> { + fn new(stats: &'a Stats, timeout: Duration) -> Self { + Self { + stats, + timeout, + waiting: VecDeque::new(), + } + } + + fn is_empty(&self) -> bool { + self.waiting.is_empty() + } + + /// Sends `event` as `op` without waiting for its `OK`. + async fn send( + &mut self, + conn: &mut NostrWsConnection, + op: Op, + event: Result, + ) -> Result<(), WsClientError> { + let event = match event { + Ok(event) => event, + Err(e) => { + self.stats.err(op, &e); + return Ok(()); + } + }; + if self.waiting.len() >= MAX_UNACKED { + self.stats.err(op, "too many publishes awaiting OK"); + return Ok(()); + } + let pending = self.stats.begin(op); + let deadline = Instant::now() + self.timeout; + let frame = json!(["EVENT", event]); + match tokio::time::timeout(self.timeout, conn.send_raw(&frame)).await { + Ok(Ok(())) => {} + Ok(Err(e)) => { + // The caller records the broken socket. + pending.forget(); + return Err(e); + } + Err(_) => { + // The relay stopped reading; the caller drops the socket. + pending.err("send timeout"); + return Err(WsClientError::Timeout); + } + } + self.waiting + .push_back((event.id.to_hex(), pending, deadline)); + Ok(()) + } + + /// Settles the oldest publish `ok` answers, if any. + fn answer(&mut self, ok: &OkResponse) { + let Some(at) = self.waiting.iter().position(|(id, ..)| *id == ok.event_id) else { + return; + }; + if let Some((_, pending, _)) = self.waiting.remove(at) { + if ok.accepted { + pending.ok(); + } else { + pending.err(&ok.message); + } + } + } + + /// When the oldest publish times out. + fn deadline(&self) -> Option { + self.waiting.front().map(|(.., deadline)| *deadline) + } + + /// Fails each publish whose `OK` is overdue at `now`. + fn expire(&mut self, now: Instant) { + while let Some((.., deadline)) = self.waiting.front() { + if *deadline > now { + break; + } + if let Some((_, pending, _)) = self.waiting.pop_front() { + pending.err("OK timeout"); + } + } + } + + /// Fails every publish still awaiting its `OK` with `reason`. + fn fail(&mut self, reason: &str) { + for (_, pending, _) in self.waiting.drain(..) { + pending.err(reason); + } + } + + /// Leaves every publish unjudged once its socket is gone, which is + /// recorded as such. + fn forget(&mut self) { + for (_, pending, _) in self.waiting.drain(..) { + pending.forget(); + } + } +} + +/// 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. One HTTP action at a time +/// runs beside the socket, and socket publishes do not wait for their `OK` +/// (see [`Unacked`]), so presence, typing, frames and requests keep their +/// schedule while a request or an `OK` is slow, as a harness's do. async fn agent(shared: Arc, global: usize) { let scenario = &shared.scenario; let profile = &scenario.agent; @@ -1050,9 +1309,24 @@ async fn agent(shared: Arc, global: usize) { let auth = identity::nip_oa_tag(&owner, &keys); let channels = scenario.member_channels(community.index, Role::Agent, global); let me = keys.public_key().to_hex(); + // What HTTP actions borrow; each owns everything else it needs. + let (sh, k, a): (&Shared, &Keys, &Tag) = (&shared, &keys, &auth); 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; + let mut busy: Busy<'_> = None; while let Some(mut conn) = connect( &shared, @@ -1066,44 +1340,73 @@ 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())); + let mut unacked = Unacked::new(&shared.stats, Duration::from_secs(PUBLISH_OK_TIMEOUT_SECS)); while ok && !stopped(&shared) { - let due = [next_query, next_post, next_frame] - .into_iter() - .flatten() - .min(); + // While an HTTP action runs, only socket work is waited for. + let idle = busy.is_none(); + let if_idle = |t: Option| t.filter(|_| idle); + let due = [ + next_presence, + next_frame, + turn.as_ref().map_or(if_idle(next_turn), |t| t.due(idle)), + if_idle(next_query), + if_idle(next_post), + unacked.deadline(), + ] + .into_iter() + .flatten() + .min(); let wait = due .map(|d| d.saturating_duration_since(Instant::now())) .unwrap_or(MAX_IDLE) .min(MAX_IDLE); - match conn.next_event(wait).await { - Ok(message) => { + let received = tokio::select! { + received = conn.next_event(wait) => Some(received), + outcome = settle(&mut busy), if busy.is_some() => { + busy = None; + land(outcome, &mut turn, &mut next_turn, &profile.turn, &mut rng); + None + } + }; + match received { + Some(Ok(RelayMessage::Ok(answer))) => unacked.answer(&answer), + Some(Ok(message)) => { if live(&shared.stats, &message, subscribed, &mut last_message).is_err() { ok = false; break; } } - Err(WsClientError::Timeout) => {} - Err(e) => { + Some(Err(WsClientError::Timeout)) | None => {} + Some(Err(e)) => { shared .stats .err(Op::WsConnect, &format!("disconnected: {e}")); @@ -1112,81 +1415,142 @@ 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) { + unacked.expire(now); + let idle = busy.is_none(); + // One action per wakeup; anything else due runs on the next pass. + // An HTTP action only starts with none in flight. + 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); + let event = event.map_err(|e| format!("sign: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, event).await + } else if let Some(step) = turn.as_mut().and_then(|t| t.step(now, &profile.turn, idle)) + { + match step { + TurnStep::Typing(channel) => { + let event = EventBuilder::new(kind(KIND_TYPING_INDICATOR), "") + .tags([tag(["h", &channel])]) + .sign_with_keys(&keys) + .map_err(|e| format!("sign: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, event).await + } + TurnStep::Frame => { + let frame = observer_frame(&keys, &owner, profile.turn.frame_bytes); + let frame = frame.map_err(|e| format!("observer: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, frame).await + } + TurnStep::Read(channel) => { + let filters = json!([{ + "kinds": AGENT_READ_KINDS, + "#h": [channel], + "limit": 50, + }]); + busy = Some(Box::pin(async move { + query(sh, Op::HttpQuery, k, community, &filters, a).await; + Outcome::Done + })); + Ok(()) + } + TurnStep::End => { + if let Some(ended) = turn.take() { + busy = Some(Box::pin(async move { + end_turn(sh, community, k, a, ended).await; + Outcome::Ended + })); + } + Ok(()) + } + } + } else if idle && turn.is_none() && next_turn.is_some_and(|t| t <= now) { + next_turn = None; + let plan = TurnPlan::draw(&channels, &profile.turn, &mut rng); + busy = Some(Box::pin(async move { + Outcome::Started(Box::new(start_turn(sh, community, k, a, plan).await)) + })); + Ok(()) + } else if idle && 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 }, { "#p": [me], "since": window, "limit": 20 }, ]); - http( - &shared, - Op::HttpQuery, - &keys, - community, - "/query", - &filters, - Some(&auth), - ) - .await; - } else if next_post.is_some_and(|t| t <= now) { + busy = Some(Box::pin(async move { + query(sh, Op::HttpQuery, k, community, &filters, a).await; + Outcome::Done + })); + Ok(()) + } else if idle && 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])]); + busy = Some(Box::pin(async move { + submit(sh, community, k, a, event, Op::HttpEvents).await; + Outcome::Done + })); + Ok(()) } 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); + let frame = frame.map_err(|e| format!("observer: {e}")); + unacked.send(&mut conn, Op::PublishEphemeral, frame).await + } else { + Ok(()) + }; + if let Err(e) = sent { + shared + .stats + .err(Op::WsConnect, &format!("disconnected: {e}")); + ok = false; } } + // At the stop, publishes still awaiting their `OK` settle within + // their deadlines while the HTTP action finishes, as a human's + // publish in flight does. One never answered outlasts the shutdown + // grace and fails the run, and a socket lost before they all settle + // fails the rest. + while ok && stopped(&shared) && !unacked.is_empty() { + let wait = unacked + .deadline() + .map_or(MAX_IDLE, |d| d.saturating_duration_since(Instant::now())) + .min(MAX_IDLE); + let received = tokio::select! { + received = conn.next_event(wait) => Some(received), + outcome = settle(&mut busy), if busy.is_some() => { + busy = None; + land(outcome, &mut turn, &mut next_turn, &profile.turn, &mut rng); + None + } + }; + match received { + Some(Ok(RelayMessage::Ok(answer))) => unacked.answer(&answer), + Some(Ok(_)) | Some(Err(WsClientError::Timeout)) | None => {} + Some(Err(e)) => { + unacked.fail(&format!("disconnected before OK: {e}")); + break; + } + } + unacked.expire(Instant::now()); + } + unacked.forget(); let _ = conn.disconnect().await; + // The action in flight finishes before the agent reconnects or + // stops; each checks the stop between its requests. + if let Some(action) = busy.take() { + land( + action.await, + &mut turn, + &mut next_turn, + &profile.turn, + &mut rng, + ); + } if stopped(&shared) { return; } @@ -1198,6 +1562,210 @@ async fn agent(shared: Arc, global: usize) { } } +/// Starts a turn the way an agent harness starts a prompt. The loadgen first +/// finds the event to answer, the channel's newest message, where a harness +/// is handed it by its subscription. Then, as `turn_context`: a threaded +/// target's thread context (see [`thread_context`]), the channel's metadata, +/// its project home (two reads) and the target author's profile, and then +/// two reactions on the target. +async fn start_turn( + shared: &Shared, + community: &Community, + keys: &Keys, + auth: &Tag, + plan: TurnPlan, +) -> Turn { + let profile = &shared.scenario.agent.turn; + let channel = plan.channel; + let newest = json!([{ "kinds": AGENT_READ_KINDS, "#h": [channel], "limit": 1 }]); + let target: Option = query(shared, Op::HttpQuery, keys, community, &newest, auth) + .await + .and_then(|events| events.into_iter().max_by_key(|e| e.created_at)); + let root = target + .as_ref() + .and_then(|t| buzz_core::nip10::parse_thread_markers(&t.tags).resolve()) + .map(|(root, _)| root); + if let Some(root) = root { + thread_context(shared, community, keys, auth, &channel, &root).await; + } + let author = target.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())] }]), + ]; + for filters in reads { + query(shared, Op::TurnContext, keys, community, &filters, auth).await; + } + if let Some(target) = &target { + 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 mut reads: Vec = plan + .reads + .iter() + .map(|at| now + plan.length.mul_f64(*at)) + .collect(); + reads.sort_unstable_by(|a, b| b.cmp(a)); + Turn { + channel, + target, + ends: now + plan.length, + next_typing: every(profile.typing_secs, now).map(|_| now), + next_frame: every(profile.frame_secs, now).map(|_| now), + reads, + } +} + +/// A harness's thread context read, as `turn_context`: one `/query` for the +/// root, the thread's newest replies (one past the context limit) and the +/// agent's own newest reply, then a `/count` of the replies when the window +/// overflowed. +async fn thread_context( + shared: &Shared, + community: &Community, + keys: &Keys, + auth: &Tag, + channel: &str, + root: &str, +) { + let replies = json!({ + "kinds": AGENT_READ_KINDS, + "#e": [root], + "#h": [channel], + "limit": THREAD_CONTEXT_REPLIES + 1, + }); + let mut mine = replies.clone(); + mine["authors"] = json!([keys.public_key().to_hex()]); + mine["limit"] = json!(1); + let filters = json!([{ "ids": [root] }, replies, mine]); + let Some(events) = query(shared, Op::TurnContext, keys, community, &filters, auth).await else { + return; + }; + let fetched: HashSet<_> = events + .iter() + .map(|e| e.id.to_hex()) + .filter(|id| id != root) + .collect(); + if fetched.len() > THREAD_CONTEXT_REPLIES && !stopped(shared) { + let mut count = replies; + count["limit"] = json!(0); + http::( + shared, + Op::TurnContext, + keys, + community, + "/count", + &json!([count]), + Some(auth), + ) + .await; + } +} + +/// A `/count` response. +#[derive(serde::Deserialize)] +struct CountReply { + #[serde(rename = "count")] + _count: u64, +} + +/// Ends a turn: a timed reply in the answered message's thread, 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 { + // As a harness answers a person: a threaded target is answered at + // its thread's first level, under the root, and a top-level one + // becomes the root of a new thread. Replies never nest deeper, so + // agents answering agents can't climb to the relay's depth limit. + let root = buzz_core::nip10::parse_thread_markers(&target.tags) + .resolve() + .map_or_else(|| target.id.to_hex(), |(root, _)| root); + tags.extend(reply_tags(&root, &root)); + } + 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 else { + return; + }; + let mine = json!([{ + "kinds": [KIND_REACTION], + "authors": [me], + "#e": [target.id.to_hex()], + }]); + for _ in TURN_REACTIONS { + let found = query(shared, Op::HttpQuery, keys, community, &mine, auth) + .await + .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()) +} + +/// Signs and submits `event` through HTTP `/events` as `op`; as with +/// [`query`], none starts after the stop. 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, +) { + if stopped(shared) { + return; + } + 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. @@ -1281,7 +1849,7 @@ async fn hot_author(shared: Arc, index: usize) { } } else { next_fetch = rng.after(profile.fetches_per_hour); - http( + http::>( &shared, Op::ThreadFetch, &keys, @@ -1295,7 +1863,10 @@ async fn hot_author(shared: Arc, index: usize) { } } -async fn http( +/// An HTTP POST of `body` to `path` as `op`; its response decoded as `T`. +/// The op succeeds only when the relay answered 2xx with a body that +/// decodes, so a malformed answer is an error rather than an empty result. +async fn http( shared: &Shared, op: Op, keys: &Keys, @@ -1303,7 +1874,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 +1882,44 @@ 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() => match serde_json::from_str(&text) { + Ok(decoded) => { + pending.ok(); + Some(decoded) + } + Err(e) => { + pending.err(&format!("decode {path}: {e}")); + None + } + }, + 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 events it returned. None starts +/// after the stop, so an action under way at the stop sends nothing more +/// than the request already in flight. +async fn query( + shared: &Shared, + op: Op, + keys: &Keys, + community: &Community, + filters: &Value, + auth: &Tag, +) -> Option> { + if stopped(shared) { + return None; + } + http(shared, op, keys, community, "/query", filters, Some(auth)).await +} + #[cfg(test)] mod tests { use super::*; @@ -1620,6 +2223,11 @@ hold_secs = 1 history_delay: Duration, /// From this long after the handshake, each `OK` takes this long. slow_ok: Option<(Duration, Duration)>, + /// Each EVENT's `OK` follows this long after `ok_delay`, without + /// holding up the frames after it. + ok_lag: Duration, + /// Drops the socket this long after the handshake. + close_after: Option, } impl FakeRelay { @@ -1628,6 +2236,8 @@ hold_secs = 1 ok_delay: Some(Duration::ZERO), history_delay: Duration::ZERO, slow_ok: None, + ok_lag: Duration::ZERO, + close_after: None, }; } @@ -1658,44 +2268,72 @@ hold_secs = 1 if ws.send(Message::Text(challenge.into())).await.is_err() { return; } - while let Some(Ok(Message::Text(text))) = ws.next().await { - let frame: Value = serde_json::from_str(&text).unwrap(); - let reply = match frame[0].as_str() { - Some("AUTH") => json!(["OK", frame[1]["id"], true, ""]), - Some("REQ") => { - seen.reqs.fetch_add(1, Ordering::SeqCst); - let id = frame[1].as_str().unwrap_or_default(); - if id.starts_with(HISTORY_PREFIX) { - tokio::time::sleep(relay.history_delay).await; - } - json!(["EOSE", frame[1]]) + let (mut sink, mut ws) = ws.split(); + let (out, mut replies) = tokio::sync::mpsc::unbounded_channel::(); + tokio::spawn(async move { + while let Some(reply) = replies.recv().await { + let text = Message::Text(reply.to_string().into()); + if sink.send(text).await.is_err() { + return; } - Some("EVENT") => { - 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(); - seen.channels.lock().unwrap().push(h); - } - let Some(mut delay) = relay.ok_delay else { - continue; - }; - if let Some((after, slow)) = relay.slow_ok { - if opened.elapsed() >= after { - delay = slow; - } - } - tokio::time::sleep(delay).await; - json!(["OK", frame[1]["id"], true, ""]) - } - _ => continue, - }; - if ws - .send(Message::Text(reply.to_string().into())) - .await - .is_err() - { - return; } + }); + let serve = async move { + while let Some(Ok(Message::Text(text))) = ws.next().await { + let frame: Value = serde_json::from_str(&text).unwrap(); + let reply = match frame[0].as_str() { + Some("AUTH") => json!(["OK", frame[1]["id"], true, ""]), + Some("REQ") => { + seen.reqs.fetch_add(1, Ordering::SeqCst); + let id = frame[1].as_str().unwrap_or_default(); + if id.starts_with(HISTORY_PREFIX) { + tokio::time::sleep(relay.history_delay).await; + } + json!(["EOSE", frame[1]]) + } + Some("EVENT") => { + if let Some(k) = frame[1]["kind"].as_u64() { + seen.kinds.lock().unwrap().push((k, Instant::now())); + } + 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(); + seen.channels.lock().unwrap().push(h); + } + let Some(mut delay) = relay.ok_delay else { + continue; + }; + if let Some((after, slow)) = relay.slow_ok { + if opened.elapsed() >= after { + delay = slow; + } + } + tokio::time::sleep(delay).await; + let ok = json!(["OK", frame[1]["id"], true, ""]); + if !relay.ok_lag.is_zero() { + let out = out.clone(); + tokio::spawn(async move { + tokio::time::sleep(relay.ok_lag).await; + let _ = out.send(ok); + }); + continue; + } + ok + } + _ => continue, + }; + if out.send(reply).is_err() { + return; + } + } + }; + // Ending `serve` drops both halves of the socket. + match relay.close_after { + Some(after) => { + let _ = tokio::time::timeout(after, serve).await; + } + None => serve.await, } }); } @@ -1710,6 +2348,8 @@ hold_secs = 1 closed: AtomicUsize, /// The `h` tag of every EVENT that had one. channels: std::sync::Mutex>, + /// The kind of every EVENT, with when it arrived. + kinds: std::sync::Mutex>, } impl Seen { @@ -2097,6 +2737,543 @@ hold_secs = 2 channel: String, } + /// `[agent]` turns for one agent in one channel over `ws` and `http`, + /// held for 3 s, with `extra` TOML appended. + fn agent_turn_scenario(ws: &str, http: &str, extra: &str) -> 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 +{extra} +"# + )) + .unwrap() + } + + /// How long the fake relay holds each tool read. + const TOOL_READ_HOLD: Duration = Duration::from_millis(250); + + /// One agent running turns over a fake socket and a fake HTTP relay + /// that holds one channel message, a full thread window and the agent's + /// live reactions, and is slow to answer tool reads. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn agent_turns_read_react_stream_reply_and_remove_their_reactions() { + agent_turns_against(FakeRelay::PROMPT).await; + } + + /// The same turns when every socket `OK` is slower than the agent's + /// heartbeat, though well inside the abort ceiling: nothing waits for an + /// `OK`, so tool reads, typing and frames all keep their schedule. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn agent_turns_keep_moving_while_socket_oks_lag() { + let lag = Duration::from_millis(1500); + let report = agent_turns_against(FakeRelay { + ok_lag: lag, + ..FakeRelay::PROMPT + }) + .await; + let acked = &report["steps"][0]["ops"]["publish_ephemeral_ok"]; + assert!(acked["count"].as_u64() > Some(0), "{report}"); + assert_eq!(acked["errors"], 0, "{report}"); + let p50 = acked["p50_ms"].as_f64().unwrap(); + assert!(p50 >= lag.as_millis() as f64, "{report}"); + // The OKs still due at the stop settle during the shutdown. + let closing = &report["shutdown"]["ops"]["publish_ephemeral_ok"]; + assert!(closing["count"].as_u64() > Some(0), "{report}"); + assert_eq!(closing["errors"], 0, "{report}"); + } + + #[tokio::test] + async fn an_agent_publish_that_never_gets_its_ok_fails_the_run() { + let (addr, _) = fake_ws_relay(FakeRelay { + ok_delay: None, + ..FakeRelay::PROMPT + }) + .await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[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 +[[ramp]] +users = 1.0 +hold_secs = 2 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let report = tokio::time::timeout( + Duration::from_secs(20), + run_with( + scenario, + net, + Shard { index: 0, count: 1 }, + Duration::from_millis(300), + ), + ) + .await + .expect("the run returns") + .expect_err("a relay that never acknowledges a publish fails the run"); + assert_eq!(report["clients_cancelled"], 1, "{report}"); + let closing = &report["shutdown"]; + assert!( + closing["ops"]["publish_ephemeral_ok"]["errors"].as_u64() > Some(0), + "{report}" + ); + assert!( + closing["failures"] + .to_string() + .contains("unfinished at stop"), + "{report}" + ); + } + + /// A relay that drops the socket while the agent drains its publishes at + /// the stop: the ones still awaiting their `OK` fail, so the drop cannot + /// pass the run. The agent's loop wakes at least every second, so it is + /// draining well before the drop, 2 s after the 2 s hold ends. + #[tokio::test] + async fn a_socket_lost_with_publishes_awaiting_their_ok_at_the_stop_fails_the_run() { + let (addr, _) = fake_ws_relay(FakeRelay { + ok_delay: None, + close_after: Some(Duration::from_secs(4)), + ..FakeRelay::PROMPT + }) + .await; + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "ws://{addr}" +http_url = "http://{addr}" +[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 +[[ramp]] +users = 1.0 +hold_secs = 2 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let report = tokio::time::timeout( + Duration::from_secs(20), + run_with( + scenario, + net, + Shard { index: 0, count: 1 }, + Duration::from_secs(10), + ), + ) + .await + .expect("the run returns") + .expect_err("publishes lost with their socket fail the run"); + assert_eq!(report["clients_cancelled"], 0, "{report}"); + let closing = &report["shutdown"]; + assert!( + closing["ops"]["publish_ephemeral_ok"]["errors"].as_u64() > Some(0), + "{report}" + ); + assert!( + closing["failures"] + .to_string() + .contains("disconnected before OK"), + "{report}" + ); + assert!(!closing["abort_reason"].is_null(), "{report}"); + } + + /// A relay that completes the handshake and then never reads: a publish + /// that cannot be written within its timeout fails and drops the socket. + #[tokio::test] + async fn a_publish_the_relay_never_reads_times_out() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let held = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + std::future::pending::<()>().await; + drop(ws); + }); + let mut conn = NostrWsConnection::connect(&format!("ws://{addr}")) + .await + .unwrap(); + let stats = Stats::default(); + let before = stats.snapshot(); + let mut unacked = Unacked::new(&stats, Duration::from_millis(200)); + let keys = Keys::generate(); + let big = "x".repeat(1 << 20); + let mut result = Ok(()); + for _ in 0..MAX_UNACKED { + let event = EventBuilder::new(kind(KIND_TYPING_INDICATOR), big.as_str()) + .sign_with_keys(&keys) + .map_err(|e| e.to_string()); + result = unacked.send(&mut conn, Op::PublishEphemeral, event).await; + if result.is_err() { + break; + } + } + assert!( + matches!(result, Err(WsClientError::Timeout)), + "every write went through: {result:?}" + ); + let reasons = stats.snapshot().reasons_since(&before, 10); + assert!( + reasons.iter().any(|(r, _)| r.contains("send timeout")), + "{reasons:?}" + ); + unacked.forget(); + held.abort(); + } + + /// Runs one agent's turns against `relay` and checks what they sent. + async fn agent_turns_against(relay: FakeRelay) -> Value { + let (ws, seen) = fake_ws_relay(relay).await; + let channel = identity::channel_id("t", 0, 0); + let me = identity::keys("t", Role::Agent, 0).public_key().to_hex(); + // A human's thread reply: only a `reply` marker, naming the root. + let author = Keys::generate(); + let root_id = "a".repeat(64); + let target = EventBuilder::new(kind(KIND_STREAM_MESSAGE), "question") + .tags([tag(["h", &channel]), tag(["e", &root_id, "", "reply"])]) + .sign_with_keys(&author) + .unwrap(); + // One reply past the context window, so each turn counts the thread. + let window: Vec = (0..=THREAD_CONTEXT_REPLIES) + .map(|i| { + EventBuilder::new(kind(KIND_STREAM_MESSAGE), format!("reply {i}")) + .tags([tag(["h", &channel]), tag(["e", &root_id, "", "reply"])]) + .sign_with_keys(&author) + .unwrap() + }) + .collect(); + let posted: Log = Default::default(); + let queries: Log = Default::default(); + let counts: Log = Default::default(); + let held: Log<(Instant, Instant)> = Default::default(); + let (post_log, query_log, count_log, hold_log, newest) = ( + posted.clone(), + queries.clone(), + counts.clone(), + held.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(); + if path == "/count" { + count_log.lock().unwrap().push(filters); + return (200, json!({"count": 40}).to_string()); + } + query_log.lock().unwrap().push(filters.clone()); + if filters[0]["limit"] == json!(50) { + let start = Instant::now(); + tokio::task::block_in_place(|| std::thread::sleep(TOOL_READ_HOLD)); + hold_log.lock().unwrap().push((start, Instant::now())); + return (200, "[]".to_string()); + } + if filters[0].get("ids").is_some() { + return (200, json!(window).to_string()); + } + let found = match filters[0]["kinds"][0].as_u64() { + Some(k) if k == u64::from(KIND_STREAM_MESSAGE) => json!([newest]), + // 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 = agent_turn_scenario(&ws, &http, ""); + 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.iter().any(|(seen, _)| *seen == u64::from(k)), + "no kind {k} over the socket: {kinds:?}" + ); + } + // Typing and frames keep streaming while a tool read is held. + let held = held.lock().unwrap().clone(); + assert!(!held.is_empty(), "no tool read made"); + let streaming = [KIND_TYPING_INDICATOR, KIND_AGENT_OBSERVER_FRAME].map(u64::from); + let during = kinds.iter().filter(|(k, at)| { + streaming.contains(k) + && held + .iter() + .any(|(start, end)| *at > *start + TOOL_READ_HOLD / 5 && at < end) + }); + assert!( + during.count() > 0, + "nothing streamed during {held:?}: {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 at the first level of the target's + // thread, under its root, 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(root_id.as_str())); + assert_eq!(markers.reply.as_deref(), Some(root_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 thread's context as the harness does, and + // counts the thread when the window overflows. + let queries = queries.lock().unwrap().clone(); + let replies_filter = json!({ + "kinds": AGENT_READ_KINDS, + "#e": [root_id], + "#h": [channel], + "limit": THREAD_CONTEXT_REPLIES + 1, + }); + let mut mine = replies_filter.clone(); + mine["authors"] = json!([me]); + mine["limit"] = json!(1); + let context = json!([{ "ids": [root_id] }, replies_filter.clone(), mine]); + assert!(queries.contains(&context), "{queries:?}"); + let mut count = replies_filter; + count["limit"] = json!(0); + let counts = counts.lock().unwrap().clone(); + assert!(!counts.is_empty(), "no thread count"); + assert!(counts.iter().all(|c| *c == json!([count])), "{counts:?}"); + // Then the channel's metadata, project home and the author's profile. + 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:?}" + ); + report + } + + /// A relay that answers every `/query` with 200 and a body that is not + /// events: each read is an error, and a turn has nothing to answer. + #[tokio::test] + async fn agent_turn_reads_that_do_not_decode_are_errors() { + let (ws, _) = fake_ws_relay(FakeRelay::PROMPT).await; + let posted: Log = Default::default(); + let post_log = posted.clone(); + let http = super::super::seed::tests::fake_http(move |_, path, body| { + if path == "/events" { + post_log + .lock() + .unwrap() + .push(serde_json::from_slice(body).unwrap()); + return (200, json!({"accepted": true, "message": ""}).to_string()); + } + (200, json!({"events": "none"}).to_string()) + }) + .await; + let scenario = agent_turn_scenario(&ws, &http, "[abort]\nmax_error_rate = 1.0"); + 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("the run tolerates every error"); + let ops = &report["steps"][0]["ops"]; + for op in ["turn_context", "http_query"] { + assert!(ops[op]["errors"].as_u64() > Some(0), "{op}: {report}"); + assert_eq!(ops[op]["count"], 0, "{op}: {report}"); + } + // No target was found, so no turn reacted. + let posted = posted.lock().unwrap(); + assert!( + posted.iter().all(|e| e.kind != kind(KIND_REACTION)), + "{posted:?}" + ); + } + + /// Runs one agent's turns against a relay that holds one top-level + /// channel message and the agent's live reactions, and holds the first + /// request `held` picks past the stop. Returns every request's path and + /// body, in order, and where the held one is among them. + async fn turns_stopped_during(held: fn(&str, &Value) -> bool) -> (Vec<(String, Value)>, usize) { + let (ws, _) = fake_ws_relay(FakeRelay::PROMPT).await; + let channel = identity::channel_id("t", 0, 0); + let target = EventBuilder::new(kind(KIND_STREAM_MESSAGE), "question") + .tags([tag(["h", &channel])]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let log: Log<(String, Value)> = Default::default(); + let at: Arc>> = Default::default(); + let (requests, held_at) = (log.clone(), at.clone()); + // Released after the 3 s hold has stopped the run, well inside the + // shutdown grace. + let release = Instant::now() + Duration::from_millis(4500); + let http = super::super::seed::tests::fake_http(move |_, path, body| { + let body: Value = serde_json::from_slice(body).unwrap(); + let hold = { + let mut requests = requests.lock().unwrap(); + requests.push((path.to_string(), body.clone())); + let mut held_at = held_at.lock().unwrap(); + let hold = held_at.is_none() && held(path, &body); + if hold { + *held_at = Some(requests.len() - 1); + } + hold + }; + if hold { + tokio::task::block_in_place(|| { + std::thread::sleep(release.saturating_duration_since(Instant::now())); + }); + } + if path == "/events" { + return (200, json!({"accepted": true, "message": ""}).to_string()); + } + let found = match body[0]["kinds"][0].as_u64() { + Some(k) if k == u64::from(KIND_STREAM_MESSAGE) => json!([target]), + Some(k) if k == u64::from(KIND_REACTION) => { + let posted = requests.lock().unwrap(); + let reactions = posted + .iter() + .filter(|(path, e)| path == "/events" && e["kind"] == json!(KIND_REACTION)); + json!(reactions.map(|(_, e)| e).collect::>()) + } + _ => json!([]), + }; + (200, found.to_string()) + }) + .await; + let scenario = agent_turn_scenario(&ws, &http, ""); + 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("a turn cut by the stop does not fail the run"); + assert_eq!(report["clients_cancelled"], 0, "{report}"); + let log = log.lock().unwrap().clone(); + let at = at.lock().unwrap().expect("the request was made and held"); + (log, at) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_turn_stopped_while_adding_a_reaction_sends_nothing_more() { + let (log, at) = turns_stopped_during(|path, body| { + path == "/events" && body["kind"] == json!(KIND_REACTION) + }) + .await; + assert_eq!(at + 1, log.len(), "requests after the stop: {log:?}"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_turn_stopped_while_reading_its_reactions_deletes_none() { + let (log, at) = turns_stopped_during(|path, body| { + path == "/query" && body[0]["kinds"][0] == json!(KIND_REACTION) + }) + .await; + assert_eq!(at + 1, log.len(), "requests after the stop: {log:?}"); + let deleted = log.iter().any(|(_, e)| e["kind"] == json!(KIND_DELETION)); + assert!(!deleted, "{log:?}"); + } + /// Holds `/events` responses until opened, counting the requests held. #[derive(Default)] struct Gate { diff --git a/crates/buzz-test-client/src/loadgen/scenario.rs b/crates/buzz-test-client/src/loadgen/scenario.rs index aabfb899e5..87547c65aa 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,26 @@ 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 +533,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 +773,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 +797,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..15c748532a 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,55 @@ 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**: an HTTP `/query` for the channel's newest message, the one + the turn answers. A harness is handed it by its subscription, so this + read is `http_query`, not `turn_context`. Then, as `turn_context`, the + harness's own reads. For a message in a thread, one `/query` for the + root, the thread's newest 13 replies and the agent's own newest reply, + then a `/count` of the replies when all 13 came back. After that, the + channel's kind:39000 metadata, its project home (kind:30621 and + kind:30617, one read each) and the message author's profile. Then two + kind:7 reactions on the 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 through HTTP `/events`, at the first level of the + message's thread (under its root), as a harness answers a person; a + top-level message becomes the root of a new thread. Then each + reaction's removal: a `/query` for the agent's reactions on + the message and a kind:5 deletion of each one found. + +An agent makes one HTTP request sequence at a time, and the socket keeps +running beside it: presence, typing and frames are sent without waiting for +their `OK`, so they stay on schedule while a request or an `OK` is slow, as a +harness's do. A publish has 30 seconds from the start of its send: one the +socket cannot write by then fails and drops the connection, and an `OK` not +back by then is an error. A read counts as an error when its response is not +valid JSON of the expected shape. Requests and publishes do not start after +the stop; one under way finishes, and publishes awaiting their `OK` settle +within the shutdown grace or fail the run. + +The request mix is an approximation of sustained turns, not a replay of the +harness. Every turn reads as a first prompt does. A harness that keeps +working in one thread reads 25 replies instead of 13, batches profile +lookups across prompts and caches project-home lookups for 30 seconds, so +the turn context here reads profiles and project homes more often than a +busy harness does. + +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 +311,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