feat(core): project Multiverse placement in signed event order

crates/buzz-core/src/lib.rs

@@ -32,6 +32,8 @@ pub mod nip10;
 pub mod observer;
 /// NIP-AB device pairing — crypto primitives, message types, and errors.
 pub mod pairing;
+/// Arrival-independent desired placement, separate from lifecycle execution.
+pub mod placement;
 /// Presence status types shared across crates.
 pub mod presence;
 /// NIP-PMA owner-encrypted private managed-agent wire codec.

crates/buzz-core/src/placement.rs

@@ -0,0 +1,149 @@
+//! Desired placement projection, not a command executor or an admission journal.
+//!
+//! Callers must first authenticate and decode each event, authorize its owner,
+//! community, agent and host, and supply only one owner/community/agent scope.
+//! This module does not parse a wire format or confer execution authority.
+//! Historical intent may be projected, but must never be replayed as commands.
+
+use std::cmp::Ordering;
+
+use nostr::{Event, EventId, PublicKey};
+
+/// Signed-event precedence: newer sender seconds, then LOWER event ID wins.
+/// This is neither receiver-arrival order nor causal/last-click order. A future
+/// timestamp can win; no clock-skew or relay-sequencing policy is added here.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct EventOrder {
+    created_at: u64,
+    id: EventId,
+}
+
+impl EventOrder {
+    /// Extract signed fields from an event already verified by the caller.
+    /// Extraction itself does not verify the signature or authorize the event.
+    pub fn from_event(event: &Event) -> Self {
+        Self {
+            created_at: event.created_at.as_secs(),
+            id: event.id,
+        }
+    }
+
+    /// Identity of the signed event, retained unchanged on transport retry.
+    pub fn event_id(self) -> EventId {
+        self.id
+    }
+}
+
+impl Ord for EventOrder {
+    fn cmp(&self, other: &Self) -> Ordering {
+        self.created_at
+            .cmp(&other.created_at)
+            .then_with(|| other.id.cmp(&self.id))
+    }
+}
+
+impl PartialOrd for EventOrder {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+/// Only Start and Stop affect placement. Restart is a separately deduplicated
+/// current-host action. Move contributes a Start only after ordinary Stop
+/// succeeds and its still-valid coordinator actually issues destination Start.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PlacementAction {
+    /// Select this host, superseding earlier host selections.
+    Start,
+    /// Stop this host without cancelling another host's selected placement.
+    Stop,
+}
+
+/// A decoded placement contribution from one authorized signed command.
+/// The transport adapter must bind all three fields to the SAME signed event.
+/// Request identity and one-shot outcomes belong in the admission journal, not
+/// in this projection; duplicate intent is harmless but not execution dedup.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PlacementIntent {
+    /// Signed command precedence (not receipt or relay arrival time).
+    pub order: EventOrder,
+    /// Authorized executor identity, not a physical machine or process ID.
+    pub host: PublicKey,
+    /// Decoded operation; legacy exact-run Stop must not be broadened here.
+    pub action: PlacementAction,
+}
+
+/// Intent for one host. This does not describe observed process state.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TargetIntent {
+    /// No relevant intent is known. This is not permission to launch.
+    Unknown,
+    /// The selected Start is still desired; its identity guards continuations.
+    Running(EventOrder),
+    /// A local Stop or selection of another host supersedes local launch work.
+    Stopped(EventOrder),
+}
+
+/// A read-only view over the relevant valid events for one scoped agent.
+/// Uses two linear scans and constant auxiliary space, regardless of delivery
+/// order. The caller owns history completeness and replay-safe retention.
+#[derive(Debug)]
+pub struct PlacementProjection<'a> {
+    intents: &'a [PlacementIntent],
+    latest_start: Option<&'a PlacementIntent>,
+}
+
+impl<'a> PlacementProjection<'a> {
+    /// Project known intent without performing, scheduling or resuming effects.
+    pub fn new(intents: &'a [PlacementIntent]) -> Self {
+        let latest_start = intents
+            .iter()
+            .filter(|intent| intent.action == PlacementAction::Start)
+            .max_by_key(|intent| intent.order);
+        Self {
+            intents,
+            latest_start,
+        }
+    }
+
+    /// Desired host and the Start that selected it, or none.
+    /// Stopping the newest selection NEVER falls back to an earlier Start.
+    pub fn desired(&self) -> Option<(PublicKey, EventOrder)> {
+        let start = self.latest_start?;
+        match self.target(start.host) {
+            TargetIntent::Running(order) => Some((start.host, order)),
+            _ => None,
+        }
+    }
+
+    /// Project one target independently of receiver arrival. A Stop for X
+    /// remains relevant even if Start X is learned later. A Stop for X cannot
+    /// change Y's unchanged Start identity, including when Y is learned later.
+    pub fn target(&self, host: PublicKey) -> TargetIntent {
+        let stop = self
+            .intents
+            .iter()
+            .filter(|intent| intent.host == host && intent.action == PlacementAction::Stop)
+            .max_by_key(|intent| intent.order);
+        match (self.latest_start, stop) {
+            (None, None) => TargetIntent::Unknown,
+            (None, Some(stop)) => TargetIntent::Stopped(stop.order),
+            (Some(start), Some(stop)) if stop.order > start.order => {
+                TargetIntent::Stopped(stop.order)
+            }
+            (Some(start), _) if start.host == host => TargetIntent::Running(start.order),
+            (Some(start), _) => TargetIntent::Stopped(start.order),
+        }
+    }
+
+    /// Check only the placement part of a pending launch's continuation guard.
+    /// Recheck immediately before effects, alongside current authorization,
+    /// local process state and durable one-shot admission. True does NOT grant
+    /// permission to replay a Start/Restart or resume an interrupted operation.
+    pub fn retains_start(&self, host: PublicKey, start: EventOrder) -> bool {
+        self.target(host) == TargetIntent::Running(start)
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/buzz-core/src/placement/tests.rs

@@ -0,0 +1,236 @@
+use super::*;
+use nostr::{EventBuilder, Keys, Kind, Timestamp};
+
+fn host(value: u8) -> PublicKey {
+    Keys::parse(&format!("{value:064x}"))
+        .expect("fixture key")
+        .public_key()
+}
+
+fn intent(action: PlacementAction, target: u8, seconds: u64) -> PlacementIntent {
+    // Signed fixtures exercise the actual event ID/timestamp extraction, not
+    // receiver time. Wire parsing and owner authorization are separate layers.
+    let keys = Keys::parse(&format!("{:064x}", 10)).expect("owner");
+    let event = EventBuilder::new(Kind::TextNote, format!("{action:?}:{target}"))
+        .custom_created_at(Timestamp::from(seconds))
+        .sign_with_keys(&keys)
+        .expect("sign fixture");
+    crate::verify_event(&event).expect("verify fixture");
+    PlacementIntent {
+        order: EventOrder::from_event(&event),
+        host: host(target),
+        action,
+    }
+}
+
+fn start(target: u8, seconds: u64) -> PlacementIntent {
+    intent(PlacementAction::Start, target, seconds)
+}
+
+fn stop(target: u8, seconds: u64) -> PlacementIntent {
+    intent(PlacementAction::Stop, target, seconds)
+}
+
+#[test]
+fn signed_order_uses_newer_seconds_then_lower_id() {
+    let x = start(1, 100);
+    let y = start(2, 100);
+    assert_ne!(x.order.event_id(), y.order.event_id());
+    assert_eq!(
+        x.order.cmp(&y.order),
+        y.order.event_id().cmp(&x.order.event_id())
+    );
+    assert!(start(1, 101).order > x.order);
+    assert!(start(1, u64::MAX).order > start(1, 101).order);
+    assert_eq!(x.order.cmp(&x.order), Ordering::Equal);
+    assert_eq!(x.order.partial_cmp(&y.order), Some(x.order.cmp(&y.order)));
+}
+
+#[test]
+fn empty_and_stop_only_history_never_authorize_launch() {
+    let empty = PlacementProjection::new(&[]);
+    assert_eq!(empty.desired(), None);
+    assert_eq!(empty.target(host(1)), TargetIntent::Unknown);
+    let x = stop(1, 20);
+    let events = [x];
+    let view = PlacementProjection::new(&events);
+    assert_eq!(view.desired(), None);
+    assert_eq!(view.target(host(1)), TargetIntent::Stopped(x.order));
+    assert_eq!(view.target(host(2)), TargetIntent::Unknown);
+    assert!(!view.retains_start(host(1), start(1, 10).order));
+}
+
+#[test]
+fn delayed_stop_of_old_host_preserves_destination_and_its_continuation() {
+    let x = start(1, 10);
+    let y = start(2, 20);
+    let stop_x = stop(1, 30);
+    for events in [[x, y, stop_x], [stop_x, x, y], [y, stop_x, x]] {
+        let view = PlacementProjection::new(&events);
+        assert_eq!(view.desired(), Some((y.host, y.order)));
+        assert!(view.retains_start(y.host, y.order));
+        assert!(!view.retains_start(x.host, x.order));
+        assert_eq!(view.target(x.host), TargetIntent::Stopped(stop_x.order));
+    }
+}
+
+#[test]
+fn stop_of_latest_destination_never_resurrects_previous_host() {
+    let x = start(1, 10);
+    let y = start(2, 20);
+    let stop_y = stop(2, 30);
+    for events in [[x, y, stop_y], [stop_y, x, y], [y, stop_y, x]] {
+        let view = PlacementProjection::new(&events);
+        assert_eq!(view.desired(), None);
+        assert_eq!(view.target(x.host), TargetIntent::Stopped(y.order));
+        assert_eq!(view.target(y.host), TargetIntent::Stopped(stop_y.order));
+        assert!(!view.retains_start(x.host, x.order));
+        assert!(!view.retains_start(y.host, y.order));
+    }
+}
+
+#[test]
+fn older_stop_cannot_cancel_newer_start_but_new_selection_invalidates_old_work() {
+    let old = start(1, 10);
+    let stopped = stop(1, 20);
+    let new = start(1, 30);
+    let events = [new, stopped, old];
+    let view = PlacementProjection::new(&events);
+    assert_eq!(view.desired(), Some((new.host, new.order)));
+    assert!(view.retains_start(new.host, new.order));
+    assert!(!view.retains_start(old.host, old.order));
+}
+
+#[test]
+fn duplicate_and_lower_rank_backfill_do_not_change_the_selected_start() {
+    let selected = start(2, 100);
+    let events = [selected, stop(1, 200), selected, start(1, 1), stop(2, 99)];
+    let view = PlacementProjection::new(&events);
+    assert_eq!(view.desired(), Some((selected.host, selected.order)));
+    assert!(view.retains_start(selected.host, selected.order));
+}
+
+#[test]
+fn future_clock_wins_even_if_later_real_action_arrives_last() {
+    let fast = start(1, 1_000_000);
+    let events = [fast, start(2, 100), stop(1, 101)];
+    let view = PlacementProjection::new(&events);
+    assert_eq!(view.desired(), Some((fast.host, fast.order)));
+    assert!(view.retains_start(fast.host, fast.order));
+}
+
+#[test]
+fn same_second_start_stop_and_host_races_follow_lower_id() {
+    let x = start(1, 100);
+    for competitor in [stop(1, 100), start(2, 100), stop(2, 100)] {
+        for events in [[x, competitor], [competitor, x]] {
+            let view = PlacementProjection::new(&events);
+            let expected = if competitor.order > x.order {
+                match competitor.action {
+                    PlacementAction::Start => Some((competitor.host, competitor.order)),
+                    PlacementAction::Stop if competitor.host == x.host => None,
+                    PlacementAction::Stop => Some((x.host, x.order)),
+                }
+            } else {
+                Some((x.host, x.order))
+            };
+            assert_eq!(view.desired(), expected);
+        }
+    }
+}
+
+// Independent chronological reference model: conditional Stop is valid only
+// over SIGNED ORDER, never over delivery order. No effects are executed here.
+fn assert_matches_ordered_model(events: &[PlacementIntent]) {
+    let mut ordered = events.to_vec();
+    ordered.sort_by(|a, b| {
+        a.order
+            .created_at
+            .cmp(&b.order.created_at)
+            .then_with(|| b.order.id.cmp(&a.order.id))
+    });
+    ordered.dedup();
+    let hosts = [host(1), host(2), host(3)];
+    let mut desired = None;
+    let mut targets = [TargetIntent::Unknown; 3];
+    for event in ordered {
+        match event.action {
+            PlacementAction::Start => {
+                desired = Some((event.host, event.order));
+                for (target, state) in hosts.iter().zip(&mut targets) {
+                    *state = if *target == event.host {
+                        TargetIntent::Running(event.order)
+                    } else {
+                        TargetIntent::Stopped(event.order)
+                    };
+                }
+            }
+            PlacementAction::Stop => {
+                if desired.is_some_and(|(target, _)| target == event.host) {
+                    desired = None;
+                }
+                for (target, state) in hosts.iter().zip(&mut targets) {
+                    if *target == event.host {
+                        *state = TargetIntent::Stopped(event.order);
+                    }
+                }
+            }
+        }
+    }
+    let view = PlacementProjection::new(events);
+    assert_eq!(view.desired(), desired, "events: {events:?}");
+    for (target, state) in hosts.iter().zip(targets) {
+        assert_eq!(view.target(*target), state, "events: {events:?}");
+        for event in events {
+            assert_eq!(
+                view.retains_start(*target, event.order),
+                state == TargetIntent::Running(event.order)
+            );
+        }
+    }
+}
+
+fn permutations(events: &mut [PlacementIntent], offset: usize, count: &mut usize) {
+    if offset == events.len() {
+        *count += 1;
+        // Every partial delivery/backfill set is projected independently.
+        for end in 0..=events.len() {
+            assert_matches_ordered_model(&events[..end]);
+        }
+        let mut duplicated = events.to_vec();
+        duplicated.extend_from_slice(events);
+        assert_matches_ordered_model(&duplicated);
+        return;
+    }
+    for next in offset..events.len() {
+        events.swap(offset, next);
+        permutations(events, offset + 1, count);
+        events.swap(offset, next);
+    }
+}
+
+#[test]
+fn all_delivery_permutations_and_prefixes_match_signed_order_projection() {
+    for mut events in [
+        [
+            start(1, 10),
+            start(2, 20),
+            stop(1, 30),
+            stop(2, 40),
+            start(3, 50),
+            stop(3, 60),
+        ],
+        [
+            start(1, 10),
+            start(2, 10),
+            stop(1, 10),
+            stop(2, 10),
+            start(3, 10),
+            stop(3, 10),
+        ],
+    ] {
+        let mut count = 0;
+        permutations(&mut events, 0, &mut count);
+        assert_eq!(count, 720);
+    }
+}