relay: accept HTTP agent observer frames (draft handoff)

crates/buzz-conformance/TRACE_SCHEMA.md

@@ -1,6 +1,7 @@
 # Trace Schema (`buzz-conformance`)
 
-Schema version: **1** (`SCHEMA_VERSION` in `src/lib.rs`).
+Schema version: **2** (`SCHEMA_VERSION` in `src/lib.rs`). Version 2 adds
+`accept_ephemeral` to the action enum; version 1 consumers cannot decode it.
 
 This document is the contract between the relay's emitter and the
 independent replay checker. It is grounded in
@@ -65,6 +66,10 @@ exact spec line it grounds in.
   via the host-community map; no `channel` field. `claimed_community`
   recorded for the same reason as above.
 
+- **`accept_ephemeral { msg_id }`**
+  A validated ephemeral event accepted for live fan-out. No durable write
+  occurs.
+
 - **`write_duplicate { msg_id, channel, claimed_community }`**
   spec: `WriteDuplicate` (line 612). The DB returned "already present";
   no row was added. No `row_community` because no row was produced.
@@ -133,7 +138,7 @@ normalized away the violation. The checker assumes you *did not*.
 |------|---------------|
 | `crates/buzz-relay/src/conformance/mod.rs` | helpers + `EmitGuard` + `sanitized_reason_for` |
 | `crates/buzz-relay/src/conformance/tracers.rs` | `NoopTracer` (prod default), `JsonlTracer` |
-| `crates/buzz-relay/src/handlers/ingest.rs` | `AuthCheck`, `WriteInsert`, `WriteInsertGlobal`, `WriteDuplicate`, outer-wrapper `SanitizedError` |
+| `crates/buzz-relay/src/handlers/ingest.rs` | `AuthCheck`, `WriteInsert`, `WriteInsertGlobal`, `AcceptEphemeral`, `WriteDuplicate`, outer-wrapper `SanitizedError` |
 | `crates/buzz-relay/src/handlers/req.rs` | **held back** — additive patch for integration onto Max's req.rs work |
 
 ## Where the checker lives

crates/buzz-conformance/src/checker.rs

@@ -258,6 +258,20 @@ mod tests {
         check_trace(&Scenario::unstructured(trace)).expect("deny with foreign claim is in-spec");
     }
 
+    #[test]
+    fn accepted_ephemeral_event_is_valid_without_a_write() {
+        let c = cid(1);
+        let trace = vec![step(
+            TraceAction::AcceptEphemeral {
+                msg_id: OpaqueId("ephemeral".into()),
+            },
+            c,
+        )];
+
+        check_trace(&Scenario::unstructured(trace))
+            .expect("ephemeral acceptance is valid without a durable write");
+    }
+
     #[test]
     fn state_after_changing_mid_request_is_state_mismatch() {
         let c1 = cid(1);

crates/buzz-conformance/src/lib.rs

@@ -83,7 +83,7 @@ impl std::fmt::Display for CommunityLabel {
 }
 
 /// Trace schema version. Bump on any backwards-incompatible field change.
-pub const SCHEMA_VERSION: u32 = 1;
+pub const SCHEMA_VERSION: u32 = 2;
 
 /// An opaque ID derived from an event id or other secret material. Stable,
 /// no payload, no key bytes. Implementations pick a hash; the checker
@@ -165,6 +165,7 @@ pub struct AbstractState {
 /// Action vocabulary (spec actions in parentheses):
 /// - [`TraceAction::WriteInsert`] (spec `WriteInsert`, lines 514–550)
 /// - [`TraceAction::WriteInsertGlobal`] (spec `WriteInsertGlobal`, lines 559–595)
+/// - [`TraceAction::AcceptEphemeral`] (in-memory event acceptance; no modeled write)
 /// - [`TraceAction::WriteDuplicate`] (spec `WriteDuplicate`, lines 606–637)
 /// - [`TraceAction::SanitizedError`] (spec `SanitizedError`, line 778)
 /// - [`TraceAction::AuthCheck`] (spec `AuthCheck`, line 794) — M2/M8 target
@@ -199,6 +200,11 @@ pub enum TraceAction {
         /// resolver but recorded for the audit trail.
         claimed_community: Option<CommunityLabel>,
     },
+    /// Validated ephemeral event accepted for live fan-out; no durable write occurs.
+    AcceptEphemeral {
+        /// Opaque hash of the event id.
+        msg_id: OpaqueId,
+    },
     /// Channel-bearing duplicate / no-op write (spec `WriteDuplicate`,
     /// `ON CONFLICT (community_id, id)` returning a duplicate result).
     WriteDuplicate {
@@ -267,6 +273,7 @@ impl TraceAction {
         match self {
             TraceAction::WriteInsert { .. } => "write_insert",
             TraceAction::WriteInsertGlobal { .. } => "write_insert_global",
+            TraceAction::AcceptEphemeral { .. } => "accept_ephemeral",
             TraceAction::WriteDuplicate { .. } => "write_duplicate",
             TraceAction::SanitizedError { .. } => "sanitized_error",
             TraceAction::AuthCheck { .. } => "auth_check",

crates/buzz-conformance/src/transitions.rs

@@ -189,6 +189,8 @@ pub fn check_step(
         // --- Spec WriteInsertGlobal (lines 559-595) ---
         // resolved == HostCommunity[host]. Same shape as WriteInsert.
         TraceAction::WriteInsertGlobal { .. } => Ok(()),
+        // Ephemeral acceptance is an observation only; it mutates no durable model state.
+        TraceAction::AcceptEphemeral { .. } => Ok(()),
 
         // --- Spec WriteDuplicate (lines 606-637) ---
         // Carries the same host-axis obligation as WriteInsert: an A-host
@@ -323,6 +325,7 @@ pub fn action_channel(action: &TraceAction) -> Option<&ChannelLabel> {
         TraceAction::ReadMessageRows { channel, .. } => channel.as_ref(),
         TraceAction::ReadByIdRows { channel, .. } => channel.as_ref(),
         TraceAction::WriteInsertGlobal { .. }
+        | TraceAction::AcceptEphemeral { .. }
         | TraceAction::ReadHostFeedRows { .. }
         | TraceAction::SanitizedError { .. }
         | TraceAction::ImplBug { .. } => None,

crates/buzz-conformance/tests/fixtures/bad_coverage_breach.jsonl

@@ -1 +1 @@
-{"schema_version":1,"action":{"type":"impl_bug","kind":"ingest_exited_without_trace"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"impl_bug","kind":"ingest_exited_without_trace"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}

crates/buzz-conformance/tests/fixtures/bad_foreign_row_leak.jsonl

@@ -1 +1 @@
-{"schema_version":1,"action":{"type":"read_message_rows","channel":"cafe0000-0000-0000-0000-000000000010","row_communities":["bbbb0000-0000-0000-0000-000000000002"]},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"read_message_rows","channel":"cafe0000-0000-0000-0000-000000000010","row_communities":["bbbb0000-0000-0000-0000-000000000002"]},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}

crates/buzz-conformance/tests/fixtures/bad_host_channel_mismatch.jsonl

@@ -1,2 +1,2 @@
-{"schema_version":1,"action":{"type":"auth_check","channel":"dead0000-0000-0000-0000-000000000020","claimed_community":"bbbb0000-0000-0000-0000-000000000002","verdict":"allow"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
-{"schema_version":1,"action":{"type":"write_insert","msg_id":"badbadbad0000000","channel":"dead0000-0000-0000-0000-000000000020","claimed_community":"bbbb0000-0000-0000-0000-000000000002"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"auth_check","channel":"dead0000-0000-0000-0000-000000000020","claimed_community":"bbbb0000-0000-0000-0000-000000000002","verdict":"allow"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"write_insert","msg_id":"badbadbad0000000","channel":"dead0000-0000-0000-0000-000000000020","claimed_community":"bbbb0000-0000-0000-0000-000000000002"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}

crates/buzz-conformance/tests/fixtures/good.jsonl

@@ -1,3 +1,3 @@
-{"schema_version":1,"action":{"type":"auth_check","channel":"cafe0000-0000-0000-0000-000000000010","claimed_community":"aaaa0000-0000-0000-0000-000000000001","verdict":"allow"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
-{"schema_version":1,"action":{"type":"write_insert","msg_id":"d34db33fcafef00d","channel":"cafe0000-0000-0000-0000-000000000010","claimed_community":"aaaa0000-0000-0000-0000-000000000001"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
-{"schema_version":1,"action":{"type":"read_message_rows","channel":"cafe0000-0000-0000-0000-000000000010","row_communities":["aaaa0000-0000-0000-0000-000000000001","aaaa0000-0000-0000-0000-000000000001"]},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"auth_check","channel":"cafe0000-0000-0000-0000-000000000010","claimed_community":"aaaa0000-0000-0000-0000-000000000001","verdict":"allow"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"write_insert","msg_id":"d34db33fcafef00d","channel":"cafe0000-0000-0000-0000-000000000010","claimed_community":"aaaa0000-0000-0000-0000-000000000001"},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}
+{"schema_version":2,"action":{"type":"read_message_rows","channel":"cafe0000-0000-0000-0000-000000000010","row_communities":["aaaa0000-0000-0000-0000-000000000001","aaaa0000-0000-0000-0000-000000000001"]},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}}

crates/buzz-relay/src/api/bridge.rs

@@ -1072,6 +1072,14 @@ async fn submit_event_authed(
                 response: e,
             }
         }
+        Err(IngestError::RateLimited(msg)) => {
+            crate::handlers::ingest::reject_with_transport("http", "rate-limited");
+            let e = api_error(StatusCode::TOO_MANY_REQUESTS, &msg);
+            SubmitOutcome::Err {
+                status: e.0,
+                response: e,
+            }
+        }
         Err(IngestError::Internal(msg)) => {
             crate::handlers::ingest::reject_with_transport("http", "error");
             let e = internal_error(&msg);

crates/buzz-relay/src/conformance/mod.rs

@@ -435,6 +435,7 @@ pub fn sanitized_reason_for(err: &crate::handlers::ingest::IngestError) -> Sanit
         E::Rejected(_) => SanitizedReason::Invalid,
         E::CanvasConflict(_) => SanitizedReason::Invalid,
         E::AuthFailed(_) => SanitizedReason::Restricted,
+        E::RateLimited(_) => SanitizedReason::Restricted,
         E::Internal(_) => SanitizedReason::ServerError,
     }
 }

crates/buzz-relay/src/handlers/command_executor.rs

@@ -1426,6 +1426,9 @@ mod postgres_tests {
                 panic!("unexpected canvas conflict: {message}")
             }
             Err(IngestError::AuthFailed(message)) => panic!("unexpected auth failure: {message}"),
+            Err(IngestError::RateLimited(message)) => {
+                panic!("unexpected rate limit: {message}")
+            }
             Err(IngestError::Internal(message)) => panic!("unexpected internal failure: {message}"),
             Ok(_) => panic!("expected revision parsing to fail"),
         }