pub struct EffectDispositionRule {
    pub effect_variant: String,
    pub disposition: EffectDisposition,  // Local | External | Routed { .. }
}

pub struct EffectDispositionRule {
    pub effect_variant: String,
    pub disposition: EffectDisposition,
    /// When set, this effect participates in an owner-handoff protocol.
    /// The named protocol must be declared as an `EffectHandoffProtocol`
    /// in every composition that includes this machine.
    ///
    /// Only meaningful for `Local` or `External` dispositions — `Routed`
    /// effects are already constrained by composition routes.
    pub handoff_protocol: Option<String>,
}

/// Declares a named actor participating in this composition.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ActorSchema {
    pub name: String,
    pub kind: ActorKind,
}

/// Distinguishes machine actors from owner actors.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ActorKind {
    /// Driven by machine transitions — deterministic given inputs.
    Machine,
    /// Represents a host/runtime that realizes effects and provides
    /// protocol-constrained feedback. Nondeterministic within protocol
    /// bounds: chooses *when* and *which* allowed feedback to submit,
    /// but cannot submit feedback outside the declared protocol.
    Owner,
}

pub struct CompositionSchema {
    // ... all existing fields unchanged ...

    /// Explicit actor declarations. Every `MachineInstance.actor` must
    /// reference an `ActorSchema.name` where `kind == Machine`.
    /// Owner actors are referenced only by `EffectHandoffProtocol.realizing_actor`.
    pub actors: Vec<ActorSchema>,
}

/// Declares the contract between a machine that emits an effect and an
/// owner actor that realizes it and provides feedback.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EffectHandoffProtocol {
    /// Protocol name — must match the `handoff_protocol` value on the
    /// producing machine's `EffectDispositionRule`.
    pub name: String,

    /// The machine instance that produces the effect.
    pub producer_instance: String,

    /// The effect variant that triggers this protocol.
    pub effect_variant: String,

    /// The owner actor that realizes the effect. Must reference an
    /// `ActorSchema` with `kind == Owner`.
    pub realizing_actor: String,

    /// Fields from the effect variant that correlate the obligation to
    /// the feedback. These become the typed obligation record in
    /// generated code.
    pub correlation_fields: Vec<String>,

    /// Machine inputs that the owner actor may submit as feedback.
    /// The owner may submit any of these, but nothing outside this set.
    pub allowed_feedback_inputs: Vec<FeedbackInputRef>,

    /// When and how the obligation must be closed.
    pub closure_policy: ClosurePolicy,

    /// Optional fairness annotation for TLA+ liveness claims.
    /// e.g., "eventual feedback under task-scheduling fairness"
    pub liveness_annotation: Option<String>,
}

/// References a specific machine input that the owner may submit.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FeedbackInputRef {
    /// The machine instance receiving the feedback.
    pub machine_instance: String,
    /// The input variant on that machine.
    pub input_variant: String,
}

/// Determines when a handoff obligation is considered closed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ClosurePolicy {
    /// Owner must acknowledge (provide at least one feedback input).
    /// Obligation remains open until feedback is received.
    AckRequired,
    /// Owner must either acknowledge or the machine must reach a
    /// terminal/abort phase. Allows cleanup without feedback.
    AckOrAbort,
    /// Obligation is closed when the machine reaches a terminal phase,
    /// regardless of whether feedback was received. Fire-and-observe.
    TerminalClosure,
}

pub struct CompositionSchema {
    // ... all existing fields unchanged ...
    pub actors: Vec<ActorSchema>,              // from step 6
    pub handoff_protocols: Vec<EffectHandoffProtocol>,  // new
}

pub enum CompositionInvariantKind {
    // ... existing variants unchanged ...

    /// Asserts that a handoff-annotated effect has a matching protocol.
    HandoffProtocolCovered {
        producer_instance: String,
        effect_variant: String,
        protocol_name: String,
    },
}

disposition_with_handoff("SpawnDrainTask", EffectDisposition::Local,
    Some("comms_drain_spawn")),
disposition("SetTurnBoundaryDrainSuppressed", EffectDisposition::Local),
    // handoff_protocol: None — local projection, no owner feedback needed
disposition_with_handoff("AbortDrainTask", EffectDisposition::Local,
    Some("comms_drain_abort")),

actors: vec![
    // ... existing Machine actors ...
    ActorSchema { name: "session_host".into(), kind: ActorKind::Owner },
],
handoff_protocols: vec![
    EffectHandoffProtocol {
        name: "comms_drain_spawn".into(),
        producer_instance: "comms_drain".into(),
        effect_variant: "SpawnDrainTask".into(),
        realizing_actor: "session_host".into(),
        correlation_fields: vec!["mode".into()],
        allowed_feedback_inputs: vec![
            FeedbackInputRef {
                machine_instance: "comms_drain".into(),
                input_variant: "TaskSpawned".into(),
            },
            FeedbackInputRef {
                machine_instance: "comms_drain".into(),
                input_variant: "TaskExited".into(),
            },
        ],
        closure_policy: ClosurePolicy::AckRequired,
        liveness_annotation: Some(
            "eventual feedback under task-scheduling fairness".into(),
        ),
    },
    EffectHandoffProtocol {
        name: "comms_drain_abort".into(),
        producer_instance: "comms_drain".into(),
        effect_variant: "AbortDrainTask".into(),
        realizing_actor: "session_host".into(),
        correlation_fields: vec![],
        allowed_feedback_inputs: vec![
            FeedbackInputRef {
                machine_instance: "comms_drain".into(),
                input_variant: "AbortObserved".into(),
            },
        ],
        closure_policy: ClosurePolicy::TerminalClosure,
        liveness_annotation: None,
    },
],

pub use composition::{
    // ... existing re-exports ...
    ActorKind, ActorSchema,                    // step 6
    ClosurePolicy, EffectHandoffProtocol,      // step 7
    FeedbackInputRef,                          // step 7
};

HandoffProtocolOnRoutedEffect {
    variant: String,
}

ActorKindMismatch {
    actor: String,
    expected: ActorKind,
    actual: ActorKind,
},
UnknownHandoffProducer {
    protocol: String,
    instance: String,
},
UnknownHandoffEffect {
    protocol: String,
    effect: String,
},
UnknownHandoffActor {
    protocol: String,
    actor: String,
},
HandoffActorNotOwner {
    protocol: String,
    actor: String,
},
UnknownHandoffFeedbackMachine {
    protocol: String,
    machine: String,
},
UnknownHandoffFeedbackInput {
    protocol: String,
    machine: String,
    input: String,
},
HandoffProtocolMismatch {
    protocol: String,
    effect_variant: String,
    expected_protocol: String,
},
MissingHandoffProtocol {
    from_instance: String,
    effect_variant: String,
    expected_protocol: String,
},
TerminalClosureRequiresTerminalPhases {
    protocol: String,
    producer_instance: String,
},
UnknownHandoffCorrelationField {
    protocol: String,
    field: String,
},

field("pending_op_ids", TypeRef::Option(Box::new(
    TypeRef::Seq(Box::new(TypeRef::Named("OperationId".into())))
))),

field("pending_op_refs", TypeRef::Option(Box::new(
    TypeRef::Seq(Box::new(TypeRef::Named("AsyncOpRef".into())))
))),

field("barrier_satisfied", TypeRef::Bool),

VariantSchema {
    name: "OpsBarrierSatisfied".into(),
    fields: vec![
        field("run_id", TypeRef::Named("RunId".into())),
        field("barrier_token", TypeRef::Named("BarrierToken".into())),
    ],
},

Guard {
    name: "barrier_satisfied".into(),
    expr: Expr::Eq(
        Box::new(Expr::Field("barrier_satisfied".into())),
        Box::new(Expr::Bool(true)),
    ),
},

TransitionSchema {
    name: "OpsBarrierSatisfied".into(),
    from: vec!["WaitingForOps".into()],
    on: InputMatch {
        variant: "OpsBarrierSatisfied".into(),
        bindings: vec!["run_id".into(), "barrier_token".into()],
    },
    guards: vec![
        Guard {
            name: "matches_active_run".into(),
            expr: Expr::Eq(
                Box::new(Expr::Field("active_run".into())),
                Box::new(Expr::Some(Box::new(Expr::Binding("run_id".into())))),
            ),
        },
    ],
    updates: vec![Update::Assign {
        field: "barrier_satisfied".into(),
        expr: Expr::Bool(true),
    }],
    to: "WaitingForOps".into(),
    emit: vec![],
}

Update::Assign {
    field: "barrier_satisfied".into(),
    expr: Expr::ContainsWhere {
        seq: Box::new(Expr::Binding("op_refs".into())),
        predicate: "wait_policy_is_barrier".into(),
        then: Expr::Bool(false),
        otherwise: Expr::Bool(true),
    },
},

# Schema validation
./scripts/repo-cargo nextest run -p meerkat-machine-schema

# Machine TLC verification
cargo run -p xtask -- machine-verify --all

# Drift check
cargo run -p xtask -- machine-check-drift --all

# RMAT audit (includes new structural seam rules)
cargo run -p xtask -- rmat-audit

# Full test suite
./scripts/repo-cargo nextest run --workspace --show-progress none \
    --status-level none --final-status-level fail