From c9e0f9c85df3dc9be9d02aca64d80081b0f15b67 Mon Sep 17 00:00:00 2001
From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com>
Date: Tue, 14 Jul 2026 20:16:39 -0700
Subject: [PATCH 1/5] test(runtime): failing properties for delayed-message
 ownership (ARN-236)

RED: Scheduler::tick() both enqueues a due message into the target mailbox
AND returns a clone; the simulation drivers process the returned clones,
never drain mailboxes, and end each iteration with a bare tick() whose
deliveries are discarded; a rejected integration callback is silently
dropped. Three seeded properties pin the consequences on main: a fault-free
run leaves every processed message queued (30 in-mailbox after 30
applications); a 50-seed delay-fault sweep loses trailing deliveries (seed
0: 44 sent, 30 applied); a callback mapped to an always-failing action
still yields a green result.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../src/scheduler/sim_actor_system.rs         | 182 ++++++++++++++++++
 1 file changed, 182 insertions(+)

diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system.rs b/crates/temper-runtime/src/scheduler/sim_actor_system.rs
index 8009277c4..31b12d3a6 100644
--- a/crates/temper-runtime/src/scheduler/sim_actor_system.rs
+++ b/crates/temper-runtime/src/scheduler/sim_actor_system.rs
@@ -707,6 +707,188 @@ fn evaluate_spec_assert(
 mod tests {
     use super::*;
 
+    // ── ARN-236: delayed-message ownership properties ─────────────────────
+    //
+    // Scheduler::tick() both enqueues a due message into the target mailbox
+    // AND returns a clone; the drivers process the returned clones and never
+    // drain mailboxes, and each loop iteration ends with a bare tick() whose
+    // returned deliveries are discarded. Consequences these tests pin:
+    // processed messages remain queued forever, deliveries surfaced only by
+    // the trailing tick are never applied, and a failing integration
+    // callback still yields a green run.
+
+    /// Accepts every action, counts applications, and (optionally) emits a
+    /// callback trigger whose configured action always fails.
+    struct CountingHandler {
+        applications: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+        emit_trigger: bool,
+        fired: bool,
+    }
+
+    impl SimActorHandler for CountingHandler {
+        fn init(&mut self) -> Result<serde_json::Value, String> {
+            Ok(serde_json::json!({"status": "Ready"}))
+        }
+        fn handle_message(
+            &mut self,
+            action: &str,
+            _params: &str,
+        ) -> Result<serde_json::Value, String> {
+            if action == "AlwaysFails" {
+                return Err("callback action rejected".to_string());
+            }
+            self.applications
+                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            if self.emit_trigger {
+                self.fired = true;
+            }
+            Ok(serde_json::json!({"status": "Ready"}))
+        }
+        fn current_status(&self) -> String {
+            "Ready".to_string()
+        }
+        fn current_item_count(&self) -> usize {
+            0
+        }
+        fn event_count(&self) -> usize {
+            self.applications.load(std::sync::atomic::Ordering::SeqCst)
+        }
+        fn valid_actions(&self) -> Vec<String> {
+            vec!["Step".to_string()]
+        }
+        fn events_json(&self) -> serde_json::Value {
+            serde_json::json!([])
+        }
+        fn pending_callbacks(&self) -> Vec<String> {
+            if self.fired {
+                vec!["boom_trigger".to_string()]
+            } else {
+                Vec::new()
+            }
+        }
+    }
+
+    fn counting_system(
+        seed: u64,
+        faults: FaultConfig,
+    ) -> (
+        SimActorSystem,
+        std::sync::Arc<std::sync::atomic::AtomicUsize>,
+    ) {
+        let applications = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let config = SimActorSystemConfig {
+            seed,
+            max_ticks: 200,
+            faults,
+            max_actions_per_actor: 30,
+        };
+        let mut system = SimActorSystem::new(config);
+        system.register_actor(
+            "counter",
+            Box::new(CountingHandler {
+                applications: applications.clone(),
+                emit_trigger: false,
+                fired: false,
+            }),
+        );
+        (system, applications)
+    }
+
+    /// No processed message may remain queued: after a run, every mailbox is
+    /// empty and the scheduler is quiescent.
+    #[test]
+    fn arn236_no_processed_message_remains_queued() {
+        let faults = FaultConfig {
+            message_delay_prob: 0.0,
+            max_delay_ticks: 0,
+            message_drop_prob: 0.0,
+            actor_crash_prob: 0.0,
+            actor_restart_prob: 0.0,
+        };
+        let (mut system, _applications) = counting_system(7, faults);
+        let result = system.run_random();
+        assert!(result.messages > 0, "the run must exercise messages");
+
+        assert_eq!(
+            system.scheduler.mailbox_depth("counter"),
+            0,
+            "a processed message must not remain queued in its mailbox \
+             (single-ownership: applied messages are consumed, not cloned)"
+        );
+        assert!(
+            system.scheduler.is_quiescent(),
+            "after a fault-free run every delivered message must be consumed"
+        );
+    }
+
+    /// Every scheduled message is applied exactly once — deliveries surfaced
+    /// by the loop's trailing tick must not be silently discarded. With
+    /// message delays (no drops, no crashes), every sent message is
+    /// eventually due, so applications must equal sends across all seeds.
+    #[test]
+    fn arn236_every_scheduled_message_is_applied_exactly_once() {
+        for seed in 0..50u64 {
+            let faults = FaultConfig {
+                message_delay_prob: 0.5,
+                max_delay_ticks: 8,
+                message_drop_prob: 0.0,
+                actor_crash_prob: 0.0,
+                actor_restart_prob: 0.0,
+            };
+            let (mut system, applications) = counting_system(seed, faults);
+            let result = system.run_random();
+            let applied = applications.load(std::sync::atomic::Ordering::SeqCst);
+            assert_eq!(
+                applied as u64, result.messages,
+                "seed {seed}: every scheduled message must be applied exactly \
+                 once ({} sent, {applied} applied) — a delivery surfaced only \
+                 by the trailing tick must not be discarded, and none may \
+                 apply twice",
+                result.messages
+            );
+        }
+    }
+
+    /// A failing integration callback must fail the run, not vanish.
+    #[test]
+    fn arn236_callback_failure_is_part_of_the_simulation_result() {
+        let applications = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let config = SimActorSystemConfig {
+            seed: 11,
+            max_ticks: 50,
+            faults: FaultConfig {
+                message_delay_prob: 0.0,
+                max_delay_ticks: 0,
+                message_drop_prob: 0.0,
+                actor_crash_prob: 0.0,
+                actor_restart_prob: 0.0,
+            },
+            max_actions_per_actor: 3,
+        };
+        let mut system = SimActorSystem::new(config);
+        system.set_integration_responses(SimIntegrationResponses::new().on_trigger(
+            "counter",
+            "boom_trigger",
+            "AlwaysFails",
+        ));
+        system.register_actor(
+            "counter",
+            Box::new(CountingHandler {
+                applications: applications.clone(),
+                emit_trigger: true,
+                fired: false,
+            }),
+        );
+
+        let result = system.run_random();
+        assert!(
+            !result.all_invariants_held || !result.violations.is_empty(),
+            "a rejected integration callback must surface in the simulation \
+             result — a green run that silently discarded a callback failure \
+             does not faithfully exercise the schedule"
+        );
+    }
+
     #[test]
     fn integration_responses_empty_returns_none() {
         let responses = SimIntegrationResponses::new();

From 7d634fe1b5feb2a131fc66ec49da26a4adee6222 Mon Sep 17 00:00:00 2001
From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com>
Date: Tue, 14 Jul 2026 20:49:59 -0700
Subject: [PATCH 2/5] fix(runtime): single-ownership delivery for scheduled
 simulation messages (ARN-236)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

GREEN: Scheduler::tick() now advances logical time and enqueues only — the
clone-return is deleted. A message is owned by exactly one place at every
instant: the pending queue, a mailbox, or the consumer that drained it. The
new drain_ready() removes and returns all queued messages in deterministic
order (actor-id order, FIFO per mailbox) and is the drivers' single
consumption path; per-message application is extracted (apply_delivered_message
in the runtime driver, apply_to_model in the verifier) so the main loop and
the new budgeted flush share one exactly-once path. The flush ticks and
drains until quiescent (budget = max_ticks), so deliveries pushed past the
last driver iteration by delay faults are applied instead of discarded. A
rejected integration callback is recorded as a violation on the simulation
result instead of vanishing behind a let-underscore.

The verifier and runtime drivers now exercise the same delivery contract.
Same-seed determinism holds (drain order is BTreeMap + FIFO; RNG draw order
unchanged); recorded traces differ from pre-fix traces because previously
lost deliveries are now applied — the point of the change. ADR-0165.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 crates/temper-runtime/src/scheduler/core.rs   |  45 +++--
 .../src/scheduler/sim_actor_system.rs         | 168 ++++++++++--------
 crates/temper-verify/src/simulation.rs        | 101 ++++++++---
 .../0165-sim-delivery-single-ownership.md     |  75 ++++++++
 4 files changed, 282 insertions(+), 107 deletions(-)
 create mode 100644 docs/adrs/0165-sim-delivery-single-ownership.md

diff --git a/crates/temper-runtime/src/scheduler/core.rs b/crates/temper-runtime/src/scheduler/core.rs
index 75b5cf3b4..3224cece8 100644
--- a/crates/temper-runtime/src/scheduler/core.rs
+++ b/crates/temper-runtime/src/scheduler/core.rs
@@ -135,14 +135,23 @@ impl SimScheduler {
         });
     }
 
-    /// Advance one tick: deliver all messages due at current_time + 1.
-    /// Returns the messages delivered this tick.
-    pub fn tick(&mut self) -> Vec<SimMessage> {
+    /// Advance one tick: enqueue all messages due at the new current time
+    /// into their target mailboxes.
+    ///
+    /// ARN-236 single-ownership contract: `tick` advances logical time and
+    /// enqueues ONLY — it does not hand deliveries to the caller. The one
+    /// consumption path is [`Self::drain_ready`] (or [`Self::receive`] for a
+    /// single actor), which REMOVES messages from mailboxes. A message is
+    /// therefore owned by exactly one place at every instant: the pending
+    /// queue, a mailbox, or the consumer that drained it. (Previously `tick`
+    /// both enqueued and returned clones; drivers processed the clones,
+    /// mailboxes grew forever, and deliveries surfaced by a discarded tick
+    /// were lost.)
+    pub fn tick(&mut self) {
         self.current_time += 1;
         self.ticks += 1;
-        let mut delivered_this_tick = Vec::new();
 
-        // Deliver all messages due at or before current time
+        // Enqueue all messages due at or before current time
         while let Some(msg) = self.pending.peek() {
             if msg.deliver_at <= self.current_time {
                 let msg = self.pending.pop().unwrap(); // ci-ok: guarded by peek() above
@@ -152,9 +161,8 @@ impl SimScheduler {
                 let actor_state = self.actor_states.get(&to).cloned();
                 match actor_state {
                     Some(SimActorState::Running) => {
-                        self.mailboxes.entry(to).or_default().push_back(msg.clone());
-                        delivered_this_tick.push(msg.clone());
-                        self.delivered.push(msg);
+                        self.delivered.push(msg.clone());
+                        self.mailboxes.entry(to).or_default().push_back(msg);
                     }
                     Some(SimActorState::Crashed) => {
                         // Actor is crashed — message is lost (or could be re-queued)
@@ -175,7 +183,7 @@ impl SimScheduler {
             }
         }
 
-        // Maybe crash an actor after delivery
+        // Maybe crash an actor after enqueue
         if self.rng.chance(self.fault_config.actor_crash_prob) {
             let running: Vec<String> = self
                 .actor_states
@@ -189,8 +197,25 @@ impl SimScheduler {
                     .insert(running[idx].clone(), SimActorState::Crashed);
             }
         }
+    }
 
-        delivered_this_tick
+    /// Remove and return every queued message, in deterministic order
+    /// (actor id order, FIFO within each mailbox).
+    ///
+    /// This is the single delivery-consumption path for drivers (ARN-236):
+    /// a drained message has left the scheduler entirely and is applied by
+    /// the caller exactly once.
+    ///
+    /// `drain_ready` does not consult `actor_states`: a driver that ticks
+    /// several times before draining would apply messages to actors that
+    /// crashed after enqueue. Both current drivers drain immediately after
+    /// every tick, so the drained set is exactly that tick's enqueues.
+    pub fn drain_ready(&mut self) -> Vec<SimMessage> {
+        let mut ready = Vec::new();
+        for queue in self.mailboxes.values_mut() {
+            ready.extend(queue.drain(..));
+        }
+        ready
     }
 
     /// Take the next message from an actor's mailbox.
diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system.rs b/crates/temper-runtime/src/scheduler/sim_actor_system.rs
index 31b12d3a6..7e7e6a884 100644
--- a/crates/temper-runtime/src/scheduler/sim_actor_system.rs
+++ b/crates/temper-runtime/src/scheduler/sim_actor_system.rs
@@ -17,7 +17,7 @@ use super::clock::{LogicalClock, SimClock};
 use super::context::{SimContextGuard, install_sim_context};
 use super::id_gen::DeterministicIdGen;
 use super::sim_handler::SimActorHandler;
-use super::{DeterministicRng, FaultConfig, SimScheduler};
+use super::{DeterministicRng, FaultConfig, SimMessage, SimScheduler};
 
 /// Configures how integration callbacks are delivered in simulation.
 ///
@@ -355,6 +355,49 @@ impl SimActorSystem {
     // Random Mode
     // ===================================================================
 
+    /// Apply one drained message to its actor: run the handler, record the
+    /// transition, check invariants, and schedule integration callbacks.
+    /// The exactly-once application step of the ARN-236 ownership model.
+    fn apply_delivered_message(&mut self, msg: &SimMessage) {
+        if let Some(handler) = self.actors.get_mut(&msg.to) {
+            let status_before = handler.current_status();
+
+            match handler.handle_message(&msg.msg_type, &msg.payload) {
+                Ok(_) => {
+                    let status_after = handler.current_status();
+                    let item_count = handler.current_item_count();
+                    let tick = self.clock.tick();
+                    *self.action_counts.get_mut(&msg.to).unwrap() += 1; // ci-ok: actor always in action_counts
+                    self.total_transitions += 1;
+
+                    // Record the transition
+                    self.recorded_transitions.push((
+                        tick,
+                        msg.to.clone(),
+                        msg.msg_type.clone(),
+                        status_before.clone(),
+                        status_after.clone(),
+                    ));
+
+                    self.check_invariants(
+                        &msg.to,
+                        &msg.msg_type,
+                        &status_before,
+                        &status_after,
+                        item_count,
+                        tick,
+                    );
+
+                    // Schedule integration callbacks for any custom effects
+                    self.schedule_integration_callbacks(&msg.to);
+                }
+                Err(_) => {
+                    // Action failed — expected for invalid transitions
+                }
+            }
+        }
+    }
+
     /// Run random exploration with fault injection.
     ///
     /// The RNG picks actors and actions. The scheduler delays/drops/crashes.
@@ -394,57 +437,38 @@ impl SimActorSystem {
             self.scheduler.send("sim-driver", &actor_id, &action, "{}");
             self.total_messages += 1;
 
-            let delivered = self.scheduler.tick();
+            self.scheduler.tick();
             self.clock.advance();
 
-            // Process delivered messages
+            // Single-ownership delivery (ARN-236): drain_ready removes the
+            // messages from their mailboxes; each is applied exactly once.
+            let delivered = self.scheduler.drain_ready();
             for msg in &delivered {
-                if let Some(handler) = self.actors.get_mut(&msg.to) {
-                    let status_before = handler.current_status();
-
-                    match handler.handle_message(&msg.msg_type, &msg.payload) {
-                        Ok(_) => {
-                            let status_after = handler.current_status();
-                            let item_count = handler.current_item_count();
-                            let tick = self.clock.tick();
-                            *self.action_counts.get_mut(&msg.to).unwrap() += 1; // ci-ok: actor always in action_counts
-                            self.total_transitions += 1;
-
-                            // Record the transition
-                            self.recorded_transitions.push((
-                                tick,
-                                msg.to.clone(),
-                                msg.msg_type.clone(),
-                                status_before.clone(),
-                                status_after.clone(),
-                            ));
-
-                            self.check_invariants(
-                                &msg.to,
-                                &msg.msg_type,
-                                &status_before,
-                                &status_after,
-                                item_count,
-                                tick,
-                            );
-
-                            // Schedule integration callbacks for any custom effects
-                            self.schedule_integration_callbacks(&msg.to);
-                        }
-                        Err(_) => {
-                            // Action failed — expected for invalid transitions
-                        }
-                    }
-                }
+                self.apply_delivered_message(msg);
             }
 
             // Deliver any pending integration callbacks
             if !self.pending_integration_callbacks.is_empty() {
                 self.deliver_integration_callbacks();
             }
+        }
 
-            // Drain any remaining scheduled messages
+        // Flush the schedule: delay faults can push deliveries past the last
+        // driver iteration. Budgeted (never unbounded), and every flushed
+        // message goes through the same exactly-once drain path — previously
+        // a bare tick() discarded these deliveries entirely (ARN-236).
+        let mut flush_budget = self.config.max_ticks;
+        while !self.scheduler.is_quiescent() && flush_budget > 0 {
+            flush_budget -= 1;
             self.scheduler.tick();
+            self.clock.advance();
+            let delivered = self.scheduler.drain_ready();
+            for msg in &delivered {
+                self.apply_delivered_message(msg);
+            }
+            if !self.pending_integration_callbacks.is_empty() {
+                self.deliver_integration_callbacks();
+            }
         }
 
         let actor_states: Vec<_> = self
@@ -569,12 +593,32 @@ impl SimActorSystem {
     }
 
     /// Deliver any pending integration callbacks by executing them as actions.
+    ///
+    /// A rejected callback is part of the simulation result (ARN-236): the
+    /// configured callback action failing to apply means the simulated
+    /// integration contract is broken, and the run must say so rather than
+    /// stay green.
     fn deliver_integration_callbacks(&mut self) {
         let callbacks: Vec<(String, String)> =
             self.pending_integration_callbacks.drain(..).collect();
         for (actor_id, callback_action) in callbacks {
             // Execute the callback as a regular step (this checks invariants too)
-            let _ = self.step(&actor_id, &callback_action, "{}");
+            if let Err(error) = self.step(&actor_id, &callback_action, "{}") {
+                let tick = self.clock.tick();
+                let status = self
+                    .actors
+                    .get(&actor_id)
+                    .map(|h| h.current_status())
+                    .unwrap_or_default();
+                self.violations.push(ActorInvariantViolation {
+                    actor_id,
+                    action: callback_action,
+                    status_before: status.clone(),
+                    status_after: status,
+                    description: format!("integration callback rejected: {error}"),
+                    tick,
+                });
+            }
         }
     }
 
@@ -705,6 +749,8 @@ fn evaluate_spec_assert(
 
 #[cfg(test)]
 mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
     use super::*;
 
     // ── ARN-236: delayed-message ownership properties ─────────────────────
@@ -720,7 +766,7 @@ mod tests {
     /// Accepts every action, counts applications, and (optionally) emits a
     /// callback trigger whose configured action always fails.
     struct CountingHandler {
-        applications: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+        applications: Arc<AtomicUsize>,
         emit_trigger: bool,
         fired: bool,
     }
@@ -737,8 +783,7 @@ mod tests {
             if action == "AlwaysFails" {
                 return Err("callback action rejected".to_string());
             }
-            self.applications
-                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            self.applications.fetch_add(1, Ordering::SeqCst);
             if self.emit_trigger {
                 self.fired = true;
             }
@@ -751,7 +796,7 @@ mod tests {
             0
         }
         fn event_count(&self) -> usize {
-            self.applications.load(std::sync::atomic::Ordering::SeqCst)
+            self.applications.load(Ordering::SeqCst)
         }
         fn valid_actions(&self) -> Vec<String> {
             vec!["Step".to_string()]
@@ -768,14 +813,8 @@ mod tests {
         }
     }
 
-    fn counting_system(
-        seed: u64,
-        faults: FaultConfig,
-    ) -> (
-        SimActorSystem,
-        std::sync::Arc<std::sync::atomic::AtomicUsize>,
-    ) {
-        let applications = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    fn counting_system(seed: u64, faults: FaultConfig) -> (SimActorSystem, Arc<AtomicUsize>) {
+        let applications = Arc::new(AtomicUsize::new(0));
         let config = SimActorSystemConfig {
             seed,
             max_ticks: 200,
@@ -798,14 +837,7 @@ mod tests {
     /// empty and the scheduler is quiescent.
     #[test]
     fn arn236_no_processed_message_remains_queued() {
-        let faults = FaultConfig {
-            message_delay_prob: 0.0,
-            max_delay_ticks: 0,
-            message_drop_prob: 0.0,
-            actor_crash_prob: 0.0,
-            actor_restart_prob: 0.0,
-        };
-        let (mut system, _applications) = counting_system(7, faults);
+        let (mut system, _applications) = counting_system(7, FaultConfig::none());
         let result = system.run_random();
         assert!(result.messages > 0, "the run must exercise messages");
 
@@ -837,7 +869,7 @@ mod tests {
             };
             let (mut system, applications) = counting_system(seed, faults);
             let result = system.run_random();
-            let applied = applications.load(std::sync::atomic::Ordering::SeqCst);
+            let applied = applications.load(Ordering::SeqCst);
             assert_eq!(
                 applied as u64, result.messages,
                 "seed {seed}: every scheduled message must be applied exactly \
@@ -852,17 +884,11 @@ mod tests {
     /// A failing integration callback must fail the run, not vanish.
     #[test]
     fn arn236_callback_failure_is_part_of_the_simulation_result() {
-        let applications = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let applications = Arc::new(AtomicUsize::new(0));
         let config = SimActorSystemConfig {
             seed: 11,
             max_ticks: 50,
-            faults: FaultConfig {
-                message_delay_prob: 0.0,
-                max_delay_ticks: 0,
-                message_drop_prob: 0.0,
-                actor_crash_prob: 0.0,
-                actor_restart_prob: 0.0,
-            },
+            faults: FaultConfig::none(),
             max_actions_per_actor: 3,
         };
         let mut system = SimActorSystem::new(config);
diff --git a/crates/temper-verify/src/simulation.rs b/crates/temper-verify/src/simulation.rs
index f68cb27a1..0b7315191 100644
--- a/crates/temper-verify/src/simulation.rs
+++ b/crates/temper-verify/src/simulation.rs
@@ -208,37 +208,45 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
         );
         total_messages += 1;
 
-        let delivered = sched.tick();
-
-        for msg in &delivered {
-            let target_idx = actor_states.iter().position(|(id, _)| id == &msg.to);
-            let Some(idx) = target_idx else { continue };
-
-            let (ref target_id, ref state_before) = actor_states[idx];
+        sched.tick();
 
-            let action: TemperModelAction = match serde_json::from_str(&msg.payload) {
-                Ok(a) => a,
-                Err(_) => continue,
-            };
+        // Single-ownership delivery (ARN-236): drain_ready removes messages
+        // from mailboxes; each is applied to the model exactly once.
+        let delivered = sched.drain_ready();
 
-            if let Some(new_state) = model.next_state(state_before, action.clone()) {
-                check_invariants_on_state(
-                    model,
-                    target_id,
-                    &action.name,
-                    state_before,
-                    &new_state,
-                    tick,
-                    &mut violations,
-                );
-
-                actor_states[idx].1 = new_state;
-                actor_action_counts[idx] += 1;
-                total_transitions += 1;
-            }
+        for msg in &delivered {
+            apply_to_model(
+                model,
+                &mut actor_states,
+                &mut actor_action_counts,
+                &mut violations,
+                &mut total_transitions,
+                tick,
+                msg,
+            );
         }
+    }
 
+    // Flush the schedule: delay faults can push deliveries past the last
+    // loop iteration. Budgeted; every flushed message runs through the same
+    // exactly-once path (previously a bare tick() discarded them — ARN-236).
+    let mut flush_budget = config.max_ticks;
+    let mut tick = config.max_ticks.saturating_sub(1);
+    while !sched.is_quiescent() && flush_budget > 0 {
+        flush_budget -= 1;
+        tick += 1;
         sched.tick();
+        for msg in &sched.drain_ready() {
+            apply_to_model(
+                model,
+                &mut actor_states,
+                &mut actor_action_counts,
+                &mut violations,
+                &mut total_transitions,
+                tick,
+                msg,
+            );
+        }
     }
 
     // Post-simulation liveness checks
@@ -314,6 +322,47 @@ fn check_liveness_post_simulation(
     violations
 }
 
+/// Apply one drained message to the model: parse the action, advance the
+/// target actor's state, check invariants, and update counters. The
+/// exactly-once application step shared by the driver loop and the flush
+/// (ARN-236).
+#[allow(clippy::too_many_arguments)]
+fn apply_to_model(
+    model: &TemperModel,
+    actor_states: &mut [(String, TemperModelState)],
+    actor_action_counts: &mut [usize],
+    violations: &mut Vec<InvariantViolation>,
+    total_transitions: &mut u64,
+    tick: u64,
+    msg: &temper_runtime::scheduler::SimMessage,
+) {
+    let target_idx = actor_states.iter().position(|(id, _)| id == &msg.to);
+    let Some(idx) = target_idx else { return };
+
+    let (ref target_id, ref state_before) = actor_states[idx];
+
+    let action: TemperModelAction = match serde_json::from_str(&msg.payload) {
+        Ok(a) => a,
+        Err(_) => return,
+    };
+
+    if let Some(new_state) = model.next_state(state_before, action.clone()) {
+        check_invariants_on_state(
+            model,
+            target_id,
+            &action.name,
+            state_before,
+            &new_state,
+            tick,
+            violations,
+        );
+
+        actor_states[idx].1 = new_state;
+        actor_action_counts[idx] += 1;
+        *total_transitions += 1;
+    }
+}
+
 /// Check invariants on a state using the model's resolved invariants.
 ///
 /// All invariant data comes from the spec — no hardcoded entity knowledge.
diff --git a/docs/adrs/0165-sim-delivery-single-ownership.md b/docs/adrs/0165-sim-delivery-single-ownership.md
new file mode 100644
index 000000000..b88aee8d4
--- /dev/null
+++ b/docs/adrs/0165-sim-delivery-single-ownership.md
@@ -0,0 +1,75 @@
+# ADR-0165: Simulation Delivery Has a Single Ownership Path
+
+## Status
+
+Accepted (2026-07-14)
+
+(Numbered 0165: 0156–0164 are claimed by concurrently open arena branches.)
+
+## Context
+
+The deterministic scheduler exposed two incompatible delivery contracts:
+`Scheduler::tick()` both enqueued a due message into the target mailbox AND
+returned a clone. The simulation drivers (temper-runtime's
+`SimActorSystem::run_random`, temper-verify's model-checking driver)
+processed the returned clones and never drained mailboxes — `receive()` had
+zero production callers — and each loop iteration ended with a bare
+`tick()` whose returned deliveries were discarded. Consequences (ARN-236,
+all reproduced by seeded tests): every processed message remained queued in
+its mailbox forever; deliveries surfaced only by the trailing tick (delayed
+messages coming due after the last driver iteration) were silently lost —
+seed 0 of the regression sweep loses 14 of 44 sends; and a rejected
+integration callback (`let _ = self.step(...)`) left the run green. DST and
+model-check results therefore did not faithfully exercise the schedule they
+claimed to.
+
+## Decision
+
+1. **`tick()` advances logical time and enqueues only.** It returns
+   nothing. A message is owned by exactly one place at every instant: the
+   pending queue, a mailbox, or the consumer that drained it.
+2. **One consumption path.** `drain_ready()` removes and returns all queued
+   messages in deterministic order (actor-id order, FIFO per mailbox);
+   `receive()` remains for single-actor consumption. Drivers apply each
+   drained message exactly once (`apply_delivered_message`, extracted so
+   the main loop and the flush share one application path). The clone-return
+   processing path is deleted.
+3. **Budgeted schedule flush.** After the driver's action loop, a bounded
+   flush (budget = `max_ticks`) ticks and drains until quiescent, so
+   delay-faulted deliveries due after the last iteration are applied through
+   the same exactly-once path instead of being discarded.
+4. **Callback failures are part of the result.** A rejected integration
+   callback is recorded as a violation (`integration callback rejected:
+   ...`) on the simulation result; a run that discards one cannot stay
+   green. Both drivers share the runtime scheduler, so the verifier and
+   runtime exercise the same delivery contract.
+
+## Consequences
+
+- Exactly-once delivery is now a tested property: a 50-seed sweep with delay
+  faults asserts applications == sends; a fault-free run asserts empty
+  mailboxes and quiescence; a rejected callback asserts a non-green result.
+- `is_quiescent()` (pending + mailboxes empty) is now reachable in driver
+  runs — previously mailboxes never emptied, so quiescence was unreachable
+  by construction.
+- Runs get slightly longer traces: deliveries that were silently lost are
+  now applied (that is the point). Seeds produce different — now correct —
+  transition sequences than before; recorded-run comparisons across this
+  change are not byte-compatible, which is expected for a semantics fix.
+- `Scheduler::run_until_quiescent` keeps its shape (tick in a bounded loop),
+  but with tick enqueue-only nothing drains during it — once anything is
+  enqueued it degrades to "tick `max_ticks` times" and terminates via the
+  bound, never via quiescence. Its only callers are scheduler unit tests,
+  which drain after it returns.
+
+## Alternatives Considered
+
+- **Making `tick()` return owned (non-cloned) messages and deleting
+  mailboxes from the delivery path**: also a single-owner model, but it
+  removes per-actor mailbox semantics (depth inspection, per-actor
+  `receive`, crash-time mailbox behavior) that other tests and fault
+  injection rely on. Enqueue-then-drain keeps those observables.
+- **Draining inside `tick()` (tick returns owned mailbox contents)**:
+  conflates time advance with consumption; a driver that wants to tick
+  several times before applying (e.g. coalescing) couldn't. Separate verbs
+  keep the contract explicit.

From 0b6d09bf4ed9b9082f0d69a9724c75995df273ee Mon Sep 17 00:00:00 2001
From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com>
Date: Tue, 14 Jul 2026 21:16:21 -0700
Subject: [PATCH 3/5] fix(verify): reaches-liveness is eventually-visited along
 the trace (ARN-236)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The corrected traces exposed a second defect the delayed-message bug had
been masking: L2's `reaches` liveness was checked against the final state
at the simulation horizon, which wrongly flags cyclic specs — the Ticket
fixture's Resolve -> Reopen cycle "failed" EventuallyResolved whenever the
random walk stopped mid-cycle, and had only ever passed because lost
trailing deliveries biased where traces ended. The check was calibrated
against corrupted traces.

Each actor now records every status it visits (seeded with the initial
status); `reaches` is satisfied the moment a target was ever visited, and
still violates when no trace ever reaches one. Both directions are pinned
by new unit tests (a cyclic spec whose only first action is the resolving
one, and a spec whose declared target has no inbound action).
no_deadlock liveness is unchanged. Residual: `reaches` properties whose
`from` is a non-initial state are still never armed — pre-existing, filed
as a Linear follow-up.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 crates/temper-verify/src/simulation.rs        | 147 +++++++++++++++++-
 .../0165-sim-delivery-single-ownership.md     |   9 ++
 2 files changed, 151 insertions(+), 5 deletions(-)

diff --git a/crates/temper-verify/src/simulation.rs b/crates/temper-verify/src/simulation.rs
index 0b7315191..74fd3772a 100644
--- a/crates/temper-verify/src/simulation.rs
+++ b/crates/temper-verify/src/simulation.rs
@@ -9,6 +9,8 @@
 //! - Any failure is reproducible by replaying the same seed
 //! - Specification invariants are checked after every transition
 
+use std::collections::{BTreeMap, BTreeSet};
+
 use temper_runtime::scheduler::{DeterministicRng, FaultConfig, SimActorState, SimScheduler};
 
 use stateright::Model;
@@ -159,11 +161,22 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
     // Initialize actors
     let mut actor_states: Vec<(String, TemperModelState)> = Vec::new();
     let mut actor_action_counts: Vec<usize> = Vec::new();
+    // Statuses each actor has EVER been in (ARN-236): liveness `reaches` is
+    // an eventually-visited property along the trace, not a state-at-horizon
+    // property — a cyclic spec (Resolve → Reopen) satisfies "eventually
+    // resolved" the moment it visits a target, wherever the random walk
+    // happens to stop. (The pre-fix final-state check only passed because
+    // lost trailing deliveries biased where traces ended.)
+    let mut visited_statuses: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
 
     for i in 0..config.num_actors {
         let actor_id = format!("entity-{i}");
         sched.register_actor(&actor_id);
         let initial = model.init_states()[0].clone();
+        visited_statuses
+            .entry(actor_id.clone())
+            .or_default()
+            .insert(model.initial_status.clone());
         actor_states.push((actor_id, initial));
         actor_action_counts.push(0);
     }
@@ -223,6 +236,7 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
                 &mut total_transitions,
                 tick,
                 msg,
+                &mut visited_statuses,
             );
         }
     }
@@ -245,12 +259,14 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
                 &mut total_transitions,
                 tick,
                 msg,
+                &mut visited_statuses,
             );
         }
     }
 
     // Post-simulation liveness checks
-    let liveness_violations = check_liveness_post_simulation(model, &actor_states);
+    let liveness_violations =
+        check_liveness_post_simulation(model, &actor_states, &visited_statuses);
 
     SimulationResult {
         all_invariants_held: violations.is_empty(),
@@ -273,6 +289,7 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
 fn check_liveness_post_simulation(
     model: &TemperModel,
     actor_states: &[(String, TemperModelState)],
+    visited_statuses: &BTreeMap<String, BTreeSet<String>>,
 ) -> Vec<LivenessViolation> {
     let mut violations = Vec::new();
 
@@ -300,15 +317,24 @@ fn check_liveness_post_simulation(
                     if targets.is_empty() {
                         continue;
                     }
-                    // If the actor started from a "from" state, it should have
-                    // reached a target state by the end of simulation.
+                    // Eventually-visited along the trace (ARN-236): the actor
+                    // satisfies `reaches` the moment it has EVER been in a
+                    // target status. Checking only the horizon final state
+                    // wrongly flags cyclic specs (Resolve -> Reopen) whose
+                    // random walk stops mid-cycle — and only ever passed
+                    // before because lost trailing deliveries biased where
+                    // traces ended.
                     let started_from = from.is_empty() || from.contains(&model.initial_status);
-                    if started_from && !targets.contains(&final_state.status) {
+                    let visited_target = visited_statuses
+                        .get(actor_id)
+                        .is_some_and(|seen| targets.iter().any(|t| seen.contains(t)));
+                    if started_from && !visited_target {
                         violations.push(LivenessViolation {
                             actor_id: actor_id.clone(),
                             property: live.name.clone(),
                             description: format!(
-                                "actor did not reach target states {:?}, stuck at '{}'",
+                                "actor never reached target states {:?} at any point in the \
+                                 trace, ending at '{}'",
                                 targets, final_state.status
                             ),
                             final_state: final_state.clone(),
@@ -335,6 +361,7 @@ fn apply_to_model(
     total_transitions: &mut u64,
     tick: u64,
     msg: &temper_runtime::scheduler::SimMessage,
+    visited_statuses: &mut BTreeMap<String, BTreeSet<String>>,
 ) {
     let target_idx = actor_states.iter().position(|(id, _)| id == &msg.to);
     let Some(idx) = target_idx else { return };
@@ -358,6 +385,10 @@ fn apply_to_model(
         );
 
         actor_states[idx].1 = new_state;
+        visited_statuses
+            .entry(actor_states[idx].0.clone())
+            .or_default()
+            .insert(actor_states[idx].1.status.clone());
         actor_action_counts[idx] += 1;
         *total_transitions += 1;
     }
@@ -467,6 +498,112 @@ mod tests {
 
     const ORDER_IOA: &str = include_str!("../../../test-fixtures/specs/order.ioa.toml");
 
+    /// Cyclic spec: Resolve moves to the target, Reopen leaves it again.
+    /// `reaches` liveness must be satisfied by EVER visiting the target,
+    /// wherever the bounded random walk happens to stop (ARN-236).
+    const CYCLIC_IOA: &str = r#"
+[automaton]
+name = "Loop"
+states = ["Open", "Resolved"]
+initial = "Open"
+
+[[action]]
+name = "Resolve"
+kind = "input"
+from = ["Open"]
+to = "Resolved"
+hint = "Resolve it."
+
+[[action]]
+name = "Reopen"
+kind = "input"
+from = ["Resolved"]
+to = "Open"
+hint = "Reopen it."
+
+[[liveness]]
+name = "EventuallyResolved"
+from = ["Open"]
+reaches = ["Resolved"]
+"#;
+
+    /// The target status exists but no action ever moves into it: the
+    /// `reaches` property is genuinely unreachable and must still violate.
+    const UNREACHABLE_IOA: &str = r#"
+[automaton]
+name = "Stuck"
+states = ["Open", "Parked", "Resolved"]
+initial = "Open"
+
+[[action]]
+name = "Park"
+kind = "input"
+from = ["Open"]
+to = "Parked"
+hint = "Park it."
+
+[[action]]
+name = "Unpark"
+kind = "input"
+from = ["Parked"]
+to = "Open"
+hint = "Unpark it."
+
+[[liveness]]
+name = "EventuallyResolved"
+from = ["Open"]
+reaches = ["Resolved"]
+"#;
+
+    fn liveness_config(seed: u64) -> SimConfig {
+        SimConfig {
+            seed,
+            max_ticks: 200,
+            num_actors: 2,
+            max_actions_per_actor: 20,
+            max_counter: 2,
+            faults: FaultConfig::none(),
+        }
+    }
+
+    /// ARN-236: a cyclic model that VISITS the target mid-trace satisfies
+    /// `reaches`, even when the walk stops outside the target.
+    #[test]
+    fn reaches_liveness_satisfied_by_ever_visiting_the_target() {
+        for seed in [7u64, 21, 99] {
+            let result = run_simulation_from_ioa(CYCLIC_IOA, &liveness_config(seed)).unwrap();
+            assert!(
+                result.total_transitions > 0,
+                "seed {seed}: the walk must actually move"
+            );
+            let reaches_violations: Vec<_> = result
+                .liveness_violations
+                .iter()
+                .filter(|v| v.property == "EventuallyResolved")
+                .collect();
+            assert!(
+                reaches_violations.is_empty(),
+                "seed {seed}: a trace that visited the target must satisfy \
+                 `reaches`, got: {reaches_violations:?}"
+            );
+        }
+    }
+
+    /// ARN-236 (the not-weakened direction): a target no action can ever
+    /// enter must still violate `reaches`.
+    #[test]
+    fn reaches_liveness_still_fails_when_target_is_never_visited() {
+        let result = run_simulation_from_ioa(UNREACHABLE_IOA, &liveness_config(7)).unwrap();
+        assert!(
+            result
+                .liveness_violations
+                .iter()
+                .any(|v| v.property == "EventuallyResolved"),
+            "a genuinely unreachable target must still violate, got: {:?}",
+            result.liveness_violations
+        );
+    }
+
     #[test]
     fn test_simulation_no_faults() {
         let config = SimConfig {
diff --git a/docs/adrs/0165-sim-delivery-single-ownership.md b/docs/adrs/0165-sim-delivery-single-ownership.md
index b88aee8d4..44f27f0c5 100644
--- a/docs/adrs/0165-sim-delivery-single-ownership.md
+++ b/docs/adrs/0165-sim-delivery-single-ownership.md
@@ -56,6 +56,15 @@ claimed to.
   now applied (that is the point). Seeds produce different — now correct —
   transition sequences than before; recorded-run comparisons across this
   change are not byte-compatible, which is expected for a semantics fix.
+- **L2 `reaches` liveness is now eventually-visited along the trace, not
+  final-state-at-horizon.** The corrected traces exposed that the old check
+  wrongly flagged cyclic specs (the Ticket fixture's Resolve → Reopen cycle
+  "failed" TicketEventuallyResolved whenever the random walk stopped
+  mid-cycle) — it had only ever passed because lost trailing deliveries
+  biased where traces ended. Each actor now records every status it visits;
+  `reaches` is satisfied the moment a target status was ever visited. Specs
+  that genuinely never reach a target still fail, and `no_deadlock`
+  liveness is unchanged.
 - `Scheduler::run_until_quiescent` keeps its shape (tick in a bounded loop),
   but with tick enqueue-only nothing drains during it — once anything is
   enqueued it degrades to "tick `max_ticks` times" and terminates via the

From f96c7d36b10c3801e054a164eaeca1f9d753ac38 Mon Sep 17 00:00:00 2001
From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com>
Date: Tue, 14 Jul 2026 22:10:17 -0700
Subject: [PATCH 4/5] refactor(verify): bundle simulation run state; document
 run_until_quiescent (ARN-236)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The dedicated PR reviewer's P1: the liveness commit silenced an 8-argument
function with #[allow(too_many_arguments)], regressing the readability
ratchet's allow count and turning the Integrity CI gate red — while the PR
body still claimed the ratchet clean from a pre-commit run. Root fix, no
allow, no baseline bump: the five mutable per-run values move into a
ModelRunState struct (actor states, action counts, violations, transition
count, visited statuses) and apply_to_model takes the run state as one
argument. Pure mechanical refactor; no behavioral change.

P2: run_until_quiescent's method doc now states that with enqueue-only
tick it terminates via the tick bound once anything is enqueued, and that
its unit-test callers drain after it returns.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 crates/temper-runtime/src/scheduler/core.rs |  5 ++
 crates/temper-verify/src/simulation.rs      | 90 ++++++++++-----------
 2 files changed, 49 insertions(+), 46 deletions(-)

diff --git a/crates/temper-runtime/src/scheduler/core.rs b/crates/temper-runtime/src/scheduler/core.rs
index 3224cece8..99855487a 100644
--- a/crates/temper-runtime/src/scheduler/core.rs
+++ b/crates/temper-runtime/src/scheduler/core.rs
@@ -229,6 +229,11 @@ impl SimScheduler {
     }
 
     /// Run until quiescent or max ticks reached. Returns total ticks.
+    ///
+    /// With enqueue-only `tick` (ARN-236) nothing drains inside this loop,
+    /// so once any message is enqueued it terminates via the tick bound,
+    /// never via quiescence. Its callers (scheduler unit tests) drain via
+    /// [`Self::receive`]/[`Self::drain_ready`] after it returns.
     pub fn run_until_quiescent(&mut self, max_ticks: u64) -> u64 {
         for _ in 0..max_ticks {
             if self.is_quiescent() {
diff --git a/crates/temper-verify/src/simulation.rs b/crates/temper-verify/src/simulation.rs
index 74fd3772a..4e511f630 100644
--- a/crates/temper-verify/src/simulation.rs
+++ b/crates/temper-verify/src/simulation.rs
@@ -181,20 +181,25 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
         actor_action_counts.push(0);
     }
 
-    let mut violations = Vec::new();
-    let mut total_transitions: u64 = 0;
+    let mut run = ModelRunState {
+        actor_states,
+        actor_action_counts,
+        violations: Vec::new(),
+        total_transitions: 0,
+        visited_statuses,
+    };
     let mut total_messages: u64 = 0;
 
     // Main simulation loop
     for tick in 0..config.max_ticks {
-        if actor_states.is_empty() {
+        if run.actor_states.is_empty() {
             break;
         }
 
-        let actor_idx = rng.next_bound(actor_states.len());
-        let (ref actor_id, ref current_state) = actor_states[actor_idx];
+        let actor_idx = rng.next_bound(run.actor_states.len());
+        let (ref actor_id, ref current_state) = run.actor_states[actor_idx];
 
-        if actor_action_counts[actor_idx] >= config.max_actions_per_actor {
+        if run.actor_action_counts[actor_idx] >= config.max_actions_per_actor {
             continue;
         }
 
@@ -228,16 +233,7 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
         let delivered = sched.drain_ready();
 
         for msg in &delivered {
-            apply_to_model(
-                model,
-                &mut actor_states,
-                &mut actor_action_counts,
-                &mut violations,
-                &mut total_transitions,
-                tick,
-                msg,
-                &mut visited_statuses,
-            );
+            apply_to_model(model, &mut run, tick, msg);
         }
     }
 
@@ -251,33 +247,24 @@ fn run_simulation_impl(model: &TemperModel, config: &SimConfig) -> SimulationRes
         tick += 1;
         sched.tick();
         for msg in &sched.drain_ready() {
-            apply_to_model(
-                model,
-                &mut actor_states,
-                &mut actor_action_counts,
-                &mut violations,
-                &mut total_transitions,
-                tick,
-                msg,
-                &mut visited_statuses,
-            );
+            apply_to_model(model, &mut run, tick, msg);
         }
     }
 
     // Post-simulation liveness checks
     let liveness_violations =
-        check_liveness_post_simulation(model, &actor_states, &visited_statuses);
+        check_liveness_post_simulation(model, &run.actor_states, &run.visited_statuses);
 
     SimulationResult {
-        all_invariants_held: violations.is_empty(),
+        all_invariants_held: run.violations.is_empty(),
         ticks: config.max_ticks.min(sched.current_time()),
-        total_transitions,
+        total_transitions: run.total_transitions,
         total_messages,
         total_dropped: sched.total_dropped() as u64,
-        violations,
+        violations: run.violations,
         liveness_violations,
         seed: config.seed,
-        actor_final_states: actor_states,
+        actor_final_states: run.actor_states,
     }
 }
 
@@ -348,25 +335,36 @@ fn check_liveness_post_simulation(
     violations
 }
 
+/// Mutable per-run simulation state shared by the driver loop, the flush,
+/// and the post-simulation checks (ARN-236).
+struct ModelRunState {
+    /// Current state per actor.
+    actor_states: Vec<(String, TemperModelState)>,
+    /// Applied-action count per actor (parallel to `actor_states`).
+    actor_action_counts: Vec<usize>,
+    /// Invariant violations found so far.
+    violations: Vec<InvariantViolation>,
+    /// Total successful transitions.
+    total_transitions: u64,
+    /// Statuses each actor has ever been in — the eventually-visited data
+    /// `reaches` liveness checks against.
+    visited_statuses: BTreeMap<String, BTreeSet<String>>,
+}
+
 /// Apply one drained message to the model: parse the action, advance the
 /// target actor's state, check invariants, and update counters. The
 /// exactly-once application step shared by the driver loop and the flush
 /// (ARN-236).
-#[allow(clippy::too_many_arguments)]
 fn apply_to_model(
     model: &TemperModel,
-    actor_states: &mut [(String, TemperModelState)],
-    actor_action_counts: &mut [usize],
-    violations: &mut Vec<InvariantViolation>,
-    total_transitions: &mut u64,
+    run: &mut ModelRunState,
     tick: u64,
     msg: &temper_runtime::scheduler::SimMessage,
-    visited_statuses: &mut BTreeMap<String, BTreeSet<String>>,
 ) {
-    let target_idx = actor_states.iter().position(|(id, _)| id == &msg.to);
+    let target_idx = run.actor_states.iter().position(|(id, _)| id == &msg.to);
     let Some(idx) = target_idx else { return };
 
-    let (ref target_id, ref state_before) = actor_states[idx];
+    let (ref target_id, ref state_before) = run.actor_states[idx];
 
     let action: TemperModelAction = match serde_json::from_str(&msg.payload) {
         Ok(a) => a,
@@ -381,16 +379,16 @@ fn apply_to_model(
             state_before,
             &new_state,
             tick,
-            violations,
+            &mut run.violations,
         );
 
-        actor_states[idx].1 = new_state;
-        visited_statuses
-            .entry(actor_states[idx].0.clone())
+        run.actor_states[idx].1 = new_state;
+        run.visited_statuses
+            .entry(run.actor_states[idx].0.clone())
             .or_default()
-            .insert(actor_states[idx].1.status.clone());
-        actor_action_counts[idx] += 1;
-        *total_transitions += 1;
+            .insert(run.actor_states[idx].1.status.clone());
+        run.actor_action_counts[idx] += 1;
+        run.total_transitions += 1;
     }
 }
 

From 59f04b894401e018029533c455d261d40af8306e Mon Sep 17 00:00:00 2001
From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com>
Date: Tue, 14 Jul 2026 22:31:13 -0700
Subject: [PATCH 5/5] test(runtime): callback trigger fires per action, not
 forever (ARN-236)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Greptile P2 on the test fixture: CountingHandler.fired latched true after
the first successful action, so every later pending_callbacks() call
re-scheduled the callback — harmless for the assertion (any failure count
proves the property) but wrong per the trait contract ("emitted by the
LAST action"). Cell<bool> with take semantics: each successful trigger
emission schedules exactly one callback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../temper-runtime/src/scheduler/sim_actor_system.rs   | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/crates/temper-runtime/src/scheduler/sim_actor_system.rs b/crates/temper-runtime/src/scheduler/sim_actor_system.rs
index 7e7e6a884..f33d8c229 100644
--- a/crates/temper-runtime/src/scheduler/sim_actor_system.rs
+++ b/crates/temper-runtime/src/scheduler/sim_actor_system.rs
@@ -768,7 +768,7 @@ mod tests {
     struct CountingHandler {
         applications: Arc<AtomicUsize>,
         emit_trigger: bool,
-        fired: bool,
+        fired: std::cell::Cell<bool>,
     }
 
     impl SimActorHandler for CountingHandler {
@@ -785,7 +785,7 @@ mod tests {
             }
             self.applications.fetch_add(1, Ordering::SeqCst);
             if self.emit_trigger {
-                self.fired = true;
+                self.fired.set(true);
             }
             Ok(serde_json::json!({"status": "Ready"}))
         }
@@ -805,7 +805,7 @@ mod tests {
             serde_json::json!([])
         }
         fn pending_callbacks(&self) -> Vec<String> {
-            if self.fired {
+            if self.fired.take() {
                 vec!["boom_trigger".to_string()]
             } else {
                 Vec::new()
@@ -827,7 +827,7 @@ mod tests {
             Box::new(CountingHandler {
                 applications: applications.clone(),
                 emit_trigger: false,
-                fired: false,
+                fired: std::cell::Cell::new(false),
             }),
         );
         (system, applications)
@@ -902,7 +902,7 @@ mod tests {
             Box::new(CountingHandler {
                 applications: applications.clone(),
                 emit_trigger: true,
-                fired: false,
+                fired: std::cell::Cell::new(false),
             }),
         );