From 0d098ee408ba284f88068e8c9f6bc127b2d6c36d Mon Sep 17 00:00:00 2001 From: acamilo Date: Tue, 22 Sep 2026 13:00:14 +0000 Subject: [PATCH] session: both transports for every integration test, and the mid-step pause written into step-v1 Review found two Memory-only tests behind a README claim that every integration test runs twice. Both are now generated by both_transports!: the one-agent composition, and the old-epoch refusal, which is one of the three tests the old-worker-replies row cites. The README says which single test walks both transports inside itself instead, because it compares their traces against each other. step-v1 section 2 gains the mid-step pause line and a dated amendment saying why it is not a new edge: a pause requested during a transition is the ordinary Committing(k) -> Ready(k+1) edge followed by Ready(k+1) -> Paused(k+1), because section 6 requires the transition to finish first, so the only boundary such a pause can land on is the one the transition just committed. --- docs/design/session-framework/step-v1.md | 7 +++++++ services/flysim/crates/fly-session/README.md | 10 +++++++--- .../flysim/crates/fly-session/tests/failures.rs | 17 +++++++---------- .../flysim/crates/fly-session/tests/session.rs | 6 +++--- 4 files changed, 24 insertions(+), 16 deletions(-) diff --git a/docs/design/session-framework/step-v1.md b/docs/design/session-framework/step-v1.md index 3dea51f..ab8554b 100644 --- a/docs/design/session-framework/step-v1.md +++ b/docs/design/session-framework/step-v1.md @@ -30,6 +30,7 @@ Starting → Ready(k) → Preparing(k) → Applying(k) → Observing(k+1) Ready(k) → Paused(k) → Ready(k) Ready(k) / Paused(k) → Capturing(k) → same boundary +pause requested mid-step → Committing(k) → Ready(k+1) → Paused(k+1) any unresolved partial failure → Failed → Restoring(new epoch) → Paused(k) terminal episode → Paused(k) → Resetting(new epoch) → Ready(0) ``` @@ -38,6 +39,12 @@ terminal episode → Paused(k) → Resetting(new epoch) → Ready(0) transition carry `scope.step=k`; result fields identify `nextStep=k+1` where applicable. Do not send Agent.Commit with step `k+1` merely because the observation is newer. +**Amendment, 2026-09-22.** The mid-step pause line above adds no new edge: a pause requested +during a transition is served by the ordinary `Committing(k) → Ready(k+1)` edge followed by +`Ready(k+1) → Paused(k+1)`. It is written into the machine because section 6 requires the +transition to finish first, so the only boundary such a pause can land on is the one the +transition just committed. + ## 3. Transaction sequence ### Phase A: prepare all agents concurrently diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md index 9a0e3ca..6821bd6 100644 --- a/services/flysim/crates/fly-session/README.md +++ b/services/flysim/crates/fly-session/README.md @@ -114,7 +114,9 @@ harness.shutdown().await; libraries, so their types live here rather than in the payload contract. An intent is a `PortControl` without its port, and only the coordinator adds the port. - **The phase machine.** `step-v1` section 2 is this crate's, not the contract crate's; the - trace's phase path is recorded beside the contract's `TransitionTrace`. + trace's phase path is recorded beside the contract's `TransitionTrace`. The mid-step pause + it takes -- the transition finishes, then the session pauses at the boundary it just + committed -- is now written into the section 2 machine as a dated amendment. ## Limitations @@ -136,8 +138,10 @@ cargo test -p fly-session # unit + both integ cargo run -p fly-session --example session # the runnable synthetic session ``` -Every integration test runs twice, once over the in-memory transport and once over a Unix -socket, through the same router code: +Every integration test runs over both transports, through the same router code: all but one +are generated twice by `both_transports!`, and +`sequential_concurrent_and_reversed_orders_agree` walks both transports inside one test +because it compares their behaviour traces against each other. - `tests/session.rs`: one world advance per complete batch; every agent Prepared before the advance; one task evaluation per transition; every agent committed before the next Prepare or diff --git a/services/flysim/crates/fly-session/tests/failures.rs b/services/flysim/crates/fly-session/tests/failures.rs index a3d960d..60e7b66 100644 --- a/services/flysim/crates/fly-session/tests/failures.rs +++ b/services/flysim/crates/fly-session/tests/failures.rs @@ -12,7 +12,7 @@ use common::{Fixture, at, fly_a, fly_b, within}; use fly_session::agent::AgentFaults; use fly_session::coordinator::Injections; use fly_session::environment::EnvironmentFaults; -use fly_session::harness::{HarnessConfig, SessionHarness, Via}; +use fly_session::harness::{HarnessConfig, Via}; use fly_session::phase::Phase; use fly_session::types::*; @@ -27,6 +27,7 @@ both_transports!( a_reply_from_another_incarnation_is_rejected, a_world_that_advanced_without_sensory_data_fails_the_transition, an_exact_duplicate_of_a_running_operation_is_in_progress, + an_old_epoch_operation_is_refused_with_stale_epoch, ); const STEPS: u64 = 4; @@ -338,13 +339,10 @@ async fn an_exact_duplicate_of_a_running_operation_is_in_progress(via: Via) { f.shutdown().await; } -/// A restarted worker under a new epoch refuses an operation from the old one. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn an_old_epoch_operation_is_refused_with_stale_epoch() { - let dir = tempfile::tempdir().unwrap(); - let mut harness = SessionHarness::start(Via::Memory, dir.path(), HarnessConfig::default()) - .await - .unwrap(); +/// A live worker under this epoch refuses an operation naming another one. +async fn an_old_epoch_operation_is_refused_with_stale_epoch(via: Via) { + let mut f = clean_fixture(via).await; + let harness = &mut f.harness; within("bootstrap", harness.coordinator.bootstrap()).await.unwrap(); within("step", harness.coordinator.step()).await.unwrap(); @@ -371,6 +369,5 @@ async fn an_old_epoch_operation_is_refused_with_stale_epoch() { .expect_err("an old epoch cannot mutate this worker"); assert_eq!(err.code, ErrorCode::StaleEpoch); assert_eq!(harness.coordinator.stats().advances, 1); - harness.shutdown().await; - drop(dir); + f.shutdown().await; } diff --git a/services/flysim/crates/fly-session/tests/session.rs b/services/flysim/crates/fly-session/tests/session.rs index 1b0acfe..feb5d7f 100644 --- a/services/flysim/crates/fly-session/tests/session.rs +++ b/services/flysim/crates/fly-session/tests/session.rs @@ -26,6 +26,7 @@ both_transports!( a_terminal_episode_pauses_at_its_own_boundary, status_answers_with_the_committed_boundary, a_worker_refuses_a_second_initialize, + a_single_agent_composition_runs_the_same_transaction, ); const STEPS: u64 = 3; @@ -376,8 +377,7 @@ async fn sequential_concurrent_and_reversed_orders_agree() { /// A one-agent composition still runs the same transaction, so the barrier is not two-agent /// specific. -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn a_single_agent_composition_runs_the_same_transaction() { +async fn a_single_agent_composition_runs_the_same_transaction(via: Via) { let config = HarnessConfig { agents: vec![AgentSpec { agent_id: id("fly-a"), @@ -387,7 +387,7 @@ async fn a_single_agent_composition_runs_the_same_transaction() { }], ..HarnessConfig::default() }; - let mut f: Fixture = fixture(Via::Memory, config).await; + let mut f: Fixture = fixture(via, config).await; within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); within("run", f.harness.coordinator.run(3)).await.unwrap(); assert_eq!(f.harness.coordinator.stats().advances, 3);