feat(server): commit intrinsic event message text

crates/bacnet-server/src/server/event_confirmed_routing_tests.rs

@@ -13,6 +13,7 @@
 //! feed acks through the same correlation entry point the dispatch loop uses.
 
 use super::device_bindings::{DeviceBindingTable, OBSERVED_BINDING_TTL};
+use super::event_notifications::CommittedIntrinsicTransition;
 use super::event_notifications_tests::local_broadcast_destination;
 use super::event_recipient_routing_tests::{address_recipient, destination_for};
 use super::*;
@@ -161,6 +162,48 @@ impl Harness {
         }
     }
 
+    async fn commit_transition(
+        &self,
+        from: EventState,
+        to: EventState,
+    ) -> CommittedIntrinsicTransition {
+        let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap();
+        let mut db = self.db.write().await;
+        BACnetServer::<RecordingTransport>::commit_intrinsic_transition(
+            &mut db,
+            &oid,
+            bacnet_objects::event::TransitionOutcome {
+                change: EventStateChange { from, to },
+                event_type: EventType::OUT_OF_RANGE,
+                distribute: true,
+            },
+        )
+        .expect("the built-in transition must commit")
+    }
+
+    async fn distribute_committed(&self) -> ObjectIdentifier {
+        let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap();
+        let committed = self
+            .commit_transition(EventState::NORMAL, EventState::HIGH_LIMIT)
+            .await;
+        BACnetServer::<RecordingTransport>::build_and_send_event_notification_with_bindings(
+            &self.db,
+            &self.network,
+            &self.comm_state,
+            &self.server_tsm,
+            &self.notification_transactions,
+            &self.device_bindings,
+            &oid,
+            committed,
+            self.retry_timeout_ms,
+        )
+        .await;
+        for _ in 0..16 {
+            tokio::task::yield_now().await;
+        }
+        oid
+    }
+
     fn broadcast_frames(&self) -> Vec<Bytes> {
         self.broadcasts.lock().unwrap().clone()
     }
@@ -284,6 +327,70 @@ async fn configured_routed_device_retries_unicast_to_router_and_correlates_by_fi
     );
 }
 
+#[tokio::test(start_paused = true)]
+async fn confirmed_retry_reuses_committed_message_bytes_after_history_changes() {
+    let identifier = ObjectIdentifier::new(ObjectType::DEVICE, 92).unwrap();
+    let mut bindings = DeviceBindingTable::new();
+    bindings
+        .insert_configured(
+            DeviceBinding::routed(identifier, 1002, RECIPIENT, ROUTER_A).unwrap(),
+            |_| false,
+        )
+        .unwrap();
+    let harness = Harness::new_with_bindings(
+        vec![destination_for(BACnetRecipient::Device(identifier), true)],
+        1_000,
+        bindings,
+    )
+    .await;
+    let oid = harness.distribute_committed().await;
+
+    let first = harness.unicast_frames();
+    assert_eq!(first.len(), 1);
+    let first_frame = first[0].1.clone();
+    let (_, first_request) = decode_confirmed(&first_frame);
+    let notification = EventNotificationRequest::decode(&first_request.service_request).unwrap();
+    assert_eq!(
+        notification.message_text,
+        Some("ANALOG_INPUT,1: NORMAL -> HIGH_LIMIT".into())
+    );
+
+    harness
+        .commit_transition(EventState::HIGH_LIMIT, EventState::NORMAL)
+        .await;
+    harness
+        .commit_transition(EventState::NORMAL, EventState::LOW_LIMIT)
+        .await;
+    let PropertyValue::CharacterString(current_message) = harness
+        .db
+        .read()
+        .await
+        .get(&oid)
+        .unwrap()
+        .read_property(PropertyIdentifier::EVENT_MESSAGE_TEXTS, Some(1))
+        .unwrap()
+    else {
+        panic!("Event_Message_Texts coordinate must be a character string");
+    };
+    assert_eq!(current_message, "ANALOG_INPUT,1: NORMAL -> LOW_LIMIT");
+
+    tokio::time::advance(Duration::from_secs(1)).await;
+    for _ in 0..16 {
+        tokio::task::yield_now().await;
+    }
+    let retried = harness.unicast_frames();
+    assert!(retried.len() >= 2, "silence must trigger a retry");
+    assert!(
+        retried.iter().all(|(_, frame)| frame == &first_frame),
+        "every retry must reuse the originally committed encoded bytes"
+    );
+    assert!(
+        harness
+            .ack_routed(ROUTER_A, 1002, RECIPIENT, first_request.invoke_id)
+            .await
+    );
+}
+
 #[tokio::test(start_paused = true)]
 async fn observed_routed_device_stops_emitting_when_retry_reaches_expiry() {
     let identifier = ObjectIdentifier::new(ObjectType::DEVICE, 91).unwrap();

crates/bacnet-server/src/server/event_message_policy.rs

@@ -0,0 +1,10 @@
+use bacnet_objects::event::EventStateChange;
+use bacnet_types::primitives::ObjectIdentifier;
+
+/// Select the server-owned message for one built-in intrinsic transition.
+pub(super) fn intrinsic_event_message_text(
+    object_identifier: &ObjectIdentifier,
+    change: &EventStateChange,
+) -> String {
+    format!("{object_identifier}: {} -> {}", change.from, change.to)
+}

crates/bacnet-server/src/server/event_message_policy_tests.rs

@@ -0,0 +1,189 @@
+use super::*;
+use crate::server::event_notifications::CommittedIntrinsicTransition;
+use bacnet_objects::analog::{AnalogInputObject, AnalogOutputObject, AnalogValueObject};
+use bacnet_objects::binary::{BinaryInputObject, BinaryOutputObject, BinaryValueObject};
+use bacnet_objects::event::TransitionOutcome;
+use bacnet_objects::multistate::{
+    MultiStateInputObject, MultiStateOutputObject, MultiStateValueObject,
+};
+use bacnet_objects::traits::BACnetObject;
+use bacnet_types::enums::{EventState, EventType};
+
+fn builtin_intrinsic_objects() -> Vec<Box<dyn BACnetObject>> {
+    vec![
+        Box::new(AnalogInputObject::new(41, "localized analog input", 0).unwrap()),
+        Box::new(AnalogOutputObject::new(41, "localized analog output", 0).unwrap()),
+        Box::new(AnalogValueObject::new(41, "localized analog value", 0).unwrap()),
+        Box::new(BinaryInputObject::new(41, "localized binary input").unwrap()),
+        Box::new(BinaryOutputObject::new(41, "localized binary output").unwrap()),
+        Box::new(BinaryValueObject::new(41, "localized binary value").unwrap()),
+        Box::new(MultiStateInputObject::new(41, "localized multistate input", 3).unwrap()),
+        Box::new(MultiStateOutputObject::new(41, "localized multistate output", 3).unwrap()),
+        Box::new(MultiStateValueObject::new(41, "localized multistate value", 3).unwrap()),
+    ]
+}
+
+fn commit(
+    db: &mut ObjectDatabase,
+    oid: ObjectIdentifier,
+    from: EventState,
+    to: EventState,
+    distribute: bool,
+) -> Option<CommittedIntrinsicTransition> {
+    BACnetServer::<RecordingTransport>::commit_intrinsic_transition(
+        db,
+        &oid,
+        TransitionOutcome {
+            change: EventStateChange { from, to },
+            event_type: EventType::OUT_OF_RANGE,
+            distribute,
+        },
+    )
+}
+
+fn message_slots(db: &ObjectDatabase, oid: ObjectIdentifier) -> [String; 3] {
+    std::array::from_fn(|index| {
+        let PropertyValue::CharacterString(text) = db
+            .get(&oid)
+            .unwrap()
+            .read_property(
+                PropertyIdentifier::EVENT_MESSAGE_TEXTS,
+                Some(index as u32 + 1),
+            )
+            .unwrap()
+        else {
+            panic!("Event_Message_Texts coordinate must be a character string");
+        };
+        text
+    })
+}
+
+fn committed_properties(db: &ObjectDatabase, oid: ObjectIdentifier) -> Vec<PropertyValue> {
+    let object = db.get(&oid).unwrap();
+    let mut values = vec![
+        object
+            .read_property(PropertyIdentifier::EVENT_STATE, None)
+            .unwrap(),
+        object
+            .read_property(PropertyIdentifier::ACKED_TRANSITIONS, None)
+            .unwrap(),
+    ];
+    for property in [
+        PropertyIdentifier::EVENT_TIME_STAMPS,
+        PropertyIdentifier::EVENT_MESSAGE_TEXTS,
+    ] {
+        values.extend((1..=3).map(|index| object.read_property(property, Some(index)).unwrap()));
+    }
+    values
+}
+
+#[test]
+fn all_nine_builtin_families_store_each_policy_message_in_only_its_coordinate() {
+    let objects = builtin_intrinsic_objects();
+    assert_eq!(objects.len(), 9);
+
+    for object in objects {
+        let oid = object.object_identifier();
+        let mut db = ObjectDatabase::new();
+        db.add(object).unwrap();
+        let cases = [
+            (EventState::NORMAL, EventState::OFFNORMAL),
+            (EventState::OFFNORMAL, EventState::FAULT),
+            (EventState::FAULT, EventState::NORMAL),
+        ];
+        let mut expected = std::array::from_fn(|_| String::new());
+
+        for (index, (from, to)) in cases.into_iter().enumerate() {
+            assert!(
+                commit(&mut db, oid, from, to, false).is_some(),
+                "{oid} must use the built-in atomic commit path"
+            );
+            expected[index] = format!("{oid}: {from} -> {to}");
+            assert_eq!(
+                message_slots(&db, oid),
+                expected,
+                "{oid} must update only transition coordinate {index}"
+            );
+        }
+    }
+}
+
+#[test]
+fn policy_format_uses_object_and_state_display_including_unknown_state_numbers() {
+    let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 7).unwrap();
+    let unknown = EventState::from_raw(65_535);
+    let mut db = ObjectDatabase::new();
+    db.add(Box::new(
+        AnalogInputObject::new(7, "name is not policy", 0).unwrap(),
+    ))
+    .unwrap();
+
+    assert!(commit(&mut db, oid, EventState::NORMAL, unknown, true).is_some());
+    assert_eq!(
+        message_slots(&db, oid),
+        [
+            "ANALOG_INPUT,7: NORMAL -> 65535".into(),
+            String::new(),
+            String::new(),
+        ]
+    );
+}
+
+#[test]
+fn stale_commit_does_not_mutate_committed_event_properties() {
+    let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 8).unwrap();
+    let mut db = ObjectDatabase::new();
+    db.add(Box::new(AnalogInputObject::new(8, "AI-8", 0).unwrap()))
+        .unwrap();
+    assert!(commit(
+        &mut db,
+        oid,
+        EventState::NORMAL,
+        EventState::HIGH_LIMIT,
+        true,
+    )
+    .is_some());
+    let before = committed_properties(&db, oid);
+
+    assert!(commit(&mut db, oid, EventState::NORMAL, EventState::FAULT, true,).is_none());
+    assert_eq!(committed_properties(&db, oid), before);
+}
+
+#[tokio::test]
+async fn outbound_message_text_equals_the_committed_history_coordinate() {
+    let db = db_with_high_limit_transition(0x80);
+    let sent = broadcasts_from_per_write_path(&db, 0).await;
+    let notification = decode_broadcast_notification(&StdMutex::new(sent));
+    let expected = "ANALOG_INPUT,1: NORMAL -> HIGH_LIMIT";
+    let history = message_slots(
+        &*db.read().await,
+        ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap(),
+    );
+
+    assert_eq!(history, [expected.into(), String::new(), String::new()]);
+    assert_eq!(notification.message_text, Some(expected.into()));
+}
+
+#[tokio::test]
+async fn event_enable_and_dcc_suppression_still_commit_the_policy_message() {
+    for (event_enable, dcc, label) in [
+        (0x00, 0, "Event_Enable"),
+        (0x80, 1, "device communication control"),
+    ] {
+        let db = db_with_high_limit_transition(event_enable);
+        let sent = broadcasts_from_per_write_path(&db, dcc).await;
+        assert!(sent.is_empty(), "{label} must suppress distribution");
+        assert_eq!(
+            message_slots(
+                &*db.read().await,
+                ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap(),
+            ),
+            [
+                "ANALOG_INPUT,1: NORMAL -> HIGH_LIMIT".into(),
+                String::new(),
+                String::new(),
+            ],
+            "{label} must not suppress the local message-history commit"
+        );
+    }
+}

crates/bacnet-server/src/server/event_notifications.rs

@@ -1,3 +1,4 @@
+use super::event_message_policy::intrinsic_event_message_text;
 use super::event_recipient_route::{ConfirmedRecipientRoute, RecipientRoute};
 use super::event_timestamp::{
     confirm_event_timestamp, sample_event_timestamp, stage_event_timestamp, SampledEventClock,
@@ -265,12 +266,13 @@ impl<T: TransportPort + 'static> BACnetServer<T> {
             .unwrap_or(0);
         let (_, ack_required) = resolve_transition_priority_ack(db, notification_class, coordinate);
         let staged_timestamp = stage_event_timestamp(db);
+        let message_text = intrinsic_event_message_text(oid, &outcome.change);
         let commit = EventTransitionCommit {
             change: outcome.change.clone(),
             coordinate,
             ack_required,
             timestamp: staged_timestamp.sample.timestamp.clone(),
-            message_text: None,
+            message_text: Some(message_text),
         };
 
         if let Err(error) = db.get_mut(oid)?.commit_event_transition_internal(commit) {
@@ -378,8 +380,8 @@ impl<T: TransportPort + 'static> BACnetServer<T> {
         // transition actions, none of which it governs.
         //
         // The shared commit kernel has also stored the selected timestamp and
-        // updated Acked_Transitions from the Notification Class policy. Message
-        // text is intentionally absent for this built-in path.
+        // updated Acked_Transitions from the Notification Class policy and
+        // stored the selected local message in the transition coordinate.
         if let Some(resolved) = resolved {
             if resolved.distribute() {
                 Self::build_and_send_event_notification_with_bindings(

crates/bacnet-server/src/server/event_notifications_tests.rs

@@ -15,6 +15,9 @@ mod commit_tests;
 #[path = "event_notifications_history_tests.rs"]
 mod history_tests;
 
+#[path = "event_message_policy_tests.rs"]
+mod message_policy_tests;
+
 /// A transport that records every broadcast NPDU it is asked to send and
 /// discards unicasts. Used to capture the EventNotification a server
 /// actually puts on the wire.

crates/bacnet-server/src/server/mod.rs

@@ -859,6 +859,7 @@ mod cov_notifications;
 mod device_bindings;
 mod dispatch;
 mod event_enrollment_lifecycle;
+mod event_message_policy;
 mod event_notifications;
 mod event_recipient_route;
 pub(crate) mod event_timestamp;