feat(sidecar): protocol 6 — carry the idempotency key, answer bridge.busy (Phase 11a)

PROTOCOL_VERSION 5 -> 6, and one behaviour: `bridge.busy` maps to 425 Too Early in
all three responders.

Everything else is free, and that is the point. The key rides in the command body,
which every write endpoint already passes through verbatim; `champ.boss.killed`
lands in `events` and on the feed through the generic forward path with no arm of
its own. No store migration — nothing gains a column.

Worth naming what the dumb-forwarder property means here specifically: the sidecar
makes no idempotency promise of its own. It does not dedupe, does not cache, and
does not know what a key means. The guarantee is the shard's, end to end, which is
the only place it can be.

425 rather than 409 because 409 is already the protocol-version gate's answer, and
the two want opposite dispositions from a client: a version mismatch is a
deployment fault nobody should retry, a busy shard is a retry that will succeed.
Sharing a status would make the difference readable only by inspecting the body,
which is how a retry loop ends up hiding a mismatched deployment. A replayed reply
is an ordinary 200 — `replayed: true` is for the log.

Co-Authored-By: Claude <noreply@anthropic.com>

sidecar/src/web.rs

@@ -246,13 +246,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 +304,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())
@@ -324,7 +344,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 +476,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,
@@ -978,3 +1010,67 @@ 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
+        );
+    }
+
+    /// 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
+        );
+    }
+}