polyc-turn-runner 2026.8.3

//! The fenced Execution protocol: identities, the active fence, steps,
//! deadlines, and named capability ceilings (`#1565`, D4).
//!
//! Control claims a dispatch with a fenced claim, then starts or resumes
//! Execution with a stable execution identity, the active fencing token, and
//! the last committed step. Execution labels every frame it sends. Each side
//! refuses a frame that does not belong to the attempt it is running, so an
//! orphaned Execution cannot keep proposing after another Control replica takes
//! ownership.
//!
//! # This is the State plane's fence, carried outward
//!
//! There is one fencing concept in this system, not two.
//! [`polyc_state::command::FencingToken`] names the ownership epoch, and
//! [`polyc_state::id::AttemptId`] names both the durable claim attempt carried
//! by [`ExecutionGrant`] and, separately, one transport attempt carried by
//! [`ExecutionIdentity`]. The work-claims contract (`polyc_state::claims`)
//! mints the claim attempt and fence; Control mints a globally unique transport
//! attempt per dial. The distinction lets one claim survive a re-dial without
//! ever reusing the identity of an ambiguous transport outcome.
//!
//! What this module adds is the boundary's own vocabulary: [`ExecutionId`],
//! which names the execution a turn runs as, and [`StepId`], which names one
//! step inside one attempt.
//!
//! # Why the step identity is derived, not invented
//!
//! A duplicate frame must be recognizable as a replay rather than committed
//! twice. [`StepId::derive`] is a pure function of the execution, the attempt,
//! and the step index, so a resent frame carries the identical identity. A
//! receiver recomputes the identity and refuses a sender's value that disagrees
//! — a random per-frame identity would make every replay look like new work.
//!
//! # Why the budget is a duration
//!
//! A monotonic origin belongs to the process that read it. The wire therefore
//! carries what is *left* of the end-to-end budget, and the receiving side
//! anchors it onto its own clock exactly once, when it admits the grant
//! ([`ExecutionSession::admit`]). The session then carries that framed
//! [`CallContext`] for the whole attempt. Re-anchoring per frame would hand
//! every retry a fresh full budget, which is a defect this repository has
//! already had once.
//!
//! # Versioning
//!
//! [`ExecutionProtocolVersion`] is deliberately not
//! [`polyc_state::id::ProtocolVersion`]. That version is persisted with durable
//! records and validated during replay, so raising it would make State refuse
//! its own log. Execution protocol semantics are a transport contract and
//! advance on their own. The check is exact equality: a peer at another version
//! is refused, never guessed at.

use std::{fmt, sync::OnceLock, time::Duration};

use polyc_capability::{Capability, CapabilitySet};
use polyc_proto::proto::polychrome::harness::v1::{
    ExecutionCapabilities as WireExecutionCapabilities, ExecutionGrant as WireExecutionGrant,
    ExecutionLabel as WireExecutionLabel, ExecutionStep as WireExecutionStep,
    harness_message::Frame as HarnessFrame,
};
use polyc_state::{
    cancel::CancellationToken,
    command::FencingToken,
    context::CallContext,
    deadline::{Clock, MonotonicInstant, ProductionClock},
    id::{AttemptId, OwnerId},
    model_attempt::{ModelConversationId, TenantId},
};

/// The most labeled steps one attempt may ever be granted.
///
/// The bound belongs to the contract, not to a backend: a wire value above it
/// is refused rather than clamped, because a clamped ceiling is one the sender
/// believes it did not get.
pub const MAX_LABELED_STEPS: u32 = 4096;

/// The longest end-to-end budget one attempt may be granted.
///
/// An attempt that outlives every takeover window would make the fence
/// decorative. Refused, never clamped, for the same reason as
/// [`MAX_LABELED_STEPS`].
pub const MAX_EXECUTION_BUDGET: Duration = Duration::from_hours(1);

// ---------------------------------------------------------------------------
// Failures
// ---------------------------------------------------------------------------

/// Why a fenced Execution frame was refused.
///
/// Every variant is terminal for the frame that carried it. None of them says
/// anything about work that already committed: an attempt that lost its fence
/// learns it lost, and reconciliation happens through the attempt history the
/// work-claims contract keeps.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ExecutionError {
    /// A required part of the protocol was absent or empty.
    #[error("the fenced Execution protocol requires {field}: {reason}")]
    Malformed {
        /// The field the sender left absent, empty, or out of bounds.
        field: String,
        /// Why this side cannot proceed without it.
        reason: String,
    },
    /// The peer speaks another version of these semantics.
    #[error("this build speaks Execution protocol {current}, and the peer declared {declared}")]
    VersionMismatch {
        /// The version the peer declared.
        declared: ExecutionProtocolVersion,
        /// The version this build speaks.
        current: ExecutionProtocolVersion,
    },
    /// The frame belongs to another execution or another attempt.
    #[error(
        "this frame names execution {execution} attempt {attempt}, and this session runs {expected_execution} attempt {expected_attempt}"
    )]
    ForeignAttempt {
        /// The execution the frame named.
        execution: ExecutionId,
        /// The attempt the frame named.
        attempt: AttemptId,
        /// The execution this session runs.
        expected_execution: ExecutionId,
        /// The attempt this session runs.
        expected_attempt: AttemptId,
    },
    /// The frame presents a fence that is not this session's.
    ///
    /// A lower fence is a superseded owner still talking. A higher one is a
    /// takeover this side has not been told about. Both are refused, because
    /// this side can order neither.
    #[error("this frame presents fence {presented}, and this session holds {current}")]
    StaleFence {
        /// The fence the frame presented.
        presented: FencingToken,
        /// The fence this session holds.
        current: FencingToken,
    },
    /// The frame repeats a step this side already admitted.
    #[error("step {index} was already admitted on this attempt")]
    DuplicateStep {
        /// The repeated index.
        index: u64,
    },
    /// The frame's step does not follow the one before it.
    #[error("this frame is step {index}, and the next step of this attempt is {expected}")]
    OutOfOrderStep {
        /// The index the frame carried.
        index: u64,
        /// The index this side required.
        expected: u64,
    },
    /// The frame's step identity is not the one its coordinates derive to.
    ///
    /// A sender cannot invent a step identity: the identity is what makes a
    /// resent frame recognizable as a replay.
    #[error("step {index} derives to another identity than the one this frame carries")]
    ForgedStepId {
        /// The index whose identity disagreed.
        index: u64,
    },
    /// The attempt has emitted every step its grant allows.
    #[error("this attempt is granted {granted} labeled steps, and step {index} is past that")]
    StepBudgetExhausted {
        /// The index that would have exceeded the grant.
        index: u64,
        /// The grant's ceiling.
        granted: u32,
    },
    /// The grant names a capability this build does not recognize.
    ///
    /// Refused rather than skipped: a grant silently narrowed to the names one
    /// side happens to know is a grant neither side agrees on.
    #[error("this grant names capabilities this build does not recognize: {unknown}")]
    UnknownCapability {
        /// The unrecognized names, comma separated.
        unknown: String,
    },
    /// The grant was routed to another Execution workload.
    #[error("this grant is for Execution audience {declared}, but this workload is {expected}")]
    WrongAudience {
        /// The recipient named by Control.
        declared: ExecutionAudience,
        /// The independently assigned identity of this workload.
        expected: ExecutionAudience,
    },
}

// ---------------------------------------------------------------------------
// Version
// ---------------------------------------------------------------------------

/// The version of the semantics carried by the fenced Execution protocol.
///
/// Independent of [`polyc_state::id::ProtocolVersion`], which is persisted with
/// durable records and validated on replay. See this module's documentation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ExecutionProtocolVersion(u32);

impl ExecutionProtocolVersion {
    /// The version this build speaks: fenced identities, steps, duration
    /// budgets, named capability ceilings, State-backed step receipts,
    /// Control-derived tenant and conversation authority in the opening grant,
    /// and model calls brokered by Control rather than dialled by Execution.
    ///
    /// Brokered model calls raise this number even though the frames that carry
    /// them are additive. A peer that keeps its own provider credentials speaks
    /// different semantics on the same frames, and the equality check is the
    /// only thing that refuses it. Sharing a number with that peer would let a
    /// mixed pair agree, and put provider credentials back inside Execution.
    pub const CURRENT: Self = Self(3);

    /// Wraps `version` as an Execution protocol semantics version.
    #[must_use]
    pub const fn new(version: u32) -> Self {
        Self(version)
    }

    /// Returns the numeric wire version.
    #[must_use]
    pub const fn get(self) -> u32 {
        self.0
    }
}

impl fmt::Display for ExecutionProtocolVersion {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(formatter, "v{}", self.0)
    }
}

/// Refuses a peer speaking a version of these semantics this build does not.
///
/// Exact equality, and the refusal is terminal: retrying at the same version
/// changes nothing. This is the same rule the State plane's transport boundary
/// applies, for the same reason — one lockstep-versioned product refuses rather
/// than guessing at a shape it does not know.
///
/// # Errors
///
/// Returns [`ExecutionError::VersionMismatch`] for every version but
/// [`ExecutionProtocolVersion::CURRENT`].
pub const fn check_execution_version(
    declared: ExecutionProtocolVersion,
) -> Result<(), ExecutionError> {
    if declared.get() == ExecutionProtocolVersion::CURRENT.get() {
        return Ok(());
    }
    Err(ExecutionError::VersionMismatch {
        declared,
        current: ExecutionProtocolVersion::CURRENT,
    })
}

// ---------------------------------------------------------------------------
// Identities
// ---------------------------------------------------------------------------

/// The stable identity of one execution.
///
/// An execution outlives every attempt at it: a resumed turn keeps this
/// identity and takes a new [`AttemptId`], exactly as a reclaimed work item
/// keeps its `WorkId`.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ExecutionId(String);

impl ExecutionId {
    /// Wraps `value` as an execution identity.
    #[must_use]
    pub fn new(value: impl Into<String>) -> Self {
        Self(value.into())
    }

    /// Returns the identity's textual form.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Reports whether the identity is the empty string.
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

/// The routing identity of the Execution workload a grant is allowed to open.
///
/// A per-conversation harness learns this value from its `SandboxClaim`
/// assignment at startup, independently of the Control caller. A shared
/// harness uses the deployment's fixed audience.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ExecutionAudience(String);

impl ExecutionAudience {
    /// Builds a non-empty recipient identity.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::Malformed`] when `value` is empty.
    pub fn new(value: impl Into<String>) -> Result<Self, ExecutionError> {
        let value = value.into();
        if value.is_empty() {
            return Err(ExecutionError::Malformed {
                field: "audience".to_owned(),
                reason: "a grant with no recipient can be replayed at any harness".to_owned(),
            });
        }
        Ok(Self(value))
    }

    /// Returns the recipient's canonical text.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for ExecutionAudience {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl fmt::Display for ExecutionId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

/// The identity of one step of one attempt.
///
/// Derived, never invented — see [`StepId::derive`].
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct StepId(String);

impl StepId {
    /// Derives the identity of step `index` of `attempt` of `execution`.
    ///
    /// The canonical bytes are length prefixed, so an identity containing the
    /// separator cannot be made to derive to another step's identity. The
    /// digest is hexadecimal SHA-256 over those bytes.
    #[must_use]
    pub fn derive(execution: &ExecutionId, attempt: &AttemptId, index: u64) -> Self {
        use sha2::Digest as _;
        let mut hasher = sha2::Sha256::new();
        hasher.update(b"polychrome.execution.step.v1");
        for part in [execution.as_str().as_bytes(), attempt.as_str().as_bytes()] {
            hasher.update(u64::try_from(part.len()).unwrap_or(u64::MAX).to_be_bytes());
            hasher.update(part);
        }
        hasher.update(index.to_be_bytes());
        Self(hex::encode(hasher.finalize()))
    }

    /// Returns the identity's textual form.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for StepId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

/// The fenced attempt one execution is running.
///
/// The three values travel together because they are checked together: a frame
/// belongs to this attempt only when all three agree.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecutionIdentity {
    execution: ExecutionId,
    attempt: AttemptId,
    fence: FencingToken,
}

impl ExecutionIdentity {
    /// Builds the identity one attempt runs under.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::Malformed`] for an empty execution or attempt
    /// identity, and for [`FencingToken::UNCLAIMED`] — an unfenced attempt is
    /// one no authority can order against a takeover.
    pub fn new(
        execution: ExecutionId,
        attempt: AttemptId,
        fence: FencingToken,
    ) -> Result<Self, ExecutionError> {
        if execution.is_empty() {
            return Err(ExecutionError::Malformed {
                field: "execution_id".to_owned(),
                reason: "an execution identity is what a resumed attempt is the same execution as"
                    .to_owned(),
            });
        }
        if attempt.is_empty() {
            return Err(ExecutionError::Malformed {
                field: "attempt_id".to_owned(),
                reason: "an attempt identity is what makes an ambiguous outcome reconcilable"
                    .to_owned(),
            });
        }
        if fence == FencingToken::UNCLAIMED {
            return Err(ExecutionError::Malformed {
                field: "fence".to_owned(),
                reason: "the unclaimed token names no ownership epoch, so nothing could refuse a \
                         superseded owner"
                    .to_owned(),
            });
        }
        Ok(Self {
            execution,
            attempt,
            fence,
        })
    }

    /// Returns the execution this attempt belongs to.
    #[must_use]
    pub const fn execution(&self) -> &ExecutionId {
        &self.execution
    }

    /// Returns the attempt.
    #[must_use]
    pub const fn attempt(&self) -> &AttemptId {
        &self.attempt
    }

    /// Returns the ownership epoch this attempt runs under.
    #[must_use]
    pub const fn fence(&self) -> FencingToken {
        self.fence
    }
}

// ---------------------------------------------------------------------------
// Steps
// ---------------------------------------------------------------------------

/// One step of one attempt: its position and its derived identity.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecutionStep {
    index: u64,
    id: StepId,
}

impl ExecutionStep {
    /// Builds step `index` of `identity`'s attempt, deriving its identity.
    #[must_use]
    pub fn at(identity: &ExecutionIdentity, index: u64) -> Self {
        Self {
            id: StepId::derive(identity.execution(), identity.attempt(), index),
            index,
        }
    }

    /// Returns the step's position in its attempt.
    #[must_use]
    pub const fn index(&self) -> u64 {
        self.index
    }

    /// Returns the step's derived identity.
    #[must_use]
    pub const fn id(&self) -> &StepId {
        &self.id
    }
}

// ---------------------------------------------------------------------------
// Named capability and frame ceilings
// ---------------------------------------------------------------------------

/// The named agent capabilities and labeled-frame count for one attempt.
///
/// The capability set bounds decisions in `polyc-agent`'s capability gate. It
/// does not describe operating-system authority or sandbox network routes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ExecutionCapabilities {
    granted: CapabilitySet,
    max_labeled_steps: u32,
}

impl ExecutionCapabilities {
    /// Builds the named capability and frame ceilings of one attempt.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::Malformed`] when `max_labeled_steps` is zero
    /// — a grant that permits not even the opening frame — or above
    /// [`MAX_LABELED_STEPS`]. The ceiling is refused rather than clamped.
    pub fn new(granted: CapabilitySet, max_labeled_steps: u32) -> Result<Self, ExecutionError> {
        if max_labeled_steps == 0 || max_labeled_steps > MAX_LABELED_STEPS {
            return Err(ExecutionError::Malformed {
                field: "max_labeled_steps".to_owned(),
                reason: format!(
                    "a ceiling of {max_labeled_steps} is outside 1..={MAX_LABELED_STEPS}; this \
                     contract refuses a ceiling it cannot grant rather than clamping one the \
                     sender would believe it received"
                ),
            });
        }
        Ok(Self {
            granted,
            max_labeled_steps,
        })
    }

    /// Returns the capability set this attempt is granted.
    #[must_use]
    pub const fn granted(self) -> CapabilitySet {
        self.granted
    }

    /// Reports whether the grant covers `capability`.
    #[must_use]
    pub const fn permits(self, capability: Capability) -> bool {
        self.granted.contains(capability)
    }

    /// Returns the most labeled steps this attempt may emit.
    #[must_use]
    pub const fn max_labeled_steps(self) -> u32 {
        self.max_labeled_steps
    }
}

// ---------------------------------------------------------------------------
// The grant
// ---------------------------------------------------------------------------

/// What Control hands Execution to open or resume one attempt.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecutionGrant {
    opening: ExecutionLabel,
    owner: OwnerId,
    claim_attempt: AttemptId,
    tenant: TenantId,
    conversation: ModelConversationId,
    audience: ExecutionAudience,
    last_committed_step: Option<ExecutionLabel>,
    budget: Duration,
    capabilities: ExecutionCapabilities,
}

impl ExecutionGrant {
    /// Builds the grant one attempt runs under.
    ///
    /// The grant owns the opening frame's label, so the identity a receiver
    /// reads and the identity the envelope declares come from one value rather
    /// than two that could disagree.
    ///
    /// `budget` is what is left of the end-to-end budget when the grant is
    /// sent, never an absolute instant. `last_committed_step` is [`None`] on a
    /// first attempt and names the resume point on a later one.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::Malformed`] for an empty owner identity, for a
    /// zero budget, and for a budget above [`MAX_EXECUTION_BUDGET`].
    // Each argument is a distinct field of the fenced grant contract, and every
    // caller names all of them. A parameter struct would move the same list
    // behind a type that a future field could default into silently, which is
    // the failure #1241 and #1238 record on this exact wire.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        identity: ExecutionIdentity,
        owner: OwnerId,
        claim_attempt: AttemptId,
        audience: ExecutionAudience,
        last_committed_step: Option<ExecutionLabel>,
        budget: Duration,
        capabilities: ExecutionCapabilities,
        tenant: TenantId,
        conversation: ModelConversationId,
    ) -> Result<Self, ExecutionError> {
        Self::from_declared(
            ExecutionLabel::new(identity, 0),
            owner,
            claim_attempt,
            audience,
            last_committed_step,
            budget,
            capabilities,
            tenant,
            conversation,
        )
    }

    /// Builds a grant around a label a peer declared, rather than minting one.
    ///
    /// The wire conversion uses this so a peer's declared version survives into
    /// [`ExecutionSession::admit`], which is what reports the mismatch.
    // Mirrors `new`'s argument list one for one; see its note.
    #[allow(clippy::too_many_arguments)]
    fn from_declared(
        opening: ExecutionLabel,
        owner: OwnerId,
        claim_attempt: AttemptId,
        audience: ExecutionAudience,
        last_committed_step: Option<ExecutionLabel>,
        budget: Duration,
        capabilities: ExecutionCapabilities,
        tenant: TenantId,
        conversation: ModelConversationId,
    ) -> Result<Self, ExecutionError> {
        if opening.step().index() != 0 {
            return Err(ExecutionError::OutOfOrderStep {
                index: opening.step().index(),
                expected: 0,
            });
        }
        let expected_opening = StepId::derive(
            opening.identity().execution(),
            opening.identity().attempt(),
            opening.step().index(),
        );
        if opening.step().id() != &expected_opening {
            return Err(ExecutionError::ForgedStepId {
                index: opening.step().index(),
            });
        }
        if owner.is_empty() {
            return Err(ExecutionError::Malformed {
                field: "owner_id".to_owned(),
                reason: "the owner is who holds the fence this attempt runs under".to_owned(),
            });
        }
        if claim_attempt.is_empty() {
            return Err(ExecutionError::Malformed {
                field: "claim_attempt_id".to_owned(),
                reason: "the durable claim attempt is the proof this dispatch was authorized"
                    .to_owned(),
            });
        }
        if tenant.is_empty() || conversation.is_empty() {
            return Err(ExecutionError::Malformed {
                field: "model_authority".to_owned(),
                reason: "Control must explicitly bind verified tenant and conversation authority"
                    .to_owned(),
            });
        }
        if let Some(committed) = &last_committed_step {
            check_execution_version(committed.version())?;
            if committed.step().index() == 0 {
                return Err(ExecutionError::Malformed {
                    field: "last_committed_step".to_owned(),
                    reason: "step zero opens an attempt and is not a committed outcome".to_owned(),
                });
            }
            if committed.identity().execution() != opening.identity().execution() {
                return Err(ExecutionError::ForeignAttempt {
                    execution: committed.identity().execution().clone(),
                    attempt: committed.identity().attempt().clone(),
                    expected_execution: opening.identity().execution().clone(),
                    expected_attempt: opening.identity().attempt().clone(),
                });
            }
            if committed.identity().attempt() == opening.identity().attempt() {
                return Err(ExecutionError::Malformed {
                    field: "last_committed_step.attempt_id".to_owned(),
                    reason: "a resumed attempt must be fresh, not resume itself".to_owned(),
                });
            }
            if committed.identity().fence() > opening.identity().fence() {
                return Err(ExecutionError::StaleFence {
                    presented: committed.identity().fence(),
                    current: opening.identity().fence(),
                });
            }
            let expected = StepId::derive(
                committed.identity().execution(),
                committed.identity().attempt(),
                committed.step().index(),
            );
            if committed.step().id() != &expected {
                return Err(ExecutionError::ForgedStepId {
                    index: committed.step().index(),
                });
            }
        }
        if budget.is_zero() || budget > MAX_EXECUTION_BUDGET {
            return Err(ExecutionError::Malformed {
                field: "budget_nanos".to_owned(),
                reason: format!(
                    "a budget of {budget:?} is outside 1ns..={MAX_EXECUTION_BUDGET:?}; an attempt \
                     that outlives every takeover window makes the fence decorative"
                ),
            });
        }
        Ok(Self {
            opening,
            owner,
            claim_attempt,
            tenant,
            conversation,
            audience,
            last_committed_step,
            budget,
            capabilities,
        })
    }

    /// Returns the opening frame's label.
    #[must_use]
    pub const fn opening(&self) -> &ExecutionLabel {
        &self.opening
    }

    /// Returns the fenced attempt this grant opens.
    #[must_use]
    pub const fn identity(&self) -> &ExecutionIdentity {
        self.opening.identity()
    }

    /// Returns the claim owner Control holds the fence as.
    #[must_use]
    pub const fn owner(&self) -> &OwnerId {
        &self.owner
    }

    /// Returns the durable State claim attempt authorizing this dispatch.
    #[must_use]
    pub const fn claim_attempt(&self) -> &AttemptId {
        &self.claim_attempt
    }

    /// Returns the verified ingress tenant Control bound to this execution.
    #[must_use]
    pub const fn tenant(&self) -> &TenantId {
        &self.tenant
    }

    /// Returns the trusted conversation Control bound to this execution.
    #[must_use]
    pub const fn conversation(&self) -> &ModelConversationId {
        &self.conversation
    }

    /// Returns the independently assigned Execution recipient.
    #[must_use]
    pub const fn audience(&self) -> &ExecutionAudience {
        &self.audience
    }

    /// Returns the last step this execution durably committed, if any.
    #[must_use]
    pub const fn last_committed_step(&self) -> Option<&ExecutionLabel> {
        self.last_committed_step.as_ref()
    }

    /// Returns what is left of the end-to-end budget.
    ///
    /// A remaining duration with no clock domain. [`ExecutionSession::admit`]
    /// anchors it once.
    #[must_use]
    pub const fn budget(&self) -> Duration {
        self.budget
    }

    /// Returns the named capability-gate and frame ceilings for this attempt.
    #[must_use]
    pub const fn capabilities(&self) -> ExecutionCapabilities {
        self.capabilities
    }
}

// ---------------------------------------------------------------------------
// The label
// ---------------------------------------------------------------------------

/// The label every frame carries, in both directions.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecutionLabel {
    version: ExecutionProtocolVersion,
    identity: ExecutionIdentity,
    step: ExecutionStep,
}

impl ExecutionLabel {
    /// Builds the label for step `index` of `identity`'s attempt, at the
    /// version this build speaks.
    #[must_use]
    pub fn new(identity: ExecutionIdentity, index: u64) -> Self {
        let step = ExecutionStep::at(&identity, index);
        Self {
            version: ExecutionProtocolVersion::CURRENT,
            identity,
            step,
        }
    }

    /// Returns the semantics version the sender declared.
    #[must_use]
    pub const fn version(&self) -> ExecutionProtocolVersion {
        self.version
    }

    /// Returns the fenced attempt this frame belongs to.
    #[must_use]
    pub const fn identity(&self) -> &ExecutionIdentity {
        &self.identity
    }

    /// Returns the step this frame is.
    #[must_use]
    pub const fn step(&self) -> &ExecutionStep {
        &self.step
    }
}

// ---------------------------------------------------------------------------
// Frame classes
// ---------------------------------------------------------------------------

/// What a frame is, for the purpose of the step sequence.
///
/// The three classes advance the sequence differently, so the rule lives in one
/// place rather than at each admission site.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameClass {
    /// The frame proposes something the fence orders: the turn input, a
    /// terminal batch or failure, a tool request, or an effect proposal. It
    /// takes the next step of the attempt.
    Proposal,
    /// The frame is display-only and never persisted. It names the step it
    /// precedes and takes none.
    Ephemeral,
    /// The frame answers a proposal. It repeats that proposal's step.
    Reply,
}

/// Classifies a frame for the step sequence.
///
/// Exhaustive on purpose — no wildcard — so a new frame kind forces a decision
/// here instead of defaulting into a class that silently changes the sequence.
#[must_use]
pub const fn classify_frame(frame: &HarnessFrame) -> FrameClass {
    match frame {
        HarnessFrame::Input(_)
        | HarnessFrame::Batch(_)
        | HarnessFrame::Failed(_)
        | HarnessFrame::PaidFetchRequest(_)
        | HarnessFrame::ControlPlaneToolRequest(_)
        | HarnessFrame::DispatchMutationRequest(_)
        | HarnessFrame::PeerCallRequest(_)
        | HarnessFrame::StepOutcomeProposal(_)
        | HarnessFrame::ModelRequest(_) => FrameClass::Proposal,
        HarnessFrame::Delta(_) => FrameClass::Ephemeral,
        HarnessFrame::PaidFetchResponse(_)
        | HarnessFrame::ControlPlaneToolResponse(_)
        | HarnessFrame::DispatchMutationResponse(_)
        | HarnessFrame::PeerCallResponse(_)
        | HarnessFrame::StepCommitReceipt(_)
        | HarnessFrame::ModelResponse(_) => FrameClass::Reply,
    }
}

// ---------------------------------------------------------------------------
// The session
// ---------------------------------------------------------------------------

/// One side's view of one fenced attempt.
///
/// Both sides run the same state machine over one shared step sequence: the
/// opening turn input is step 0 and Control stamps it, and every later proposal
/// takes the next index. `high` is therefore the highest proposal index this
/// side has stamped or admitted, whichever came last.
///
/// The session also holds the attempt's budget, anchored once onto this side's
/// clock when the session opened. Nothing re-anchors it.
#[derive(Debug, Clone)]
pub struct ExecutionSession {
    identity: ExecutionIdentity,
    capabilities: ExecutionCapabilities,
    context: CallContext,
    high: u64,
}

impl ExecutionSession {
    /// Opens Control's side of an attempt.
    ///
    /// The budget is anchored onto `now` here and never again. The opening
    /// frame's label is [`ExecutionGrant::opening`].
    #[must_use]
    pub fn open(grant: &ExecutionGrant, now: MonotonicInstant) -> Self {
        Self {
            identity: grant.identity().clone(),
            capabilities: grant.capabilities(),
            context: CallContext::from_origin_relative_budget(
                grant.budget(),
                CancellationToken::new(),
            )
            .in_frame(now),
            // A replacement attempt opens with its own step-zero envelope, but
            // its first *proposal* must follow the last outcome State accepted
            // for this execution.  `last_committed_step` is resume input only:
            // it changes neither this attempt's identity nor the State fence
            // Control later selects for the step-commit door.
            high: grant
                .last_committed_step()
                .map_or(0, |committed| committed.step().index()),
        }
    }

    /// Opens Execution's side of an attempt from the grant it was handed.
    ///
    /// Fail-closed: the version the grant declares is checked before a session
    /// exists. The budget is anchored onto `now` here and never again.
    ///
    /// A caller follows this with [`Self::check_opening_envelope`] before it
    /// decodes any payload, including the buffered transport.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::VersionMismatch`] when the peer speaks another
    /// version of these semantics.
    pub fn admit(
        grant: &ExecutionGrant,
        expected_audience: &ExecutionAudience,
        now: MonotonicInstant,
    ) -> Result<Self, ExecutionError> {
        check_execution_version(grant.opening().version())?;
        if grant.audience() != expected_audience {
            return Err(ExecutionError::WrongAudience {
                declared: grant.audience().clone(),
                expected: expected_audience.clone(),
            });
        }
        Ok(Self::open(grant, now))
    }

    /// Refuses an opening frame whose envelope label is not the grant's.
    ///
    /// The identity a receiver acts on comes from the grant. This is what makes
    /// the envelope's copy of it unable to say something else.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::VersionMismatch`], [`ExecutionError::ForeignAttempt`],
    /// [`ExecutionError::StaleFence`], or [`ExecutionError::OutOfOrderStep`],
    /// whichever the envelope disagreed on first.
    pub fn check_opening_envelope(&self, label: &ExecutionLabel) -> Result<(), ExecutionError> {
        check_execution_version(label.version())?;
        self.check_identity(label.identity())?;
        self.check_step_id(label.step())?;
        if label.step().index() != 0 {
            return Err(ExecutionError::OutOfOrderStep {
                index: label.step().index(),
                expected: 0,
            });
        }
        Ok(())
    }

    /// Returns the fenced attempt this session runs.
    #[must_use]
    pub const fn identity(&self) -> &ExecutionIdentity {
        &self.identity
    }

    /// Returns the named capability-gate and frame ceilings for this attempt.
    #[must_use]
    pub const fn capabilities(&self) -> ExecutionCapabilities {
        self.capabilities
    }

    /// Reports whether this attempt's named gate ceiling covers `capability`.
    ///
    /// This is the query the model and tool brokers ask before they act on a
    /// proposal. The brokers themselves arrive in later chunks; the grant they
    /// read is this one.
    #[must_use]
    pub const fn permits(&self, capability: Capability) -> bool {
        self.capabilities.permits(capability)
    }

    /// Returns the budget this attempt runs under, already in this side's
    /// clock frame.
    ///
    /// Carry the returned context through subcalls rather than rebuilding one
    /// from the grant: a second anchoring would mint the original budget again.
    #[must_use]
    pub const fn call_context(&self) -> &CallContext {
        &self.context
    }

    /// Cancels the attempt's carried context.
    ///
    /// Transport owners call this before aborting work so nested consumers of
    /// the framed context observe the same withdrawal.
    pub fn cancel(&self) {
        self.context.cancellation().cancel();
    }

    /// Returns the highest proposal step this attempt has reached.
    #[must_use]
    pub const fn high_step(&self) -> u64 {
        self.high
    }

    /// Stamps the label of the next proposal this side sends.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::StepBudgetExhausted`] when the attempt has
    /// emitted every step its grant allows.
    pub fn stamp_proposal(&mut self) -> Result<ExecutionLabel, ExecutionError> {
        let index = self.high.saturating_add(1);
        self.check_within_grant(index)?;
        self.high = index;
        Ok(ExecutionLabel::new(self.identity.clone(), index))
    }

    /// Stamps the label of a display-only frame.
    ///
    /// The frame names the step it precedes and takes none, so a turn that
    /// streams a thousand deltas spends one step, not a thousand.
    #[must_use]
    pub fn stamp_ephemeral(&self) -> ExecutionLabel {
        ExecutionLabel::new(self.identity.clone(), self.high.saturating_add(1))
    }

    /// Stamps the label of a reply to `proposal`.
    ///
    /// A reply repeats the step it answers, so a response can be matched to its
    /// request by the protocol and not only by a tool-call identifier.
    #[must_use]
    pub fn stamp_reply(&self, proposal: &ExecutionLabel) -> ExecutionLabel {
        ExecutionLabel::new(self.identity.clone(), proposal.step().index())
    }

    /// Admits an inbound frame's label and advances the sequence.
    ///
    /// Checks, in order: the version, the attempt identity, the fence, the
    /// derived step identity, and the step's place in the sequence for its
    /// class. A proposal advances the sequence; an ephemeral or reply frame
    /// does not.
    ///
    /// # Errors
    ///
    /// Returns the [`ExecutionError`] naming the first check that refused. A
    /// refusal is terminal for the frame: nothing about it is retried at the
    /// same coordinates.
    pub fn admit_frame(
        &mut self,
        label: &ExecutionLabel,
        class: FrameClass,
    ) -> Result<(), ExecutionError> {
        check_execution_version(label.version())?;
        self.check_identity(label.identity())?;
        self.check_step_id(label.step())?;

        let index = label.step().index();
        match class {
            FrameClass::Proposal => {
                let expected = self.high.saturating_add(1);
                if index == self.high {
                    return Err(ExecutionError::DuplicateStep { index });
                }
                if index != expected {
                    return Err(ExecutionError::OutOfOrderStep { index, expected });
                }
                self.check_within_grant(index)?;
                self.high = index;
            }
            FrameClass::Ephemeral => {
                let expected = self.high.saturating_add(1);
                if index != expected {
                    return Err(ExecutionError::OutOfOrderStep { index, expected });
                }
            }
            FrameClass::Reply => {
                if index == 0 || index > self.high {
                    return Err(ExecutionError::OutOfOrderStep {
                        index,
                        expected: self.high,
                    });
                }
            }
        }
        Ok(())
    }

    /// Admits a reply only when its label names the exact proposal being
    /// answered.
    ///
    /// Multiple tool proposals may be outstanding at once, so `high` alone is
    /// insufficient: the payload's correlation id selects `expected_step`, and
    /// the repeated label must agree with it.
    ///
    /// # Errors
    ///
    /// Returns whatever [`Self::admit_frame`] refuses the label with, plus
    /// [`ExecutionError::OutOfOrderStep`] when the label repeats a step other
    /// than the outstanding proposal this reply claims to answer.
    pub fn admit_reply(
        &mut self,
        label: &ExecutionLabel,
        expected_step: u64,
    ) -> Result<(), ExecutionError> {
        self.admit_frame(label, FrameClass::Reply)?;
        if label.step().index() != expected_step {
            return Err(ExecutionError::OutOfOrderStep {
                index: label.step().index(),
                expected: expected_step,
            });
        }
        Ok(())
    }

    /// Refuses a label that does not name this session's fenced attempt.
    fn check_identity(&self, presented: &ExecutionIdentity) -> Result<(), ExecutionError> {
        if presented.execution() != self.identity.execution()
            || presented.attempt() != self.identity.attempt()
        {
            return Err(ExecutionError::ForeignAttempt {
                execution: presented.execution().clone(),
                attempt: presented.attempt().clone(),
                expected_execution: self.identity.execution().clone(),
                expected_attempt: self.identity.attempt().clone(),
            });
        }
        if presented.fence() != self.identity.fence() {
            return Err(ExecutionError::StaleFence {
                presented: presented.fence(),
                current: self.identity.fence(),
            });
        }
        Ok(())
    }

    /// Refuses a step identity that is not the one its coordinates derive to.
    fn check_step_id(&self, step: &ExecutionStep) -> Result<(), ExecutionError> {
        if *step.id()
            != StepId::derive(
                self.identity.execution(),
                self.identity.attempt(),
                step.index(),
            )
        {
            return Err(ExecutionError::ForgedStepId {
                index: step.index(),
            });
        }
        Ok(())
    }

    /// Refuses a step past the grant's ceiling.
    ///
    /// The opening frame is step 0, so a grant of `n` allows steps `0..n`
    /// exclusive of `n`.
    fn check_within_grant(&self, index: u64) -> Result<(), ExecutionError> {
        if index >= u64::from(self.capabilities.max_labeled_steps()) {
            return Err(ExecutionError::StepBudgetExhausted {
                index,
                granted: self.capabilities.max_labeled_steps(),
            });
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Wire conversions
// ---------------------------------------------------------------------------
//
// Every conversion names every field. A `..Default::default()` spread would
// zero a field the other side later gains instead of failing to compile, which
// is the bug class issues #1241 and #1238 recorded on this exact boundary.

impl From<&ExecutionStep> for WireExecutionStep {
    fn from(value: &ExecutionStep) -> Self {
        Self {
            index: value.index(),
            step_id: value.id().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl From<&WireExecutionStep> for ExecutionStep {
    /// Reads a step off the wire without checking its identity.
    ///
    /// The identity is checked where the attempt is known —
    /// [`ExecutionSession::admit_frame`] recomputes it and refuses a value that
    /// disagrees. Reading and checking are separate so a malformed identity is
    /// reported as [`ExecutionError::ForgedStepId`] rather than as a parse
    /// failure that names no step.
    fn from(value: &WireExecutionStep) -> Self {
        Self {
            index: value.index,
            id: StepId(value.step_id.clone()),
        }
    }
}

impl From<&ExecutionCapabilities> for WireExecutionCapabilities {
    fn from(value: &ExecutionCapabilities) -> Self {
        Self {
            granted: value
                .granted()
                .names()
                .into_iter()
                .map(str::to_owned)
                .collect(),
            max_labeled_steps: value.max_labeled_steps(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<&WireExecutionCapabilities> for ExecutionCapabilities {
    type Error = ExecutionError;

    /// Reads named capability and frame ceilings off the wire.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::UnknownCapability`] when the grant names a
    /// capability this build does not recognize, and
    /// [`ExecutionError::Malformed`] when the step ceiling is outside
    /// <code>1..=[MAX_LABELED_STEPS]</code>.
    fn try_from(value: &WireExecutionCapabilities) -> Result<Self, Self::Error> {
        let (granted, unknown) =
            CapabilitySet::from_names(value.granted.iter().map(String::as_str));
        if !unknown.is_empty() {
            return Err(ExecutionError::UnknownCapability {
                unknown: unknown.join(", "),
            });
        }
        Self::new(granted, value.max_labeled_steps)
    }
}

impl From<&ExecutionGrant> for WireExecutionGrant {
    fn from(value: &ExecutionGrant) -> Self {
        Self {
            opening: buffa::MessageField::some(WireExecutionLabel::from(value.opening())),
            owner_id: value.owner().as_str().to_owned(),
            last_committed_step: value
                .last_committed_step()
                .map(WireExecutionLabel::from)
                .into(),
            budget_nanos: u64::try_from(value.budget().as_nanos()).unwrap_or(u64::MAX),
            capabilities: buffa::MessageField::some(WireExecutionCapabilities::from(
                &value.capabilities(),
            )),
            claim_attempt_id: value.claim_attempt().as_str().to_owned(),
            audience: value.audience().as_str().to_owned(),
            tenant_id: value.tenant().as_str().to_owned(),
            conversation_id: value.conversation().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<&WireExecutionGrant> for ExecutionGrant {
    type Error = ExecutionError;

    /// Reads the grant off the wire.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::Malformed`] when the capabilities are absent,
    /// the owner is empty, or the budget is outside
    /// <code>1ns..=[MAX_EXECUTION_BUDGET]</code>, and
    /// [`ExecutionError::UnknownCapability`] when the grant names a capability
    /// this build does not recognize.
    fn try_from(value: &WireExecutionGrant) -> Result<Self, Self::Error> {
        let capabilities =
            value
                .capabilities
                .as_option()
                .ok_or_else(|| ExecutionError::Malformed {
                    field: "capabilities".to_owned(),
                    reason: "an attempt with no declared grant would hold whatever it could reach"
                        .to_owned(),
                })?;
        let opening = value
            .opening
            .as_option()
            .ok_or_else(|| ExecutionError::Malformed {
                field: "opening".to_owned(),
                reason: "the grant names the execution, attempt, and fence this turn runs as"
                    .to_owned(),
            })?;
        Self::from_declared(
            ExecutionLabel::try_from(opening)?,
            OwnerId::new(value.owner_id.clone()),
            AttemptId::new(value.claim_attempt_id.clone()),
            ExecutionAudience::new(value.audience.clone())?,
            value
                .last_committed_step
                .as_option()
                .map(ExecutionLabel::try_from)
                .transpose()?,
            Duration::from_nanos(value.budget_nanos),
            ExecutionCapabilities::try_from(capabilities)?,
            TenantId::new(value.tenant_id.clone()),
            ModelConversationId::new(value.conversation_id.clone()),
        )
    }
}

impl From<&ExecutionLabel> for WireExecutionLabel {
    fn from(value: &ExecutionLabel) -> Self {
        Self {
            protocol_version: value.version().get(),
            execution_id: value.identity().execution().as_str().to_owned(),
            attempt_id: value.identity().attempt().as_str().to_owned(),
            fence: value.identity().fence().get(),
            step: buffa::MessageField::some(WireExecutionStep::from(value.step())),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<&WireExecutionLabel> for ExecutionLabel {
    type Error = ExecutionError;

    /// Reads a frame label off the wire.
    ///
    /// The version is carried through rather than checked here, so
    /// [`ExecutionSession::admit_frame`] reports a mismatch as
    /// [`ExecutionError::VersionMismatch`] naming both versions.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::Malformed`] when the step is absent, when
    /// either identity is empty, or when the fence is
    /// [`FencingToken::UNCLAIMED`].
    fn try_from(value: &WireExecutionLabel) -> Result<Self, Self::Error> {
        let step = value
            .step
            .as_option()
            .ok_or_else(|| ExecutionError::Malformed {
                field: "step".to_owned(),
                reason: "an unnumbered frame has no place in the attempt's sequence".to_owned(),
            })?;
        Ok(Self {
            version: ExecutionProtocolVersion::new(value.protocol_version),
            identity: ExecutionIdentity::new(
                ExecutionId::new(value.execution_id.clone()),
                AttemptId::new(value.attempt_id.clone()),
                FencingToken::new(value.fence),
            )?,
            step: ExecutionStep::from(step),
        })
    }
}

// ---------------------------------------------------------------------------
// Process clock and transport mapping
// ---------------------------------------------------------------------------

/// Returns this process's monotonic instant.
///
/// One clock for the whole process, started on the first read, so every session
/// frames its budget against the same origin and a later check sees real
/// elapsed time. A clock built per call would report zero elapsed on every
/// read, and a budget framed against it could never expire.
#[must_use]
pub fn now() -> MonotonicInstant {
    static CLOCK: OnceLock<ProductionClock> = OnceLock::new();
    CLOCK.get_or_init(ProductionClock::new).now()
}

/// Maps a refusal onto the transport status the peer reads.
///
/// Two classes, and the split is what a caller does next. A shape this build
/// cannot read is an argument fault: the same bytes fail again. A refusal that
/// depends on the attempt's state — a lost fence, a step out of sequence, a
/// spent grant — is a precondition fault: the attempt is over, and continuing
/// means claiming again, not resending.
#[must_use]
pub fn connect_error_from_execution(error: &ExecutionError) -> connectrpc::ConnectError {
    let message = error.to_string();
    match error {
        ExecutionError::Malformed { .. }
        | ExecutionError::VersionMismatch { .. }
        | ExecutionError::UnknownCapability { .. }
        | ExecutionError::ForgedStepId { .. }
        | ExecutionError::WrongAudience { .. } => {
            connectrpc::ConnectError::invalid_argument(message)
        }
        ExecutionError::ForeignAttempt { .. }
        | ExecutionError::StaleFence { .. }
        | ExecutionError::DuplicateStep { .. }
        | ExecutionError::OutOfOrderStep { .. }
        | ExecutionError::StepBudgetExhausted { .. } => {
            connectrpc::ConnectError::failed_precondition(message)
        }
    }
}

/// Reads the label a frame carries, refusing one that omits it.
///
/// # Errors
///
/// Returns [`ExecutionError::Malformed`] when the field is absent. An unlabeled
/// frame is one no fence orders, which is the state this protocol exists to
/// remove.
pub fn required_label(
    label: Option<&WireExecutionLabel>,
) -> Result<ExecutionLabel, ExecutionError> {
    let wire = label.ok_or_else(|| ExecutionError::Malformed {
        field: "label".to_owned(),
        reason: "every frame declares the execution, attempt, fence, and step it belongs to"
            .to_owned(),
    })?;
    ExecutionLabel::try_from(wire)
}

/// Reads the grant a turn input carries, refusing one that omits it.
///
/// # Errors
///
/// Returns [`ExecutionError::Malformed`] when the field is absent, and whatever
/// [`ExecutionGrant`]'s conversion reports otherwise.
pub fn required_grant(
    grant: Option<&WireExecutionGrant>,
) -> Result<ExecutionGrant, ExecutionError> {
    let wire = grant.ok_or_else(|| ExecutionError::Malformed {
        field: "execution".to_owned(),
        reason: "a turn that runs under no grant holds no bounded capability and no budget"
            .to_owned(),
    })?;
    ExecutionGrant::try_from(wire)
}

#[cfg(test)]
mod tests;