From 6d67fa7ed293ca7c00a9019ab90ece740400ee39 Mon Sep 17 00:00:00 2001
From: acamilo <dev@localhost>
Date: Tue, 22 Sep 2026 18:47:42 +0000
Subject: [PATCH 1/5] session: a retried Acknowledge is not a failed epoch

The bound test failed two runs in thirty under load, and not on the
bound it was testing. Both failures were bootstrap: "a worker did not
acknowledge every lifecycle reply".

ipc-v1 section 5 says already released or unknown ids are ignored, and
the contract type already holds the acknowledged list to a subset of
the request. So the second Acknowledge of the same ids answers with an
empty list by design -- and the section 6 resolution produces exactly
that second Acknowledge whenever the first reply is slower than the
probe. Demanding the whole list back turned a safe, contract-sanctioned
retry into a failed epoch, which is a defect in the coordinator rather
than in the test: a slow lifecycle reply would do it to a real session
too.

The test made itself easy to hit by installing a fifty-millisecond
probe before bootstrap, so bootstrap's own lifecycle calls ran under a
budget meant for the step under test. It now bootstraps at ordinary
deadlines and tightens them afterwards.

The two bounds are also separated by construction rather than by clock.
Each half puts the bound it is not testing out of reach -- u32::MAX
attempts against a fifth of a second, three attempts against an hour --
so no scheduling delay can flip which one fires, and the silent
participant is ten minutes slow against a twenty-second test timeout,
so returning at all proves a bound ended it. The wall-clock assertion
is gone and the attempt count is asserted instead, which
last_resolution_attempts now records. One agent per composition, so the
participant the failure names is not a race either.
---
 .../crates/fly-session/src/coordinator.rs     | 21 +++--
 .../crates/fly-session/tests/processes.rs     | 85 ++++++++++++-------
 2 files changed, 68 insertions(+), 38 deletions(-)

diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs
index e7e4bb7..825a366 100644
--- a/services/flysim/crates/fly-session/src/coordinator.rs
+++ b/services/flysim/crates/fly-session/src/coordinator.rs
@@ -290,6 +290,8 @@ pub struct Coordinator {
     pub resolutions: u64,
     /// How the last resolution ended, so a test or a supervisor can tell which bound fired.
     pub last_resolution: Option<ResolutionEnd>,
+    /// How many attempts the last resolution spent. Counted, not inferred from the clock.
+    pub last_resolution_attempts: u32,
     /// The caller-side failure-detection budgets of `ipc-v1` section 6.
     pub deadlines: Deadlines,
     /// Per-method and critical-path latency samples. Local synthetic timings, never a
@@ -357,6 +359,7 @@ impl Coordinator {
             in_progress_replies: 0,
             resolutions: 0,
             last_resolution: None,
+            last_resolution_attempts: 0,
             deadlines: Deadlines::default(),
             metrics: Metrics::default(),
             blame: None,
@@ -840,14 +843,15 @@ impl Coordinator {
             let reply = self
                 .call(&worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[])
                 .await?;
-            let result: AcknowledgeResult =
+            // The reply lists what this call released, which is not always everything it
+            // asked about: `ipc-v1` section 5 says "Already released/unknown IDs are
+            // ignored", and the contract type already holds the list to a subset of the
+            // request. A second Acknowledge therefore answers with an empty list by design --
+            // and the section 6 resolution produces exactly that second Acknowledge whenever
+            // the first one's reply was slow. Demanding the whole list back turned a safe,
+            // contract-sanctioned retry into a failed epoch.
+            let _: AcknowledgeResult =
                 reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?;
-            if result.acknowledged.len() != ids.len() {
-                return Err(self.fail_now(
-                    DomainError::invalid("a worker did not acknowledge every lifecycle reply"),
-                    "acknowledge",
-                ));
-            }
         }
         self.audit.push("acknowledge.lifecycle".to_owned());
         Ok(())
@@ -1147,11 +1151,14 @@ impl Coordinator {
         // The budget is the working limit and the attempt count is a guard; whichever runs
         // out is recorded, so "it gave up" is never an unexplained number.
         let mut end = ResolutionEnd::AttemptsExhausted;
+        let mut spent = 0u32;
         for _ in 0..attempts {
             if started.elapsed() >= budget {
                 end = ResolutionEnd::BudgetExpired;
                 break;
             }
+            spent += 1;
+            self.last_resolution_attempts = spent;
             let outcome = call_owned(
                 self.bus.clone(),
                 worker.clone(),
diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs
index d2dcd53..77014c5 100644
--- a/services/flysim/crates/fly-session/tests/processes.rs
+++ b/services/flysim/crates/fly-session/tests/processes.rs
@@ -16,7 +16,7 @@ use common::{at, count, fly_a, fly_b, mode_fixture, within};
 use fly_session::agent::AgentFaults;
 use fly_session::coordinator::{DispatchOrder, Injections};
 use fly_session::environment::EnvironmentFaults;
-use fly_session::harness::{ExecutionMode, HarnessConfig, Via};
+use fly_session::harness::{AgentSpec, ExecutionMode, HarnessConfig, Via};
 use fly_session::launcher::{ReapOutcome, ThreadBudget};
 use fly_session::ResolutionEnd;
 use fly_session::phase::Phase;
@@ -114,15 +114,19 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode)
     config.environment_faults =
         EnvironmentFaults { advance_delay_ms: 500, ..EnvironmentFaults::default() };
     let mut f = mode_fixture(mode, config).await;
+    // Bootstrap first, at ordinary deadlines: its lifecycle calls are not what this test is
+    // about, and squeezing them through the probe below only tests the machine's luck.
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     // A probe well inside both delays, and a resolution budget well outside them: the point is
-    // a call that expires and an operation that is nevertheless fine.
+    // a call that expires and an operation that is nevertheless fine. The guard is out of
+    // reach so the budget is the only bound in play, and the budget is far above what the
+    // delays need, so neither ends this resolution -- the answer does.
     f.harness.coordinator.deadlines = fly_session::Deadlines {
         probe: Duration::from_millis(120),
-        resolve: Duration::from_secs(20),
-        resolve_attempts: 4096,
+        resolve: Duration::from_secs(15),
+        resolve_attempts: u32::MAX,
         boot: Duration::from_secs(30),
     };
-    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let reports = within("run", f.harness.coordinator.run(2))
         .await
         .expect("a slow participant is resolved, not failed");
@@ -174,26 +178,44 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode)
     f.shutdown().await;
 }
 
+/// One agent, one port, and a participant that will not answer this side of the test's own
+/// timeout. The composition for the bound tests: one participant means one possible name in
+/// the failure, so which agent is blamed is not a race.
+fn one_silent_agent(mode: ExecutionMode) -> HarnessConfig {
+    HarnessConfig {
+        agents: vec![AgentSpec {
+            // Ten minutes. The suite's own `within` gives up at twenty seconds, so if the step
+            // returns at all, a bound ended it and not the participant. That is a claim about
+            // the code rather than about how fast this machine happens to be.
+            faults: AgentFaults { prepare_delay_ms: 600_000, ..AgentFaults::default() },
+            ..AgentSpec::new("fly-a", "p1", 7)
+        }],
+        mode,
+        ..HarnessConfig::default()
+    }
+}
+
 /// The resolution has two bounds, and which one ended it is never left to be guessed.
 ///
-/// `resolve` is the working limit at the default values -- the attempt guard is over sixteen
-/// seconds of pauses against an eight-second budget -- so an unresponsive participant runs the
-/// budget out. Setting the guard low instead ends the same resolution the other way, and the
-/// failure says so both in `last_resolution` and in its own message.
+/// Both halves are arranged so the bound under test is the only one that *can* fire: the
+/// other is set orders of magnitude out of reach, so no amount of scheduling delay flips them.
+/// The claim is the contract's -- a resolution ends by budget or by guard, records which, and
+/// names it in the failure -- and nothing here is timed.
+///
+/// The deadlines are installed after `bootstrap`, deliberately. Bootstrap makes lifecycle
+/// calls of its own, and squeezing them through a fifty-millisecond probe tests the harness's
+/// luck rather than the resolution.
 async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode) {
-    // The budget is what ends it at ordinary settings: a generous attempt guard, a short
-    // budget, and a participant far slower than either.
-    let mut config = two_agents(mode);
-    config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() };
-    let mut f = mode_fixture(mode, config).await;
+    // Half one: the budget fires, because the guard cannot. `u32::MAX` attempts at the two
+    // millisecond pause is over ninety days; the budget is a fifth of a second.
+    let mut f = mode_fixture(mode, one_silent_agent(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     f.harness.coordinator.deadlines = fly_session::Deadlines {
         probe: Duration::from_millis(50),
-        resolve: Duration::from_millis(300),
-        resolve_attempts: 8192,
+        resolve: Duration::from_millis(200),
+        resolve_attempts: u32::MAX,
         boot: Duration::from_secs(30),
     };
-    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
-    let started = Instant::now();
     let failure = within("step", f.harness.coordinator.step())
         .await
         .expect_err("a participant that never answers exhausts the resolution");
@@ -202,36 +224,37 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode)
         failure.error.message.contains("resolution budget"),
         "the message names the bound that fired: {failure}"
     );
-    assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str()));
-    assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
     assert!(
-        started.elapsed() < Duration::from_secs(20),
-        "the budget, not the 30-second participant, is what ended it"
+        f.harness.coordinator.last_resolution_attempts < u32::MAX,
+        "the budget ended it with attempts still in hand, which is what makes it the budget"
     );
+    assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
+    assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
     assert!(f.harness.coordinator.is_fenced());
     f.shutdown().await;
 
-    // The guard is what ends it when it is set below the budget: three attempts against a
-    // budget the participant could never reach anyway.
-    let mut config = two_agents(mode);
-    config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() };
-    let mut f = mode_fixture(mode, config).await;
+    // Half two: the guard fires, because the budget cannot. Three attempts against an hour.
+    let mut f = mode_fixture(mode, one_silent_agent(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     f.harness.coordinator.deadlines = fly_session::Deadlines {
         probe: Duration::from_millis(50),
-        resolve: Duration::from_secs(600),
+        resolve: Duration::from_secs(3_600),
         resolve_attempts: 3,
         boot: Duration::from_secs(30),
     };
-    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let failure = within("step", f.harness.coordinator.step())
         .await
         .expect_err("three attempts are not enough to resolve a silent participant");
     assert_eq!(f.harness.coordinator.last_resolution, Some(ResolutionEnd::AttemptsExhausted));
     assert!(
-        failure.error.message.contains("attempt guard") && failure.error.message.contains("3 attempts"),
+        failure.error.message.contains("attempt guard")
+            && failure.error.message.contains("3 attempts"),
         "the message names the bound that fired and its size: {failure}"
     );
-    assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str()));
+    // Counted, not timed: the guard was spent exactly, and the hour never came near.
+    assert_eq!(f.harness.coordinator.last_resolution_attempts, 3);
+    assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
+    assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
     f.shutdown().await;
 }
 

From f6baeb5cfa13479a79a6772eb2d4570496f64bcf Mon Sep 17 00:00:00 2001
From: acamilo <dev@localhost>
Date: Tue, 22 Sep 2026 19:03:18 +0000
Subject: [PATCH 2/5] session: a test for the acknowledge rule, not only for
 the flake

The short-list acknowledgment is a contract rule, so it gets a test that
says so rather than one that depends on a worker being slow.
acknowledge_replies carries the ipc-v1 section 5 sentence in its doc
comment and returns what the worker actually released;
acknowledge_lifecycle calls it, so bootstrap and the test exercise the
same path.

an_acknowledge_that_releases_nothing_is_not_a_failure drives the case
directly, once per execution mode: bootstrap releases every lifecycle
reply, the test asks for those ids again, the worker ignores them and
releases nothing, and the coordinator must accept the empty list, stay
unfenced, stay at its boundary and still play the next transition. It
fails if the equality check returns.
---
 .../crates/fly-session/src/coordinator.rs     | 43 ++++++++++++-----
 .../crates/fly-session/tests/processes.rs     | 47 +++++++++++++++++++
 2 files changed, 77 insertions(+), 13 deletions(-)

diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs
index 825a366..f9d74c5 100644
--- a/services/flysim/crates/fly-session/src/coordinator.rs
+++ b/services/flysim/crates/fly-session/src/coordinator.rs
@@ -839,24 +839,41 @@ impl Coordinator {
                 .push(request_id);
         }
         for (worker, ids) in by_worker.into_values() {
-            let params = AcknowledgeParams { request_ids: ids.clone() };
-            let reply = self
-                .call(&worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[])
-                .await?;
-            // The reply lists what this call released, which is not always everything it
-            // asked about: `ipc-v1` section 5 says "Already released/unknown IDs are
-            // ignored", and the contract type already holds the list to a subset of the
-            // request. A second Acknowledge therefore answers with an empty list by design --
-            // and the section 6 resolution produces exactly that second Acknowledge whenever
-            // the first one's reply was slow. Demanding the whole list back turned a safe,
-            // contract-sanctioned retry into a failed epoch.
-            let _: AcknowledgeResult =
-                reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?;
+            self.acknowledge_replies(&worker, &ids).await?;
         }
         self.audit.push("acknowledge.lifecycle".to_owned());
         Ok(())
     }
 
+    /// Releases a worker's retained lifecycle replies, and accepts a short answer.
+    ///
+    /// **`ipc-v1` section 5: "Already released/unknown IDs are ignored."** The reply lists what
+    /// *this* call released, which is not always everything it asked about, and the contract
+    /// type already holds that list to a subset of the request. So a second Acknowledge of the
+    /// same ids answers with an empty list by design, and an empty list is success.
+    ///
+    /// This matters beyond tidiness. The `ipc-v1` section 6 resolution turns any Acknowledge
+    /// whose reply is slower than the probe into a second Acknowledge of the same ids, so the
+    /// short answer is not an edge case -- it is what the contract produces on an ordinarily
+    /// slow worker. Requiring the whole list back made the contract's own idempotence a failed
+    /// epoch, which is what
+    /// `an_acknowledge_that_releases_nothing_is_not_a_failure` guards against.
+    ///
+    /// Returns the ids the worker actually released.
+    pub async fn acknowledge_replies(
+        &mut self,
+        worker: &WorkerRef,
+        request_ids: &[DomainRequestId],
+    ) -> Outcome<Vec<DomainRequestId>> {
+        let params = AcknowledgeParams { request_ids: request_ids.to_vec() };
+        let reply = self
+            .call(worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[])
+            .await?;
+        let result: AcknowledgeResult =
+            reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?;
+        Ok(result.acknowledged)
+    }
+
     /// Queries one worker's status without waiting for its current mutation.
     pub async fn status(&mut self, worker: &WorkerRef) -> Outcome<StatusResult> {
         let reply = self
diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs
index 77014c5..6cdd3b5 100644
--- a/services/flysim/crates/fly-session/tests/processes.rs
+++ b/services/flysim/crates/fly-session/tests/processes.rs
@@ -23,6 +23,7 @@ use fly_session::phase::Phase;
 use fly_session::types::*;
 
 all_modes!(
+    an_acknowledge_that_releases_nothing_is_not_a_failure,
     a_slow_participant_is_resolved_rather_than_failed,
     a_resolution_says_which_of_its_two_bounds_ended_it,
     a_delayed_one_agent_result_holds_the_world,
@@ -84,6 +85,52 @@ async fn sequential_reversed_and_parallel_completion_agree() {
     }
 }
 
+// -------------------------------------------------------------------------------------------
+// ipc-v1 section 5: an Acknowledge that releases nothing is success
+
+/// `ipc-v1` section 5: "Already released/unknown IDs are ignored."
+///
+/// A second `Worker.Acknowledge` of ids the worker has already released answers with an empty
+/// list. That is the contract working, not a worker misbehaving, and the coordinator must
+/// accept it and carry on. The session's own bootstrap releases every lifecycle reply, so
+/// asking again for the same ids is exactly that case -- driven directly here rather than by
+/// making something slow, because it is a rule about the reply and not about timing.
+///
+/// The rule has teeth because of section 6: any Acknowledge whose reply outruns the probe is
+/// resolved, and the resolution *is* a second Acknowledge of the same ids. A coordinator that
+/// demands the whole list back therefore fences a healthy session the first time a worker is
+/// slow to answer. It did, on this branch's parent; this test fails if that check returns.
+async fn an_acknowledge_that_releases_nothing_is_not_a_failure(mode: ExecutionMode) {
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    // Bootstrap acknowledges every lifecycle reply, so afterwards the worker holds none.
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap();
+
+    // The ids bootstrap already released. The worker ignores them and releases nothing.
+    let already: Vec<DomainRequestId> =
+        (1..=3).map(DomainRequestId::from_serial).collect();
+    let released = within(
+        "acknowledge",
+        f.harness.coordinator.acknowledge_replies(&worker, &already),
+    )
+    .await
+    .expect("a second Acknowledge of released ids is success, not a failed epoch");
+    assert!(
+        released.is_empty(),
+        "already released ids are ignored, so this call released nothing: {released:?}"
+    );
+
+    // The session is untouched by it: not fenced, still at its boundary, and still plays.
+    assert!(!f.harness.coordinator.is_fenced(), "an empty acknowledgment is not a fault");
+    assert_eq!(f.harness.coordinator.phase(), Phase::Ready(0));
+    let report = within("step", f.harness.coordinator.step())
+        .await
+        .expect("the session continues after an Acknowledge that released nothing");
+    assert_eq!(report.boundary, 1);
+    assert_eq!(f.harness.coordinator.stats().advances, 1);
+    f.shutdown().await;
+}
+
 // -------------------------------------------------------------------------------------------
 // ipc-v1 section 6: an uncertain call is resolved, not failed
 

From ef82a05a3a745a4ef89f4c3b3287abcae4b8e91a Mon Sep 17 00:00:00 2001
From: acamilo <dev@localhost>
Date: Tue, 22 Sep 2026 19:05:58 +0000
Subject: [PATCH 3/5] docs: which replies permit a subset, and which do not

The sweep behind this branch found one check demanding an exact match
where the contract permits a short answer, and four that were right to
demand one. The question that separates them belongs where the next
check gets written, not only in a run report: is the far side reporting
what it did, or being held to a requirement?

Worker.Acknowledge is the only reply of the first kind here, because
ipc-v1 section 5 makes it idempotent. The commit and batch checks are
the second kind and are named so nobody loosens them later in the name
of tolerance; they are what make a partial commit and an incomplete
batch fail.
---
 services/flysim/crates/fly-session/README.md | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md
index 8e114e2..1d3b0ae 100644
--- a/services/flysim/crates/fly-session/README.md
+++ b/services/flysim/crates/fly-session/README.md
@@ -178,6 +178,22 @@ harness.shutdown().await;
   event ids derived from epoch, source step, rule and ordinal.
 - **Executors.** The stateless identity executor only, as v1 specifies.
 
+## Before you add a check to a reply
+
+Ask which kind of reply it is. Is the far side **reporting what it did**, in which case a
+subset or an empty answer is permitted and must be accepted? Or is it **being held to a
+requirement**, in which case exactness is the rule and must be enforced? `Worker.Acknowledge`
+is the only reply of the first kind in this crate, because `ipc-v1` section 5 explicitly makes
+it idempotent -- "Already released/unknown IDs are ignored" -- so a second one legitimately
+releases nothing, and the section 6 resolution turns any slow Acknowledge into exactly that
+second one. Demanding the whole list back there fenced healthy sessions until
+`an_acknowledge_that_releases_nothing_is_not_a_failure` was written.
+
+The commit and batch checks are the second kind and must stay exact: `commit_all` requires
+every agent (`step-v1` section 3 phase D, section 7) and `check_batch` requires every declared
+port (`workers-v1` section 3). Loosening those in the name of tolerance is the same mistake
+pointing the other way -- they are what make a partial commit and an incomplete batch fail.
+
 ## Where this crate narrows or adds to the contract crate
 
 - **Required views.** `WorldObservation::validate_against` checks the views a result carries

From c1a972548bcc86663c100cd7f15fb8b87290bd5b Mon Sep 17 00:00:00 2001
From: acamilo <dev@localhost>
Date: Tue, 22 Sep 2026 19:48:10 +0000
Subject: [PATCH 4/5] session: the death rows kill once the victim provably has
 the work

Both death rows killed after a fixed sleep, so under load the kill could
land before the call was dispatched. The bus then reports not-dispatched
and MutationCertainty::None, which is correct -- the participant never
received anything -- while the test demanded unknown. A full-workspace
run caught it: "left: None, right: None" at the certainty assertion.

kill_once_it_is_working polls the victim's own Worker.Status until it is
provably inside the operation before killing: the agent until it is
Preparing with an active request id, the world until it has recorded the
batch, which the arena does before its injected delay. Dispatch has then
demonstrably happened and unknown is the only correct certainty.

The wall-clock boundedness assertions are gone with them. The suite's
own `within` is the bound, and the victim is five seconds slow against
its twenty, so returning at all is the claim.
---
 .../crates/fly-session/tests/processes.rs     | 59 +++++++++++++++----
 1 file changed, 46 insertions(+), 13 deletions(-)

diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs
index 6cdd3b5..e8e19f5 100644
--- a/services/flysim/crates/fly-session/tests/processes.rs
+++ b/services/flysim/crates/fly-session/tests/processes.rs
@@ -358,6 +358,35 @@ async fn a_delayed_one_agent_result_holds_the_world(mode: ExecutionMode) {
 // -------------------------------------------------------------------------------------------
 // Acceptance: worker or helper death has a bounded diagnosed outcome
 
+/// Waits until `worker` is provably inside the operation, then kills it.
+///
+/// Sleeping a fixed time before the kill asserts a race: under load the kill can land before
+/// the call is even dispatched, and then `MutationCertainty::None` is the *correct* answer
+/// because the participant never received anything. The certainty the death rows are about --
+/// `unknown`, because the participant died with work in its hands -- only holds if the work
+/// reached it, so the test waits for the worker's own status to say so rather than guessing
+/// from the clock.
+async fn kill_once_it_is_working(
+    launcher: &mut fly_session::Launcher,
+    worker: &Id,
+    inside: impl Fn(&StatusResult) -> bool,
+) -> ReapOutcome {
+    let deadline = Instant::now() + Duration::from_secs(15);
+    loop {
+        if let Ok(status) = launcher.health_check(worker).await
+            && inside(&status)
+        {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "{worker} never reported itself inside the operation"
+        );
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+    launcher.kill(worker).await
+}
+
 /// One agent dies in the middle of its Prepare. The epoch fails with a typed cause naming
 /// that agent, within the caller's own budget, and nothing continues on the remainder.
 async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
@@ -367,20 +396,21 @@ async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let started = Instant::now();
 
+    let victim = fly_b();
     let (coordinator, launcher) = f.harness.parts();
     let (stepped, reaped) = tokio::join!(
         async { within("step", coordinator.step()).await },
-        async {
-            tokio::time::sleep(Duration::from_millis(80)).await;
-            launcher.kill(&fly_b()).await
-        }
+        // Killed once it has the Prepare in its hands, not after a fixed sleep: the row is
+        // about a participant that dies *with work*, so the work has to have reached it.
+        kill_once_it_is_working(launcher, &victim, |status| {
+            status.state == WorkerState::Preparing && status.active_request_id.is_some()
+        })
     );
     assert_eq!(reaped, ReapOutcome::Terminated);
     let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one");
-    assert!(
-        started.elapsed() < Duration::from_secs(20),
-        "the outcome must be bounded, not a hang"
-    );
+    // Boundedness is the suite's own `within` above: the participant is five seconds slow and
+    // `within` gives up at twenty, so returning at all is the claim.
+    let _ = started;
     assert_eq!(
         failure.participant.as_deref(),
         Some(fly_b().as_str()),
@@ -419,14 +449,17 @@ async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     let (coordinator, launcher) = f.harness.parts();
     let (stepped, reaped) = tokio::join!(
         async { within("step", coordinator.step()).await },
-        async {
-            tokio::time::sleep(Duration::from_millis(200)).await;
-            launcher.kill(&environment).await
-        }
+        // Killed once the world has recorded the batch, which the arena does before its
+        // injected delay. So the Advance provably reached it and the certainty is `unknown`
+        // rather than `none`; a fixed sleep could land before dispatch under load, and then
+        // `none` would be right and this row would be asserting a race.
+        kill_once_it_is_working(launcher, &environment, |status| {
+            status.last_batch_id.is_some()
+        })
     );
     assert_eq!(reaped, ReapOutcome::Terminated);
     let failure = stepped.expect_err("a dead world is a failed epoch");
-    assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang");
+    let _ = started;
     assert_eq!(
         failure.participant.as_deref(),
         Some(environment.as_str()),

From 50ab3d47ba94cafddb931154c804f4801681cfd0 Mon Sep 17 00:00:00 2001
From: acamilo <dev@localhost>
Date: Tue, 22 Sep 2026 21:03:09 +0000
Subject: [PATCH 5/5] session: hold an Acknowledge to the request it answered

Dropping the length check left nothing checking the reply against the
request at all: AcknowledgeResult::validate_against existed with no
caller, so a worker could acknowledge ids this session never asked
about. That is the other half of the rule the README states. A short
list is the worker reporting what it released and is accepted; an id
from outside the request is the worker reporting about someone else's
cache and is refused, named, before any mutation.

The bootstrap-path regression is now covered too. The earlier test calls
acknowledge_replies directly, which guards the check where it lives but
not where it lived, so a length check put back into acknowledge_lifecycle
left it green. The duplicate_lifecycle_acknowledge injection releases the
ids first, out of sight, so the call that method makes and checks is
already the second one -- the shape the section 6 resolution produces.
Verified by putting the old check back: three tests fail with it, none
without.

Also: last_resolution_attempts is cleared with last_resolution, so a
resolution ending before its first attempt no longer reports the
previous count; the guard half of the bound test asserts the fence like
the budget half; the attempts assertion checks a real bound rather than
u32::MAX; and two dead Instant bindings are gone.
---
 .../flysim/crates/fly-session/src/agent.rs    |  7 ++
 services/flysim/crates/fly-session/src/cli.rs |  3 +
 .../crates/fly-session/src/coordinator.rs     | 37 +++++++++
 .../flysim/crates/fly-session/src/launcher.rs |  3 +
 .../flysim/crates/fly-session/src/worker.rs   | 23 +++++-
 .../crates/fly-session/tests/processes.rs     | 77 ++++++++++++++++---
 6 files changed, 138 insertions(+), 12 deletions(-)

diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs
index 22b5ac5..0802c44 100644
--- a/services/flysim/crates/fly-session/src/agent.rs
+++ b/services/flysim/crates/fly-session/src/agent.rs
@@ -199,6 +199,9 @@ pub struct AgentFaults {
     /// Refuse `State.ActivateRestore` after this worker has already staged, so a group meets
     /// a failure halfway through activation.
     pub fail_activate_restore: bool,
+    /// Add this id to every `Worker.Acknowledge` reply, so the caller meets a worker
+    /// reporting about an id it was never asked about.
+    pub acknowledge_extra_id: Option<Id>,
 }
 
 /// One fake agent worker's configuration.
@@ -689,6 +692,10 @@ impl WorkerEndpoint for FakeAgentWorker {
         self.config.worker_threads as u64
     }
 
+    fn acknowledge_extra_id(&self) -> Option<Id> {
+        self.config.faults.acknowledge_extra_id.clone()
+    }
+
     fn methods(&self) -> Vec<&'static str> {
         vec![
             "Agent.Initialize",
diff --git a/services/flysim/crates/fly-session/src/cli.rs b/services/flysim/crates/fly-session/src/cli.rs
index bbd6bbf..92d426d 100644
--- a/services/flysim/crates/fly-session/src/cli.rs
+++ b/services/flysim/crates/fly-session/src/cli.rs
@@ -212,6 +212,9 @@ fn serve(role: &str, options: &Options) -> Result<(), String> {
                 commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?,
                 fail_stage_restore: options.flag(flags::FAIL_STAGE_RESTORE)?,
                 fail_activate_restore: options.flag(flags::FAIL_ACTIVATE_RESTORE)?,
+                // A worker process is never asked to misbehave this way: the subset refusal
+                // is a caller-side check and its test runs the worker in-process.
+                acknowledge_extra_id: None,
             },
             client_id: client_id.clone(),
             service: service.clone(),
diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs
index 78f7858..eae9fb4 100644
--- a/services/flysim/crates/fly-session/src/coordinator.rs
+++ b/services/flysim/crates/fly-session/src/coordinator.rs
@@ -54,6 +54,10 @@ pub struct Injections {
     pub altered_advance_controls: bool,
     /// Read and release the Advance result's frame, then replay the same operation.
     pub consume_advance_artifact_then_retry: bool,
+    /// Acknowledge the lifecycle replies twice, which is what the `ipc-v1` section 6
+    /// resolution does to any Acknowledge whose first reply outran the probe. The second one
+    /// legitimately releases nothing, and bootstrap must accept it.
+    pub duplicate_lifecycle_acknowledge: bool,
 }
 
 /// What an injection produced, for a test to assert on.
@@ -895,6 +899,21 @@ impl Coordinator {
                 .push(request_id);
         }
         for (worker, ids) in by_worker.into_values() {
+            if self.injections.duplicate_lifecycle_acknowledge {
+                // Release them first, out of sight, so the call this method then makes and
+                // checks is already the *second* one -- which is the shape the section 6
+                // resolution produces when an Acknowledge's first reply outruns the probe,
+                // and the shape the original defect fenced a healthy session on. Adding a
+                // second call after the checked one would not reproduce it: the first reply
+                // is always complete, so a length check on it would pass.
+                let first = self.acknowledge_replies(&worker, &ids).await?;
+                if first.len() != ids.len() {
+                    return Err(self.fail_now(
+                        DomainError::invalid("the first Acknowledge did not release everything"),
+                        "acknowledge",
+                    ));
+                }
+            }
             self.acknowledge_replies(&worker, &ids).await?;
         }
         self.audit.push("acknowledge.lifecycle".to_owned());
@@ -915,6 +934,12 @@ impl Coordinator {
     /// epoch, which is what
     /// `an_acknowledge_that_releases_nothing_is_not_a_failure` guards against.
     ///
+    /// A short list is accepted; a list about something else is not. The worker reports what
+    /// *it* released, so fewer ids than asked for is success -- but it is still only entitled
+    /// to report about the ids it was asked about, and an id outside the request is a worker
+    /// talking about another caller's cache. That half is exact-demanded, and
+    /// `AcknowledgeResult::validate_against` is what says so.
+    ///
     /// Returns the ids the worker actually released.
     pub async fn acknowledge_replies(
         &mut self,
@@ -927,6 +952,15 @@ impl Coordinator {
             .await?;
         let result: AcknowledgeResult =
             reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?;
+        if let Err(e) = result.validate_against(&params) {
+            return Err(self.fail_now(
+                DomainError::before(
+                    ErrorCode::IdentityMismatch,
+                    format!("a worker acknowledged an id this session never asked about: {e}"),
+                ),
+                "acknowledge",
+            ));
+        }
         Ok(result.acknowledged)
     }
 
@@ -1221,7 +1255,10 @@ impl Coordinator {
         let attempts = self.deadlines.resolve_attempts;
         let started = Instant::now();
         self.resolutions += 1;
+        // Both, together: a resolution that ends before its first attempt would otherwise
+        // report the previous one's count.
         self.last_resolution = None;
+        self.last_resolution_attempts = 0;
         self.audit.push(format!("resolve:{}:{method}", worker.worker_id));
         // The budget is the working limit and the attempt count is a guard; whichever runs
         // out is recorded, so "it gave up" is never an unexplained number.
diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs
index 43d6705..a19ab4d 100644
--- a/services/flysim/crates/fly-session/src/launcher.rs
+++ b/services/flysim/crates/fly-session/src/launcher.rs
@@ -1511,6 +1511,9 @@ mod flag_tests {
                 commit_delay_ms: 2,
                 fail_stage_restore: true,
                 fail_activate_restore: true,
+                // Argv carries the injected faults a worker process can have; this one is
+                // in-process only, because the check it drives is the caller's.
+                acknowledge_extra_id: None,
             },
             client_id: "worker-fly-a".to_owned(),
             service: "agent.fly-a".to_owned(),
diff --git a/services/flysim/crates/fly-session/src/worker.rs b/services/flysim/crates/fly-session/src/worker.rs
index d717878..e361af3 100644
--- a/services/flysim/crates/fly-session/src/worker.rs
+++ b/services/flysim/crates/fly-session/src/worker.rs
@@ -194,6 +194,13 @@ pub trait WorkerEndpoint: Send + 'static {
     /// allocation" can read the allocation instead of being told it out of band.
     fn worker_threads(&self) -> u64;
 
+    /// An id this worker will add to every `Worker.Acknowledge` reply, for a test that needs a
+    /// worker reporting about something it was never asked about. `None` for a worker that
+    /// behaves.
+    fn acknowledge_extra_id(&self) -> Option<Id> {
+        None
+    }
+
     /// The domain methods this endpoint implements, beyond the common `Worker.*` set.
     /// Anything else returns UNSUPPORTED without entering the endpoint.
     fn methods(&self) -> Vec<&'static str>;
@@ -281,7 +288,8 @@ async fn run<E: WorkerEndpoint>(
 ) {
     // Identity and capabilities are fixed for the endpoint's lifetime, so the shell reads them
     // once and never takes the endpoint mutex to answer Hello or Status.
-    let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads) = {
+    #[allow(clippy::type_complexity)]
+    let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads, extra_ack) = {
         let e = endpoint.lock().await;
         (
             e.worker_id(),
@@ -292,6 +300,7 @@ async fn run<E: WorkerEndpoint>(
             e.status_cell(),
             e.methods(),
             e.worker_threads(),
+            e.acknowledge_extra_id(),
         )
     };
     let mut running: Vec<tokio::task::JoinHandle<()>> = Vec::new();
@@ -345,7 +354,7 @@ async fn run<E: WorkerEndpoint>(
                 continue;
             }
             "Worker.Acknowledge" => {
-                let outcome = match acknowledge(&request, &cache).await {
+                let outcome = match acknowledge(&request, &cache, extra_ack.as_ref()).await {
                     Ok(result) => success(&request, &worker_id, &incarnation_id, result),
                     Err(e) => {
                         failure(&request.request_id, &worker_id, &incarnation_id, request.scope.clone(), e)
@@ -717,16 +726,24 @@ fn hello(
 async fn acknowledge(
     request: &SessionRpcRequest,
     cache: &Arc<tokio::sync::Mutex<ResultCache>>,
+    extra: Option<&Id>,
 ) -> DomainResult<Map<String, Value>> {
     let params: AcknowledgeParams = AcknowledgeParams::from_json(&request.params)
         .map_err(|e| DomainError::invalid(format!("Worker.Acknowledge: {e}")))?;
     if params.request_ids.is_empty() || params.request_ids.len() > MAX_ACKNOWLEDGE {
         return Err(DomainError::invalid("Worker.Acknowledge takes 1..=16 request ids"));
     }
-    let acknowledged = {
+    let mut acknowledged = {
         let mut c = cache.lock().await;
         c.acknowledge(&params.request_ids)
     };
+    // A deliberately misbehaving worker, for the caller-side subset check to refuse.
+    if let Some(extra) = extra
+        && let Ok(id) = DomainRequestId::parse(extra)
+        && !params.request_ids.contains(&id)
+    {
+        acknowledged.push(id);
+    }
     let result = AcknowledgeResult { acknowledged };
     Ok(object(result.to_json()))
 }
diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs
index 980ef3d..76f853d 100644
--- a/services/flysim/crates/fly-session/tests/processes.rs
+++ b/services/flysim/crates/fly-session/tests/processes.rs
@@ -24,6 +24,7 @@ use fly_session::types::*;
 
 all_modes!(
     an_acknowledge_that_releases_nothing_is_not_a_failure,
+    bootstrap_survives_the_second_acknowledge_its_resolution_makes,
     a_slow_participant_is_resolved_rather_than_failed,
     a_resolution_says_which_of_its_two_bounds_ended_it,
     a_delayed_one_agent_result_holds_the_world,
@@ -131,6 +132,65 @@ async fn an_acknowledge_that_releases_nothing_is_not_a_failure(mode: ExecutionMo
     f.shutdown().await;
 }
 
+/// The same rule, on the path `bootstrap` actually uses.
+///
+/// The test above calls `acknowledge_replies` directly, which guards the check where it lives
+/// now but not where it lived before: a length check reintroduced into `acknowledge_lifecycle`
+/// after that call would leave it green. This one drives bootstrap itself, with the
+/// `duplicate_lifecycle_acknowledge` injection doing exactly what the section 6 resolution
+/// does -- the same ids again, to a worker that has already released them -- so the second,
+/// empty answer has to be accepted by every check on bootstrap's path.
+async fn bootstrap_survives_the_second_acknowledge_its_resolution_makes(mode: ExecutionMode) {
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    f.harness.coordinator.injections = Injections {
+        duplicate_lifecycle_acknowledge: true,
+        ..Injections::default()
+    };
+    within("bootstrap", f.harness.coordinator.bootstrap())
+        .await
+        .expect("bootstrap accepts the second, empty acknowledgment of its own lifecycle ids");
+    assert!(!f.harness.coordinator.is_fenced());
+    assert_eq!(f.harness.coordinator.phase(), Phase::Ready(0));
+    let report = within("step", f.harness.coordinator.step()).await.expect("and still plays");
+    assert_eq!(report.boundary, 1);
+    f.shutdown().await;
+}
+
+/// The other half of the rule: a short list is accepted, an id outside the request is not.
+///
+/// A worker reports what *it* released, so fewer ids than asked for is success -- but it is
+/// only entitled to report about the ids it was asked about. An id from outside the request is
+/// a worker talking about another caller's cache, and
+/// `AcknowledgeResult::validate_against` is what refuses it. Without this, dropping the length
+/// check left nothing checking the reply against the request at all.
+///
+/// Not generated per mode, deliberately. The check is the *caller's*, so the mode of the
+/// worker that misbehaves is irrelevant to it, and the alternative -- carrying the
+/// misbehaviour to a separate process over argv -- would put a flag in the shipped binary
+/// whose only purpose is to make a worker lie about its acknowledgments.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_acknowledged_id_outside_the_request_is_refused() {
+    let mode = ExecutionMode::InProcess;
+    let mut config = two_agents(mode);
+    // This worker adds an id nobody asked about to every acknowledgment.
+    config.agents[0].faults = AgentFaults {
+        acknowledge_extra_id: Some(id("req-9999")),
+        ..AgentFaults::default()
+    };
+    let mut f = mode_fixture(mode, config).await;
+    let failure = within("bootstrap", f.harness.coordinator.bootstrap())
+        .await
+        .expect_err("a worker may not acknowledge an id this session never asked about");
+    assert_eq!(failure.error.code, ErrorCode::IdentityMismatch);
+    assert!(
+        failure.error.message.contains("never asked about"),
+        "the refusal says what was wrong: {failure}"
+    );
+    assert_eq!(failure.detail, "acknowledge");
+    assert_eq!(failure.error.mutation, MutationCertainty::None, "refused before any mutation");
+    f.shutdown().await;
+}
+
 // -------------------------------------------------------------------------------------------
 // ipc-v1 section 6: an uncertain call is resolved, not failed
 
@@ -275,10 +335,12 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode)
         failure.error.message.contains("resolution budget"),
         "the message names the bound that fired: {failure}"
     );
-    assert!(
-        f.harness.coordinator.last_resolution_attempts < u32::MAX,
-        "the budget ended it with attempts still in hand, which is what makes it the budget"
-    );
+    // The budget ended it with attempts still in hand, which is what makes it the budget. A
+    // 200 ms budget at a 50 ms probe cannot spend more than a handful, and `u32::MAX` was
+    // never in reach; asserting against the guard's own size would be vacuous.
+    let spent = f.harness.coordinator.last_resolution_attempts;
+    assert!(spent >= 1, "the resolution made at least one attempt");
+    assert!(spent < 100, "and nowhere near its guard: {spent}");
     assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
     assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
     assert!(f.harness.coordinator.is_fenced());
@@ -308,6 +370,7 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode)
     assert_eq!(f.harness.coordinator.last_resolution_attempts, 3);
     assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
     assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
+    assert!(f.harness.coordinator.is_fenced(), "an exhausted guard fences the epoch too");
     f.shutdown().await;
 }
 
@@ -400,7 +463,6 @@ async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     config.agents[1].faults = AgentFaults { prepare_delay_ms: 5_000, ..AgentFaults::default() };
     let mut f = mode_fixture(mode, config).await;
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
-    let started = Instant::now();
 
     let victim = fly_b();
     let (coordinator, launcher) = f.harness.parts();
@@ -413,10 +475,9 @@ async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
         })
     );
     assert_eq!(reaped, ReapOutcome::Terminated);
-    let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one");
     // Boundedness is the suite's own `within` above: the participant is five seconds slow and
     // `within` gives up at twenty, so returning at all is the claim.
-    let _ = started;
+    let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one");
     assert_eq!(
         failure.participant.as_deref(),
         Some(fly_b().as_str()),
@@ -450,7 +511,6 @@ async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     let mut f = mode_fixture(mode, config).await;
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let environment = f.harness.environment_id();
-    let started = Instant::now();
 
     let (coordinator, launcher) = f.harness.parts();
     let (stepped, reaped) = tokio::join!(
@@ -465,7 +525,6 @@ async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     );
     assert_eq!(reaped, ReapOutcome::Terminated);
     let failure = stepped.expect_err("a dead world is a failed epoch");
-    let _ = started;
     assert_eq!(
         failure.participant.as_deref(),
         Some(environment.as_str()),