Merge pull request 'feat(sidecar): protocol 7 — the Event System's command plane (Phase 16b cutover, 1 of 6)' (#40) from edge into main

Reviewed-on: #40
Reviewed-by: Colby Whitlock <whitlocktech@gmail.com>

.gitea/workflows/pr-checks.yml

@@ -17,10 +17,12 @@
 #   so let this workflow run on one PR first. The `PR Checks / *` glob matches
 #   without needing the dropdown.
 #
-# Scope note: this gates PRs into `main` only. Feature work that lands on an
-# integration branch first (e.g. `edge`) is still caught on the branch's PR into
-# `main`. To gate that earlier hop too, add the branch to the `branches:` list
-# below — nothing else needs to change.
+# Scope note: `edge` is gated as well as `main`. Multi-phase work lands there
+# first, so gating only the `main` hop would run these checks for the first time
+# at the cutover — the one moment a red build is most expensive to discover. This
+# is the same call `RunicGateway/installer` made for the same reason, and it was
+# taken here after a nine-PR Android workstream landed on an ungated `edge` with
+# no CI at all. Adding a branch to the `branches:` list is the whole change.
 #
 # Runner: the same self-hosted `ubuntu-latest` runner release.yml uses. Rust is
 # not assumed to be preinstalled, so the toolchain step bootstraps it the same
@@ -31,7 +33,7 @@ name: PR Checks
 
 on:
   pull_request:
-    branches: [main]
+    branches: [main, edge]
 
 # A newer push to the same PR cancels the in-flight run.
 concurrency:

.gitea/workflows/release.yml

@@ -149,6 +149,39 @@ jobs:
             fi
           fi
 
+          # ── Orphan sweep ────────────────────────────────────────────────
+          #
+          # The check above is VERSION-SCOPED: it only ever asks about the one
+          # version this run computed. That is enough to recover an orphan on
+          # the very next run, and useless afterwards — once any releasable
+          # commit lands, the next run computes a NEW version, never looks at
+          # the old tag again, and the orphan becomes permanent and silent.
+          #
+          # servuo-plugins v0.1.0 is the proof, and the proof is pointed: the
+          # commit that ADDED the recovery above was itself typed
+          # `fix(release): ... recover the orphaned v0.1.0 tag`, so it bumped to
+          # v0.1.1 — and the run that introduced the recovery stepped straight
+          # past the tag it was written to rescue. That tag is still orphaned.
+          #
+          # So every v* tag is checked, and anything missing a release is
+          # WARNED about. Deliberately not recovered: publishing an old version
+          # would mean building today's tree and shipping it under a tag whose
+          # tree it is not, which is worse than the inconsistency it fixes.
+          # A human decides whether to recover or drop it.
+          #
+          # Never fails the run. A sweep that can break a good release is a
+          # sweep someone will delete.
+          ORPHANS=""
+          for T in $(git tag -l 'v*' --sort=-v:refname); do
+            T_HTTP="$(curl -s -o /dev/null -w '%{http_code}' \
+              -H "Authorization: token $(printf '%s' "${REGISTRY_TOKEN:-}" | tr -d '\r\n')" \
+              "https://${GITEA_HOST}/api/v1/repos/${REPO}/releases/tags/${T}" || echo 000)"
+            [ "$T_HTTP" = "404" ] && ORPHANS="${ORPHANS} ${T}"
+          done
+          if [ -n "${ORPHANS}" ]; then
+            echo "::warning::Tags with no release:${ORPHANS} — a run failed after tagging. Publish or delete them; this job will not do either."
+          fi
+
           # Changelog range. A recovery run has nothing after the tag, so
           # summarize what the tag itself contains rather than emitting an empty
           # list: the range that produced it, i.e. previous-tag..this-tag.
@@ -342,18 +375,75 @@ jobs:
           # corrupt the Authorization header.
           CI_TOKEN="$(printf '%s' "${REGISTRY_TOKEN}" | tr -d '\r\n')"
 
-          REL_ID="$(curl -sSf -X POST "${API}/releases" \
-            -H "Authorization: token ${CI_TOKEN}" \
-            -H "Content-Type: application/json" \
-            -d "$(jq -n --arg tag "$TAG" --arg body "$BODY" \
-                   '{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')" \
-            | jq -r '.id')"
+          PAYLOAD="$(jq -n --arg tag "$TAG" --arg body "$BODY" \
+                       '{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')"
+
+          # This POST is the step that orphaned tag v0.1.1 (run 75): it landed one
+          # second after the tag push and Gitea answered 500, having not finished
+          # processing the pushed tag. Re-running the workflow published the same
+          # four assets untouched, so the failure was a race, not a bad request.
+          #
+          # Two things went wrong there, and both are fixed here.
+          #
+          # 1. `curl -sSf` prints NO response body on an error status, so all the
+          #    log carried was "curl: (22) ... error: 500" and the cause had to be
+          #    inferred from timestamps. Capture the body and print it.
+          # 2. Nothing retried, so a transient 5xx became a permanent orphan tag.
+          #    The plan step CAN recover one, but only on a run that reaches it --
+          #    and a later push with no releasable commits stands down before it
+          #    gets there, so in practice the tag sits until a human notices.
+          #
+          # 4xx is deliberately NOT retried: a bad token or a malformed body does
+          # not improve by being sent again, and retrying only turns a clear
+          # failure into a slow one.
+          REL_ID=""
+          for attempt in 1 2 3 4 5; do
+            HTTP="$(curl -s -o /tmp/rel.json -w '%{http_code}' -X POST "${API}/releases" \
+              -H "Authorization: token ${CI_TOKEN}" \
+              -H "Content-Type: application/json" \
+              -d "${PAYLOAD}" || echo 000)"
+
+            if [ "$HTTP" = "201" ] || [ "$HTTP" = "200" ]; then
+              REL_ID="$(jq -r '.id' /tmp/rel.json)"
+              break
+            fi
+
+            echo "::warning::POST /releases attempt ${attempt} returned HTTP ${HTTP}"
+            echo "--- response body ---"
+            cat /tmp/rel.json || true
+            echo
+            echo "---------------------"
+
+            case "$HTTP" in
+              4*) echo "::error::HTTP ${HTTP} is a client error - not retrying."; exit 1 ;;
+            esac
+
+            if [ "$attempt" = 5 ]; then
+              echo "::error::POST /releases still failing after 5 attempts. Tag ${TAG} is pushed but has no release."
+              echo "::error::Re-run this workflow - the plan step detects the orphan tag and republishes it."
+              exit 1
+            fi
+            sleep $(( attempt * 5 ))
+          done
+
+          if [ -z "$REL_ID" ] || [ "$REL_ID" = "null" ]; then
+            echo "::error::Release created but no id came back; refusing to upload assets blind."
+            exit 1
+          fi
           echo "Created release ${TAG} (id=${REL_ID})"
 
           for f in "${BIN}-linux-x86_64" "${BIN}-linux-aarch64" "${BIN}-windows-x86_64.exe" SHA256SUMS; do
-            curl -sSf -X POST "${API}/releases/${REL_ID}/assets?name=${f}" \
+            # Same treatment. An upload that fails quietly leaves a release whose
+            # SHA256SUMS does not cover every binary it advertises, which is worse
+            # than no release at all -- that file IS the trust anchor.
+            HTTP="$(curl -s -o /tmp/asset.json -w '%{http_code}' -X POST "${API}/releases/${REL_ID}/assets?name=${f}" \
               -H "Authorization: token ${CI_TOKEN}" \
-              -F "attachment=@dist/${f}" >/dev/null
+              -F "attachment=@dist/${f}" || echo 000)"
+            if [ "$HTTP" != "201" ] && [ "$HTTP" != "200" ]; then
+              echo "::error::uploading ${f} returned HTTP ${HTTP}"
+              cat /tmp/asset.json || true
+              exit 1
+            fi
             echo "  uploaded ${f}"
           done
 

README.md

@@ -12,11 +12,34 @@ ServUO plugin (C#, net48)  ──loopback TCP, newline-JSON──►  Rust sidec
 The shard never speaks WebSocket and exposes no port of its own — the sidecar is the only
 network-facing component, which is what keeps the game unreachable from the internet.
 
+## Running a shard? Don't build this
+
+The [**Runic Gateway installer**](https://gitea.whitlocktech.com/RunicGateway/installer) installs
+this sidecar for you — the released binary, its config, a hardened service account and the service
+registration — alongside the shard plugin, in one run, on Linux or Windows:
+
+```bash
+sudo ./runicgateway-installer-linux-x86_64 install
+```
+
+It ends by printing the base URL, WebSocket URL, protocol version and auth token to paste into
+**Admin → Shard** on your site. Guide:
+[installer/INSTALL.md](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/installer/INSTALL.md).
+
+Installing it yourself is supported too — the release binaries on this repo's
+[releases page](https://gitea.whitlocktech.com/RunicGateway/link/releases) are the same ones the
+installer fetches, and
+[INSTALL.md Appendix A3–A4](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/installer/INSTALL.md#a3-install-the-sidecar)
+covers placing the binary and registering the service by hand.
+
+Everything below this line is for **developing on the sidecar**.
+
 ## Related repos
 
 | Repo | What |
 |------|------|
 | **this** — `RunicGateway/link` | The Rust sidecar (`sidecar/`). |
+| [RunicGateway/installer](https://gitea.whitlocktech.com/RunicGateway/installer) | The **installer** — deploys this sidecar and the plugin onto a shard host. The supported way to set one up. |
 | [RunicGateway/servuo-plugins](https://gitea.whitlocktech.com/RunicGateway/servuo-plugins) | The **C# ServUO plugin** — the shard side of the bridge (`overlay/`, `patches/`, `deploy.ps1`, test scaffolding). |
 | [RunicGateway/docs](https://gitea.whitlocktech.com/RunicGateway/docs) | All project documentation — design docs, protocol spec, integration guide, research. |
 
@@ -28,9 +51,10 @@ network-facing component, which is what keeps the game unreachable from the inte
 | `.gitea/workflows/pr-checks.yml` | Gates every PR into `main` on `cargo fmt --check`, `cargo clippy -D warnings`, and `cargo test`. |
 | `.gitea/workflows/release.yml` | Builds + releases the sidecar binary (Linux + Windows) on every merge to `main`. |
 
-## Build & run
+## Build & run (development)
 
-The sidecar is a standard cargo crate:
+Building from source is for working *on* the sidecar; a deployment gets its binary from a release,
+via the installer or by hand. The sidecar is a standard cargo crate:
 
 ```bash
 cd sidecar
@@ -39,7 +63,7 @@ cp sidecar.toml.example sidecar.toml   # then edit
 cargo run --release
 ```
 
-Deploying it rather than developing on it: `--config <PATH>` names the config file (as does
+Deploying it by hand rather than developing on it: `--config <PATH>` names the config file (as does
 `$UOLINK_CONFIG`), and `--print-config` prints the resolved settings — **including the auth token
 the website needs** — as JSON, provisioning the config file on first run. That is the supported way
 to read the token back; it is not meant to be scraped from the log.

sidecar/src/app.rs

@@ -14,11 +14,13 @@
 //! - `shutdown` is whatever "stop" means on this host: Ctrl-C and `SIGTERM` on Unix, the SCM's
 //!   `Stop` control on Windows.
 
+use std::collections::HashMap;
 use std::future::Future;
 use std::sync::atomic::AtomicI64;
 use std::sync::Arc;
 use std::time::Instant;
 
+use serde_json::Value;
 use tokio::sync::{broadcast, mpsc};
 use tracing::info;
 
@@ -88,6 +90,9 @@ where
     let last_event_ts = last_event.clone();
     let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
     let mut total: u64 = 0;
+    // Partly-received guild rosters, keyed by guild id. Lives in the event-loop task, so it needs
+    // no lock and dies with the loop. See `accumulate_roster`.
+    let mut roster_parts: HashMap<i64, RosterParts> = HashMap::new();
     tokio::spawn(async move {
         while let Some(ev) = event_rx.recv().await {
             // Any line from the shard — including pong heartbeats — is a sign of life.
@@ -161,6 +166,43 @@ where
                             }
                         }
                     }
+                    // Guild roster (Protocol 4): the member list that `guild.update`'s counts cannot
+                    // express. It writes a *different column* of the same row, so it never races
+                    // guild.update. `guild.leave` deliberately has no arm here — the sidecar
+                    // forwards it (persisted and broadcast below, like any event) and the board's
+                    // roster self-corrects on the next `guild.roster`, which the shard re-emits
+                    // whenever the member set changes. Keeping the delta out of the board is what
+                    // keeps the sidecar a forwarder rather than a thing that maintains state.
+                    //
+                    // A roster over the shard's per-line cap arrives as several frames, so it is
+                    // reassembled before it is stored — see `accumulate_roster` for why that happens
+                    // here rather than by appending to the column.
+                    "guild.roster" => {
+                        if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
+                            let seq = ev.value.get("seq").and_then(|v| v.as_i64()).unwrap_or(0);
+                            let more = ev
+                                .value
+                                .get("more")
+                                .and_then(|v| v.as_bool())
+                                .unwrap_or(false);
+                            let members = ev
+                                .value
+                                .get("members")
+                                .and_then(|m| m.as_array())
+                                .cloned()
+                                .unwrap_or_default();
+
+                            if let Some(complete) =
+                                accumulate_roster(&mut roster_parts, id, seq, more, members)
+                            {
+                                let json = serde_json::Value::Array(complete).to_string();
+                                if let Err(e) = event_store.upsert_guild_roster(id, &json, t).await
+                                {
+                                    tracing::warn!(error = %e, "failed to upsert guild roster");
+                                }
+                            }
+                        }
+                    }
                     "guild.remove" => {
                         if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
                             if let Err(e) = event_store.delete_guild(id).await {
@@ -279,6 +321,10 @@ where
             // with announce=false so a restart does not re-proclaim every article at once. news.add
             // is idempotent by id, so replaying to a still-populated shard is harmless.
             if ev.kind == "server.hello" {
+                // A (re)connected shard restarts every roster from `seq` 0, so any half-received
+                // one belongs to the previous connection and can never be completed.
+                roster_parts.clear();
+
                 match event_store.news_all().await {
                     Ok(items) => {
                         for mut item in items {
@@ -327,3 +373,199 @@ pub fn now_ms() -> i64 {
         .map(|d| d.as_millis() as i64)
         .unwrap_or(0)
 }
+
+/// A guild roster that has arrived in part: the `seq` expected next, and what has accumulated.
+struct RosterParts {
+    next_seq: i64,
+    members: Vec<Value>,
+}
+
+/// Refuses to accumulate a roster past this many members. The shard caps its own frames, so
+/// exceeding this means a shard that is buggy or not what it claims to be — and the one thing this
+/// buffer must not do is grow without bound on its say-so.
+const MAX_ROSTER_MEMBERS: usize = 50_000;
+
+/// Reassembles a `guild.roster` that the shard split across frames, returning the whole member list
+/// once the final frame arrives and `None` while one is still incomplete.
+///
+/// Reassembly happens **here, in memory, before the store** rather than by appending to the
+/// `members` column, for two reasons. Appending would make the write a read-modify-write — the exact
+/// thing the two-column board design exists to avoid — and it would publish a torn roster: a reader
+/// hitting `GET /guilds` between frames would see a partial member list as though it were the truth.
+/// Buffering keeps the store's write a single atomic upsert of a complete roster.
+///
+/// This is transport-level reassembly, not domain state: it is the same category of work as turning
+/// bytes into a line, and it holds nothing once a roster is complete. That is what keeps it
+/// compatible with the sidecar being a forwarder.
+///
+/// The ordinary case — a guild inside the shard's per-line cap, which is every realistic one —
+/// arrives as `seq` 0 with `more` false and is returned immediately without ever touching the map.
+fn accumulate_roster(
+    parts: &mut HashMap<i64, RosterParts>,
+    id: i64,
+    seq: i64,
+    more: bool,
+    members: Vec<Value>,
+) -> Option<Vec<Value>> {
+    if seq == 0 {
+        // A fresh roster supersedes any partial one: the shard restarts at 0 every time it emits,
+        // so a leftover buffer is from an emission that was interrupted and will never finish.
+        parts.remove(&id);
+
+        if !more {
+            return Some(members);
+        }
+
+        parts.insert(
+            id,
+            RosterParts {
+                next_seq: 1,
+                members,
+            },
+        );
+        return None;
+    }
+
+    let entry = match parts.get_mut(&id) {
+        Some(entry) => entry,
+        // A continuation with nothing to continue: the sidecar started, or the shard reconnected,
+        // midway through an emission. Dropping it is right — the next full roster is complete.
+        None => {
+            tracing::debug!(
+                guild = id,
+                seq,
+                "roster continuation with no start; ignoring"
+            );
+            return None;
+        }
+    };
+
+    if entry.next_seq != seq {
+        tracing::warn!(
+            guild = id,
+            expected = entry.next_seq,
+            got = seq,
+            "roster frames out of order; discarding the partial roster"
+        );
+        parts.remove(&id);
+        return None;
+    }
+
+    entry.members.extend(members);
+
+    if entry.members.len() > MAX_ROSTER_MEMBERS {
+        tracing::warn!(
+            guild = id,
+            len = entry.members.len(),
+            "roster exceeded the reassembly cap; discarding"
+        );
+        parts.remove(&id);
+        return None;
+    }
+
+    if more {
+        entry.next_seq = seq + 1;
+        return None;
+    }
+
+    parts.remove(&id).map(|done| done.members)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn members(names: &[&str]) -> Vec<Value> {
+        names
+            .iter()
+            .map(|n| serde_json::json!({"name": n}))
+            .collect()
+    }
+
+    fn names(vs: &[Value]) -> Vec<String> {
+        vs.iter()
+            .map(|v| v["name"].as_str().unwrap_or_default().to_string())
+            .collect()
+    }
+
+    #[test]
+    fn a_single_frame_roster_is_returned_immediately() {
+        // Every realistic guild takes this path, and it must not depend on the buffer at all.
+        let mut parts = HashMap::new();
+        let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Ada", "Bo"]));
+
+        assert_eq!(names(&out.expect("complete")), ["Ada", "Bo"]);
+        assert!(parts.is_empty(), "nothing should be buffered");
+    }
+
+    #[test]
+    fn a_chunked_roster_reassembles_in_order() {
+        // The case the live rig caught: without this, only the final frame survived and a
+        // 155-member guild appeared on the board with 3 members.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
+        assert!(accumulate_roster(&mut parts, 1, 1, true, members(&["Bo"])).is_none());
+        let out = accumulate_roster(&mut parts, 1, 2, false, members(&["Cy"]));
+
+        assert_eq!(names(&out.expect("complete")), ["Ada", "Bo", "Cy"]);
+        assert!(parts.is_empty(), "buffer is released once complete");
+    }
+
+    #[test]
+    fn a_restarted_roster_supersedes_a_partial_one() {
+        // A shard that reconnects mid-emission starts again at seq 0. The abandoned frames must not
+        // end up spliced onto the front of the new roster.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Stale"])).is_none());
+        let out = accumulate_roster(&mut parts, 1, 0, false, members(&["Fresh"]));
+
+        assert_eq!(names(&out.expect("complete")), ["Fresh"]);
+    }
+
+    #[test]
+    fn an_out_of_order_frame_discards_the_partial_roster() {
+        // Better to publish nothing and wait for the next full emission than to store a roster with
+        // a hole in it that nothing downstream could detect.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["Ada"])).is_none());
+        assert!(accumulate_roster(&mut parts, 1, 2, false, members(&["Skipped"])).is_none());
+        assert!(parts.is_empty());
+    }
+
+    #[test]
+    fn a_continuation_with_no_start_is_ignored() {
+        // The sidecar restarting midway through a shard's emission.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 3, false, members(&["Orphan"])).is_none());
+        assert!(parts.is_empty());
+    }
+
+    #[test]
+    fn two_guilds_reassemble_independently() {
+        // Rosters for different guilds interleave freely — the sweep emits one guild after another
+        // and nothing serialises them on the wire.
+        let mut parts = HashMap::new();
+
+        assert!(accumulate_roster(&mut parts, 1, 0, true, members(&["A1"])).is_none());
+        assert!(accumulate_roster(&mut parts, 2, 0, true, members(&["B1"])).is_none());
+        let g2 = accumulate_roster(&mut parts, 2, 1, false, members(&["B2"]));
+        let g1 = accumulate_roster(&mut parts, 1, 1, false, members(&["A2"]));
+
+        assert_eq!(names(&g2.expect("guild 2")), ["B1", "B2"]);
+        assert_eq!(names(&g1.expect("guild 1")), ["A1", "A2"]);
+    }
+
+    #[test]
+    fn an_empty_roster_is_a_complete_roster() {
+        // A guild whose last member left emits one frame with an empty array. Treating that as
+        // "nothing to store" would leave the board showing the roster it had before.
+        let mut parts = HashMap::new();
+        let out = accumulate_roster(&mut parts, 1, 0, false, vec![]);
+
+        assert_eq!(out.expect("complete").len(), 0);
+    }
+}

sidecar/src/main.rs

@@ -46,7 +46,52 @@ use tracing_subscriber::EnvFilter;
 /// `vendor.listing.remove`, with the `GET /ruleset`, `/points` and `/market` reads that serve them
 /// from the store. Same shape as the v2 bump — the kinds are additive, the endpoints are not — and
 /// there is deliberately no feature-negotiation array: v3 implies all three kinds.
-pub const PROTOCOL_VERSION: u32 = 3;
+///
+/// v4 (Protocol 4): adds `guild.roster` and `guild.leave`, giving the guild board a real member list
+/// instead of the member *count* that was all v2 could express. Additive in the same way again — the
+/// kinds are new, `GET /guilds` grows a `roster` key, and nothing existing changed shape. This is the
+/// first bump that also needed a **store migration** (`guilds.members`), because it is the first to
+/// add a column to a table that already exists rather than a whole new table; see `store::migrate`.
+///
+/// v5 (Protocol 5): three enrichments that are additive in the same way again, bumped together
+/// rather than one at a time because a protocol bump is not cheap here — it costs a sidecar
+/// release, a republished bundle and an operator update on every shard, so a field left out costs
+/// a whole second round of that rather than a follow-up commit. They are:
+///
+///   * `house.decay` gains `ownerName` and a decay SCHEDULE — `nextStage`, `decayPeriodSec`,
+///     `dynamicDecay`, and `estimatedCollapse` only where it is exactly knowable (at IDOC under
+///     dynamic decay; at any stage under static decay, which has no randomness to wait out).
+///   * `vendor.listing` gains `ownerAcct` — without which the frame names an owner nobody can
+///     resolve to a person — and a `fees` object carrying the charge, the funds, the pay interval
+///     and the resolved `dismissalAt`.
+///   * `account.login.result` is a NEW kind: the verdict of a login, which the pre-existing
+///     `account.login.attempt` structurally cannot carry (its EventSink fires before the auth
+///     decision is made).
+///
+/// **No store migration this time**, unlike v4. Every frame is persisted whole and the board tables
+/// index only the columns they already had, so the new fields ride inside the stored JSON and the
+/// new kind lands in `events` like any other. That is the dumb-forwarder property doing its job:
+/// the sidecar defines no schema for a frame's contents and so needs no change when they grow.
+///
+/// v6 (Protocol 6): the first bump that is about a GUARANTEE rather than about data, and the first
+/// the sidecar mostly gets for free. Two things:
+///
+///   * **`idempotencyKey` on inbound commands.** A command that carries one is executed by the shard
+///     at most once; a repeat is answered with the original reply rather than re-run. That is what
+///     makes a world-writing verb retryable at all — until now a lost acknowledgement was
+///     indistinguishable from a command that never applied, so the website had to declare every
+///     write un-retryable and accept losing one rather than risk doubling it. The sidecar's part is
+///     to CARRY the key (it rides in the command body, which every write endpoint already passes
+///     through verbatim) and to understand the one new answer the shard can now give: `bridge.busy`,
+///     meaning a command under that key is still in flight. See `web::respond`.
+///   * **`champ.boss.killed` is a new kind**: a champion's defeat, with the damage table only the
+///     shard ever sees. It was previously inferable from `champ.update` going `bossUp` true then
+///     false alongside a nearby `mob.killed`, which is fragile and says nothing about who did the
+///     work. It lands in `events` and on the feed like any other kind, with no code here at all —
+///     the dumb-forwarder property again.
+///
+/// **No store migration.** Nothing gains a column; the new kind is persisted whole like every other.
+pub const PROTOCOL_VERSION: u32 = 7;
 
 // Not `#[tokio::main]`: on Windows the SCM dispatcher takes over this thread and starts the runtime
 // itself, on its own thread, once the service actually begins. The runtime is built by whichever

sidecar/src/store.rs

@@ -45,6 +45,7 @@ impl Store {
             .await?;
 
         sqlx::query(SCHEMA).execute(&pool).await?;
+        migrate(&pool).await?;
         info!(%path, "store ready");
         Ok(Self { pool })
     }
@@ -236,12 +237,61 @@ impl Store {
         Ok(())
     }
 
+    /// Upserts one guild's member roster (Protocol 4), keyed by guild id, touching **only** the
+    /// `members` column.
+    ///
+    /// Deliberately not a write to `json`. That column holds the verbatim `guild.update` line, and a
+    /// roster arriving as its own event must not clobber the snapshot — name, abbreviation, leader,
+    /// online count — that `guild.update` owns. Splitting the two writers across two columns of one
+    /// row is what lets both be plain upserts: neither needs to read the other's value first, so
+    /// there is no read-modify-write and no ordering requirement between the two kinds.
+    ///
+    /// The `INSERT` half is not redundant: a roster can arrive before the first `guild.update` for a
+    /// guild, and the row it creates then carries `'{}'` until that update fills it in.
+    pub async fn upsert_guild_roster(
+        &self,
+        id: i64,
+        members_json: &str,
+        t: i64,
+    ) -> anyhow::Result<()> {
+        sqlx::query(
+            "INSERT INTO guilds (id, name, json, updated_t, members) VALUES (?, NULL, '{}', ?, ?)
+             ON CONFLICT(id) DO UPDATE SET members = excluded.members, updated_t = excluded.updated_t",
+        )
+        .bind(id)
+        .bind(t)
+        .bind(members_json)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
     /// The full guild board: every guild's latest snapshot, ordered by name.
+    ///
+    /// The roster is stored in its own column (see [`Self::upsert_guild_roster`]) and folded into
+    /// the projected object as `roster` here, at read time. A guild that has had a `guild.update`
+    /// but no `guild.roster` yet simply has no `roster` key, which is the honest representation of
+    /// "not known" and distinct from a guild whose roster is genuinely empty.
     pub async fn guilds_all(&self) -> anyhow::Result<Vec<Value>> {
-        let rows = sqlx::query("SELECT json FROM guilds ORDER BY name, id")
+        let rows = sqlx::query("SELECT json, members FROM guilds ORDER BY name, id")
             .fetch_all(&self.pool)
             .await?;
-        Ok(parse_json_column(rows))
+
+        Ok(rows
+            .into_iter()
+            .filter_map(|r| {
+                let mut v: Value = serde_json::from_str(&r.get::<String, _>("json")).ok()?;
+                let members: Option<String> = r.get("members");
+
+                if let (Some(obj), Some(raw)) = (v.as_object_mut(), members) {
+                    if let Ok(list) = serde_json::from_str::<Value>(&raw) {
+                        obj.insert("roster".into(), list);
+                    }
+                }
+
+                Some(v)
+            })
+            .collect())
     }
 
     // ---- governor board (Protocol 2.0) ----
@@ -513,6 +563,75 @@ impl Store {
     }
 }
 
+/// The schema version this build expects. Bump it, and add the matching arm to [`migrate`], for
+/// every change that `SCHEMA` alone cannot make to a database that already exists.
+const SCHEMA_VERSION: i64 = 1;
+
+/// Brings an existing database forward to [`SCHEMA_VERSION`].
+///
+/// `SCHEMA` is `CREATE TABLE IF NOT EXISTS` only, which is enough to *add a table* but cannot add a
+/// column to a table that is already there. Every schema change up to and including Protocol 3.0
+/// happened to add whole tables, so this never mattered and `ALTER TABLE` appears nowhere in this
+/// repo's history. `guilds.members` (Protocol 4) is the first column added to an existing table, so
+/// the mechanism has to exist now.
+///
+/// The version counter is SQLite's own `PRAGMA user_version`: an integer in the database header, so
+/// it needs no table of its own and cannot be separated from the file it describes. Each step runs
+/// in a transaction **together with** the bump that records it, so a step either lands completely or
+/// not at all, and an interrupted run resumes at the right place rather than re-applying half of one.
+///
+/// A failure here propagates and aborts startup, deliberately. A half-migrated store answers the
+/// website with confusing partial data, which is worse than being plainly absent — and the shard
+/// dials *out* to the sidecar, so a sidecar that refuses to start never stalls the game.
+async fn migrate(pool: &SqlitePool) -> anyhow::Result<()> {
+    let mut version: i64 = sqlx::query_scalar("PRAGMA user_version")
+        .fetch_one(pool)
+        .await?;
+
+    // A database written by a *newer* sidecar than this binary. This is not an error: every step
+    // here is additive, so a newer schema has only columns and tables an older reader ignores, and
+    // refusing to start would turn "roll the binary back" — a recovery path — into a dead end.
+    if version > SCHEMA_VERSION {
+        tracing::warn!(
+            found = version,
+            expected = SCHEMA_VERSION,
+            "store was written by a newer sidecar; continuing, as migrations are additive"
+        );
+        return Ok(());
+    }
+
+    while version < SCHEMA_VERSION {
+        let next = version + 1;
+        let mut tx = pool.begin().await?;
+
+        match next {
+            // Protocol 4: the guild board carries a member roster. Its own column rather than a
+            // field folded into `json`, because `json` holds the verbatim `guild.update` line and
+            // the two writers must not overwrite each other — see `upsert_guild_roster`.
+            1 => {
+                sqlx::query("ALTER TABLE guilds ADD COLUMN members TEXT")
+                    .execute(&mut *tx)
+                    .await?;
+            }
+            // Unreachable while SCHEMA_VERSION and this match are edited together, which is the
+            // point of failing loudly rather than silently leaving the counter short.
+            n => anyhow::bail!("no migration step defined for schema version {n}"),
+        }
+
+        // `PRAGMA` takes no bind parameters, so this is formatted — safe because `next` is an i64
+        // this loop produced, never anything from outside the process.
+        sqlx::query(&format!("PRAGMA user_version = {next}"))
+            .execute(&mut *tx)
+            .await?;
+
+        tx.commit().await?;
+        info!(version = next, "schema migration applied");
+        version = next;
+    }
+
+    Ok(())
+}
+
 fn parse_json_column(rows: Vec<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
     rows.into_iter()
         .filter_map(|r| serde_json::from_str(&r.get::<String, _>("json")).ok())
@@ -611,3 +730,195 @@ CREATE TABLE IF NOT EXISTS ruleset (
     updated_t INTEGER NOT NULL
 );
 "#;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A unique scratch database path. Matches `config`'s idiom — `std::env::temp_dir()` plus the
+    /// test name — so the cases stay independent under the parallel test runner.
+    fn scratch(name: &str) -> String {
+        let dir = std::env::temp_dir().join(format!("uo-link-store-test-{name}"));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).expect("create scratch dir");
+        dir.join("uo-link.db").to_string_lossy().into_owned()
+    }
+
+    /// The `guilds` table exactly as a pre-Protocol-4 sidecar left it: no `members` column, and
+    /// `user_version` still 0. This is the shape a real operator's database is in before an update,
+    /// and the only starting point where the migration does anything.
+    async fn legacy_db(path: &str) -> SqlitePool {
+        let pool = SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect_with(
+                SqliteConnectOptions::new()
+                    .filename(path)
+                    .create_if_missing(true),
+            )
+            .await
+            .expect("open legacy db");
+
+        sqlx::query(
+            "CREATE TABLE guilds (
+                 id        INTEGER PRIMARY KEY,
+                 name      TEXT,
+                 json      TEXT NOT NULL,
+                 updated_t INTEGER NOT NULL
+             )",
+        )
+        .execute(&pool)
+        .await
+        .expect("create legacy guilds table");
+
+        pool.close().await;
+        pool
+    }
+
+    async fn user_version(store: &Store) -> i64 {
+        sqlx::query_scalar("PRAGMA user_version")
+            .fetch_one(&store.pool)
+            .await
+            .expect("read user_version")
+    }
+
+    async fn guild_columns(store: &Store) -> Vec<String> {
+        sqlx::query("PRAGMA table_info(guilds)")
+            .fetch_all(&store.pool)
+            .await
+            .expect("table_info")
+            .into_iter()
+            .map(|r| r.get::<String, _>("name"))
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn an_existing_pre_protocol_4_database_gains_the_members_column() {
+        // The case that matters: `SCHEMA`'s CREATE TABLE IF NOT EXISTS is a no-op against a table
+        // that is already there, so without `migrate` this database would never get the column and
+        // every roster write would fail against a live install.
+        let path = scratch("legacy-upgrade");
+        legacy_db(&path).await;
+
+        let store = Store::open(&path).await.expect("open migrates");
+
+        assert!(
+            guild_columns(&store).await.contains(&"members".to_string()),
+            "the migration must add guilds.members to a database that already had the table"
+        );
+        assert_eq!(user_version(&store).await, SCHEMA_VERSION);
+    }
+
+    #[tokio::test]
+    async fn a_fresh_database_lands_at_the_current_version() {
+        let path = scratch("fresh");
+        let store = Store::open(&path).await.expect("open");
+
+        assert!(guild_columns(&store).await.contains(&"members".to_string()));
+        assert_eq!(user_version(&store).await, SCHEMA_VERSION);
+    }
+
+    #[tokio::test]
+    async fn reopening_an_already_migrated_database_is_a_no_op() {
+        // Every sidecar restart re-runs this path, so a second run must not attempt the ALTER again
+        // — which would fail with "duplicate column name" and, since a migration failure aborts
+        // startup, would leave the sidecar unable to start at all after its first upgrade.
+        let path = scratch("idempotent");
+        legacy_db(&path).await;
+
+        Store::open(&path).await.expect("first open");
+        let store = Store::open(&path).await.expect("second open must succeed");
+
+        assert_eq!(user_version(&store).await, SCHEMA_VERSION);
+    }
+
+    #[tokio::test]
+    async fn a_roster_does_not_clobber_the_guild_update_snapshot() {
+        // The invariant the two-column split exists to give. `json` holds the verbatim guild.update
+        // line; if a roster write touched it, name/abbr/online would vanish from the board.
+        let path = scratch("no-clobber");
+        let store = Store::open(&path).await.expect("open");
+
+        store
+            .upsert_guild(
+                7,
+                Some("The Cartographers"),
+                r#"{"kind":"guild.update","id":7,"name":"The Cartographers","abbr":"MAP","members":2,"online":1}"#,
+                100,
+            )
+            .await
+            .expect("upsert guild");
+
+        store
+            .upsert_guild_roster(
+                7,
+                r#"[{"serial":"0x1","name":"Ada"},{"serial":"0x2","name":"Bo"}]"#,
+                200,
+            )
+            .await
+            .expect("upsert roster");
+
+        let guilds = store.guilds_all().await.expect("read board");
+        assert_eq!(guilds.len(), 1);
+        let g = &guilds[0];
+
+        assert_eq!(
+            g["name"], "The Cartographers",
+            "guild.update's name survived"
+        );
+        assert_eq!(g["abbr"], "MAP", "guild.update's abbr survived");
+        assert_eq!(g["online"], 1, "guild.update's online count survived");
+        assert_eq!(g["roster"].as_array().expect("roster is an array").len(), 2);
+        assert_eq!(g["roster"][0]["name"], "Ada");
+    }
+
+    #[tokio::test]
+    async fn the_two_writers_are_order_independent() {
+        // A roster can arrive before the first guild.update for a guild — on a reconnect the shard
+        // re-emits both and nothing orders them. Neither write may depend on the other's row.
+        let path = scratch("either-order");
+        let store = Store::open(&path).await.expect("open");
+
+        store
+            .upsert_guild_roster(9, r#"[{"serial":"0x3","name":"Cy"}]"#, 100)
+            .await
+            .expect("roster first");
+        store
+            .upsert_guild(
+                9,
+                Some("Late Arrivals"),
+                r#"{"kind":"guild.update","id":9,"name":"Late Arrivals","abbr":"LTE"}"#,
+                200,
+            )
+            .await
+            .expect("update second");
+
+        let guilds = store.guilds_all().await.expect("read board");
+        assert_eq!(guilds.len(), 1, "one row, not two");
+        assert_eq!(guilds[0]["name"], "Late Arrivals");
+        assert_eq!(guilds[0]["roster"].as_array().expect("roster").len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_guild_with_no_roster_yet_has_no_roster_key() {
+        // "Not known" and "known to be empty" are different, and the board must not conflate them:
+        // a website reading `roster: []` would render an empty roster as fact.
+        let path = scratch("absent-roster");
+        let store = Store::open(&path).await.expect("open");
+
+        store
+            .upsert_guild(
+                11,
+                Some("Unswept"),
+                r#"{"kind":"guild.update","id":11,"name":"Unswept"}"#,
+                100,
+            )
+            .await
+            .expect("upsert guild");
+
+        let guilds = store.guilds_all().await.expect("read board");
+        assert!(
+            guilds[0].get("roster").is_none(),
+            "a guild with no roster event must not grow a roster key"
+        );
+    }
+}

sidecar/src/web.rs

@@ -72,6 +72,41 @@ pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
         .route("/admin/ban", post(admin_ban))
         .route("/admin/unban", post(admin_unban))
         .route("/admin/broadcast", post(admin_broadcast))
+        // The event plane (protocol 6, EVENTS_PLAN.md Phase 11b). Leases are a live config value
+        // the website holds for a bounded time; the shard restores baseline when the deadline
+        // passes whether or not anyone asks it to. GET lists the whole catalog with current values,
+        // which is the one read both `read()` and `inForce()` on the website's side are served by.
+        .route("/lease", get(lease_list).post(lease_apply))
+        .route("/lease/release", post(lease_release))
+        // The run-scoped participation ledger. `snapshot` is a POST despite being a read: it
+        // carries the caller's `idempotencyKey`, and on a well-attended run the shard walks its
+        // members across ticks rather than in one inbound call -- so a repeat arriving mid-walk is
+        // answered `bridge.busy`, and a read that can be refused as a repeat is not a GET.
+        .route("/participation", post(participation_open))
+        .route(
+            "/participation/:run_id/snapshot",
+            post(participation_snapshot),
+        )
+        .route("/participation/:run_id/close", post(participation_close))
+        // The world verbs (protocol 7, EVENTS_PLAN.md Phase 12a). Five things an event author
+        // can place -- creatures, an enhanced "boss", an oracle NPC, a temporary gate,
+        // decoration -- and ONE command family, because each of them ends in "an object exists
+        // and this run owns it". POST places, GET says what the run still owns, POST .../despawn
+        // gives it back. Ownership is held on the shard, so despawn cannot be pointed at a serial
+        // the run did not create.
+        .route("/world", post(world_spawn))
+        .route("/world/:run_id", get(world_owned))
+        .route("/world/:run_id/despawn", post(world_despawn))
+        // The one-shots (protocol 7 part b, EVENTS_PLAN.md Phase 12b). Neither owned nor
+        // borrowed: an item put into somebody's hands, and a world save. Both are `done is
+        // done`, which is why they are not in the world family -- there is nothing to give
+        // back and no ledger row core would come back for.
+        //
+        // `GET /items` is the shard's own grant allowlist, so the website's dropdown offers
+        // what this shard will actually build rather than what a module guessed.
+        .route("/items", get(item_catalog))
+        .route("/items/grant", post(item_grant))
+        .route("/world/save", post(world_save))
         // Help-page (support) queue: snapshot the open queue, respond to / close a page.
         .route("/pages", get(pages_list))
         .route("/pages/:id/respond", post(page_respond))
@@ -246,13 +281,31 @@ fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
 
 // ---- shared reply handling ----
 
+/// Protocol 6. `bridge.busy` says a command carrying this `idempotencyKey` is already in flight on
+/// the shard: nothing was run, and the caller should come back.
+///
+/// It maps to **425 Too Early**, which is what that status is for — a server unwilling to risk
+/// processing a request that might be a replay. The obvious alternative, 409, is already the
+/// protocol-version gate's answer, and those two want opposite dispositions from a client: a version
+/// mismatch is a deployment fault nobody should retry, and a busy shard is a retry that should
+/// succeed on its own. Sharing a status would have made the difference readable only by inspecting
+/// the body, which is exactly how a retry loop ends up hiding a mismatched deployment.
+///
+/// It is checked BEFORE the `.error` suffix test in each responder below, and it is deliberately not
+/// spelled `bridge.busy.error`: nothing is wrong. The work is happening.
+const BUSY_KIND: &str = "bridge.busy";
+const BUSY_STATUS: StatusCode = StatusCode::TOO_EARLY;
+
 /// Turns an RPC result into an HTTP response. A `bridge.error` reply from the shard becomes a 4xx;
-/// a real reply is returned as-is; transport failures map to 503/504.
+/// a `bridge.busy` reply becomes a 425; a real reply is returned as-is; transport failures map to
+/// 503/504.
 fn respond(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
     match result {
         Ok(value) => {
             let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
-            if kind == "bridge.error" || kind.ends_with(".error") {
+            if kind == BUSY_KIND {
+                (BUSY_STATUS, Json(value))
+            } else if kind == "bridge.error" || kind.ends_with(".error") {
                 let reason = value
                     .get("reason")
                     .and_then(|r| r.as_str())
@@ -286,7 +339,9 @@ fn respond_admin(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
     match result {
         Ok(value) => {
             let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
-            if kind == "admin.error" {
+            if kind == BUSY_KIND {
+                (BUSY_STATUS, Json(value))
+            } else if kind == "admin.error" {
                 let reason = value
                     .get("reason")
                     .and_then(|r| r.as_str())
@@ -317,6 +372,70 @@ fn respond_admin(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
     }
 }
 
+/// Like `respond`, but for the event plane: leases and the participation ledger.
+///
+/// Two mappings are the point of it existing rather than reusing `respond`.
+///
+/// **`lease.drifted` is a 200.** The shard was asked to compare and set, it compared, and it
+/// refused to overwrite somebody's deliberate change -- that is the mechanism working, not a
+/// failure, and `cleanup.js` on the website treats `drifted` as a distinct successful outcome
+/// rather than an error. It is also why this is not a 409: 409 is the protocol-version gate's, and
+/// a version mismatch and a drifted lease want opposite dispositions from a caller. The same
+/// argument protocol 6 made for `bridge.busy` being a 425.
+///
+/// **The event plane being switched off is a 403**, not the 400 the generic responder's
+/// reason-sniffing would produce. `Bridge.EventsEnabled` is an operator's deliberate refusal to let
+/// the website change the world on a schedule, and telling the website it sent a bad request would
+/// send an administrator hunting a bug in a step that is written correctly.
+fn respond_event(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
+    match result {
+        Ok(value) => {
+            let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
+            if kind == BUSY_KIND {
+                (BUSY_STATUS, Json(value))
+            } else if kind.ends_with(".error") {
+                let reason = value
+                    .get("reason")
+                    .and_then(|r| r.as_str())
+                    .unwrap_or("request rejected");
+                let code = if reason.contains("disabled") {
+                    StatusCode::FORBIDDEN
+                } else if reason.contains("no lease is offered")
+                    || reason.contains("not counting")
+                    // Phase 12b. A grant against a run this shard has never been told to count
+                    // is the same shape as an unknown lease key: the caller named something that
+                    // does not exist here, which is a 404 and never a retry. It is deliberately
+                    // NOT the same as a run whose ledger is open and empty -- that is a 200 with
+                    // `granted: 0`, because "nobody came" is a result rather than a mistake.
+                    || reason.contains("no participation ledger")
+                {
+                    StatusCode::NOT_FOUND
+                } else if reason.contains("saves at most every") {
+                    // A save refused because one just happened is the shard's rate limit, and it
+                    // is TRANSIENT in a way nothing else on this plane is: the same request will
+                    // succeed once the interval passes. 429 says exactly that, and keeps it out of
+                    // the module's permanent-status set so a phase boundary is retried rather than
+                    // abandoned.
+                    StatusCode::TOO_MANY_REQUESTS
+                } else {
+                    StatusCode::BAD_REQUEST
+                };
+                (code, Json(value))
+            } else {
+                (StatusCode::OK, Json(value))
+            }
+        }
+        Err(RpcError::NoShard) => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            Json(json!({"error": "shard not connected"})),
+        ),
+        Err(RpcError::Timeout) => (
+            StatusCode::GATEWAY_TIMEOUT,
+            Json(json!({"error": "shard did not reply in time"})),
+        ),
+    }
+}
+
 /// Like `respond`, but for the account-provisioning plane. Maps an `account.error` reply to a
 /// status by its reason: a name clash is a 409, the per-IP cap is a 429, a disabled/protected/
 /// refused action is a 403, an unknown target or "not linked" is a 404, anything else a 400.
@@ -324,7 +443,9 @@ fn respond_account(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>)
     match result {
         Ok(value) => {
             let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
-            if kind == "account.error" {
+            if kind == BUSY_KIND {
+                (BUSY_STATUS, Json(value))
+            } else if kind == "account.error" {
                 let reason = value
                     .get("reason")
                     .and_then(|r| r.as_str())
@@ -454,6 +575,16 @@ async fn link_delete(
 /// Forwards a staff moderation command to the shard, correlated on a fresh reqId. Injects `kind`
 /// and `reqId`, requiring the caller-supplied `actor` up front (the shard enforces it too). The
 /// body's remaining fields (account/serial/durationSec/reason/text/hue) pass straight through.
+///
+/// **Protocol 6: `idempotencyKey` is one of those remaining fields**, and passing it through is the
+/// whole of the sidecar's part in the guarantee. It is worth stating rather than leaving to the
+/// word "remaining", because a later refactor that narrowed this to a known field list would quietly
+/// turn every retried world write back into a possible duplicate, and nothing here would fail.
+///
+/// The key belongs to the CALLER's unit of work — the website's event step — so the sidecar neither
+/// generates one nor validates it. Note also that `reqId` is regenerated on every call: a retry
+/// carries the same idempotency key under a NEW correlation id, which is exactly why the shard
+/// re-stamps a replayed reply rather than echoing the id the first attempt used.
 async fn admin_call(st: &AppState, kind: &str, body: Value) -> (StatusCode, Json<Value>) {
     let mut obj = match body {
         Value::Object(m) => m,
@@ -504,6 +635,241 @@ async fn admin_broadcast(State(st): State<AppState>, Json(body): Json<Value>) ->
     admin_call(&st, "admin.broadcast", body).await
 }
 
+// ---- event plane handlers (protocol 6, Phase 11b) ----
+
+/// Forwards an event-plane command to the shard, correlated on a fresh reqId.
+///
+/// Deliberately NOT `admin_call`: that one requires an `actor`, because every verb behind it is a
+/// staff member pressing a button and the shard's audit trail has to name them. An event verb's
+/// author is a RUN, which the body already carries as `runId` -- and demanding an actor here would
+/// have the runner inventing a human name for something no human is doing.
+///
+/// Everything else about it is the same, and the `idempotencyKey` passthrough matters for the same
+/// reason it does there: the key is one of the body's remaining fields, and a refactor that
+/// narrowed this to a known field list would silently make every retried lease a possible duplicate.
+async fn event_call(st: &AppState, kind: &str, body: Value) -> (StatusCode, Json<Value>) {
+    let mut obj = match body {
+        Value::Object(m) => m,
+        Value::Null => serde_json::Map::new(),
+        _ => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "body must be a JSON object"})),
+            )
+        }
+    };
+
+    let req_id = st.rpc.next_req_id();
+    obj.insert("kind".to_string(), json!(kind));
+    obj.insert("reqId".to_string(), json!(req_id));
+
+    respond_event(st.rpc.call(&st.shard, Value::Object(obj), &req_id).await)
+}
+
+/// Every lease this shard offers, with what each is worth right now and what is holding it.
+///
+/// One read answers both questions the website asks about a lease: `read()` wants the current value
+/// before it applies anything, and `inForce()` wants to know whether the shard still has a record
+/// of the hold. Splitting them would be two round trips for one key.
+///
+/// **`held` means "the shard still has a record of this lease", not "the value is still
+/// overridden".** A lease whose deadline has already fired stays listed, with `expired: true`,
+/// until teardown collects its verdict -- otherwise a reconcile in that window would report it gone
+/// and the website would write off a correctly-working backstop as an orphaned resource.
+/// **`?key=` and `?target=` narrow it to one row, and a targeted key needs them** (protocol 7 part
+/// b). `Spawner.MaxCount` is one capability over thousands of spawners, so it has no single
+/// "current" and the catalog walk cannot fill one in -- while the website's `read()` needs exactly
+/// one value for exactly one target before it applies anything. Naming both answers that.
+///
+/// The frame also carries `holds`: every lease this shard is actually holding, whatever key or
+/// target it is on. A catalog walk can enumerate the KEYS but never the holds on a targeted one --
+/// there is no list of spawners to walk -- so without it a reconcile after an outage would have no
+/// way to ask "what are you still holding?".
+async fn lease_list(State(st): State<AppState>, Query(q): Query<LeaseQuery>) -> impl IntoResponse {
+    let mut body = serde_json::Map::new();
+    if let Some(key) = q.key {
+        body.insert("key".to_string(), json!(key));
+    }
+    if let Some(target) = q.target {
+        body.insert("target".to_string(), json!(target));
+    }
+    let arg = if body.is_empty() {
+        Value::Null
+    } else {
+        Value::Object(body)
+    };
+    event_call(&st, "lease.list", arg).await
+}
+
+/// Narrowing for `GET /lease`. Both optional: absent means the whole catalog, as before.
+#[derive(Debug, Deserialize)]
+struct LeaseQuery {
+    key: Option<String>,
+    target: Option<String>,
+}
+
+/// Body: {"key":"...","value":"...","holdMs":<ms>,"untilMs":<opt>,"runId":<opt>,"idempotencyKey":<opt>}.
+///
+/// **`holdMs` is authoritative and `untilMs` is carried for display.** An absolute deadline computed
+/// on the website and honoured on the shard is a deadline measured against two clocks, and a shard
+/// running ten minutes fast would restore a ten-minute lease the moment it took it. A duration is
+/// immune to that; the absolute time is still worth sending so a console can say when the hold ends.
+///
+/// Values cross as TEXT whatever the lease's declared type, because JSON would otherwise decide for
+/// us: `1200` and `1200.0` are one number to a parser and two strings to a compare-and-set.
+async fn lease_apply(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    event_call(&st, "lease.apply", body).await
+}
+
+/// Body: {"key":"...","expected":"...","baseline":"...","idempotencyKey":<opt>}.
+///
+/// `expected` is what the event applied and `baseline` is what to put back, both out of the
+/// website's ledger rather than the shard's memory -- so a release still works after a reconnect,
+/// and a shard that has forgotten the lease entirely (a restart, which reverts every lease anyway)
+/// can answer honestly instead of refusing.
+///
+/// A mismatch comes back `lease.drifted` with a **200**: see `respond_event`.
+async fn lease_release(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    event_call(&st, "lease.release", body).await
+}
+
+/// Body: {"runId":"...","map":"Felucca","x":N,"y":N,"radius":N,"holdMs":<opt>}.
+///
+/// Declares where a run happens and starts counting who is there. The area is a map, a point and a
+/// radius rather than a region name, because protocol 6's own live walk established that the most
+/// specific region containing an event is routinely anonymous.
+async fn participation_open(
+    State(st): State<AppState>,
+    Json(body): Json<Value>,
+) -> impl IntoResponse {
+    event_call(&st, "participation.open", body).await
+}
+
+/// Body: {"idempotencyKey":<opt>}. Answers the run's tally, best-effort resolved to accounts.
+///
+/// A POST for a read, and the reason is worth keeping: on a well-attended run the shard walks its
+/// members in chunks across Core ticks rather than handing the whole resolve to one inbound call,
+/// so the handler completes after its call returned and a repeat arriving in between is answered
+/// `bridge.busy`. A read that can legitimately be refused as a repeat in flight is not a GET.
+async fn participation_snapshot(
+    State(st): State<AppState>,
+    Path(run_id): Path<String>,
+    Json(body): Json<Value>,
+) -> impl IntoResponse {
+    let mut obj = match body {
+        Value::Object(m) => m,
+        _ => serde_json::Map::new(),
+    };
+    obj.insert("runId".to_string(), json!(run_id));
+    event_call(&st, "participation.snapshot", Value::Object(obj)).await
+}
+
+/// Body: {"idempotencyKey":<opt>}. Stops counting; the tally stays readable through the shard's
+/// grace window, because closing an event and collecting its results are two steps and either can
+/// be retried.
+async fn participation_close(
+    State(st): State<AppState>,
+    Path(run_id): Path<String>,
+    Json(body): Json<Value>,
+) -> impl IntoResponse {
+    let mut obj = match body {
+        Value::Object(m) => m,
+        _ => serde_json::Map::new(),
+    };
+    obj.insert("runId".to_string(), json!(run_id));
+    event_call(&st, "participation.close", Value::Object(obj)).await
+}
+
+/// Body: {"runId":"...","what":"creature|boss|npc|gate|decor","map":"...","x":N,"y":N,...}.
+///
+/// One route for five author-facing verbs. The `what` discriminator is a wire detail: the
+/// differences between them -- a boss's multipliers, an oracle's lines, a gate's destination and
+/// `holdMs` -- are fields on one command rather than five commands, so there is one ledger shape,
+/// one teardown path and one reconcile instead of five near-identical ones in three repos.
+///
+/// The shard registers every serial it places against the run and PERSISTS that registry beside
+/// the world save, which is what makes `world_despawn` below safe: a spawned creature survives a
+/// restart, so an in-memory registry would leave the website holding serials the shard would not
+/// vouch for.
+async fn world_spawn(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    event_call(&st, "world.spawn", body).await
+}
+
+/// What the run still owns, and the answer the website's `reconcile()` is built on.
+///
+/// A GET, unlike `participation_snapshot`: it carries no idempotency key and the shard answers it
+/// in one pass, pruning rows whose object the world has already lost as it walks. Anything not
+/// listed is gone -- which is the shape core wants, because it takes a row out of its ledger only
+/// on an explicit reply and this is that reply.
+async fn world_owned(State(st): State<AppState>, Path(run_id): Path<String>) -> impl IntoResponse {
+    event_call(&st, "world.owned", json!({ "runId": run_id })).await
+}
+
+/// Body: {"serials":[...]} -- or no serials at all, which means everything the run owns and is the
+/// call teardown actually makes.
+///
+/// Three answers, and the split is why the shard keeps a registry at all. `removed` was found and
+/// deleted; `gone` was owned but already absent, which is what happens when a player kills an event
+/// creature and is a SUCCESS; `refused` was never this run's to delete, and is the only answer here
+/// that means somebody asked for something they should not have.
+async fn world_despawn(
+    State(st): State<AppState>,
+    Path(run_id): Path<String>,
+    Json(body): Json<Value>,
+) -> impl IntoResponse {
+    let mut obj = match body {
+        Value::Object(m) => m,
+        _ => serde_json::Map::new(),
+    };
+    obj.insert("runId".to_string(), json!(run_id));
+    event_call(&st, "world.despawn", Value::Object(obj)).await
+}
+
+// ---- the one-shots (protocol 7 part b) ----
+
+/// What this shard is willing to grant, and the bounds it will grant within.
+///
+/// A read, so the website's option source offers what this shard will actually build. The module
+/// holds the same list, which is two copies of a short allowlist on purpose and exactly how the
+/// lease bounds are already carried: the module's copy is what makes a bad value a refusal on a
+/// form, and this one is what is true when the website is wrong.
+async fn item_catalog(State(st): State<AppState>) -> impl IntoResponse {
+    event_call(&st, "item.catalog", Value::Null).await
+}
+
+/// Body: {"runId":"...","item":"gold","amount":N,"hue":<opt>,"name":<opt>,"where":<opt>,"idempotencyKey":<opt>}.
+///
+/// **The recipients are not in the body, and that is the design.** The shard already holds the
+/// run's participation ledger (protocol 6 part b), keyed by the same character serials the
+/// website's `member_key` holds, so the grant names a run and the shard resolves who was there.
+/// Sending a list would mean the same list crossing the wire twice with a window in which the two
+/// disagree -- and it would have needed a core surface handing a module core's own participants.
+///
+/// A run with no ledger open is a 404, not an empty success: "nobody came" and "you never told me
+/// to count" are different facts, and only the first is a result a run should record.
+///
+/// **Retryable, and protocol 6 is why.** `EVENTS.md` §G called a grant un-retryable because a lost
+/// acknowledgement and a grant that never applied looked the same -- exactly the argument that made
+/// `uo.broadcast` answer `retry: false` in Phase 9. An `idempotencyKey` closes that: a repeat is
+/// answered by the original reply, so a retried grant cannot be one winner receiving two.
+async fn item_grant(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    event_call(&st, "item.grant", body).await
+}
+
+/// Body: {"idempotencyKey":<opt>}. Starts a world save, useful as a phase boundary.
+///
+/// The reply says the save was STARTED and nothing more. What actually happened rides
+/// `world.save.before` / `world.save.after`, which have been on the event stream since protocol 2 --
+/// so this route asserts nothing it cannot know, and a caller that needs the completion watches the
+/// stream it is already connected to.
+///
+/// **A save too soon after the last one is refused, not queued**, and the shard counts ServUO's own
+/// autosave as the last one. A save stops the world; a queued one would land at a moment nobody
+/// chose, in the middle of whatever the next step is doing.
+async fn world_save(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    event_call(&st, "world.save", body).await
+}
+
 // ---- help-page queue handlers ----
 
 /// The open help-page queue, correlated on reqId. Returns a pages.list.
@@ -978,3 +1344,272 @@ async fn ws_client(mut socket: WebSocket, state: AppState) {
 
     info!("ws client disconnected");
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn reply(kind: &str) -> Result<Value, RpcError> {
+        Ok(json!({"t": 1, "kind": kind, "reqId": "r-9"}))
+    }
+
+    /// Protocol 6. Every responder must recognise `bridge.busy`, because every write plane can be
+    /// retried: the staff plane, the account plane and the plain command plane all reach handlers
+    /// that a keyed retry can arrive at. A responder that missed it would return 200 with a body
+    /// saying nothing happened, which is the worst of the three possible answers.
+    #[test]
+    fn busy_maps_to_425_on_every_plane() {
+        assert_eq!(respond(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
+        assert_eq!(respond_admin(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
+        assert_eq!(
+            respond_account(reply("bridge.busy")).0,
+            StatusCode::TOO_EARLY
+        );
+        assert_eq!(respond_event(reply("bridge.busy")).0, StatusCode::TOO_EARLY);
+    }
+
+    /// The event plane is the FIRST place `bridge.busy` is reachable on a live shard rather than
+    /// only in a unit test: `participation.snapshot` walks a well-attended run's members across
+    /// Core ticks, so it completes after its inbound call returned and a repeat can genuinely land
+    /// mid-flight. 11a built the door and had nothing to walk through it.
+    #[test]
+    fn a_drifted_lease_is_a_200_not_a_409() {
+        let value =
+            json!({"kind": "lease.drifted", "key": "PlayerCaps.SkillCap", "current": "1300"});
+        let (status, body) = respond_event(Ok(value));
+
+        // The shard was asked to compare and set, it compared, and it declined to overwrite
+        // somebody's deliberate change. That is the mechanism working; the website records
+        // `drifted` as a distinct successful outcome rather than an error.
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body.0.get("current").and_then(|v| v.as_str()), Some("1300"));
+
+        // And explicitly not the version gate's status, for the reason 425 is not either: a
+        // mismatched deployment and a moved value want opposite dispositions from a caller.
+        assert_ne!(status, StatusCode::CONFLICT);
+    }
+
+    /// The event plane being switched off is an operator's refusal, not a malformed request. A 400
+    /// would send an administrator hunting a bug in a step that is written correctly.
+    #[test]
+    fn the_event_gate_being_off_is_a_403() {
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "lease.error",
+                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
+            })))
+            .0,
+            StatusCode::FORBIDDEN
+        );
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "participation.error",
+                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
+            })))
+            .0,
+            StatusCode::FORBIDDEN
+        );
+    }
+
+    /// Protocol 7's world verbs go through the same responder, and this pins the two mappings
+    /// they depend on rather than trusting that the reason-sniffing above keeps covering a kind
+    /// it was written before.
+    ///
+    /// A CEILING refusal is a 400 on purpose. It is permanent -- retrying "you asked for 80
+    /// creatures and this shard places 30" gets the same answer forever -- and it is the module's
+    /// `PERMANENT_STATUSES` that has to see it as such, so classifying it as anything retryable
+    /// would put a run in a loop against a limit that will never move.
+    #[test]
+    fn a_world_refusal_is_a_400_and_the_gate_is_still_a_403() {
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "world.error",
+                "action": "spawn",
+                "reason": "this shard places 1 to 30 of 'creature' at a time, and 80 was asked for"
+            })))
+            .0,
+            StatusCode::BAD_REQUEST
+        );
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "world.error",
+                "action": "spawn",
+                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
+            })))
+            .0,
+            StatusCode::FORBIDDEN
+        );
+    }
+
+    /// A run the shard has no registry rows for answers with an EMPTY hand, not a 404, and the
+    /// distinction is load-bearing for reconcile.
+    ///
+    /// "This run owns nothing" and "I have never heard of this run" are the same fact once the
+    /// registry is the only record of ownership, and they stay the same fact across a restart:
+    /// the registry is written by `EventSink.WorldSave`, so it and the objects it describes are
+    /// saved and lost together. A 404 here would make the website treat a run that legitimately
+    /// owns nothing as a shard it could not reach.
+    #[test]
+    fn a_run_owning_nothing_is_an_empty_list_not_a_404() {
+        let (status, body) = respond_event(Ok(json!({
+            "kind": "world.owned.ok",
+            "runId": "77",
+            "owned": [],
+            "pruned": 0
+        })));
+
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(
+            body.0
+                .get("owned")
+                .and_then(|v| v.as_array())
+                .map(|a| a.len()),
+            Some(0)
+        );
+    }
+
+    /// An unknown lease key and an unknown run are not-founds; anything else the shard refuses is a
+    /// bad request. The catalog is short and a typo in a step is the likely cause of both.
+    #[test]
+    fn unknown_lease_and_run_are_404s() {
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "lease.error",
+                "reason": "no lease is offered for key 'Loot.MaxProps'"
+            })))
+            .0,
+            StatusCode::NOT_FOUND
+        );
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "participation.error",
+                "reason": "this shard is not counting run '42'"
+            })))
+            .0,
+            StatusCode::NOT_FOUND
+        );
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "lease.error",
+                "reason": "a lease needs a positive holdMs"
+            })))
+            .0,
+            StatusCode::BAD_REQUEST
+        );
+    }
+
+    /// A run this shard was never told to count is a 404; a run that WAS counted and had no
+    /// attendees is a 200. Protocol 7 part b.
+    #[test]
+    fn an_uncounted_run_is_a_404_and_an_empty_one_is_not() {
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "oneshot.error",
+                "reason": "run 42 has no participation ledger open on this shard"
+            })))
+            .0,
+            StatusCode::NOT_FOUND
+        );
+        // The distinction the 404 exists to preserve. "Nobody came" is a RESULT -- an event
+        // nobody attended still happened -- and answering it as a failure would have the module
+        // retry a grant against a ledger that will be just as empty next time.
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "item.grant.ok",
+                "runId": "42",
+                "granted": 0,
+                "missed": []
+            })))
+            .0,
+            StatusCode::OK
+        );
+    }
+
+    /// The save rate limit is the one refusal on this plane that the same request will get past
+    /// by waiting, so it is a 429 rather than the 400 every other refusal is.
+    #[test]
+    fn a_save_refused_for_coming_too_soon_is_a_429() {
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "oneshot.error",
+                "reason": "this shard saves at most every 300 seconds, and the last save was 12 seconds ago"
+            })))
+            .0,
+            StatusCode::TOO_MANY_REQUESTS
+        );
+        // And an ordinary refusal on the same plane is still a 400, so the 429 is not swallowing
+        // the class it sits beside: a grant this shard does not offer will never succeed, however
+        // long the caller waits.
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "oneshot.error",
+                "reason": "this shard does not grant 'castle'"
+            })))
+            .0,
+            StatusCode::BAD_REQUEST
+        );
+        // The event gate being off stays a 403 on this plane too -- it is an operator's deliberate
+        // refusal, not a bad request.
+        assert_eq!(
+            respond_event(Ok(json!({
+                "kind": "oneshot.error",
+                "reason": "the event plane is disabled on this shard (Bridge.EventsEnabled)"
+            })))
+            .0,
+            StatusCode::FORBIDDEN
+        );
+    }
+
+    /// A lease taken, a tally answered: an ordinary success carries straight through.
+    #[test]
+    fn event_successes_are_200s() {
+        assert_eq!(respond_event(reply("lease.ok")).0, StatusCode::OK);
+        assert_eq!(respond_event(reply("lease.list.ok")).0, StatusCode::OK);
+        assert_eq!(respond_event(reply("participation.ok")).0, StatusCode::OK);
+        assert_eq!(
+            respond_event(reply("participation.snapshot.ok")).0,
+            StatusCode::OK
+        );
+    }
+
+    /// 425 must not collide with the protocol-version gate's 409: a mismatch is a deployment fault
+    /// nobody should retry, a busy shard is a retry that will succeed. Same-status would make the
+    /// two readable only by inspecting the body.
+    #[test]
+    fn busy_is_not_the_version_gates_status() {
+        assert_ne!(BUSY_STATUS, StatusCode::CONFLICT);
+    }
+
+    /// A replayed reply is an ordinary success. The shard marks it `replayed: true` for the log, and
+    /// the caller must be able to treat it exactly as it would have treated the answer it lost.
+    #[test]
+    fn a_replayed_reply_is_still_a_200() {
+        let value = json!({"t": 1, "kind": "admin.ok", "reqId": "r-9", "replayed": true});
+        let (status, body) = respond_admin(Ok(value));
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body.0.get("replayed").and_then(|v| v.as_bool()), Some(true));
+    }
+
+    /// The error mapping the busy arm is threaded in front of must be untouched by it.
+    #[test]
+    fn errors_still_map_as_before() {
+        assert_eq!(
+            respond(Ok(
+                json!({"kind": "bridge.error", "reason": "unknown account"})
+            ))
+            .0,
+            StatusCode::NOT_FOUND
+        );
+        assert_eq!(
+            respond_admin(Ok(json!({"kind": "admin.error", "reason": "protected"}))).0,
+            StatusCode::FORBIDDEN
+        );
+        assert_eq!(
+            respond_account(Ok(
+                json!({"kind": "account.error", "reason": "already exists"})
+            ))
+            .0,
+            StatusCode::CONFLICT
+        );
+    }
+}