aion-rs 0.31.0

//! Continue-as-new lifecycle transition behaviour, including the aion#213
//! single-recorder boundary and its run-scoped refusal.

use std::sync::Arc;

use aion_core::{ActivityId, Event, Payload, WorkflowStatus};
use aion_package::ContentHash;
use aion_store::visibility::VisibilityStore;
use aion_store::{EventStore, InMemoryStore};
use serde_json::json;

use super::{ContinueAsNewContext, ContinueAsNewRequest, continue_as_new};
use crate::EngineError;
use crate::durability::Recorder;
use crate::loader::WorkflowCatalog;
use crate::registry::{
    CompletionNotifier, HandleResidency, Registry, TerminalOutcome, WorkflowHandle,
    WorkflowHandleParts,
};
use crate::runtime::{RuntimeConfig, RuntimeHandle};
use crate::supervision::SupervisionTree;

struct ActiveWorkflow {
    store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
    catalog: Arc<WorkflowCatalog>,
    runtime: Arc<RuntimeHandle>,
    supervision: Arc<SupervisionTree>,
    registry: Arc<Registry>,
    handle: WorkflowHandle,
}

fn payload(label: &str) -> Result<Payload, aion_core::PayloadError> {
    Payload::from_json(&json!({ "label": label }))
}

/// The declared workflow timeout the fixture's package commits to. Long
/// enough that nothing here ever reaches it: what these tests measure is
/// which deadline EVENTS the transition records, never a fire.
const DECLARED_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3_600);

fn workflow_catalog() -> Arc<WorkflowCatalog> {
    let catalog = Arc::new(WorkflowCatalog::new());
    catalog.note_loaded_workflow_with_timeout_for_test(
        "checkout",
        "checkout_deployed_v1",
        "run",
        ContentHash::from_bytes([3; 32]),
        DECLARED_TIMEOUT,
    );
    catalog.note_loaded_workflow_with_timeout_for_test(
        "checkout",
        "checkout_deployed_v2",
        "run",
        ContentHash::from_bytes([4; 32]),
        DECLARED_TIMEOUT,
    );
    catalog.note_loaded_workflow_for_test(
        "fulfillment",
        "fulfillment_deployed",
        "run",
        ContentHash::from_bytes([5; 32]),
    );
    catalog
}

async fn active_workflow() -> Result<ActiveWorkflow, Box<dyn std::error::Error>> {
    let backing = Arc::new(InMemoryStore::default());
    let store: Arc<dyn EventStore> = Arc::clone(&backing) as Arc<dyn EventStore>;
    let visibility_store: Arc<dyn VisibilityStore> = backing;
    let catalog = workflow_catalog();
    let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(
        Some(1),
        crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT,
    ))?);
    runtime.register_waiting_test_module("checkout_deployed_v1", "run");
    runtime.register_waiting_test_module("checkout_deployed_v2", "run");
    runtime.register_waiting_test_module("fulfillment_deployed", "run");
    let supervision = Arc::new(SupervisionTree::new());
    let registry = Arc::new(Registry::default());
    let workflow_id = aion_core::WorkflowId::new_v4();
    let run_id = aion_core::RunId::new_v4();
    let mut recorder = Recorder::new(workflow_id.clone(), Arc::clone(&store))
        .with_visibility(run_id.clone(), Arc::clone(&visibility_store));
    recorder
        .record_workflow_started(
            chrono::Utc::now(),
            crate::durability::WorkflowStartRecord {
                workflow_type: "checkout".to_owned(),
                input: payload("input")?,
                run_id: run_id.clone(),
                parent_run_id: None,
                parent_workflow_id: None,
                package_version: aion_core::PackageVersion::new("a".repeat(64)),
            },
        )
        .await?;
    // The predecessor's declared-timeout deadline, exactly as the start path
    // records it. Without it the transition would have no deadline to retire
    // and the D5 half of the batch would be untested.
    let deadline_id = crate::time::deadline_timer_id(&run_id)?;
    recorder
        .record_timer_started(
            chrono::Utc::now(),
            deadline_id,
            chrono::Utc::now() + chrono::Duration::hours(1),
        )
        .await?;
    // The successor arms a live deadline, so the transition needs a real timer
    // bridge to arm it against; without one the start would fail honestly
    // rather than exercising the boundary.
    crate::runtime::nif_timer_bridge::install_timer_nif_bridge(
        runtime.nif_state(),
        Arc::clone(&registry),
        Arc::clone(&store),
        tokio::runtime::Handle::current(),
        runtime.signal_delivery(),
    );
    let pid = runtime.spawn_test_process_with_trap_exit(true)?;
    let handle = WorkflowHandle::new(WorkflowHandleParts {
        workflow_id: workflow_id.clone(),
        run_id: run_id.clone(),
        pid,
        workflow_type: "checkout".to_owned(),
        namespace: String::from("default"),
        loaded_version: ContentHash::from_bytes([3; 32]),
        cached_status: WorkflowStatus::Running,
        residency: HandleResidency::Resident,
        recorder,
        completion: CompletionNotifier::new(),
    });
    registry.insert((workflow_id, run_id), handle.clone())?;

    Ok(ActiveWorkflow {
        store,
        visibility_store,
        catalog,
        runtime,
        supervision,
        registry,
        handle,
    })
}

fn context(active: &ActiveWorkflow) -> ContinueAsNewContext<'_> {
    ContinueAsNewContext {
        store: Arc::clone(&active.store),
        visibility_store: Arc::clone(&active.visibility_store),
        catalog: Arc::clone(&active.catalog),
        runtime: &active.runtime,
        supervision: Arc::clone(&active.supervision),
        registry: &active.registry,
        search_attribute_schema: Arc::new(aion_core::SearchAttributeSchema::new()),
    }
}

#[tokio::test]
async fn pending_activity_rejects_without_terminal_event() -> Result<(), Box<dyn std::error::Error>>
{
    let active = active_workflow().await?;
    {
        let recorder = active.handle.recorder();
        let mut recorder = recorder.lock().await;
        recorder
            .record_activity_scheduled(
                chrono::Utc::now(),
                ActivityId::from_sequence_position(2),
                "charge-card".to_owned(),
                payload("activity")?,
                String::from("default"),
                None,
            )
            .await?;
    }

    let result = continue_as_new(
        context(&active),
        active.handle.workflow_id(),
        active.handle.run_id(),
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await;

    assert!(matches!(
        result,
        Err(EngineError::Runtime { reason }) if reason.contains("pending work")
    ));
    let history = active
        .store
        .read_history(active.handle.workflow_id())
        .await?;
    assert!(!matches!(
        history.last(),
        Some(Event::WorkflowContinuedAsNew { .. })
    ));
    assert_eq!(
        active
            .registry
            .get(active.handle.workflow_id(), active.handle.run_id())?,
        Some(active.handle.clone())
    );
    active.runtime.shutdown()?;
    Ok(())
}

#[tokio::test]
async fn pending_child_rejects_without_terminal_event() -> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;
    {
        let recorder = active.handle.recorder();
        let mut recorder = recorder.lock().await;
        recorder
            .record_child_workflow_started(
                chrono::Utc::now(),
                aion_core::WorkflowId::new_v4(),
                "fulfillment".to_owned(),
                payload("child")?,
                aion_core::PackageVersion::new("a".repeat(64)),
            )
            .await?;
    }

    let result = continue_as_new(
        context(&active),
        active.handle.workflow_id(),
        active.handle.run_id(),
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await;

    assert!(matches!(
        result,
        Err(EngineError::Runtime { reason }) if reason.contains("pending work")
    ));
    let history = active
        .store
        .read_history(active.handle.workflow_id())
        .await?;
    assert!(!matches!(
        history.last(),
        Some(Event::WorkflowContinuedAsNew { .. })
    ));
    assert_eq!(
        active
            .registry
            .get(active.handle.workflow_id(), active.handle.run_id())?,
        Some(active.handle.clone())
    );
    active.runtime.shutdown()?;
    Ok(())
}

#[tokio::test]
async fn success_records_notifies_deregisters_and_starts_new_run()
-> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;
    let old_workflow_id = active.handle.workflow_id().clone();
    let old_run_id = active.handle.run_id().clone();
    let input = payload("next")?;
    let mut receiver = active.handle.completion().subscribe();

    let new_handle = continue_as_new(
        context(&active),
        &old_workflow_id,
        &old_run_id,
        ContinueAsNewRequest {
            input: input.clone(),
            workflow_type: None,
        },
    )
    .await?;
    receiver.changed().await?;

    assert_eq!(new_handle.workflow_id(), &old_workflow_id);
    assert_ne!(new_handle.run_id(), &old_run_id);
    assert_eq!(new_handle.workflow_type(), "checkout");
    assert_eq!(
        new_handle.loaded_version(),
        &ContentHash::from_bytes([4; 32]),
        "the continue-as-new successor must take the latest loaded version (D1)"
    );
    assert_eq!(active.registry.get(&old_workflow_id, &old_run_id)?, None);
    assert_eq!(
        active.registry.get(&old_workflow_id, new_handle.run_id())?,
        Some(new_handle.clone())
    );
    assert_eq!(
        receiver.borrow().clone(),
        Some(TerminalOutcome::ContinuedAsNew {
            input: input.clone(),
            workflow_type: None,
            parent_run_id: old_run_id.clone(),
        })
    );

    // aion#213 R1: the whole transition is ONE contiguous batch, in this order.
    let history = active.store.read_history(&old_workflow_id).await?;
    assert_transition_batch(&history, &input, &old_run_id, &new_handle)?;
    // aion#213 R1: ONE recorder for the workflow, carried across the boundary.
    // Not "a recorder that happens to be at the right head" — the same
    // instance, which is what makes a late predecessor append serialize with
    // the successor's rather than race it.
    assert!(
        Arc::ptr_eq(&active.handle.recorder(), &new_handle.recorder()),
        "the successor must carry the predecessor's own recorder"
    );
    active.runtime.shutdown()?;
    Ok(())
}

/// The batch aion#213 R1 specifies, asserted as an ORDER rather than a set:
/// the order is the contract the recovery sweeps and the deadline repair chain
/// read, and a boundary that interleaved anything between the terminal and the
/// successor's start would be the split transition this fix removed.
fn assert_transition_batch(
    history: &[Event],
    input: &Payload,
    old_run_id: &aion_core::RunId,
    new_handle: &WorkflowHandle,
) -> Result<(), Box<dyn std::error::Error>> {
    let predecessor_deadline = crate::time::deadline_timer_id(old_run_id)?;
    let successor_deadline = crate::time::deadline_timer_id(new_handle.run_id())?;
    match history {
        [
            Event::WorkflowStarted { .. },
            Event::TimerStarted {
                timer_id: armed_predecessor,
                ..
            },
            Event::WorkflowContinuedAsNew {
                input: continued_input,
                workflow_type,
                parent_run_id,
                ..
            },
            Event::TimerCancelled {
                timer_id: retired,
                cause: aion_core::TimerCancelCause::WorkflowIntent,
                ..
            },
            Event::WorkflowStarted {
                input: started_input,
                workflow_type: started_type,
                run_id: started_run_id,
                parent_run_id: started_parent,
                ..
            },
            Event::TimerStarted {
                timer_id: armed_successor,
                ..
            },
        ] => {
            assert_eq!(armed_predecessor, &predecessor_deadline);
            assert_eq!(continued_input, input);
            assert_eq!(workflow_type, &None);
            assert_eq!(parent_run_id, old_run_id);
            assert_eq!(
                retired, &predecessor_deadline,
                "the predecessor's deadline is retired IN the batch (D5)"
            );
            assert_eq!(started_input, input);
            assert_eq!(started_type, "checkout");
            assert_eq!(started_run_id, new_handle.run_id());
            assert_eq!(started_parent, &Some(old_run_id.clone()));
            assert_eq!(armed_successor, &successor_deadline);
        }
        other => {
            return Err(format!("expected continue-as-new history, found {other:?}").into());
        }
    }
    assert_eq!(
        crate::time::outstanding_deadline_timer(history, old_run_id),
        None,
        "the predecessor deadline is retired, so failover re-arm cannot resurrect it"
    );
    assert_eq!(
        crate::time::outstanding_deadline_timer(history, new_handle.run_id()),
        Some(successor_deadline),
        "the successor's own deadline stays live"
    );
    // ONE batch, minted from ONE clock read: the four boundary events occupy
    // consecutive sequences and share a single `recorded_at`. A transition
    // split back into two appends takes a second `Utc::now()` and leaves a gap
    // where anything else could be sequenced, so both facts break together.
    let boundary = &history[2..];
    let stamp = *boundary
        .first()
        .ok_or("the boundary must have a first event")?
        .recorded_at();
    for (offset, event) in boundary.iter().enumerate() {
        let expected_seq = u64::try_from(offset)?
            .checked_add(3)
            .ok_or("seq overflow")?;
        assert_eq!(
            event.seq(),
            expected_seq,
            "the boundary occupies consecutive sequences: {history:#?}"
        );
        assert_eq!(
            *event.recorded_at(),
            stamp,
            "the boundary is minted from one clock read: {history:#?}"
        );
    }
    Ok(())
}

#[tokio::test]
async fn recorded_terminal_rejects_continue_without_second_terminal_event()
-> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;
    {
        let recorder = active.handle.recorder();
        let mut recorder = recorder.lock().await;
        recorder
            .record_workflow_cancelled(
                chrono::Utc::now(),
                "caller requested cancellation".to_owned(),
            )
            .await?;
    }

    let result = continue_as_new(
        context(&active),
        active.handle.workflow_id(),
        active.handle.run_id(),
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await;

    assert!(matches!(
        result,
        Err(EngineError::Runtime { reason })
            if reason.contains("already recorded a terminal event")
    ));
    let history = active
        .store
        .read_history(active.handle.workflow_id())
        .await?;
    assert!(
        matches!(
            history.as_slice(),
            [
                Event::WorkflowStarted { .. },
                Event::TimerStarted { .. },
                Event::WorkflowCancelled { .. }
            ]
        ),
        "the refused transition appended nothing: {history:#?}"
    );
    active.runtime.shutdown()?;
    Ok(())
}

#[tokio::test]
async fn different_replacement_type_rejects_before_terminal_mutation()
-> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;

    let result = continue_as_new(
        context(&active),
        active.handle.workflow_id(),
        active.handle.run_id(),
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: Some("fulfillment".to_owned()),
        },
    )
    .await;

    assert!(matches!(
        result,
        Err(EngineError::Runtime { reason })
            if reason.contains("must restart the same workflow type")
    ));
    let history = active
        .store
        .read_history(active.handle.workflow_id())
        .await?;
    assert!(
        matches!(
            history.as_slice(),
            [
                Event::WorkflowStarted { workflow_type, .. },
                Event::TimerStarted { .. }
            ] if workflow_type == "checkout"
        ),
        "the refused transition appended nothing: {history:#?}"
    );
    assert_eq!(
        active
            .registry
            .get(active.handle.workflow_id(), active.handle.run_id())?,
        Some(active.handle.clone())
    );
    active.runtime.shutdown()?;
    Ok(())
}

// --- aion#213: the double-writer window across continue-as-new. -------------

/// The exact incident, reproduced through the timer bridge's own interleaving
/// seam.
///
/// # What the defect was
///
/// The transition used to be two lock scopes: the terminal and the deadline
/// retirement under the predecessor's recorder, and then —
/// lock released — `start_workflow_with_options`, which read the history head
/// UNLOCKED and built a SECOND `Recorder` for the same workflow id. The
/// predecessor's process is still alive across that gap. Its in-flight `sleep`
/// arm appended a `TimerStarted` through the OLD recorder, moving the durable
/// head past the value the new recorder had just read, and the successor's
/// first append died on `SequenceConflict { expected: 4, found: 5 }`.
///
/// # How this test reaches it
///
/// [`TimerNifBridge::arm_timer`]'s test-only `arm_interleave` hook fires from
/// inside an arm — after the arming run has already recorded its
/// `TimerStarted`, before the wheel entry exists — which is precisely the
/// window the predecessor occupied. Running the whole transition from that
/// hook puts the boundary INSIDE a predecessor arm rather than beside it, so
/// the two are genuinely concurrent on the one history.
///
/// # Killing mutations
///
/// - Give the successor its own recorder (revert to `start_workflow_with_options`):
///   the arm's `TimerStarted` at the head the successor read makes the
///   successor's `WorkflowStarted` a `SequenceConflict`, and the transition
///   returns `Err`.
/// - Delete the recorder's run guard: the late predecessor append below is
///   sequenced, and a `TimerStarted` for the OLD run lands after its own
///   terminal, inside the successor's segment.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_predecessor_arm_across_the_transition_neither_conflicts_nor_lands()
-> Result<(), Box<dyn std::error::Error>> {
    use crate::durability::RunAdmission;
    use crate::engine_seam::{EngineHandle, TimerWheelEntry, WorkflowProcessHandle};

    let active = active_workflow().await?;
    let workflow_id = active.handle.workflow_id().clone();
    let predecessor_run = active.handle.run_id().clone();
    let bridge =
        crate::runtime::nif_timer_bridge::installed_timer_bridge(active.runtime.nif_state())
            .map_err(|error| format!("the timer bridge must be installed: {error}"))?;

    // The predecessor's `sleep`, in the order the NIF performs it: the durable
    // `TimerStarted` first, then the wheel arm. Both happen while this run is
    // still the workflow's live generation, so both must succeed.
    let sleep_timer = aion_core::TimerId::anonymous(9);
    let fire_at = chrono::Utc::now() + chrono::Duration::hours(2);
    {
        let recorder = active.handle.recorder();
        let mut recorder = recorder.lock().await;
        recorder
            .record_timer_started(chrono::Utc::now(), sleep_timer.clone(), fire_at)
            .await?;
    }

    // THE INTERLEAVING: run the transition from inside the arm.
    let transition = install_transition_interleave(&active, &bridge)?;

    bridge.arm_timer(TimerWheelEntry {
        process: WorkflowProcessHandle::new(active.handle.pid()),
        timer_id: sleep_timer.clone(),
        fire_at,
    })?;

    let successor_run = transition
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone()
        .ok_or("the interleave hook never ran, so nothing was measured")?
        .map_err(|error| format!("the transition must not fail: {error}"))?;

    // (a) NO SequenceConflict: the transition succeeded, which it could not
    //     have done through a second recorder seeded before the arm landed.
    assert_ne!(successor_run, predecessor_run);

    // (b) The predecessor's NEXT durable append — the one that would land
    //     AFTER the transition — is refused, and appends nothing.
    let head_before = {
        let recorder = active.handle.recorder();
        let recorder = recorder.lock().await;
        assert_eq!(
            recorder.admit_run_append(&predecessor_run).await?,
            RunAdmission::RefusedTerminal,
            "the predecessor's generation is closed, so its late appends are refused"
        );
        assert_eq!(
            recorder.admit_run_append(&successor_run).await?,
            RunAdmission::Open,
            "control: the successor's own appends are still admitted"
        );
        recorder.current_head()
    };
    let history = active.store.read_history(&workflow_id).await?;
    assert_eq!(
        history.iter().map(aion_core::Event::seq).max(),
        Some(head_before),
        "the refused append moved neither the tracked head nor the durable one"
    );

    // (c) No `TimerStarted` for the old run after its own terminal, and the
    //     successor's segment is exactly the batch R1 specifies.
    let continued = history
        .iter()
        .position(|event| matches!(event, Event::WorkflowContinuedAsNew { .. }))
        .ok_or("no WorkflowContinuedAsNew in history")?;
    let predecessor_deadline = crate::time::deadline_timer_id(&predecessor_run)?;
    let successor_deadline = crate::time::deadline_timer_id(&successor_run)?;
    match &history[continued..] {
        [
            Event::WorkflowContinuedAsNew { .. },
            Event::TimerCancelled {
                timer_id: retired, ..
            },
            Event::WorkflowStarted {
                run_id: started, ..
            },
            Event::TimerStarted {
                timer_id: armed, ..
            },
        ] => {
            assert_eq!(retired, &predecessor_deadline);
            assert_eq!(started, &successor_run);
            assert_eq!(armed, &successor_deadline);
        }
        other => {
            return Err(format!(
                "the transition must be one contiguous batch with nothing after it, found {other:?}"
            )
            .into());
        }
    }
    assert!(
        !history[continued..].iter().any(|event| matches!(
            event,
            Event::TimerStarted { timer_id, .. } if timer_id == &sleep_timer
        )),
        "the predecessor's sleep must not be armed into the successor's segment: {history:#?}"
    );

    active.runtime.shutdown()?;
    Ok(())
}

/// The predecessor's late append and the transition, run genuinely
/// concurrently.
///
/// One recorder means one mutex, so the interleaving space is FINITE — the
/// arm is either sequenced before the boundary or refused after it — and both
/// outcomes are asserted here rather than one being hoped for. What is never
/// legal, in either order, is a `SequenceConflict`: that is the store's
/// double-writer alarm, and after this fix there is no second writer to raise
/// it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_transition_and_a_concurrent_predecessor_append_are_serialised_never_conflicting()
-> Result<(), Box<dyn std::error::Error>> {
    use crate::durability::RunAdmission;

    let active = active_workflow().await?;
    let workflow_id = active.handle.workflow_id().clone();
    let predecessor_run = active.handle.run_id().clone();
    let sleep_timer = aion_core::TimerId::anonymous(11);

    let arming = {
        let recorder = active.handle.recorder();
        let predecessor_run = predecessor_run.clone();
        let sleep_timer = sleep_timer.clone();
        tokio::spawn(async move {
            let mut recorder = recorder.lock().await;
            match recorder.admit_run_append(&predecessor_run).await {
                Ok(RunAdmission::RefusedTerminal) => Ok(false),
                Ok(RunAdmission::Open) => recorder
                    .record_timer_started(
                        chrono::Utc::now(),
                        sleep_timer,
                        chrono::Utc::now() + chrono::Duration::hours(2),
                    )
                    .await
                    .map(|_| true),
                Err(error) => Err(error),
            }
        })
    };

    let successor = continue_as_new(
        context(&active),
        &workflow_id,
        &predecessor_run,
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await?;
    let armed = arming.await??;

    let history = active.store.read_history(&workflow_id).await?;
    let continued = history
        .iter()
        .position(|event| matches!(event, Event::WorkflowContinuedAsNew { .. }))
        .ok_or("no WorkflowContinuedAsNew in history")?;
    let sleep_position = history.iter().position(
        |event| matches!(event, Event::TimerStarted { timer_id, .. } if timer_id == &sleep_timer),
    );
    match (armed, sleep_position) {
        // The arm won the lock: its `TimerStarted` belongs to the
        // predecessor's own segment, strictly before the terminal.
        (true, Some(position)) => assert!(
            position < continued,
            "an admitted predecessor arm belongs before its own terminal: {history:#?}"
        ),
        // The boundary won the lock: the arm is refused and nothing is
        // appended for it at all.
        (false, None) => {}
        (armed, position) => {
            return Err(format!(
                "an arm reported as armed={armed} left its TimerStarted at {position:?}, which is \
                 neither of the two orders one recorder allows: {history:#?}"
            )
            .into());
        }
    }
    // Whichever order won, the successor's segment holds exactly its own
    // start and its own deadline: no predecessor event ever lands in it.
    let successor_deadline = crate::time::deadline_timer_id(successor.run_id())?;
    let started = history
        .iter()
        .position(|event| {
            matches!(event, Event::WorkflowStarted { run_id, .. } if run_id == successor.run_id())
        })
        .ok_or("the successor was not started")?;
    match &history[started..] {
        [
            Event::WorkflowStarted { .. },
            Event::TimerStarted { timer_id, .. },
        ] => assert_eq!(timer_id, &successor_deadline),
        other => {
            return Err(format!("unexpected successor segment: {other:?}").into());
        }
    }
    active.runtime.shutdown()?;
    Ok(())
}

/// The registry holds ONE handle for the workflow at every instant of the
/// transition, and it is the successor's when the transition returns.
///
/// The predecessor is not left registered (which is what let two handles for
/// one workflow id exist at once, so a by-workflow-id resolver could pick
/// either), and the workflow is never momentarily absent — `rekey_generation`
/// removes and inserts under one lock.
#[tokio::test]
async fn the_transition_leaves_exactly_one_handle_and_it_is_the_successors()
-> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;
    let workflow_id = active.handle.workflow_id().clone();
    let predecessor_run = active.handle.run_id().clone();

    let successor = continue_as_new(
        context(&active),
        &workflow_id,
        &predecessor_run,
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await?;

    assert_eq!(active.registry.get(&workflow_id, &predecessor_run)?, None);
    assert_eq!(
        active.registry.get(&workflow_id, successor.run_id())?,
        Some(successor.clone())
    );
    assert_eq!(
        active.registry.sole_handle(&workflow_id)?,
        Some(successor.clone()),
        "the workflow resolves to ONE writer, with no ambiguity to break"
    );
    assert_eq!(
        active.registry.live_run_pid(&workflow_id)?,
        Some((successor.run_id().clone(), successor.pid())),
        "the live-pid index follows the successor"
    );
    active.runtime.shutdown()?;
    Ok(())
}

/// A start that reuses a live workflow's id is refused before it can seed a
/// rival recorder (aion#213 R1).
///
/// The refusal is the general guard behind the specific fix: continue-as-new
/// is no longer the only way an id could acquire a second append authority,
/// and the start path is where every other way would come through.
#[tokio::test]
async fn starting_a_fresh_run_under_a_live_workflow_id_is_refused()
-> Result<(), Box<dyn std::error::Error>> {
    use crate::lifecycle::start::{StartWorkflowContext, StartWorkflowOptions};

    let active = active_workflow().await?;
    let workflow_id = active.handle.workflow_id().clone();

    let result = crate::lifecycle::start::start_workflow_with_options(
        StartWorkflowContext {
            store: Arc::clone(&active.store),
            visibility_store: Arc::clone(&active.visibility_store),
            catalog: Arc::clone(&active.catalog),
            runtime: Arc::clone(&active.runtime),
            supervision: Arc::clone(&active.supervision),
            registry: Arc::clone(&active.registry),
            signal_handoff: None,
            search_attribute_schema: Arc::new(aion_core::SearchAttributeSchema::new()),
            monitor_tokio_handle: tokio::runtime::Handle::current(),
        },
        "checkout",
        payload("rival")?,
        StartWorkflowOptions {
            workflow_id: Some(workflow_id.clone()),
            ..StartWorkflowOptions::default()
        },
    )
    .await;

    match result {
        Err(EngineError::WorkflowIdAlreadyLive {
            holder_run_id,
            holder_pid,
            ..
        }) => {
            assert_eq!(holder_run_id, active.handle.run_id().to_string());
            assert_eq!(holder_pid, active.handle.pid());
        }
        other => {
            return Err(format!("a rival start must be refused, got {other:?}").into());
        }
    }
    let history = active.store.read_history(&workflow_id).await?;
    assert_eq!(
        history.len(),
        2,
        "the refusal must record nothing: {history:#?}"
    );
    active.runtime.shutdown()?;
    Ok(())
}

/// The transition never releases the recorder between the predecessor's
/// terminal and the successor's start — the window aion#213 lived in.
///
/// # How the window is watched
///
/// `tokio::sync::Mutex` is FAIR, so an observer that releases and immediately
/// re-queues is granted the lock at every point the transition gives it up.
/// Under the fix the transition holds the lock ONCE across the whole boundary,
/// so every snapshot the observer takes is either the pre-transition history
/// or the complete post-transition one. There is no third state to see.
///
/// Killing mutation: split the boundary the way it used to be — record the
/// terminal, drop the lock, then open the successor. The observer is granted
/// the lock in that gap and snapshots a history whose last event is the
/// terminal (or its deadline retirement) with no successor behind it, which is
/// the state a second recorder used to be seeded from.
///
/// # The observer is watching BEFORE the transition starts
///
/// `tokio::spawn` schedules; it does not run. Nothing about a spawned task
/// orders its first poll before the spawner's next `.await`, and on a loaded
/// box the transition can run to completion — and `watching` flip to false —
/// before the observer is polled once. Then the observer takes zero snapshots
/// and the vacuity guard below fires: measured on main 2026-09-03 (canvas
/// battery 3 on 205, `FAIL [0.035s]`, panic at the `!snapshots.is_empty()`
/// assertion), a red that read as the fix regressing when the test had simply
/// not measured. So the observer reports its FIRST snapshot on a oneshot and
/// the test awaits that report before calling `continue_as_new`: from then on
/// the observer is queued on the fair mutex at every point the transition
/// could give it up, which is the only ordering the measurement needs.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_recorder_is_never_released_between_the_terminal_and_the_successor()
-> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;
    let workflow_id = active.handle.workflow_id().clone();
    let predecessor_run = active.handle.run_id().clone();

    let watching = Arc::new(std::sync::atomic::AtomicBool::new(true));
    let (first_snapshot_taken, observer_is_watching) = tokio::sync::oneshot::channel::<()>();
    let observer = {
        let recorder = active.handle.recorder();
        let store = Arc::clone(&active.store);
        let workflow_id = workflow_id.clone();
        let watching = Arc::clone(&watching);
        tokio::spawn(async move {
            let mut snapshots = Vec::new();
            let mut first_snapshot_taken = Some(first_snapshot_taken);
            while watching.load(std::sync::atomic::Ordering::SeqCst) {
                let held = recorder.lock().await;
                snapshots.push(store.read_history(&workflow_id).await);
                drop(held);
                if let Some(report) = first_snapshot_taken.take()
                    && report.send(()).is_err()
                {
                    // The test body is gone (it failed before the transition
                    // started); there is nobody to measure for.
                    return snapshots;
                }
                tokio::task::yield_now().await;
            }
            snapshots
        })
    };
    // Not before the observer has been granted the lock once: see the doc
    // comment. An observer that stopped without reporting is a fault here.
    observer_is_watching.await?;

    let successor = continue_as_new(
        context(&active),
        &workflow_id,
        &predecessor_run,
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await?;
    watching.store(false, std::sync::atomic::Ordering::SeqCst);
    let snapshots = observer.await?;

    assert!(
        !snapshots.is_empty(),
        "the observer must have taken at least one snapshot, or it measured nothing"
    );
    for snapshot in snapshots {
        let snapshot = snapshot?;
        let has_terminal = snapshot
            .iter()
            .any(|event| matches!(event, Event::WorkflowContinuedAsNew { .. }));
        let has_successor = snapshot.iter().any(|event| {
            matches!(event, Event::WorkflowStarted { run_id, .. } if run_id == successor.run_id())
        });
        assert!(
            !has_terminal || has_successor,
            "the recorder was released with a terminal and no successor behind it — the window a \
             second recorder was seeded from: {snapshot:#?}"
        );
    }
    active.runtime.shutdown()?;
    Ok(())
}

/// An activity an EARLIER generation left unsettled must not refuse the CURRENT
/// generation's continuation — while an unsettled activity in the current
/// generation still does (aion#213 R7).
///
/// # 🔴 A WHOLE-HISTORY PENDING-WORK SCAN ON A CHAIN THAT IS DEFINITIONALLY MULTI-GENERATION
///
/// `guard_no_pending_work` accumulates unsettled activities and children by
/// forward-scanning whatever slice it is handed, and `WorkflowStarted` is in its
/// NO-OP arm: a generation boundary does not clear the pending sets. Handed the
/// whole history, one activity left behind by generation 7 refuses the
/// continuation of generation 4,000 — permanently, with a refusal naming work
/// from generations ago — and the scan grows without bound with the chain's age.
/// `workloop/iteration.rs` diagnosed and fixed exactly this one module over;
/// continue-as-new produces the same chain in the same history.
///
/// The stale activity here is not contrived: the workflow-code path records its
/// `WorkflowContinuedAsNew` inside the NIF with NO pending-work guard, and the
/// exit monitor opens the successor afterwards. That is the history this test
/// builds, through the same primitive the monitor calls.
///
/// Killing mutation: pass `history` instead of `run_segment(history, predecessor_run)`
/// at the guard in `admit_transition`. The first continuation below is then
/// refused for an activity that belongs to a generation that is already over.
#[tokio::test]
async fn an_earlier_generations_unsettled_activity_does_not_refuse_this_generation()
-> Result<(), Box<dyn std::error::Error>> {
    let active = active_workflow().await?;
    let workflow_id = active.handle.workflow_id().clone();
    let second_run = open_generation_over_an_unsettled_activity(&active).await?;

    // THE PROPERTY: generation 1's activity is still unsettled in this history,
    // and generation 2 continues anyway.
    let history = active.store.read_history(&workflow_id).await?;
    assert!(
        history.iter().any(|event| matches!(
            event,
            Event::ActivityScheduled { activity_id, .. }
                if activity_id == &ActivityId::from_sequence_position(3)
        )),
        "the fixture must leave a genuinely unsettled activity behind, or this test measures \
         nothing: {history:#?}"
    );
    let third = continue_as_new(
        context(&active),
        &workflow_id,
        &second_run,
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await?;

    // THE CONTROL: an activity unsettled in the CURRENT generation still
    // refuses, so the scope change did not simply disable the guard.
    {
        let recorder = third.recorder();
        let mut recorder = recorder.lock().await;
        recorder
            .record_activity_scheduled(
                chrono::Utc::now(),
                ActivityId::from_sequence_position(99),
                "ship-order".to_owned(),
                payload("live")?,
                String::from("default"),
                None,
            )
            .await?;
    }
    let refused = continue_as_new(
        context(&active),
        &workflow_id,
        third.run_id(),
        ContinueAsNewRequest {
            input: payload("next")?,
            workflow_type: None,
        },
    )
    .await;
    assert!(
        matches!(
            refused,
            Err(EngineError::Runtime { ref reason }) if reason.contains("pending work")
        ),
        "an unsettled activity in the CURRENT generation must still refuse: {refused:?}"
    );

    active.runtime.shutdown()?;
    Ok(())
}

/// Build the history the workflow-code path leaves behind: generation 1 with an
/// activity it never settled, its own `WorkflowContinuedAsNew` recorded (the NIF
/// runs no pending-work guard), and generation 2 opened from that terminal by
/// the same primitive the exit monitor calls. Returns generation 2's run id.
async fn open_generation_over_an_unsettled_activity(
    active: &ActiveWorkflow,
) -> Result<aion_core::RunId, Box<dyn std::error::Error>> {
    use crate::lifecycle::continuation::{
        ContinuationOrigin, ContinuationOutcome, ContinuationRequest, open_successor_generation,
    };
    use crate::lifecycle::start::StartWorkflowContext;

    let first_run = active.handle.run_id().clone();
    {
        let recorder = active.handle.recorder();
        let mut recorder = recorder.lock().await;
        recorder
            .record_activity_scheduled(
                chrono::Utc::now(),
                ActivityId::from_sequence_position(3),
                "charge-card".to_owned(),
                payload("stranded")?,
                String::from("default"),
                None,
            )
            .await?;
        recorder
            .record_workflow_continued_as_new(
                chrono::Utc::now(),
                payload("carry")?,
                None,
                first_run.clone(),
            )
            .await?;
    }

    let outcome = open_successor_generation(
        &StartWorkflowContext {
            store: Arc::clone(&active.store),
            visibility_store: Arc::clone(&active.visibility_store),
            catalog: Arc::clone(&active.catalog),
            runtime: Arc::clone(&active.runtime),
            supervision: Arc::clone(&active.supervision),
            registry: Arc::clone(&active.registry),
            signal_handoff: None,
            search_attribute_schema: Arc::new(aion_core::SearchAttributeSchema::new()),
            monitor_tokio_handle: tokio::runtime::Handle::current(),
        },
        &active.handle,
        ContinuationRequest {
            predecessor_run: first_run,
            origin: ContinuationOrigin::TerminalAlreadyRecorded,
            workflow_type: String::from("checkout"),
            input: payload("carry")?,
        },
    )
    .await?;
    match outcome {
        ContinuationOutcome::Opened(handle) => Ok(handle.run_id().clone()),
        ContinuationOutcome::AlreadyOpen => Err("the successor must be opened by this call".into()),
    }
}

/// Install the [`arm_interleave`] hook that runs the whole continue-as-new
/// transition from inside a predecessor arm, and hand back the slot its
/// outcome lands in.
///
/// The hook is synchronous and the caller's runtime is already driving it, so
/// the wait moves to a scoped thread — the same shape
/// `nif_timer_bridge::run_blocking` uses, for the same reason.
type InterleavedTransition = Arc<std::sync::Mutex<Option<Result<aion_core::RunId, String>>>>;

fn install_transition_interleave(
    active: &ActiveWorkflow,
    bridge: &Arc<crate::runtime::nif_timer_bridge::TimerNifBridge>,
) -> Result<InterleavedTransition, Box<dyn std::error::Error>> {
    let store = Arc::clone(&active.store);
    let visibility_store = Arc::clone(&active.visibility_store);
    let catalog = Arc::clone(&active.catalog);
    let runtime = Arc::clone(&active.runtime);
    let supervision = Arc::clone(&active.supervision);
    let registry = Arc::clone(&active.registry);
    let tokio_handle = tokio::runtime::Handle::current();
    let workflow_id = active.handle.workflow_id().clone();
    let predecessor_run = active.handle.run_id().clone();
    let input = payload("next")?;
    let outcome: InterleavedTransition = Arc::new(std::sync::Mutex::new(None));
    let reported = Arc::clone(&outcome);
    bridge.set_arm_interleave(Arc::new(move || {
        let context = ContinueAsNewContext {
            store: Arc::clone(&store),
            visibility_store: Arc::clone(&visibility_store),
            catalog: Arc::clone(&catalog),
            runtime: &runtime,
            supervision: Arc::clone(&supervision),
            registry: &registry,
            search_attribute_schema: Arc::new(aion_core::SearchAttributeSchema::new()),
        };
        let request = ContinueAsNewRequest {
            input: input.clone(),
            workflow_type: None,
        };
        let result = std::thread::scope(|scope| {
            match scope
                .spawn(|| {
                    tokio_handle.block_on(continue_as_new(
                        context,
                        &workflow_id,
                        &predecessor_run,
                        request,
                    ))
                })
                .join()
            {
                Ok(result) => result.map(|handle| handle.run_id().clone()),
                Err(_) => Err(EngineError::Runtime {
                    reason: String::from("the transition thread panicked"),
                }),
            }
        });
        *reported
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) =
            Some(result.map_err(|error| error.to_string()));
    }));
    Ok(outcome)
}