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.
This commit is contained in:
acamilo 2026-09-22 13:00:14 +00:00
parent 2ef842ed90
commit 0d098ee408
4 changed files with 24 additions and 16 deletions

View file

@ -30,6 +30,7 @@ Starting → Ready(k) → Preparing(k) → Applying(k) → Observing(k+1)
Ready(k) → Paused(k) → Ready(k) Ready(k) → Paused(k) → Ready(k)
Ready(k) / Paused(k) → Capturing(k) → same boundary 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) any unresolved partial failure → Failed → Restoring(new epoch) → Paused(k)
terminal episode → Paused(k) → Resetting(new epoch) → Ready(0) 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. 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. 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 ## 3. Transaction sequence
### Phase A: prepare all agents concurrently ### Phase A: prepare all agents concurrently

View file

@ -114,7 +114,9 @@ harness.shutdown().await;
libraries, so their types live here rather than in the payload contract. An intent is a 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. `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 - **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 ## Limitations
@ -136,8 +138,10 @@ cargo test -p fly-session # unit + both integ
cargo run -p fly-session --example session # the runnable synthetic session 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 Every integration test runs over both transports, through the same router code: all but one
socket, through the same router code: 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 - `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 advance; one task evaluation per transition; every agent committed before the next Prepare or

View file

@ -12,7 +12,7 @@ use common::{Fixture, at, fly_a, fly_b, within};
use fly_session::agent::AgentFaults; use fly_session::agent::AgentFaults;
use fly_session::coordinator::Injections; use fly_session::coordinator::Injections;
use fly_session::environment::EnvironmentFaults; 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::phase::Phase;
use fly_session::types::*; use fly_session::types::*;
@ -27,6 +27,7 @@ both_transports!(
a_reply_from_another_incarnation_is_rejected, a_reply_from_another_incarnation_is_rejected,
a_world_that_advanced_without_sensory_data_fails_the_transition, a_world_that_advanced_without_sensory_data_fails_the_transition,
an_exact_duplicate_of_a_running_operation_is_in_progress, an_exact_duplicate_of_a_running_operation_is_in_progress,
an_old_epoch_operation_is_refused_with_stale_epoch,
); );
const STEPS: u64 = 4; 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; f.shutdown().await;
} }
/// A restarted worker under a new epoch refuses an operation from the old one. /// A live worker under this epoch refuses an operation naming another one.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn an_old_epoch_operation_is_refused_with_stale_epoch(via: Via) {
async fn an_old_epoch_operation_is_refused_with_stale_epoch() { let mut f = clean_fixture(via).await;
let dir = tempfile::tempdir().unwrap(); let harness = &mut f.harness;
let mut harness = SessionHarness::start(Via::Memory, dir.path(), HarnessConfig::default())
.await
.unwrap();
within("bootstrap", harness.coordinator.bootstrap()).await.unwrap(); within("bootstrap", harness.coordinator.bootstrap()).await.unwrap();
within("step", harness.coordinator.step()).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"); .expect_err("an old epoch cannot mutate this worker");
assert_eq!(err.code, ErrorCode::StaleEpoch); assert_eq!(err.code, ErrorCode::StaleEpoch);
assert_eq!(harness.coordinator.stats().advances, 1); assert_eq!(harness.coordinator.stats().advances, 1);
harness.shutdown().await; f.shutdown().await;
drop(dir);
} }

View file

@ -26,6 +26,7 @@ both_transports!(
a_terminal_episode_pauses_at_its_own_boundary, a_terminal_episode_pauses_at_its_own_boundary,
status_answers_with_the_committed_boundary, status_answers_with_the_committed_boundary,
a_worker_refuses_a_second_initialize, a_worker_refuses_a_second_initialize,
a_single_agent_composition_runs_the_same_transaction,
); );
const STEPS: u64 = 3; 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 /// A one-agent composition still runs the same transaction, so the barrier is not two-agent
/// specific. /// specific.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_single_agent_composition_runs_the_same_transaction(via: Via) {
async fn a_single_agent_composition_runs_the_same_transaction() {
let config = HarnessConfig { let config = HarnessConfig {
agents: vec![AgentSpec { agents: vec![AgentSpec {
agent_id: id("fly-a"), agent_id: id("fly-a"),
@ -387,7 +387,7 @@ async fn a_single_agent_composition_runs_the_same_transaction() {
}], }],
..HarnessConfig::default() ..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("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
within("run", f.harness.coordinator.run(3)).await.unwrap(); within("run", f.harness.coordinator.run(3)).await.unwrap();
assert_eq!(f.harness.coordinator.stats().advances, 3); assert_eq!(f.harness.coordinator.stats().advances, 3);