polyc-turn-runner 2026.8.3

//! The D4 proof set for the fenced Execution protocol.
//!
//! Every case drives the same functions a real peer reaches: a label is built
//! by an [`ExecutionSession`], round-tripped through the generated wire type,
//! and admitted by the other side's session. Nothing pins a value a caller
//! could not produce.

#![allow(clippy::pedantic, clippy::nursery, missing_docs)]

use super::*;
use polyc_proto::proto::polychrome::harness::v1::{TextDelta, TurnBatch, TurnInput};
use polyc_state::deadline::MonotonicInstant;

fn identity(fence: u64) -> ExecutionIdentity {
    ExecutionIdentity::new(
        ExecutionId::new("exec-1"),
        AttemptId::new("attempt-1"),
        FencingToken::new(fence),
    )
    .expect("a fenced identity")
}

fn capabilities() -> ExecutionCapabilities {
    ExecutionCapabilities::new(
        CapabilitySet::of(Capability::LocalRead).with(Capability::FixedConnectorRead),
        16,
    )
    .expect("bounded capabilities")
}

fn audience() -> ExecutionAudience {
    ExecutionAudience::new("harness-a").expect("an audience")
}

fn grant_for(identity: ExecutionIdentity) -> ExecutionGrant {
    ExecutionGrant::new(
        identity,
        OwnerId::new("control-a"),
        AttemptId::new("claim-attempt-1"),
        audience(),
        None,
        Duration::from_secs(30),
        capabilities(),
        TenantId::new("test-tenant"),
        ModelConversationId::new("test-conversation"),
    )
    .expect("a grant")
}

fn grant() -> ExecutionGrant {
    grant_for(identity(7))
}

/// Both sides of one attempt, already past the opening frame: Control has
/// stamped step 0 and Execution has admitted it.
fn paired() -> (ExecutionSession, ExecutionSession) {
    let grant = grant();
    let control = ExecutionSession::open(&grant, MonotonicInstant::ORIGIN);
    let execution = ExecutionSession::admit(&grant, &audience(), MonotonicInstant::ORIGIN)
        .expect("the grant opens the attempt");
    execution
        .check_opening_envelope(&over_the_wire(grant.opening()))
        .expect("the envelope agrees with the grant");
    (control, execution)
}

#[test]
fn a_grant_for_another_execution_audience_is_refused() {
    let error = ExecutionSession::admit(
        &grant(),
        &ExecutionAudience::new("harness-b").expect("an audience"),
        MonotonicInstant::ORIGIN,
    )
    .unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::WrongAudience { declared, expected }
            if declared.as_str() == "harness-a" && expected.as_str() == "harness-b"
    ));
}

#[test]
fn a_forged_opening_step_inside_the_grant_is_refused() {
    let mut wire = WireExecutionGrant::from(&grant());
    wire.opening
        .as_option_mut()
        .expect("an opening")
        .step
        .as_option_mut()
        .expect("a step")
        .step_id = "forged".to_owned();

    let error = ExecutionGrant::try_from(&wire).unwrap_err();
    assert!(matches!(error, ExecutionError::ForgedStepId { index: 0 }));
}

#[test]
fn a_forged_last_committed_step_is_refused() {
    let committed = ExecutionLabel::new(identity(7), 3);
    let resumed = ExecutionGrant::new(
        ExecutionIdentity::new(
            ExecutionId::new("exec-1"),
            AttemptId::new("attempt-2"),
            FencingToken::new(7),
        )
        .expect("an identity"),
        OwnerId::new("control-a"),
        AttemptId::new("claim-attempt-1"),
        audience(),
        Some(committed),
        Duration::from_secs(30),
        capabilities(),
        TenantId::new("test-tenant"),
        ModelConversationId::new("test-conversation"),
    )
    .expect("a grant");
    let mut wire = WireExecutionGrant::from(&resumed);
    wire.last_committed_step
        .as_option_mut()
        .expect("a committed label")
        .step
        .as_option_mut()
        .expect("a step")
        .step_id = "forged".to_owned();

    let error = ExecutionGrant::try_from(&wire).unwrap_err();
    assert!(matches!(error, ExecutionError::ForgedStepId { index: 3 }));
}

/// A label as the peer would actually see it: through the wire type, not the
/// in-memory value.
fn over_the_wire(label: &ExecutionLabel) -> ExecutionLabel {
    let wire = WireExecutionLabel::from(label);
    ExecutionLabel::try_from(&wire).expect("a label this build wrote is one it reads")
}

// ── Wire parity ─────────────────────────────────────────────────────────────

#[test]
fn every_protocol_value_survives_the_wire_unchanged() {
    let prior = ExecutionIdentity::new(
        ExecutionId::new("exec-1"),
        AttemptId::new("attempt-0"),
        FencingToken::new(7),
    )
    .unwrap();
    let resumed = ExecutionGrant::new(
        identity(7),
        OwnerId::new("control-b"),
        AttemptId::new("claim-attempt-1"),
        audience(),
        Some(ExecutionLabel::new(prior, 4)),
        Duration::from_millis(1500),
        ExecutionCapabilities::new(CapabilitySet::all(), 32).expect("bounded capabilities"),
        TenantId::new("test-tenant"),
        ModelConversationId::new("test-conversation"),
    )
    .expect("a grant");

    let wire = WireExecutionGrant::from(&resumed);
    let read = ExecutionGrant::try_from(&wire).expect("a grant this build wrote is one it reads");
    assert_eq!(read, resumed);
    assert_eq!(read.budget(), Duration::from_millis(1500));
    assert_eq!(read.tenant().as_str(), "test-tenant");
    assert_eq!(read.conversation().as_str(), "test-conversation");
    assert_eq!(
        read.last_committed_step().map(|label| label.step().index()),
        Some(4)
    );
    assert_eq!(read.capabilities().granted(), CapabilitySet::all());

    let opening = read.opening();
    assert_eq!(over_the_wire(opening), *opening);
    assert_eq!(opening.step().index(), 0);
    assert_eq!(opening.identity().fence(), FencingToken::new(7));
}

#[test]
fn a_frame_with_no_label_is_refused() {
    let error = required_label(None).unwrap_err();
    assert!(matches!(error, ExecutionError::Malformed { ref field, .. } if field == "label"));
}

#[test]
fn a_turn_input_with_no_grant_is_refused() {
    let error = required_grant(None).unwrap_err();
    assert!(matches!(error, ExecutionError::Malformed { ref field, .. } if field == "execution"));
}

// ── Incompatible version ────────────────────────────────────────────────────

#[test]
fn a_peer_at_another_version_is_refused_rather_than_guessed_at() {
    assert!(check_execution_version(ExecutionProtocolVersion::CURRENT).is_ok());

    // A grant from a peer at the next version, read exactly as a receiver reads
    // one: off the wire.
    let mut wire_grant = WireExecutionGrant::from(&grant());
    let mut wire_label = wire_grant.opening.as_option().expect("an opening").clone();
    wire_label.protocol_version = ExecutionProtocolVersion::CURRENT.get() + 1;
    wire_grant.opening = buffa::MessageField::some(wire_label.clone());
    let ahead = ExecutionGrant::try_from(&wire_grant).expect("the grant still parses");

    let error = ExecutionSession::admit(&ahead, &audience(), MonotonicInstant::ORIGIN).unwrap_err();
    // Derived from CURRENT, never a literal. A literal silently stopped
    // proving anything the moment CURRENT reached it: D6 moved the execution
    // protocol to 2, and this case had pinned 2 as "another version".
    assert!(matches!(
        error,
        ExecutionError::VersionMismatch { declared, current }
            if declared.get() == ExecutionProtocolVersion::CURRENT.get() + 1
                && current == ExecutionProtocolVersion::CURRENT
    ));

    let stale = ExecutionLabel::try_from(&wire_label).expect("the label still parses");

    // And on a later frame too, not only at admission.
    let (mut control, _) = paired();
    let error = control
        .admit_frame(&stale, FrameClass::Proposal)
        .unwrap_err();
    assert!(matches!(error, ExecutionError::VersionMismatch { .. }));
}

/// Pins the exact Execution protocol version this build speaks.
///
/// D4, D5, D7, and D6 all land on version 2, so the wire carries that number.
/// The control plane and the harness compare it for exact equality and fail
/// closed, which is why the two must deploy together. The literal below is
/// deliberate: a self-referential comparison against `CURRENT` would survive a
/// revert of the bump and prove nothing. This assertion fails the moment
/// `CURRENT` returns to 1, which is exactly the review event a protocol change
/// must trigger.
///
/// The paired integration proof is
/// `crates/harness/tests/execution_fence.rs`'s
/// `every_frame_a_served_turn_sends_is_labeled_under_the_admitted_attempt`,
/// which reads the same number back off a real served turn's frames.
#[test]
fn the_execution_protocol_version_is_pinned_where_both_planes_compare_it() {
    assert_eq!(
        ExecutionProtocolVersion::CURRENT.get(),
        3,
        "brokered model calls change what the same frames mean, so Execution \
         and Control must not share a number with a peer that dials providers \
         itself"
    );

    // The number a peer actually reads, taken off the wire rather than from
    // the constant, so the conversion seam is covered too.
    let wire = WireExecutionGrant::from(&grant());
    assert_eq!(
        wire.opening
            .as_option()
            .expect("an opening label")
            .protocol_version,
        3,
        "the wire carries the version both planes compare"
    );
}

/// Pins both numeric protocol constants at their production conversion seam.
///
/// The Execution version moved to 2 for D4, D5, D7, and D6 while the
/// persisted version stayed at 1. This test holds them apart, so a later
/// change to either lifetime is an explicit review event.
#[test]
fn the_execution_and_persisted_protocol_versions_are_pinned_independently() {
    // The number the wire actually carries, read back off the wire rather than
    // from the constant: this is what a peer checks for equality.
    let wire = WireExecutionGrant::from(&grant());
    let declared = wire
        .opening
        .as_option()
        .expect("an opening label")
        .protocol_version;
    assert_eq!(
        declared,
        ExecutionProtocolVersion::CURRENT.get(),
        "the wire carries the Execution protocol version"
    );

    // And the persisted one, which State validates while it replays its own
    // durable log. Raising the Execution protocol version must not move it, so
    // this pins its value: a change to `ProtocolVersion::CURRENT` fails here
    // and forces the replay question to be answered deliberately.
    assert_eq!(
        polyc_state::id::ProtocolVersion::CURRENT.get(),
        1,
        "the persisted command protocol version is unchanged by this protocol"
    );
}

// ── Stale fence ─────────────────────────────────────────────────────────────

#[test]
fn a_superseded_owner_cannot_get_a_frame_admitted() {
    let (mut control, _) = paired();

    // The same execution and attempt, one ownership epoch behind.
    let superseded_grant = grant_for(identity(6));
    let mut superseded = ExecutionSession::open(&superseded_grant, MonotonicInstant::ORIGIN);
    let orphan = superseded.stamp_proposal().expect("a proposal");

    let error = control
        .admit_frame(&over_the_wire(&orphan), FrameClass::Proposal)
        .unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::StaleFence { presented, current }
            if presented == FencingToken::new(6) && current == FencingToken::new(7)
    ));
    assert_eq!(
        control.high_step(),
        0,
        "a refused frame advances nothing on the receiving side"
    );
}

#[test]
fn a_fence_from_a_takeover_this_side_has_not_seen_is_refused_too() {
    let (mut control, _) = paired();
    let ahead_grant = grant_for(identity(8));
    let mut ahead = ExecutionSession::open(&ahead_grant, MonotonicInstant::ORIGIN);
    let frame = ahead.stamp_proposal().expect("a proposal");

    let error = control
        .admit_frame(&over_the_wire(&frame), FrameClass::Proposal)
        .unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::StaleFence { presented, .. } if presented == FencingToken::new(8)
    ));
}

#[test]
fn the_unclaimed_token_never_names_an_attempt() {
    let error = ExecutionIdentity::new(
        ExecutionId::new("exec-1"),
        AttemptId::new("attempt-1"),
        FencingToken::UNCLAIMED,
    )
    .unwrap_err();
    assert!(matches!(error, ExecutionError::Malformed { ref field, .. } if field == "fence"));
}

#[test]
fn a_frame_from_another_execution_or_attempt_is_refused() {
    let (mut control, _) = paired();

    for foreign in [
        ExecutionIdentity::new(
            ExecutionId::new("exec-2"),
            AttemptId::new("attempt-1"),
            FencingToken::new(7),
        )
        .expect("a fenced identity"),
        ExecutionIdentity::new(
            ExecutionId::new("exec-1"),
            AttemptId::new("attempt-2"),
            FencingToken::new(7),
        )
        .expect("a fenced identity"),
    ] {
        let foreign_grant = grant_for(foreign);
        let mut other = ExecutionSession::open(&foreign_grant, MonotonicInstant::ORIGIN);
        let frame = other.stamp_proposal().expect("a proposal");
        let error = control
            .admit_frame(&over_the_wire(&frame), FrameClass::Proposal)
            .unwrap_err();
        assert!(matches!(error, ExecutionError::ForeignAttempt { .. }));
    }
}

// ── Duplicate frame and out-of-order step ───────────────────────────────────

#[test]
fn a_duplicate_frame_is_recognized_by_its_derived_step_identity() {
    let (mut control, mut execution) = paired();
    let first = execution.stamp_proposal().expect("a proposal");
    let resent = over_the_wire(&first);

    assert_eq!(
        resent.step().id(),
        first.step().id(),
        "a resent frame carries the identical step identity, so a receiver can tell it is a replay"
    );

    control
        .admit_frame(&resent, FrameClass::Proposal)
        .expect("the first arrival is admitted");
    let error = control
        .admit_frame(&resent, FrameClass::Proposal)
        .unwrap_err();
    assert!(matches!(error, ExecutionError::DuplicateStep { index } if index == 1));
}

#[test]
fn a_step_that_skips_its_predecessor_is_refused() {
    let (mut control, mut execution) = paired();
    let _skipped = execution.stamp_proposal().expect("step 1");
    let second = execution.stamp_proposal().expect("step 2");

    let error = control
        .admit_frame(&over_the_wire(&second), FrameClass::Proposal)
        .unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::OutOfOrderStep { index, expected } if index == 2 && expected == 1
    ));
}

#[test]
fn a_sender_cannot_invent_a_step_identity() {
    let (mut control, mut execution) = paired();
    let proposal = execution.stamp_proposal().expect("a proposal");
    let mut wire = WireExecutionLabel::from(&proposal);
    wire.step = buffa::MessageField::some(WireExecutionStep {
        index: 1,
        step_id: "a value the sender chose".to_owned(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    });
    let forged = ExecutionLabel::try_from(&wire).expect("the label still parses");

    let error = control
        .admit_frame(&forged, FrameClass::Proposal)
        .unwrap_err();
    assert!(matches!(error, ExecutionError::ForgedStepId { index } if index == 1));
}

#[test]
fn an_ephemeral_frame_names_the_step_it_precedes_and_takes_none() {
    let (mut control, mut execution) = paired();

    for _ in 0..3 {
        let delta = execution.stamp_ephemeral();
        control
            .admit_frame(&over_the_wire(&delta), FrameClass::Ephemeral)
            .expect("a display-only frame rides the step in flight");
    }
    assert_eq!(
        execution.high_step(),
        0,
        "a thousand deltas spend one step, not a thousand"
    );

    let batch = execution.stamp_proposal().expect("the terminal batch");
    control
        .admit_frame(&over_the_wire(&batch), FrameClass::Proposal)
        .expect("the batch takes the step the deltas announced");
    assert_eq!(control.high_step(), 1);
}

#[test]
fn a_reply_repeats_the_step_it_answers() {
    let (mut control, mut execution) = paired();
    let request = execution.stamp_proposal().expect("a tool request");
    control
        .admit_frame(&over_the_wire(&request), FrameClass::Proposal)
        .expect("the request is admitted");

    let reply = control.stamp_reply(&request);
    assert_eq!(reply.step(), request.step());
    execution
        .admit_frame(&over_the_wire(&reply), FrameClass::Reply)
        .expect("the reply answers a step this side emitted");

    // A reply naming a step nobody proposed is refused.
    let ahead_grant = grant();
    let mut ahead = ExecutionSession::open(&ahead_grant, MonotonicInstant::ORIGIN);
    ahead.stamp_proposal().expect("step 1");
    ahead.stamp_proposal().expect("step 2");
    let unanswerable = ahead.stamp_reply(&ExecutionLabel::new(identity(7), 2));
    let error = execution
        .admit_frame(&over_the_wire(&unanswerable), FrameClass::Reply)
        .unwrap_err();
    assert!(matches!(error, ExecutionError::OutOfOrderStep { index, .. } if index == 2));
}

#[test]
fn a_reply_cannot_borrow_another_outstanding_proposals_step() {
    let grant = grant();
    let mut execution =
        ExecutionSession::admit(&grant, &audience(), MonotonicInstant::ORIGIN).unwrap();
    let mut control = ExecutionSession::open(&grant, MonotonicInstant::ORIGIN);
    let first = execution.stamp_proposal().unwrap();
    let _second = execution.stamp_proposal().unwrap();
    control.admit_frame(&first, FrameClass::Proposal).unwrap();
    control.admit_frame(&_second, FrameClass::Proposal).unwrap();

    let error = execution.admit_reply(&first, 2).unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::OutOfOrderStep {
            index: 1,
            expected: 2
        }
    ));
}

/// The class of a frame decides how it moves the sequence, so the mapping is
/// pinned here rather than re-derived at each admission site.
#[test]
fn each_frame_kind_has_one_class() {
    use polyc_proto::proto::polychrome::harness::v1::{
        ControlPlaneToolRequest, ControlPlaneToolResponse, DispatchMutationRequest,
        DispatchMutationResponse, PaidFetchRequest, PaidFetchResponse, PeerCallRequest,
        PeerCallResponse, TurnFailure,
    };

    let proposals = [
        HarnessFrame::from(TurnInput::default()),
        HarnessFrame::from(TurnBatch::default()),
        HarnessFrame::from(TurnFailure::default()),
        HarnessFrame::from(PaidFetchRequest::default()),
        HarnessFrame::from(ControlPlaneToolRequest::default()),
        HarnessFrame::from(DispatchMutationRequest::default()),
        HarnessFrame::from(PeerCallRequest::default()),
    ];
    for frame in &proposals {
        assert_eq!(classify_frame(frame), FrameClass::Proposal, "{frame:?}");
    }

    assert_eq!(
        classify_frame(&HarnessFrame::from(TextDelta::default())),
        FrameClass::Ephemeral
    );

    let replies = [
        HarnessFrame::from(PaidFetchResponse::default()),
        HarnessFrame::from(ControlPlaneToolResponse::default()),
        HarnessFrame::from(DispatchMutationResponse::default()),
        HarnessFrame::from(PeerCallResponse::default()),
    ];
    for frame in &replies {
        assert_eq!(classify_frame(frame), FrameClass::Reply, "{frame:?}");
    }
}

// ── Bounded capabilities ────────────────────────────────────────────────────

#[test]
fn a_grant_naming_a_capability_this_build_does_not_know_is_refused_whole() {
    let mut wire = WireExecutionCapabilities::from(&capabilities());
    wire.granted.push("teleport".to_owned());

    let error = ExecutionCapabilities::try_from(&wire).unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::UnknownCapability { ref unknown } if unknown == "teleport"
    ));
}

/// The never-granted markers have no name `Capability::from_name` recognizes,
/// so no wire grant can seed them however the sender spells them.
#[test]
fn the_never_granted_capabilities_are_unnameable_on_the_wire() {
    for name in ["grant-access", "revoke-access", "manage-admin"] {
        let wire = WireExecutionCapabilities {
            granted: vec![name.to_owned()],
            max_labeled_steps: 16,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let error = ExecutionCapabilities::try_from(&wire).unwrap_err();
        assert!(
            matches!(error, ExecutionError::UnknownCapability { .. }),
            "{name} must not parse into a grant"
        );
    }
}

#[test]
fn a_step_ceiling_outside_the_contract_is_refused_never_clamped() {
    for ceiling in [0, MAX_LABELED_STEPS + 1] {
        let error = ExecutionCapabilities::new(CapabilitySet::EMPTY, ceiling).unwrap_err();
        assert!(
            matches!(error, ExecutionError::Malformed { ref field, .. } if field == "max_labeled_steps"),
            "{ceiling}"
        );
    }
    assert_eq!(
        ExecutionCapabilities::new(CapabilitySet::EMPTY, MAX_LABELED_STEPS)
            .expect("the ceiling itself is grantable")
            .max_labeled_steps(),
        MAX_LABELED_STEPS
    );
}

#[test]
fn an_attempt_stops_at_the_steps_its_grant_allows() {
    let narrow = ExecutionGrant::new(
        identity(7),
        OwnerId::new("control-a"),
        AttemptId::new("claim-attempt-1"),
        audience(),
        None,
        Duration::from_secs(30),
        ExecutionCapabilities::new(CapabilitySet::EMPTY, 2).expect("bounded capabilities"),
        TenantId::new("test-tenant"),
        ModelConversationId::new("test-conversation"),
    )
    .expect("a grant");
    let mut control = ExecutionSession::open(&narrow, MonotonicInstant::ORIGIN);
    let mut execution = ExecutionSession::admit(&narrow, &audience(), MonotonicInstant::ORIGIN)
        .expect("the attempt opens");

    let first = execution
        .stamp_proposal()
        .expect("step 1 is within the grant");
    control
        .admit_frame(&over_the_wire(&first), FrameClass::Proposal)
        .expect("step 1 is admitted");

    let error = execution.stamp_proposal().unwrap_err();
    assert!(matches!(
        error,
        ExecutionError::StepBudgetExhausted { index, granted } if index == 2 && granted == 2
    ));
}

#[test]
fn the_grant_is_what_a_broker_asks() {
    let (_, execution) = paired();
    assert!(execution.permits(Capability::LocalRead));
    assert!(execution.permits(Capability::FixedConnectorRead));
    assert!(!execution.permits(Capability::ArbitraryEgress));
    assert!(!execution.permits(Capability::MutateExternal));
}

// ── Deadlines ───────────────────────────────────────────────────────────────

/// The budget is anchored once, when the session opens, and the session then
/// spends it as its own clock advances. This is the 4-pre rule carried onto
/// this boundary: a value re-anchored per frame would hand every frame a fresh
/// full budget.
#[test]
fn the_budget_is_anchored_once_and_then_spent() {
    // A receiver whose clock is nowhere near the sender's origin.
    let receiver_now = MonotonicInstant::from_nanos(1_786_386_272_000_000_000);
    let execution =
        ExecutionSession::admit(&grant(), &audience(), receiver_now).expect("the attempt opens");

    let family = polyc_state::id::OperationFamily::new("execution.turn");
    assert_eq!(
        execution.call_context().remaining(receiver_now),
        Duration::from_secs(30),
        "the whole wire budget is available at the instant it was framed"
    );

    let later = receiver_now.saturating_add(Duration::from_secs(10));
    assert_eq!(
        execution.call_context().remaining(later),
        Duration::from_secs(20),
        "the framed budget is spent by the receiver's own advancing clock"
    );

    let past = receiver_now.saturating_add(Duration::from_secs(31));
    assert!(matches!(
        execution.call_context().check(past, &family),
        Err(polyc_state::error::StateError::DeadlineExpired { .. })
    ));
}

#[test]
fn a_budget_outside_the_contract_is_refused_never_clamped() {
    for budget in [
        Duration::ZERO,
        MAX_EXECUTION_BUDGET + Duration::from_secs(1),
    ] {
        let error = ExecutionGrant::new(
            identity(7),
            OwnerId::new("control-a"),
            AttemptId::new("claim-attempt-1"),
            audience(),
            None,
            budget,
            capabilities(),
            TenantId::new("test-tenant"),
            ModelConversationId::new("test-conversation"),
        )
        .unwrap_err();
        assert!(
            matches!(error, ExecutionError::Malformed { ref field, .. } if field == "budget_nanos"),
            "{budget:?}"
        );
    }
}

// ── Reconnect and takeover ──────────────────────────────────────────────────

/// A reconnect keeps the execution identity and takes a fresh attempt and a
/// fresh fence. The old attempt's frames stop being admitted the moment the new
/// session exists, and the new attempt resumes after the last committed step
/// rather than repeating it.
#[test]
fn a_reconnect_keeps_the_execution_and_supersedes_the_old_attempt() {
    let (mut first_control, mut first_execution) = paired();
    let committed = first_execution.stamp_proposal().expect("step 1");
    first_control
        .admit_frame(&over_the_wire(&committed), FrameClass::Proposal)
        .expect("step 1 commits");

    let resumed_identity = ExecutionIdentity::new(
        ExecutionId::new("exec-1"),
        AttemptId::new("attempt-2"),
        FencingToken::new(8),
    )
    .expect("a fenced identity");
    let resumed_grant = ExecutionGrant::new(
        resumed_identity,
        OwnerId::new("control-b"),
        AttemptId::new("claim-attempt-2"),
        audience(),
        Some(ExecutionLabel::new(first_control.identity().clone(), 1)),
        Duration::from_secs(30),
        capabilities(),
        TenantId::new("test-tenant"),
        ModelConversationId::new("test-conversation"),
    )
    .expect("a grant");
    let mut second_control = ExecutionSession::open(&resumed_grant, MonotonicInstant::ORIGIN);

    assert_eq!(
        second_control.identity().execution(),
        first_control.identity().execution(),
        "the execution outlives the attempt"
    );
    assert_ne!(
        second_control.identity().attempt(),
        first_control.identity().attempt(),
        "an attempt is never reused, so an ambiguous outcome stays reconcilable"
    );
    assert_eq!(
        resumed_grant
            .last_committed_step()
            .map(|label| label.step().index()),
        Some(1),
        "the resumed attempt is told where to continue from"
    );
    assert_eq!(
        second_control
            .stamp_proposal()
            .expect("the successor may propose after State's step")
            .step()
            .index(),
        2,
        "a successor never reuses the State-committed proposal index"
    );

    let orphan = first_execution
        .stamp_proposal()
        .expect("the orphan keeps going");
    let error = second_control
        .admit_frame(&over_the_wire(&orphan), FrameClass::Proposal)
        .unwrap_err();
    // The orphan is a different attempt of the same execution, so it is refused
    // by attempt before the fence is even compared. A superseded owner of the
    // SAME attempt is the `StaleFence` case above.
    assert!(matches!(
        error,
        ExecutionError::ForeignAttempt { ref attempt, ref expected_attempt, .. }
            if attempt.as_str() == "attempt-1" && expected_attempt.as_str() == "attempt-2"
    ));

    let _ = ExecutionSession::admit(&resumed_grant, &audience(), MonotonicInstant::ORIGIN)
        .expect("the new attempt opens");
}

// ── Cancellation ────────────────────────────────────────────────────────────

/// Withdrawal is reported through the session's own budget, as a typed
/// outcome. It rolls nothing back: a step already admitted stays admitted.
#[test]
fn cancellation_is_reported_through_the_sessions_budget() {
    let (_, execution) = paired();
    let family = polyc_state::id::OperationFamily::new("execution.turn");
    assert!(
        execution
            .call_context()
            .check(MonotonicInstant::ORIGIN, &family)
            .is_ok()
    );

    execution.call_context().cancellation().cancel();
    assert!(matches!(
        execution
            .call_context()
            .check(MonotonicInstant::ORIGIN, &family),
        Err(polyc_state::error::StateError::Cancelled { .. })
    ));
}

// ── Identity hygiene ────────────────────────────────────────────────────────

#[test]
fn an_empty_identity_names_nothing() {
    for (execution, attempt, field) in [
        ("", "attempt-1", "execution_id"),
        ("exec-1", "", "attempt_id"),
    ] {
        let error = ExecutionIdentity::new(
            ExecutionId::new(execution),
            AttemptId::new(attempt),
            FencingToken::new(7),
        )
        .unwrap_err();
        assert!(
            matches!(error, ExecutionError::Malformed { field: ref f, .. } if f == field),
            "{execution}/{attempt}"
        );
    }

    let error = ExecutionGrant::new(
        identity(7),
        OwnerId::new(""),
        AttemptId::new("claim-attempt-1"),
        audience(),
        None,
        Duration::from_secs(30),
        capabilities(),
        TenantId::new("test-tenant"),
        ModelConversationId::new("test-conversation"),
    )
    .unwrap_err();
    assert!(matches!(error, ExecutionError::Malformed { ref field, .. } if field == "owner_id"));
}

/// The derivation is length prefixed, so an identity that contains the shape of
/// the next field cannot be made to collide with another attempt's step.
#[test]
fn the_step_derivation_cannot_be_made_to_collide() {
    let left = StepId::derive(&ExecutionId::new("ab"), &AttemptId::new("cd"), 1);
    let right = StepId::derive(&ExecutionId::new("a"), &AttemptId::new("bcd"), 1);
    assert_ne!(left, right);
    assert_eq!(left.as_str().len(), 64, "a sha-256 digest, hex encoded");

    assert_ne!(
        StepId::derive(&ExecutionId::new("ab"), &AttemptId::new("cd"), 1),
        StepId::derive(&ExecutionId::new("ab"), &AttemptId::new("cd"), 2),
    );
}