From 4cf2a3bd6684982a69bd3c44bd466cf8ce31c5b6 Mon Sep 17 00:00:00 2001
From: Luke Tornquist <tornquist@squareup.com>
Date: Fri, 30 Oct 2026 13:52:30 -0400
Subject: [PATCH] feat(loadgen): add a hot-thread stressor and teardown (#8746)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## Summary

`buzz-loadgen` spreads load evenly: every channel gets the same history
and the same short thread chains. That never puts heavy load on a single
thread. A busy conversation, such as agents talking to each other, can
give one root a huge number of replies, and thread-summary work grows
with it (see #8635). This PR adds an optional `[hot_thread]` scenario
section so a load test can measure that case.

- **Seed:** `seed` posts a hot root in community 0, channel 0, then
`seed_replies` replies directly under it. The root comes from a
dedicated identity that posts nothing else, so `run` finds it however
many other roots the channel has. The replies go out concurrently, like
the rest of the seed, so the run starts with the thread already large.
- **Run:** `authors` hot-thread clients are scaled by each ramp step's
`users`, the same way humans and agents are. Each one replies to the hot
root over HTTP `/events` and reads the thread's newest 100 replies over
HTTP `/query`, at `replies_per_hour` and `fetches_per_hour`.
- **Metrics:** two new report ops, `hot_reply` and `thread_fetch`. The
replies are timed, so humans subscribed to that channel also record
delivery latency for them.
- **No root, failed run:** an author that can't find the seeded root
counts a `thread_fetch` error on every attempt, so a run without the hot
thread fails instead of quietly measuring nothing.
- **Off by default:** the section is disabled while `seed_replies` and
`authors` are both 0, so existing scenarios are unchanged.

`perf/scenarios/local-smoke.toml` gets a small hot thread, and
`perf/LOADGEN.md` documents the knobs.

### Teardown

The PR also adds `buzz-loadgen teardown <scenario>`, so a staging
fixture can be removed after the load test.

- For each community from `target.community_host_template`, it first
calls `POST /operator/communities/archive` as the fixture owner, using
the operator key `seed` uses. It then calls `POST
/operator/communities/delete`.
- The deletion `request_id` is derived from `run_id` and the host, so a
re-run replays the same requests. That retries failures and reports
progress without a `deletion_request_conflict`.
- The relay's deletion drain does the actual deletion asynchronously.
- If the fixture owner does not own a host, the host is reported
`absent` and left untouched.
- A scenario without a host template is refused.
- Deleted hosts stay as tombstones, so `LOADGEN.md` says each fixture
needs a fresh `run_id`.

The signing has to happen as each derived owner, which is why this lives
in the generator rather than in deployment config.

## What got simpler

Nothing got simpler; this adds a scale vector. To keep it small, the PR
reuses what's already there:
- the seeder's post plan, plus `find_roots` with a content prefix
- the existing HTTP helper, whose auth tag is now optional
- humans' delivery measurement

A separate chain-depth knob and dedicated viewer clients were left out.
`shape.thread_depth` already seeds chains up to the relay's depth cap,
and humans in the hot channel already receive the replies.

## Testing

-
`loadgen::seed::tests::the_hot_thread_seeds_direct_replies_under_its_own_root`
checks that the hot root and its replies are seeded, that every reply is
direct (depth 1), and that a rejected root fails the seed.
-
`loadgen::run::tests::hot_thread_authors_reply_to_the_hot_root_and_fail_without_it`
checks several things:
  - authors post replies tagged to the hot root;
- root lookups query only the hot-root identity, and a newer
reply-shaped decoy is never taken as the root;
  - scheduled reads all target the hot root's channel and id;
  - the run fails when the root isn't seeded.
-
`loadgen::run::tests::hot_thread_authors_start_nothing_once_the_run_stops`
runs 8 hot authors directly. It holds every in-flight `/events` reply at
a gate until the stop is signalled, so each author's next reply is
already overdue when the gate opens. It then checks that exactly 8
replies were posted, none started after the stop. Removing the
post-`select!` stop check fails the test, and the gate is released
before any assertion, so a failure can't hang the job.
- The run now signals the stop before the last step's snapshot instead
of after building that step's report. Under a loaded parallel test run,
work started in that gap leaked into `shutdown`.
-
`loadgen::teardown::tests::teardown_deletes_only_the_fixture_owners_communities_and_replays_on_rerun`
runs teardown twice against a fake operator that checks the owner,
requires archive before delete, and conflicts on a different request id.
It checks that:
  - the owned host is archived and deletion-requested;
- another owner's host and an unseeded host are reported absent and left
untouched;
  - a 503 on archive is reported as a failure;
  - the second pass converges to the same result;
- a deletion already past membership purge, where archive returns 404,
is replayed and reported at its stage under `requested`;
- the same run under another host template gets its own request ids:
another owner's host is reported absent rather than conflicting with a
request id already used.

If the request id is replaced with a fresh one per run, the test fails.
-
`loadgen::teardown::tests::teardown_refuses_a_scenario_without_templated_communities`
- All of these run in the existing `loadgen::` unit lane.
- Teardown was run end to end at `28192004` against a local relay. The
relay had operator keys configured, local S3 behind it, and `buzz-admin
deletions drain`. The run covered:
  - seeding three templated communities;
- teardown, with every fixture archived and a deletion request at its
derived id;
  - a re-run that converged;
  - the drain, which tombstoned the fixtures;
- teardown again after the purge, which reported each fixture at
`retention_pending` and exited 0;
  - a re-seed of the torn-down run, which failed at provisioning.

A community outside the fixture stayed active, with no deletion request.
- At `28192004`, `cargo fmt --check`, `cargo clippy -p buzz-test-client
--all-targets -D warnings` and six parallel runs of `cargo test -p
buzz-test-client` pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/buzz-test-client/src/bin/loadgen.rs    |  20 +-
 .../buzz-test-client/src/loadgen/identity.rs  |  30 +-
 crates/buzz-test-client/src/loadgen/mod.rs    |   4 +-
 crates/buzz-test-client/src/loadgen/run.rs    | 364 +++++++++++++++++-
 .../buzz-test-client/src/loadgen/scenario.rs  |  33 ++
 crates/buzz-test-client/src/loadgen/seed.rs   | 103 ++++-
 crates/buzz-test-client/src/loadgen/stats.rs  |   8 +-
 .../buzz-test-client/src/loadgen/teardown.rs  | 345 +++++++++++++++++
 perf/LOADGEN.md                               |  33 ++
 perf/scenarios/local-smoke.toml               |   7 +
 10 files changed, 918 insertions(+), 29 deletions(-)
 create mode 100644 crates/buzz-test-client/src/loadgen/teardown.rs

diff --git a/crates/buzz-test-client/src/bin/loadgen.rs b/crates/buzz-test-client/src/bin/loadgen.rs
index aa4e302f17..42263bf2f9 100644
--- a/crates/buzz-test-client/src/bin/loadgen.rs
+++ b/crates/buzz-test-client/src/bin/loadgen.rs
@@ -3,11 +3,12 @@
 //! Usage:
 //!   buzz-loadgen seed <scenario.toml>
 //!   buzz-loadgen run <scenario.toml> [--shard i/n] [--report <path>]
+//!   buzz-loadgen teardown <scenario.toml>
 
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use buzz_test_client::loadgen::{net::Net, run, scenario::Scenario, seed, stats};
+use buzz_test_client::loadgen::{net::Net, run, scenario::Scenario, seed, stats, teardown};
 use clap::{Parser, Subcommand};
 
 #[derive(Parser)]
@@ -34,6 +35,8 @@ enum Command {
         #[arg(long)]
         report: Option<PathBuf>,
     },
+    /// Archive the scenario's communities and request their deletion.
+    Teardown { scenario: PathBuf },
 }
 
 #[tokio::main]
@@ -48,7 +51,9 @@ async fn main() -> anyhow::Result<()> {
 
     let cli = Cli::parse();
     let path = match &cli.command {
-        Command::Seed { scenario } | Command::Run { scenario, .. } => scenario,
+        Command::Seed { scenario }
+        | Command::Run { scenario, .. }
+        | Command::Teardown { scenario } => scenario,
     };
     let scenario = Scenario::parse(&std::fs::read_to_string(path)?)
         .map_err(|e| anyhow::anyhow!("{}: {e}", path.display()))?;
@@ -81,5 +86,16 @@ async fn main() -> anyhow::Result<()> {
             }
             outcome
         }
+        Command::Teardown { .. } => {
+            let report = teardown::teardown(&scenario, &net).await?;
+            println!("{}", report.to_json());
+            if !report.failures.is_empty() {
+                anyhow::bail!(
+                    "{} communities failed teardown; re-run to retry",
+                    report.failures.len()
+                );
+            }
+            Ok(())
+        }
     }
 }
diff --git a/crates/buzz-test-client/src/loadgen/identity.rs b/crates/buzz-test-client/src/loadgen/identity.rs
index 8dfbc1ced8..44f8cc8a9c 100644
--- a/crates/buzz-test-client/src/loadgen/identity.rs
+++ b/crates/buzz-test-client/src/loadgen/identity.rs
@@ -12,6 +12,10 @@ pub enum Role {
     Owner,
     Human,
     Agent,
+    HotAuthor,
+    /// Posts only hot-thread roots, so `run` finds them among its newest
+    /// posts however many other roots a channel has.
+    HotRoot,
 }
 
 impl Role {
@@ -20,6 +24,8 @@ impl Role {
             Role::Owner => "owner",
             Role::Human => "human",
             Role::Agent => "agent",
+            Role::HotAuthor => "hot-author",
+            Role::HotRoot => "hot-root",
         }
     }
 }
@@ -52,19 +58,29 @@ pub fn keys(run_id: &str, role: Role, index: usize) -> Keys {
     }
 }
 
+fn uuid(parts: &[&str]) -> uuid::Uuid {
+    let seed = digest(parts);
+    let mut bytes = [0u8; 16];
+    bytes.copy_from_slice(&seed[..16]);
+    uuid::Builder::from_random_bytes(bytes).into_uuid()
+}
+
 /// Channel UUID for channel `index` of community `community`.
 pub fn channel_id(run_id: &str, community: usize, index: usize) -> String {
-    let seed = digest(&[
+    uuid(&[
         "buzz-loadgen-channel",
         run_id,
         &community.to_string(),
         &index.to_string(),
-    ]);
-    let mut bytes = [0u8; 16];
-    bytes.copy_from_slice(&seed[..16]);
-    uuid::Builder::from_random_bytes(bytes)
-        .into_uuid()
-        .to_string()
+    ])
+    .to_string()
+}
+
+/// Deletion request id for `host` in run `run_id`, so a re-run of `teardown`
+/// replays the same request instead of conflicting with it. Keyed by host,
+/// so a changed host template never reuses another host's request id.
+pub fn teardown_request_id(run_id: &str, host: &str) -> uuid::Uuid {
+    uuid(&["buzz-loadgen-teardown", run_id, host])
 }
 
 /// NIP-OA `auth` tag in which `owner` authorizes `agent`.
diff --git a/crates/buzz-test-client/src/loadgen/mod.rs b/crates/buzz-test-client/src/loadgen/mod.rs
index 4e8bf8c17a..8bf6bda607 100644
--- a/crates/buzz-test-client/src/loadgen/mod.rs
+++ b/crates/buzz-test-client/src/loadgen/mod.rs
@@ -4,7 +4,8 @@
 //! and a population of simulated humans and agents with per-connection rates.
 //! `seed` creates the communities,
 //! channels, history and deep threads; `run` drives the population through a
-//! ramp of user multipliers and reports latency percentiles per operation.
+//! ramp of user multipliers and reports latency percentiles per operation;
+//! `teardown` archives the seeded communities and requests their deletion.
 //!
 //! Every identity, community host and channel id is derived from the
 //! scenario's `run_id`, so independent shards (`--shard i/n`) agree on the
@@ -16,3 +17,4 @@ pub mod run;
 pub mod scenario;
 pub mod seed;
 pub mod stats;
+pub mod teardown;
diff --git a/crates/buzz-test-client/src/loadgen/run.rs b/crates/buzz-test-client/src/loadgen/run.rs
index ac10d3d845..8b78b467d3 100644
--- a/crates/buzz-test-client/src/loadgen/run.rs
+++ b/crates/buzz-test-client/src/loadgen/run.rs
@@ -19,7 +19,7 @@ use tokio::sync::{watch, Mutex};
 use super::identity::{self, Role};
 use super::net::{http_reason, Net};
 use super::scenario::{Community, Scenario};
-use super::seed::find_roots;
+use super::seed::{find_roots, hot_root_keys, reply_tags, HOT_ROOT_PREFIX, ROOT_PREFIX};
 use super::stats::{abort_reason, ops_json, Op, Stats};
 
 /// Content prefix of timed messages: `lg:<send unix micros>`.
@@ -95,7 +95,7 @@ async fn run_with(
     let connect_gap = Duration::from_secs_f64(1.0 / pop.connect_rate_per_sec);
 
     let mut tasks = Vec::new();
-    let (mut humans, mut agents) = (0usize, 0usize);
+    let (mut humans, mut agents, mut authors) = (0usize, 0usize, 0usize);
     let mut steps = Vec::new();
     let mut aborted = false;
     // End of the last reported window and start of the last judged one.
@@ -105,6 +105,7 @@ async fn run_with(
     for (n, step) in scenario.ramp.iter().enumerate() {
         let want_humans = (pop.humans as f64 * step.users).ceil() as usize;
         let want_agents = (pop.agents as f64 * step.users).ceil() as usize;
+        let want_authors = (scenario.hot_thread.authors as f64 * step.users).ceil() as usize;
         // The window covers the ramp too, so connect/auth and history
         // failures from new clients count toward the abort check.
         let before = stats.snapshot();
@@ -132,16 +133,28 @@ async fn run_with(
                 humans += 1;
             }
         }
+        // Hot-thread authors hold no socket, so they start without pacing.
+        while authors < want_authors {
+            if shard.owns(authors) {
+                let (shared, index) = (shared.clone(), authors);
+                tasks.push(tokio::spawn(hot_author(shared, index)));
+            }
+            authors += 1;
+        }
         let ramp_secs = ramp_start.elapsed().as_secs_f64();
 
         let hold_start = Instant::now();
         tokio::time::sleep(Duration::from_secs(step.hold_secs)).await;
         let hold_secs = hold_start.elapsed().as_secs_f64();
-        let after = stats.snapshot();
-        let summaries = after.since(&before);
-
         // Clients only return once stopped, so a finished one panicked.
         let lost = tasks.iter().filter(|t| t.is_finished()).count();
+        // Stop before the last step's snapshot, so the shutdown settles only
+        // work already in flight when that step's window closed.
+        if n + 1 == scenario.ramp.len() {
+            let _ = stop_tx.send(true);
+        }
+        let after = stats.snapshot();
+        let summaries = after.since(&before);
         let reason = if lost > 0 {
             Some(format!("{lost} clients exited unexpectedly"))
         } else {
@@ -156,6 +169,7 @@ async fn run_with(
             "users": step.users,
             "humans": want_humans,
             "agents": want_agents,
+            "hot_authors": want_authors,
             "shard": format!("{}/{}", shard.index, shard.count),
             "ramp_secs": (ramp_secs * 10.0).round() / 10.0,
             "hold_secs": (hold_secs * 10.0).round() / 10.0,
@@ -467,7 +481,8 @@ async fn roots(
     }
     let owner = identity::keys(&shared.scenario.run_id, Role::Owner, community.index);
     let pending = shared.stats.begin(Op::HttpQuery);
-    let found = match find_roots(&shared.net, community, reader, &owner, channel).await {
+    let found = match find_roots(&shared.net, community, reader, &owner, channel, ROOT_PREFIX).await
+    {
         Ok(found) => {
             pending.ok();
             found
@@ -914,7 +929,7 @@ async fn agent(shared: Arc<Shared>, global: usize) {
                     community,
                     "/query",
                     &filters,
-                    &auth,
+                    Some(&auth),
                 )
                 .await;
             } else if next_post.is_some_and(|t| t <= now) {
@@ -985,6 +1000,103 @@ async fn agent(shared: Arc<Shared>, global: usize) {
     }
 }
 
+/// 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.
+async fn hot_author(shared: Arc<Shared>, index: usize) {
+    let scenario = &shared.scenario;
+    let profile = &scenario.hot_thread;
+    let community = &shared.communities[0];
+    let keys = identity::keys(&scenario.run_id, Role::HotAuthor, index);
+    let root_author = hot_root_keys(&scenario.run_id);
+    let channel = identity::channel_id(&scenario.run_id, 0, 0);
+    let me = keys.public_key().to_hex();
+    let mut rng = Rng::new(u64::MAX / 2 - index as u64);
+    let mut backoff = Backoff::new();
+    // A missing root is an error every attempt, so a run without the seeded
+    // hot thread fails its steps instead of measuring nothing.
+    let root = loop {
+        if stopped(&shared) {
+            return;
+        }
+        let pending = shared.stats.begin(Op::ThreadFetch);
+        let newest = find_roots(
+            &shared.net,
+            community,
+            &keys,
+            &root_author,
+            &channel,
+            HOT_ROOT_PREFIX,
+        )
+        .await
+        .map(|roots| roots.last().cloned());
+        match newest {
+            Ok(Some(root)) => {
+                pending.ok();
+                break root;
+            }
+            Ok(None) => pending.err("hot thread root not seeded"),
+            Err(e) => pending.err(&e.to_string()),
+        }
+        backoff.wait(&shared, &mut rng).await;
+    };
+    let read = json!([{
+        "kinds": [KIND_STREAM_MESSAGE],
+        "#h": [channel],
+        "#e": [root],
+        "limit": 100,
+    }]);
+    let mut next_reply = rng.after(profile.replies_per_hour);
+    let mut next_fetch = rng.after(profile.fetches_per_hour);
+    loop {
+        let Some(due) = [next_reply, next_fetch].into_iter().flatten().min() else {
+            return stopping(&shared).await;
+        };
+        tokio::select! {
+            _ = tokio::time::sleep_until(due.into()) => {}
+            _ = stopping(&shared) => return,
+        }
+        // An overdue arrival and the stop can be ready together, and `select!`
+        // picks either; start nothing new once the run is stopping.
+        if stopped(&shared) {
+            return;
+        }
+        if next_reply == Some(due) {
+            next_reply = rng.after(profile.replies_per_hour);
+            let event = EventBuilder::new(
+                kind(KIND_STREAM_MESSAGE),
+                format!("{TIMED_PREFIX}{} {me}", now_micros()),
+            )
+            .tags(
+                [tag(["h", &channel])]
+                    .into_iter()
+                    .chain(reply_tags(&root, &root)),
+            )
+            .sign_with_keys(&keys);
+            let pending = shared.stats.begin(Op::HotReply);
+            match event {
+                Ok(event) => match shared.net.submit(&keys, community, &event, None).await {
+                    Ok(()) => pending.ok(),
+                    Err(e) => pending.err(&e.to_string()),
+                },
+                Err(e) => pending.err(&format!("sign: {e}")),
+            }
+        } else {
+            next_fetch = rng.after(profile.fetches_per_hour);
+            http(
+                &shared,
+                Op::ThreadFetch,
+                &keys,
+                community,
+                "/query",
+                &read,
+                None,
+            )
+            .await;
+        }
+    }
+}
+
 async fn http(
     shared: &Shared,
     op: Op,
@@ -992,13 +1104,13 @@ async fn http(
     community: &Community,
     path: &str,
     body: &Value,
-    auth: &Tag,
+    auth: Option<&Tag>,
 ) {
     let pending = shared.stats.begin(op);
     let body = serde_json::to_vec(body).expect("serialize JSON body");
     match shared
         .net
-        .post(keys, &community.http_url, path, body, Some(auth))
+        .post(keys, &community.http_url, path, body, auth)
         .await
     {
         Ok((status, _)) if status.is_success() => pending.ok(),
@@ -1651,4 +1763,238 @@ hold_secs = 2
         );
         assert!(closing["p99_ms"].as_f64().unwrap() > 1000.0, "{report}");
     }
+
+    /// What a hot-thread run sent: its outcome, the events and `/query`
+    /// filters it posted, and the hot root's id and channel.
+    struct HotRun {
+        outcome: Result<Value, Value>,
+        posted: Vec<Event>,
+        queries: Vec<Value>,
+        root_id: String,
+        channel: String,
+    }
+
+    type Log<T> = Arc<std::sync::Mutex<Vec<T>>>;
+
+    /// A fake relay for the hot thread: the events it received, the `/query`
+    /// filters it answered, and the hot root's id and channel.
+    struct HotRelay {
+        addr: String,
+        posted: Log<Event>,
+        queries: Log<Value>,
+        root_id: String,
+        channel: String,
+    }
+
+    /// Holds `/events` responses until opened, counting the requests held.
+    #[derive(Default)]
+    struct Gate {
+        arrived: AtomicUsize,
+        open: std::sync::Mutex<bool>,
+        opened: std::sync::Condvar,
+    }
+
+    impl Gate {
+        fn release(&self) {
+            *self.open.lock().unwrap() = true;
+            self.opened.notify_all();
+        }
+    }
+
+    /// Serves a relay whose root lookup returns a newer reply-shaped decoy by
+    /// the root's author, plus the hot root itself when `seeded`. With a
+    /// `gate`, every event waits for it to open before it is accepted.
+    async fn hot_relay(seeded: bool, gate: Option<Arc<Gate>>) -> HotRelay {
+        let author = hot_root_keys("t");
+        let channel = identity::channel_id("t", 0, 0);
+        let root = EventBuilder::new(kind(KIND_STREAM_MESSAGE), HOT_ROOT_PREFIX)
+            .tags([tag(["h", &channel])])
+            .sign_with_keys(&author)
+            .unwrap();
+        let root_id = root.id.to_hex();
+        let decoy = EventBuilder::new(kind(KIND_STREAM_MESSAGE), "lg-hot-reply 0")
+            .tags(
+                [tag(["h", &channel])]
+                    .into_iter()
+                    .chain(reply_tags(&"0".repeat(64), &"0".repeat(64))),
+            )
+            .custom_created_at(Timestamp::from(root.created_at.as_secs() + 1))
+            .sign_with_keys(&author)
+            .unwrap();
+        let posted: Log<Event> = Default::default();
+        let queries: Log<Value> = Default::default();
+        let (post_log, query_log) = (posted.clone(), queries.clone());
+        let addr = super::super::seed::tests::fake_http(move |_, target, body| {
+            if target == "/events" {
+                let event: Event = serde_json::from_slice(body).unwrap();
+                post_log.lock().unwrap().push(event);
+                if let Some(gate) = &gate {
+                    gate.arrived.fetch_add(1, Ordering::SeqCst);
+                    tokio::task::block_in_place(|| {
+                        let open = gate.open.lock().unwrap();
+                        drop(gate.opened.wait_while(open, |open| !*open).unwrap());
+                    });
+                }
+                return (200, json!({"accepted": true, "message": ""}).to_string());
+            }
+            let filter: Value = serde_json::from_slice(body).unwrap();
+            let lookup = filter[0].get("authors").is_some();
+            query_log.lock().unwrap().push(filter[0].clone());
+            let found = match (lookup, seeded) {
+                (true, true) => json!([root, decoy]),
+                (true, false) => json!([decoy]),
+                (false, _) => json!([]),
+            };
+            (200, found.to_string())
+        })
+        .await;
+        HotRelay {
+            addr,
+            posted,
+            queries,
+            root_id,
+            channel,
+        }
+    }
+
+    /// A one-second scenario with `[hot_thread]` section `hot` against `addr`.
+    fn hot_scenario(addr: &str, hot: &str) -> 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 = 0
+channels_per_member = 1
+[hot_thread]
+{hot}
+[[ramp]]
+users = 1.0
+hold_secs = 1
+"#
+        ))
+        .unwrap()
+    }
+
+    /// Runs `[hot_thread]` section `hot` for a second against `hot_relay`.
+    async fn hot_thread_run(seeded: bool, hot: &str) -> HotRun {
+        let relay = hot_relay(seeded, None).await;
+        let scenario = hot_scenario(&relay.addr, hot);
+        let net = Net::new(&scenario).await.unwrap();
+        let shard = Shard::parse("0/1").unwrap();
+        let outcome = run_with(scenario, net, shard, Duration::from_secs(5)).await;
+        let posted = relay.posted.lock().unwrap().clone();
+        let queries = relay.queries.lock().unwrap().clone();
+        HotRun {
+            outcome,
+            posted,
+            queries,
+            root_id: relay.root_id,
+            channel: relay.channel,
+        }
+    }
+
+    const TWO_AUTHORS: &str = "authors = 2\nreplies_per_hour = 36000.0\nfetches_per_hour = 36000.0";
+
+    #[tokio::test]
+    async fn hot_thread_authors_reply_to_the_hot_root_and_fail_without_it() {
+        let run = hot_thread_run(true, TWO_AUTHORS).await;
+        let report = run.outcome.expect("hot-thread run passes");
+        assert_eq!(report["steps"][0]["hot_authors"], 2, "{report}");
+        assert!(!run.posted.is_empty(), "{report}");
+        for event in &run.posted {
+            let markers = buzz_core::nip10::parse_thread_markers(&event.tags);
+            assert_eq!(markers.root.as_deref(), Some(run.root_id.as_str()));
+            assert_eq!(markers.reply.as_deref(), Some(run.root_id.as_str()));
+            let h = event
+                .tags
+                .iter()
+                .find(|t| t.kind().as_str() == "h")
+                .unwrap();
+            assert_eq!(h.content(), Some(run.channel.as_str()));
+        }
+        // The root is looked up among its own author's posts, so other roots
+        // in the channel can't push it out of the window.
+        let author = hot_root_keys("t").public_key().to_hex();
+        let (lookups, reads): (Vec<&Value>, Vec<&Value>) =
+            run.queries.iter().partition(|q| q.get("authors").is_some());
+        assert!(!lookups.is_empty());
+        for lookup in lookups {
+            assert_eq!(lookup["authors"], json!([author]), "{lookup}");
+        }
+        // Scheduled reads fetch the hot thread itself.
+        let read = json!({
+            "kinds": [KIND_STREAM_MESSAGE],
+            "#h": [run.channel],
+            "#e": [run.root_id],
+            "limit": 100,
+        });
+        assert!(!reads.is_empty(), "no scheduled thread reads: {report}");
+        assert!(reads.iter().all(|q| **q == read), "{reads:?}");
+        let ops = &report["steps"][0]["ops"];
+        assert!(ops["hot_reply"]["count"].as_u64() > Some(0), "{report}");
+
+        // Without a seeded hot root the authors post nothing and the step aborts.
+        let run = hot_thread_run(false, TWO_AUTHORS).await;
+        let report = run.outcome.expect_err("a missing hot root fails the run");
+        assert!(run.posted.is_empty());
+        let reason = report["steps"][0]["abort_reason"]
+            .as_str()
+            .unwrap_or_default();
+        assert!(reason.starts_with("thread_fetch"), "{report}");
+    }
+
+    /// Every author is mid-reply at the stop, held there by the relay, with
+    /// its next reply already due. Each must finish the reply it started and
+    /// start no other: exactly one reply per author settles after the stop,
+    /// and the relay receives no more.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn hot_thread_authors_start_nothing_once_the_run_stops() {
+        let gate = Arc::new(Gate::default());
+        let relay = hot_relay(true, Some(gate.clone())).await;
+        let scenario = hot_scenario(&relay.addr, "authors = 8\nreplies_per_hour = 36000000.0");
+        let net = Net::new(&scenario).await.unwrap();
+        let stats = Arc::new(Stats::default());
+        let (stop_tx, stop) = watch::channel(false);
+        let shared = Arc::new(Shared {
+            communities: scenario.communities(),
+            scenario,
+            net,
+            stats: stats.clone(),
+            stop,
+            roots: Mutex::new(HashMap::new()),
+        });
+        let authors: Vec<_> = (0..8)
+            .map(|i| tokio::spawn(hot_author(shared.clone(), i)))
+            .collect();
+        let deadline = Duration::from_secs(10);
+        let arrived = tokio::time::timeout(deadline, async {
+            while gate.arrived.load(Ordering::SeqCst) < 8 {
+                tokio::time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await;
+
+        stop_tx.send(true).unwrap();
+        let at_stop = stats.snapshot();
+        // Release before any assertion, so a failure never leaves handlers
+        // blocked in the gate and the runtime unable to shut down.
+        gate.release();
+        arrived.expect("every author starts a reply");
+        for author in authors {
+            tokio::time::timeout(deadline, author)
+                .await
+                .expect("authors return once stopped")
+                .unwrap();
+        }
+        let settled = ops_json(&stats.snapshot().since(&at_stop), 1.0);
+        assert_eq!(settled["hot_reply"]["count"], 8, "{settled}");
+        assert_eq!(relay.posted.lock().unwrap().len(), 8, "{settled}");
+    }
 }
diff --git a/crates/buzz-test-client/src/loadgen/scenario.rs b/crates/buzz-test-client/src/loadgen/scenario.rs
index 05680cb9d8..5aeaa4ffb1 100644
--- a/crates/buzz-test-client/src/loadgen/scenario.rs
+++ b/crates/buzz-test-client/src/loadgen/scenario.rs
@@ -17,6 +17,8 @@ pub struct Scenario {
     pub human: HumanProfile,
     #[serde(default)]
     pub agent: AgentProfile,
+    #[serde(default)]
+    pub hot_thread: HotThread,
     pub ramp: Vec<RampStep>,
     #[serde(default)]
     pub abort: Abort,
@@ -154,6 +156,29 @@ impl Default for AgentProfile {
     }
 }
 
+/// 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.
+#[derive(Clone, Debug, Default, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct HotThread {
+    /// Direct replies seeded under the hot root.
+    pub seed_replies: usize,
+    /// Clients replying to the hot root at 1x; ramp steps multiply them.
+    pub authors: usize,
+    /// Replies each author posts per hour through HTTP `/events`.
+    pub replies_per_hour: f64,
+    /// Thread reads (the root's newest 100 replies) each author makes per
+    /// hour through HTTP `/query`.
+    pub fetches_per_hour: f64,
+}
+
+impl HotThread {
+    pub fn enabled(&self) -> bool {
+        self.seed_replies > 0 || self.authors > 0
+    }
+}
+
 #[derive(Clone, Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct RampStep {
@@ -271,6 +296,14 @@ 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),
+            (
+                "hot_thread.replies_per_hour",
+                self.hot_thread.replies_per_hour,
+            ),
+            (
+                "hot_thread.fetches_per_hour",
+                self.hot_thread.fetches_per_hour,
+            ),
         ];
         for (name, value) in non_negative {
             if !value.is_finite() || value < 0.0 {
diff --git a/crates/buzz-test-client/src/loadgen/seed.rs b/crates/buzz-test-client/src/loadgen/seed.rs
index 4ff8af2158..61379bc27d 100644
--- a/crates/buzz-test-client/src/loadgen/seed.rs
+++ b/crates/buzz-test-client/src/loadgen/seed.rs
@@ -16,6 +16,13 @@ use super::stats::{Op, Snapshot, Stats};
 
 /// Content prefix of seeded thread roots; `run` finds roots by it.
 pub const ROOT_PREFIX: &str = "lg-root";
+/// Content prefix of the seeded hot-thread root; `run` finds it by this.
+pub const HOT_ROOT_PREFIX: &str = "lg-hot";
+
+/// Author of every hot-thread root in run `run_id`.
+pub fn hot_root_keys(run_id: &str) -> Keys {
+    identity::keys(run_id, Role::HotRoot, 0)
+}
 
 /// Seeds every community. Idempotent for communities this run's fixture
 /// owner owns, and for channels; history and threads are appended on every
@@ -112,6 +119,12 @@ async fn post_all(
 }
 
 async fn provision(scenario: &Scenario, net: &Net, host: &str, owner: &Keys) -> anyhow::Result<()> {
+    let (base, operator) = operator(scenario)?;
+    provision_host(net, base, &operator, host, owner).await
+}
+
+/// The operator API base URL and the operator's keys, for templated communities.
+pub(super) fn operator(scenario: &Scenario) -> anyhow::Result<(&str, Keys)> {
     let base = scenario
         .target
         .operator_url
@@ -123,8 +136,7 @@ async fn provision(scenario: &Scenario, net: &Net, host: &str, owner: &Keys) ->
             scenario.target.operator_key_env
         )
     })?;
-    let operator = Keys::parse(&secret)?;
-    provision_host(net, base, &operator, host, owner).await
+    Ok((base, Keys::parse(&secret)?))
 }
 
 /// Creates `host` owned by `owner`. The request is `create_only`, so the relay
@@ -215,12 +227,20 @@ enum PostKind {
     History,
     Root,
     Reply { root: usize },
+    HotRoot,
+    HotReply,
+}
+
+/// Key of the hot root in [`RootIds`]: the root after channel 0's seeded ones.
+fn hot_key(scenario: &Scenario) -> (usize, usize, usize) {
+    (0, 0, scenario.shape.thread_roots_per_channel)
 }
 
 /// Thread roots, then history and replies (posted once root ids are known).
 /// A root's `seq` is its index within its channel. Each thread's replies form
 /// one chain, so reply `k` sits at depth `k + 1`; threads and history posts
-/// are independent of each other.
+/// are independent of each other. The hot thread's replies all answer its
+/// root directly, so they post concurrently.
 fn plan_posts(scenario: &Scenario) -> (Vec<Chain>, Vec<Chain>) {
     let shape = &scenario.shape;
     let mut roots = Vec::new();
@@ -256,6 +276,23 @@ fn plan_posts(scenario: &Scenario) -> (Vec<Chain>, Vec<Chain>) {
             }
         }
     }
+    if scenario.hot_thread.enabled() {
+        let (community, channel, seq) = hot_key(scenario);
+        roots.push(vec![Post {
+            community,
+            channel,
+            kind: PostKind::HotRoot,
+            seq,
+        }]);
+        rest.extend((0..scenario.hot_thread.seed_replies).map(|seq| {
+            vec![Post {
+                community,
+                channel,
+                kind: PostKind::HotReply,
+                seq,
+            }]
+        }));
+    }
     (roots, rest)
 }
 
@@ -293,6 +330,12 @@ impl Post {
                 tags.extend(reply_tags(root_id, parent));
                 (author, format!("lg-reply {}", self.seq))
             }
+            PostKind::HotRoot => (hot_root_keys(&scenario.run_id), HOT_ROOT_PREFIX.to_string()),
+            PostKind::HotReply => {
+                let root_id = roots.get(&hot_key(scenario))?;
+                tags.extend(reply_tags(root_id, root_id));
+                (author, format!("lg-hot-reply {}", self.seq))
+            }
         };
         let event = EventBuilder::new(Kind::Custom(9), content)
             .tags(tags)
@@ -305,21 +348,23 @@ impl Post {
 /// NIP-10 markers for a reply. `root` always names the thread root, since the
 /// relay rejects a reply whose root differs from its parent's, and `reply`
 /// names the parent, from which the relay derives depth (parent depth + 1).
-fn reply_tags(root: &str, parent: &str) -> [Tag; 2] {
+pub fn reply_tags(root: &str, parent: &str) -> [Tag; 2] {
     [
         tag(["e", root, "", "root"]),
         tag(["e", parent, "", "reply"]),
     ]
 }
 
-/// Up to the 100 newest seeded thread-root ids in `channel`, oldest first.
-/// `run` replies into these; a sample of roots is enough load.
+/// Ids of `owner`'s seeded roots in `channel` whose content starts with
+/// `prefix`, oldest first, among its 100 newest posts there. `run` replies
+/// into these; a sample of roots is enough load.
 pub async fn find_roots(
     net: &Net,
     community: &Community,
     reader: &Keys,
     owner: &Keys,
     channel: &str,
+    prefix: &str,
 ) -> anyhow::Result<Vec<String>> {
     let filter = json!([{
         "kinds": [9],
@@ -343,7 +388,7 @@ pub async fn find_roots(
     let mut roots: Vec<(u64, String)> = events
         .into_iter()
         .filter(|e| {
-            e.content.starts_with(ROOT_PREFIX) && !e.tags.iter().any(|t| t.kind().as_str() == "e")
+            e.content.starts_with(prefix) && !e.tags.iter().any(|t| t.kind().as_str() == "e")
         })
         .map(|e| (e.created_at.as_secs(), e.id.to_hex()))
         .collect();
@@ -356,13 +401,13 @@ fn tag<const N: usize>(parts: [&str; N]) -> Tag {
 }
 
 #[cfg(test)]
-mod tests {
+pub(super) mod tests {
     use super::*;
     use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 
     /// Serves HTTP/1.1 on loopback. `handler(method, target, body)` returns
     /// the status and JSON body of each response.
-    async fn fake_http<F>(handler: F) -> String
+    pub(in crate::loadgen) async fn fake_http<F>(handler: F) -> String
     where
         F: Fn(&str, &str, &[u8]) -> (u16, String) + Send + Sync + 'static,
     {
@@ -453,6 +498,12 @@ mod tests {
     }
 
     async fn seed_against(reject: &str) -> (anyhow::Result<()>, Arc<Stats>, Accepted) {
+        seed_with(reject, "").await
+    }
+
+    /// Seeds two history posts and two chains of three replies, plus `extra`
+    /// scenario TOML, against a relay that rejects content `reject`.
+    async fn seed_with(reject: &str, extra: &str) -> (anyhow::Result<()>, Arc<Stats>, Accepted) {
         let (addr, accepted) = fake_relay(reject.into()).await;
         let scenario = Scenario::parse(&format!(
             r#"
@@ -473,6 +524,7 @@ channels_per_member = 1
 [[ramp]]
 users = 1.0
 hold_secs = 1
+{extra}
 "#
         ))
         .unwrap();
@@ -519,6 +571,39 @@ hold_secs = 1
         assert_eq!(depths, [1, 1, 2, 2, 3, 3]);
     }
 
+    #[tokio::test]
+    async fn the_hot_thread_seeds_direct_replies_under_its_own_root() {
+        let hot = "[hot_thread]\nseed_replies = 4";
+        let (outcome, stats, accepted) = seed_with("none", hot).await;
+        outcome.expect("every post accepted");
+        // The base 10 posts plus the hot root and its 4 replies.
+        assert_eq!(seeded(&stats.snapshot(), &Snapshot::default()), 15);
+        let hot_depths: Vec<usize> = {
+            let accepted = accepted.lock().unwrap();
+            let roots: Vec<&String> = accepted
+                .iter()
+                .filter(|(_, (_, depth))| *depth == 0)
+                .map(|(id, _)| id)
+                .collect();
+            // Every reply under the most-replied root sits directly on it.
+            let hot_root = roots
+                .iter()
+                .max_by_key(|root| accepted.values().filter(|(r, _)| r == **root).count())
+                .unwrap();
+            accepted
+                .iter()
+                .filter(|(id, (root, _))| root == *hot_root && id != hot_root)
+                .map(|(_, (_, depth))| *depth)
+                .collect()
+        };
+        assert_eq!(hot_depths, [1, 1, 1, 1]);
+
+        // A rejected hot root leaves its replies unpostable.
+        let (outcome, _, _) = seed_with(HOT_ROOT_PREFIX, hot).await;
+        let err = outcome.expect_err("a rejected hot root fails the seed");
+        assert!(err.to_string().contains("seeded 10 of 15"), "{err}");
+    }
+
     #[test]
     fn reply_k_points_reply_at_reply_k_minus_one_and_root_at_the_root() {
         let scenario = Scenario::parse(
diff --git a/crates/buzz-test-client/src/loadgen/stats.rs b/crates/buzz-test-client/src/loadgen/stats.rs
index 08c2a47c9d..62e3f43234 100644
--- a/crates/buzz-test-client/src/loadgen/stats.rs
+++ b/crates/buzz-test-client/src/loadgen/stats.rs
@@ -19,10 +19,12 @@ pub enum Op {
     HttpEvents,
     Seed,
     Subscribe,
+    HotReply,
+    ThreadFetch,
 }
 
 impl Op {
-    pub const ALL: [Op; 9] = [
+    pub const ALL: [Op; 11] = [
         Op::WsConnect,
         Op::History,
         Op::PublishEphemeral,
@@ -32,6 +34,8 @@ impl Op {
         Op::HttpEvents,
         Op::Seed,
         Op::Subscribe,
+        Op::HotReply,
+        Op::ThreadFetch,
     ];
 
     pub fn name(self) -> &'static str {
@@ -45,6 +49,8 @@ impl Op {
             Op::HttpEvents => "http_events",
             Op::Seed => "seed",
             Op::Subscribe => "subscribe_eose",
+            Op::HotReply => "hot_reply",
+            Op::ThreadFetch => "thread_fetch",
         }
     }
 }
diff --git a/crates/buzz-test-client/src/loadgen/teardown.rs b/crates/buzz-test-client/src/loadgen/teardown.rs
new file mode 100644
index 0000000000..8b7f890419
--- /dev/null
+++ b/crates/buzz-test-client/src/loadgen/teardown.rs
@@ -0,0 +1,345 @@
+//! Archives and requests deletion of the communities `seed` provisioned.
+
+use std::collections::BTreeMap;
+
+use futures_util::stream::{self, StreamExt};
+use nostr::Keys;
+use reqwest::StatusCode;
+use serde_json::json;
+
+use super::identity::{self, Role};
+use super::net::{http_reason, Net};
+use super::scenario::Scenario;
+use super::seed;
+
+/// The acknowledgement version the relay accepts for owner deletions.
+const ACKNOWLEDGEMENT_VERSION: i32 = 1;
+
+/// What `teardown` did across the scenario's communities.
+#[derive(Debug, Default)]
+pub struct Report {
+    /// Deletion requests by the stage the relay reported.
+    pub requested: BTreeMap<String, usize>,
+    /// Communities with neither this run's fixture owner nor this run's
+    /// deletion request: never seeded, or another owner's host.
+    pub absent: usize,
+    /// One reason per community whose archive or deletion request failed.
+    pub failures: Vec<String>,
+}
+
+impl Report {
+    pub fn to_json(&self) -> serde_json::Value {
+        json!({
+            "requested": self.requested,
+            "absent": self.absent,
+            "failed": self.failures.len(),
+            "failures": self.failures.iter().take(10).collect::<Vec<_>>(),
+        })
+    }
+}
+
+enum Outcome {
+    Requested(String),
+    Absent,
+}
+
+/// Archives every templated community of the scenario as its fixture owner,
+/// then requests its deletion under a request id derived from `run_id` and
+/// the host, so a re-run replays the same requests. Deletion itself runs
+/// asynchronously on the relay's deletion drain.
+pub async fn teardown(scenario: &Scenario, net: &Net) -> anyhow::Result<Report> {
+    if scenario.target.community_host_template.is_none() {
+        anyhow::bail!(
+            "teardown deletes only communities from target.community_host_template; \
+             this scenario targets {} directly",
+            scenario.target.http_url
+        );
+    }
+    let (base, operator) = seed::operator(scenario)?;
+    Ok(teardown_with(scenario, net, base, &operator).await)
+}
+
+async fn teardown_with(scenario: &Scenario, net: &Net, base: &str, operator: &Keys) -> Report {
+    let outcomes: Vec<Result<Outcome, String>> = stream::iter(scenario.communities())
+        .filter_map(|community| async move { Some((community.index, community.host?)) })
+        .map(|(index, host)| async move {
+            teardown_community(net, base, operator, &scenario.run_id, index, &host).await
+        })
+        .buffer_unordered(scenario.shape.seed_concurrency)
+        .collect()
+        .await;
+    let mut report = Report::default();
+    for outcome in outcomes {
+        match outcome {
+            Ok(Outcome::Requested(stage)) => *report.requested.entry(stage).or_default() += 1,
+            Ok(Outcome::Absent) => report.absent += 1,
+            Err(reason) => report.failures.push(reason),
+        }
+    }
+    report
+}
+
+async fn teardown_community(
+    net: &Net,
+    base: &str,
+    operator: &Keys,
+    run_id: &str,
+    index: usize,
+    host: &str,
+) -> Result<Outcome, String> {
+    let owner = identity::keys(run_id, Role::Owner, index)
+        .public_key()
+        .to_hex();
+    let post = |path: &'static str, body: serde_json::Value| async move {
+        let body = serde_json::to_vec(&body).map_err(|e| e.to_string())?;
+        net.post(operator, base, path, body, None)
+            .await
+            .map_err(|e| format!("{path} {host}: {e}"))
+    };
+
+    let (status, text) = post(
+        "/operator/communities/archive",
+        json!({ "host": host, "owner_pubkey": owner }),
+    )
+    .await?;
+    // Not found means this owner holds no live community here. A deletion
+    // already past membership purge looks the same, so replay it below.
+    if !status.is_success() && status != StatusCode::NOT_FOUND {
+        return Err(format!("archive {host}: {}", http_reason(status, &text)));
+    }
+
+    let (status, text) = post(
+        "/operator/communities/delete",
+        json!({
+            "host": host,
+            "owner_pubkey": owner,
+            "request_id": identity::teardown_request_id(run_id, host).to_string(),
+            "acknowledgement_version": ACKNOWLEDGEMENT_VERSION,
+        }),
+    )
+    .await?;
+    if status == StatusCode::NOT_FOUND {
+        return Ok(Outcome::Absent);
+    }
+    if !status.is_success() {
+        return Err(format!("delete {host}: {}", http_reason(status, &text)));
+    }
+    let accepted: serde_json::Value = serde_json::from_str(&text)
+        .map_err(|e| format!("delete {host}: unreadable response: {e}"))?;
+    let stage = accepted["status"].as_str().unwrap_or("unknown").to_string();
+    Ok(Outcome::Requested(stage))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::loadgen::seed::tests::fake_http;
+    use std::collections::HashMap;
+    use std::sync::{Arc, Mutex};
+
+    #[derive(Debug, Default)]
+    struct Community {
+        /// Empty once deletion has purged membership.
+        owner: String,
+        archived: bool,
+        /// Request id and stage of the community's deletion request.
+        deletion: Option<(String, &'static str)>,
+    }
+
+    /// The operator's archive and delete endpoints: owner-checked, archive
+    /// before delete, and a request id replayed for its community at any
+    /// stage, even after membership purge. A request id already used for
+    /// another host, or a second id for one community, conflicts.
+    /// `unavailable` answers every archive with 503.
+    async fn fake_operator(
+        communities: Arc<Mutex<HashMap<String, Community>>>,
+        unavailable: &'static str,
+    ) -> String {
+        fake_http(move |_, target, body| {
+            let request: serde_json::Value = serde_json::from_slice(body).unwrap();
+            let host = request["host"].as_str().unwrap();
+            let owner = request["owner_pubkey"].as_str().unwrap();
+            let mut communities = communities.lock().unwrap();
+            let not_found = (404, json!({"error": "community not found"}).to_string());
+            match target {
+                "/operator/communities/archive" => {
+                    let Some(community) = communities.get_mut(host).filter(|c| c.owner == owner)
+                    else {
+                        return not_found;
+                    };
+                    if host == unavailable {
+                        return (503, json!({"error": "propagation pending"}).to_string());
+                    }
+                    community.archived = true;
+                    (200, json!({"status": "archived"}).to_string())
+                }
+                "/operator/communities/delete" => {
+                    if request["acknowledgement_version"] != 1 {
+                        return (400, json!({"error": "unsupported"}).to_string());
+                    }
+                    let id = request["request_id"].as_str().unwrap();
+                    let known = communities.iter().find_map(|(h, c)| {
+                        c.deletion
+                            .as_ref()
+                            .filter(|(known, _)| known == id)
+                            .map(|(_, stage)| (h.as_str(), *stage))
+                    });
+                    match known {
+                        Some((h, stage)) if h == host => {
+                            return (202, json!({ "status": stage }).to_string());
+                        }
+                        Some(_) => {
+                            return (
+                                409,
+                                json!({"code": "deletion_request_conflict"}).to_string(),
+                            );
+                        }
+                        None => {}
+                    }
+                    let Some(community) = communities.get_mut(host).filter(|c| c.owner == owner)
+                    else {
+                        return not_found;
+                    };
+                    if !community.archived {
+                        return (409, json!({"code": "community_not_archived"}).to_string());
+                    }
+                    if community.deletion.is_some() {
+                        return (
+                            409,
+                            json!({"code": "deletion_request_conflict"}).to_string(),
+                        );
+                    }
+                    community.deletion = Some((id.to_string(), "submitted"));
+                    (202, json!({"status": "submitted"}).to_string())
+                }
+                other => panic!("unexpected {other}"),
+            }
+        })
+        .await
+    }
+
+    fn scenario(addr: &str, template: Option<&str>) -> Scenario {
+        let template = template
+            .map(|t| format!("community_host_template = \"{t}\"\noperator_url = \"http://{addr}\""))
+            .unwrap_or_default();
+        Scenario::parse(&format!(
+            r#"
+run_id = "t"
+[target]
+ws_url = "ws://{addr}"
+http_url = "http://{addr}"
+{template}
+[shape]
+communities = {}
+channels_per_community = 1
+[population]
+humans = 1
+agents = 0
+channels_per_member = 1
+[[ramp]]
+users = 1.0
+hold_secs = 1
+"#,
+            if template.is_empty() { 1 } else { 5 }
+        ))
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn teardown_deletes_only_the_fixture_owners_communities_and_replays_on_rerun() {
+        let owner = |i| identity::keys("t", Role::Owner, i).public_key().to_hex();
+        let communities = Arc::new(Mutex::new(HashMap::from([
+            (
+                "lg-t-0.example".to_string(),
+                Community {
+                    owner: owner(0),
+                    ..Default::default()
+                },
+            ),
+            (
+                // A host this run does not own is left alone.
+                "lg-t-1.example".to_string(),
+                Community {
+                    owner: Keys::generate().public_key().to_hex(),
+                    ..Default::default()
+                },
+            ),
+            // lg-t-2 was never seeded.
+            (
+                "lg-t-3.example".to_string(),
+                Community {
+                    owner: owner(3),
+                    ..Default::default()
+                },
+            ),
+            (
+                // Deletion already purged membership, so archive finds no
+                // owner; the request is still replayed for its stage.
+                "lg-t-4.example".to_string(),
+                Community {
+                    archived: true,
+                    deletion: Some((
+                        identity::teardown_request_id("t", "lg-t-4.example").to_string(),
+                        "postgres_purged",
+                    )),
+                    ..Default::default()
+                },
+            ),
+        ])));
+        let addr = fake_operator(communities.clone(), "lg-t-3.example").await;
+        let fixture = scenario(&addr, Some("lg-{run}-{i}.example"));
+        let net = Net::new(&fixture).await.unwrap();
+        let base = format!("http://{addr}");
+        let operator = Keys::generate();
+
+        for pass in ["first", "re-run"] {
+            let report = teardown_with(&fixture, &net, &base, &operator).await;
+            assert_eq!(
+                report.requested,
+                BTreeMap::from([("postgres_purged".into(), 1), ("submitted".into(), 1)]),
+                "{pass}"
+            );
+            assert_eq!(report.absent, 2, "{pass}");
+            assert_eq!(report.failures.len(), 1, "{pass}: {:?}", report.failures);
+            assert!(
+                report.failures[0].starts_with("archive lg-t-3.example: HTTP 503"),
+                "{pass}: {:?}",
+                report.failures
+            );
+        }
+        // The same run under another host template names another owner's
+        // host; it gets its own request id and is left alone.
+        communities.lock().unwrap().insert(
+            "other-0.example".to_string(),
+            Community {
+                owner: Keys::generate().public_key().to_hex(),
+                ..Default::default()
+            },
+        );
+        let other = scenario(&addr, Some("other-{i}.example"));
+        let report = teardown_with(&other, &net, &base, &operator).await;
+        assert!(report.failures.is_empty(), "{:?}", report.failures);
+        assert_eq!(report.absent, 5);
+
+        let communities = communities.lock().unwrap();
+        assert!(!communities["other-0.example"].archived);
+        assert_eq!(
+            communities["lg-t-0.example"].deletion,
+            Some((
+                identity::teardown_request_id("t", "lg-t-0.example").to_string(),
+                "submitted"
+            ))
+        );
+        assert!(!communities["lg-t-1.example"].archived);
+        assert!(communities["lg-t-1.example"].deletion.is_none());
+        assert!(communities["lg-t-3.example"].deletion.is_none());
+    }
+
+    #[tokio::test]
+    async fn teardown_refuses_a_scenario_without_templated_communities() {
+        let scenario = scenario("127.0.0.1:9", None);
+        let net = Net::new(&scenario).await.unwrap();
+        let err = teardown(&scenario, &net).await.unwrap_err();
+        assert!(err.to_string().contains("community_host_template"), "{err}");
+    }
+}
diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md
index 4fd6e39a8c..86288ac44b 100644
--- a/perf/LOADGEN.md
+++ b/perf/LOADGEN.md
@@ -8,6 +8,7 @@ per operation for each step.
 ```bash
 cargo run --release -p buzz-test-client --bin buzz-loadgen -- seed perf/scenarios/local-smoke.toml
 cargo run --release -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios/local-smoke.toml
+cargo run --release -p buzz-test-client --bin buzz-loadgen -- teardown perf/scenarios/<templated>.toml
 ```
 
 ## Scale vectors
@@ -18,6 +19,7 @@ cargo run --release -p buzz-test-client --bin buzz-loadgen -- run perf/scenarios
 | Channels | `shape.channels_per_community`, `population.channels_per_member` |
 | History depth | `shape.history_messages_per_channel`, `human.history_limit` |
 | Deep threads | `shape.thread_roots_per_channel`, `shape.thread_depth`, `human.thread_reply_fraction` |
+| One hot thread | `[hot_thread]`: `seed_replies`, `authors`, `replies_per_hour`, `fetches_per_hour` |
 | Humans / agents | `population.humans`, `population.agents`, multiplied by each `[[ramp]]` step's `users` |
 | Throughput per client | `[human]` and `[agent]` rates |
 | Reconnect churn | `human.mean_session_secs`, `population.connect_rate_per_sec` |
@@ -32,6 +34,14 @@ example `seed_concurrency` of 0, `human.presence_interval_secs` below 1, or a
 `community_host_template` with a port while `target.connect_addr` is set (the
 override maps hostnames, so HTTP would ignore it).
 
+`[hot_thread]` models one thread that keeps growing, such as a conversation
+between busy agents. `seed` posts its root in community 0, channel 0 and
+then `seed_replies` replies directly under it, concurrently. During `run`,
+`authors` clients (multiplied by each ramp step's `users`, like humans and
+agents) reply to that root and read its newest 100 replies over HTTP. Their
+replies are timed, so humans subscribed to that channel record delivery.
+Leave `seed_replies` and `authors` at 0 to disable it.
+
 ## What each client does
 
 - **Human**: authenticates over WebSocket and subscribes to its channels'
@@ -42,6 +52,10 @@ override maps hostnames, so HTTP would ignore it).
 - **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.
+- **Hot-thread author**: finds the seeded hot root, then posts replies to it
+  through HTTP `/events` (`hot_reply`) and reads the thread through HTTP
+  `/query` (`thread_fetch`). Until the root is found every attempt counts as
+  a `thread_fetch` error, so a run without the seeded thread fails.
 
 Messages carry their send time, so every subscriber records publish→delivery
 latency (`deliver`). Across machines this depends on clock sync.
@@ -126,6 +140,25 @@ fixture owner derived from `run_id` owns it (re-running `seed` for the same
 scenario); otherwise `seed` fails. Use a fresh `run_id` or a host template
 with `{run}` to avoid colliding with another community.
 
+## Teardown
+
+`teardown <scenario>` removes the communities `seed` provisioned from
+`target.community_host_template`, using the operator key `seed` used. For each
+host it archives the community as its fixture owner, then requests owner
+deletion with a request id derived from `run_id` and the host. The
+relay's deletion drain does the deletion asynchronously; `teardown` reports
+the requests by the stage the relay returned (`requested`), hosts this run's
+fixture owner does not own (`absent`), and failures. It never touches a host
+owned by anyone else, and refuses a scenario without a host template.
+
+Re-running `teardown` replays the same request ids, so it retries failures
+and reports progress without conflicting. It exits non-zero while any
+community failed.
+
+Deleted hosts stay reserved as tombstones, so a torn-down `run_id` cannot be
+seeded again: `seed` fails on the first host. Give each fixture a new
+`run_id`.
+
 ## Safety
 
 The generator creates real communities, channels and events. Point it only at
diff --git a/perf/scenarios/local-smoke.toml b/perf/scenarios/local-smoke.toml
index d52362aefe..3364fef3fd 100644
--- a/perf/scenarios/local-smoke.toml
+++ b/perf/scenarios/local-smoke.toml
@@ -32,6 +32,13 @@ query_interval_secs = 2.0
 events_per_hour = 360.0
 observer_frames_per_hour = 360.0
 
+# One thread that keeps growing while the population runs.
+[hot_thread]
+seed_replies = 200
+authors = 3
+replies_per_hour = 1800.0
+fetches_per_hour = 360.0
+
 [[ramp]]
 users = 1.0
 hold_secs = 30