ferrum_interfaces/vnext/execution/
sequence_checkpoint.rs

1//! Plan-derived checkpoint closure. A layout authorizes byte planning only;
2//! completed-frame evidence and session transfer arbitration are still required
3//! before any state is captured or imported.
4
5use serde::{Deserialize, Serialize};
6
7use super::{
8    canonical_fingerprint, invalid_plan, AllocationLifetime, DynamicResourceDescriptor,
9    DynamicStorageContract, ExecutionPlan, NodeId, PlanHash, ProgramValueId, ResolvedTensorSpec,
10    ResourceId, StateId, StateInitialization, VNextError,
11};
12use crate::vnext::{
13    CheckpointCompletedInputCapture, CheckpointInputDependency, ContractVersion,
14    ProgramCheckpointInputs, ProviderCheckpointContract, ProviderCheckpointStateLayout, ProviderId,
15    StateCheckpointContract,
16};
17
18mod derive;
19mod output_only;
20pub(super) use derive::derive_sequence_checkpoint;
21mod ranges;
22pub use ranges::{
23    SequenceCheckpointBytePlan, SequenceCheckpointCopyRange, SequenceCheckpointResourceRanges,
24};
25mod inputs;
26pub use inputs::{CheckpointCanonicalInput, SequenceCheckpointInputIdentity};
27
28pub const SEQUENCE_CHECKPOINT_LAYOUT_VERSION: ContractVersion = ContractVersion::new(1, 0);
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
31#[serde(rename_all = "snake_case")]
32pub enum SequenceCheckpointUnsupportedReason {
33    InputsUndeclared,
34    InputCoverage,
35    OutputOnlyInputAffectsState {
36        value_id: ProgramValueId,
37        node_id: NodeId,
38        state_id: StateId,
39    },
40    MissingInput {
41        value_id: ProgramValueId,
42    },
43    InvalidInput {
44        value_id: ProgramValueId,
45    },
46    NoSequenceState,
47    StateUndeclared {
48        state_id: StateId,
49    },
50    StateLifetime {
51        state_id: StateId,
52        lifetime: AllocationLifetime,
53    },
54    StateWithoutWriter {
55        state_id: StateId,
56    },
57    ProviderUndeclared {
58        node_id: NodeId,
59        provider_id: ProviderId,
60    },
61    ProviderPersistentWorkspace {
62        node_id: NodeId,
63    },
64    BoundaryIntersection,
65    StateOracleCoverage {
66        node_id: NodeId,
67    },
68    StatePortUndeclared {
69        node_id: NodeId,
70        state_id: StateId,
71    },
72    InvalidStatePort {
73        node_id: NodeId,
74    },
75    StateLayout {
76        state_id: StateId,
77        reason: String,
78    },
79    UncoveredResource {
80        resource_id: ResourceId,
81    },
82}
83
84#[derive(Debug, Clone, Copy)]
85pub enum SequenceCheckpointCapability<'a> {
86    Enabled(&'a SequenceCheckpointLayout),
87    Unsupported(&'a [SequenceCheckpointUnsupportedReason]),
88}
89
90/// Every actual selected operation participates, including stateless operations
91/// that can affect prefix results or legality. An unselected provider cannot
92/// establish this contract for the plan.
93#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
94#[serde(deny_unknown_fields)]
95pub struct SequenceCheckpointProvider {
96    node_id: NodeId,
97    provider_id: ProviderId,
98    operation_fingerprint: String,
99    implementation_fingerprint: String,
100    contract: ProviderCheckpointContract,
101}
102
103#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
104#[serde(deny_unknown_fields)]
105pub struct SequenceCheckpointState {
106    state_id: StateId,
107    value_id: ProgramValueId,
108    writers: Vec<NodeId>,
109    semantics: StateCheckpointContract,
110    tensor: ResolvedTensorSpec,
111    resource_id: ResourceId,
112    offset_bytes: u64,
113    layout: ProviderCheckpointStateLayout,
114    storage: DynamicStorageContract,
115    initialization: StateInitialization,
116    descriptor: DynamicResourceDescriptor,
117}
118
119impl SequenceCheckpointState {
120    pub fn state_id(&self) -> &StateId {
121        &self.state_id
122    }
123    pub fn value_id(&self) -> &ProgramValueId {
124        &self.value_id
125    }
126    pub fn writers(&self) -> &[NodeId] {
127        &self.writers
128    }
129    pub fn semantics(&self) -> StateCheckpointContract {
130        self.semantics
131    }
132    pub fn tensor(&self) -> &ResolvedTensorSpec {
133        &self.tensor
134    }
135    pub fn resource_id(&self) -> &ResourceId {
136        &self.resource_id
137    }
138    pub fn storage(&self) -> &DynamicStorageContract {
139        &self.storage
140    }
141    /// The Sequence initialization cell is identified by this base resource;
142    /// every projection of that cell must be imported before it is committed.
143    pub fn initialization(&self) -> StateInitialization {
144        self.initialization
145    }
146    pub fn layout(&self) -> ProviderCheckpointStateLayout {
147        self.layout
148    }
149}
150
151#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
152#[serde(deny_unknown_fields)]
153pub(super) struct SequenceCheckpointLayoutData {
154    contract_version: ContractVersion,
155    inputs: ProgramCheckpointInputs,
156    input_dependency: CheckpointInputDependency,
157    boundaries: crate::vnext::CheckpointBoundaryConstraint,
158    #[serde(
159        default,
160        skip_serializing_if = "CheckpointCompletedInputCapture::is_unsupported"
161    )]
162    completed_input_capture: CheckpointCompletedInputCapture,
163    providers: Vec<SequenceCheckpointProvider>,
164    states: Vec<SequenceCheckpointState>,
165}
166
167impl SequenceCheckpointLayoutData {
168    pub(super) fn validate_version(&self) -> Result<(), VNextError> {
169        if self.contract_version != SEQUENCE_CHECKPOINT_LAYOUT_VERSION {
170            return Err(invalid_plan(
171                "unsupported sequence checkpoint layout version",
172            ));
173        }
174        Ok(())
175    }
176}
177
178/// Constructed only from a trusted plan build. Public deserialization cannot
179/// create this authority; plan wire data must survive a full semantic rebuild.
180#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
181#[serde(transparent)]
182pub struct SequenceCheckpointLayout {
183    data: SequenceCheckpointLayoutData,
184}
185
186impl SequenceCheckpointLayout {
187    pub fn fingerprint(&self) -> Result<String, VNextError> {
188        canonical_fingerprint(&self.data, "fingerprint sequence checkpoint layout")
189    }
190    pub fn inputs(&self) -> &ProgramCheckpointInputs {
191        &self.data.inputs
192    }
193    pub fn input_dependency(&self) -> CheckpointInputDependency {
194        self.data.input_dependency
195    }
196    /// Aggregate constraint on the actual span ending at a capture boundary.
197    pub fn capture_span_constraint(
198        &self,
199    ) -> super::super::operation::CheckpointTokenSpanConstraint {
200        self.data.boundaries.prefix()
201    }
202    /// Greatest shared boundary reachable from the source's actual retired
203    /// offset, with a legal nonempty suffix for every participant.
204    pub fn shared_prefix_boundary(
205        &self,
206        processed: u64,
207        source_prompt: u64,
208        common_prefix: u64,
209        follower_prompts: &[u64],
210    ) -> Option<u64> {
211        if self.input_dependency() != CheckpointInputDependency::ExactTokenPrefix
212            || follower_prompts.is_empty()
213        {
214            return None;
215        }
216        self.latest_reusable_boundary(processed, source_prompt, common_prefix, follower_prompts)
217    }
218
219    /// Nearest boundary before the end of this input that can serve an exact
220    /// repeat while leaving a legal nonempty suffix to execute for logits.
221    /// This is pure planning, not a capture or capacity reservation.
222    pub fn prompt_tail_boundary(&self, processed: u64, prompt: u64) -> Option<u64> {
223        if self.input_dependency() != CheckpointInputDependency::ExactTokenPrefix {
224            return None;
225        }
226        self.latest_reusable_boundary(processed, prompt, prompt, &[])
227    }
228
229    fn latest_reusable_boundary(
230        &self,
231        processed: u64,
232        source_prompt: u64,
233        common_prefix: u64,
234        follower_prompts: &[u64],
235    ) -> Option<u64> {
236        let prefix = self.data.boundaries.prefix();
237        let suffix = self.data.boundaries.suffix();
238        let prefix_alignment = prefix.alignment().get();
239        let suffix_alignment = suffix.alignment().get();
240        let suffix_residue = source_prompt % suffix_alignment;
241        let mut upper =
242            common_prefix.min(source_prompt.checked_sub(suffix.minimum_tokens().get())?);
243        for &prompt in follower_prompts {
244            // All suffixes share one alignment. Incompatible residues cannot
245            // acquire a common boundary, regardless of common-prefix length.
246            if prompt % suffix_alignment != suffix_residue {
247                return None;
248            }
249            upper = upper.min(prompt.checked_sub(suffix.minimum_tokens().get())?);
250        }
251        let lower = processed.checked_add(prefix.minimum_tokens().get())?;
252        if lower > upper {
253            return None;
254        }
255        let (mut divisor, mut remainder) = (prefix_alignment, suffix_alignment);
256        while remainder != 0 {
257            (divisor, remainder) = (remainder, divisor % remainder);
258        }
259        if processed % divisor != source_prompt % divisor {
260            return None;
261        }
262        // Search only the sparser of the two alignment lattices. This skips
263        // invalid token positions without multiplying alignments or overflowing
264        // an LCM. The aggregate contract remains the final authority.
265        let (step, residue) = if prefix_alignment >= suffix_alignment {
266            (prefix_alignment, processed % prefix_alignment)
267        } else {
268            (suffix_alignment, suffix_residue)
269        };
270        let remainder = upper % step;
271        let adjustment = if remainder >= residue {
272            remainder - residue
273        } else {
274            step - (residue - remainder)
275        };
276        let mut boundary = upper.checked_sub(adjustment)?;
277        while boundary >= lower {
278            if self.permits_capture_from(processed, boundary, source_prompt) {
279                return Some(boundary);
280            }
281            boundary = boundary.checked_sub(step)?;
282        }
283        None
284    }
285    pub fn states(&self) -> &[SequenceCheckpointState] {
286        &self.data.states
287    }
288    pub fn providers(&self) -> &[SequenceCheckpointProvider] {
289        &self.data.providers
290    }
291    pub fn permits_capture_from(&self, processed: u64, boundary: u64, prompt: u64) -> bool {
292        if boundary == prompt {
293            return self.data.completed_input_capture == CheckpointCompletedInputCapture::Supported
294                && boundary
295                    .checked_sub(processed)
296                    .is_some_and(|span| self.data.boundaries.prefix().permits(span));
297        }
298        self.data
299            .boundaries
300            .permits_from(processed, boundary, prompt)
301    }
302    pub fn completed_input_capture(&self) -> CheckpointCompletedInputCapture {
303        self.data.completed_input_capture
304    }
305    pub fn permits_suffix(&self, boundary: u64, prompt: u64) -> bool {
306        boundary > 0
307            && prompt
308                .checked_sub(boundary)
309                .is_some_and(|suffix| self.data.boundaries.suffix().permits(suffix))
310    }
311}
312
313impl SequenceCheckpointProvider {
314    pub fn node_id(&self) -> &NodeId {
315        &self.node_id
316    }
317    pub fn provider_id(&self) -> &ProviderId {
318        &self.provider_id
319    }
320    pub fn contract(&self) -> &ProviderCheckpointContract {
321        &self.contract
322    }
323}
324
325impl ExecutionPlan {
326    pub fn sequence_checkpoint_capability(&self) -> SequenceCheckpointCapability<'_> {
327        match &self.payload.sequence_checkpoint_layout {
328            Some(layout) => SequenceCheckpointCapability::Enabled(layout),
329            None => SequenceCheckpointCapability::Unsupported(&self.checkpoint_unsupported_reasons),
330        }
331    }
332
333    /// Pure, checked allocation/copy-range planning. N is not a proof that a
334    /// source has actually completed that boundary; transfer requires its own
335    /// completed-frame authority and checks the source's actual backing.
336    pub fn checkpoint_byte_plan(
337        &self,
338        boundary: u64,
339    ) -> Result<SequenceCheckpointBytePlan, VNextError> {
340        self.payload
341            .sequence_checkpoint_layout
342            .as_ref()
343            .ok_or_else(|| invalid_plan("plan does not support sequence checkpoints"))?
344            .byte_plan(self.plan_hash.clone(), boundary)
345    }
346}
ferrum_interfaces/vnext/execution/sequence_checkpoint.rs

ferrum_interfaces/vnext/execution/
sequence_checkpoint.rs
