1use crate::{
8 artifact::{
9 plan_model_preparation, ArtifactFormat, ArtifactInspection, ModelConfigurationResolver,
10 PreparationPolicy,
11 },
12 backend::{BackendProvider, ModelLoadingBackend, ModelRuntime, SessionCapabilities},
13 execution::{
14 DevicePlan, DraftingPlan, ExecutionPlan, ExpertCachePlan, ResidencyPlan,
15 DEFAULT_MAX_CACHED_SHARDS,
16 },
17 speculative::SpeculativeDraft,
18};
19use serde::{Deserialize, Serialize};
20use std::{
21 path::PathBuf,
22 sync::atomic::{AtomicU64, Ordering},
23 time::Duration,
24};
25
26static NEXT_EXECUTION_PLAN_TARGET_ID: AtomicU64 = AtomicU64::new(1);
27
28fn next_execution_plan_target_id() -> Result<u64, AutomaticPlanningError> {
29 NEXT_EXECUTION_PLAN_TARGET_ID
30 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
31 current.checked_add(1)
32 })
33 .map_err(|_| {
34 AutomaticPlanningError::Invalid(
35 "execution-plan target identity space is exhausted".into(),
36 )
37 })
38}
39
40pub const AUTOMATIC_SCHEMA_VERSION: u32 = 6;
42
43#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
45#[serde(rename_all = "snake_case")]
46pub enum ObservationKind {
47 Exact,
49 Conservative,
51 Observational,
53 Estimated,
55}
56
57#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
59#[serde(tag = "status", rename_all = "snake_case")]
60pub enum Observed<T> {
61 Available {
63 value: T,
65 kind: ObservationKind,
67 source: String,
69 },
70 Unsupported {
72 reason: String,
74 },
75 Unavailable {
77 reason: String,
79 },
80}
81
82impl<T> Observed<T> {
83 pub fn exact(value: T, source: impl Into<String>) -> Self {
85 Self::Available {
86 value,
87 kind: ObservationKind::Exact,
88 source: source.into(),
89 }
90 }
91
92 pub fn unavailable(reason: impl Into<String>) -> Self {
94 Self::Unavailable {
95 reason: reason.into(),
96 }
97 }
98
99 pub fn unsupported(reason: impl Into<String>) -> Self {
101 Self::Unsupported {
102 reason: reason.into(),
103 }
104 }
105
106 pub const fn value(&self) -> Option<&T> {
108 match self {
109 Self::Available { value, .. } => Some(value),
110 Self::Unsupported { .. } | Self::Unavailable { .. } => None,
111 }
112 }
113}
114
115fn unobserved_embedded_draft_layers() -> Observed<usize> {
116 Observed::unavailable("embedded drafting requires normalized architecture inspection")
117}
118
119#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
121pub struct ModelResourceProfile {
122 pub schema_version: u32,
124 pub path: PathBuf,
126 pub artifact_format: ArtifactFormat,
128 #[serde(skip_serializing_if = "Option::is_none")]
130 pub model_family: Option<String>,
131 #[serde(skip_serializing_if = "Option::is_none")]
133 pub architecture: Option<String>,
134 #[serde(skip_serializing_if = "Option::is_none")]
136 pub tensor_count: Option<usize>,
137 #[serde(skip_serializing_if = "Option::is_none")]
139 pub checkpoint_shards: Option<usize>,
140 #[serde(default = "unobserved_embedded_draft_layers")]
142 pub embedded_draft_layers: Observed<usize>,
143 #[serde(default = "unobserved_embedded_draft_layers")]
145 pub embedded_draft_capacity: Observed<usize>,
146 pub stored_tensor_bytes: Observed<u64>,
148 pub largest_stored_tensor_bytes: Observed<u64>,
150 pub materialized_parameter_bytes: Observed<u64>,
152 pub pinned_parameter_bytes: Observed<u64>,
154 pub largest_execution_group_bytes: Observed<u64>,
156 pub largest_adjacent_execution_groups_bytes: Observed<u64>,
158 pub expert_parameter_bytes: Observed<u64>,
160}
161
162impl ModelResourceProfile {
163 pub fn unmeasured(path: PathBuf, artifact_format: ArtifactFormat) -> Self {
165 let unavailable = || {
166 Observed::unavailable("resource value requires a validated checkpoint parameter plan")
167 };
168 Self {
169 schema_version: AUTOMATIC_SCHEMA_VERSION,
170 path,
171 artifact_format,
172 model_family: None,
173 architecture: None,
174 tensor_count: None,
175 checkpoint_shards: None,
176 embedded_draft_layers: unobserved_embedded_draft_layers(),
177 embedded_draft_capacity: unobserved_embedded_draft_layers(),
178 stored_tensor_bytes: Observed::unavailable(
179 "checkpoint tensor catalog was not established",
180 ),
181 largest_stored_tensor_bytes: Observed::unavailable(
182 "checkpoint tensor catalog was not established",
183 ),
184 materialized_parameter_bytes: unavailable(),
185 pinned_parameter_bytes: unavailable(),
186 largest_execution_group_bytes: unavailable(),
187 largest_adjacent_execution_groups_bytes: unavailable(),
188 expert_parameter_bytes: unavailable(),
189 }
190 }
191}
192
193#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
195pub struct HardwareDeviceProfile {
196 pub id: String,
198 pub family: String,
200 pub index: usize,
202 pub total_memory_bytes: Observed<u64>,
204 pub available_memory_bytes: Observed<u64>,
206}
207
208#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
210pub struct HardwareBackendProfile {
211 pub backend: crate::execution::BackendId,
213 pub available: bool,
215 #[serde(skip_serializing_if = "Option::is_none")]
217 pub detail: Option<String>,
218 pub devices: Vec<HardwareDeviceProfile>,
220}
221
222#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
224pub struct HardwareProfile {
225 pub schema_version: u32,
227 pub operating_system: String,
229 pub architecture: String,
231 pub logical_cpu_count: Observed<u64>,
233 pub physical_memory_bytes: Observed<u64>,
235 pub available_memory_bytes: Observed<u64>,
237 pub physical_memory_semantics: HardwareMemorySemantics,
239 pub backends: Vec<HardwareBackendProfile>,
241}
242
243impl HardwareProfile {
244 pub fn observe_host(
250 physical_memory_bytes: Observed<u64>,
251 available_memory_bytes: Observed<u64>,
252 physical_memory_semantics: HardwareMemorySemantics,
253 backends: Vec<HardwareBackendProfile>,
254 ) -> Self {
255 let logical_cpu_count = std::thread::available_parallelism().map_or_else(
256 |error| Observed::unavailable(error.to_string()),
257 |count| Observed::exact(count.get() as u64, "std::thread::available_parallelism"),
258 );
259 Self {
260 schema_version: AUTOMATIC_SCHEMA_VERSION,
261 operating_system: std::env::consts::OS.into(),
262 architecture: std::env::consts::ARCH.into(),
263 logical_cpu_count,
264 physical_memory_bytes,
265 available_memory_bytes,
266 physical_memory_semantics,
267 backends,
268 }
269 }
270}
271
272#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
274#[serde(rename_all = "snake_case")]
275pub enum HardwareMemorySemantics {
276 Unified,
278 SeparateTiers,
280 Unknown,
282}
283
284impl From<crate::capability::PhysicalMemorySemantics> for HardwareMemorySemantics {
285 fn from(value: crate::capability::PhysicalMemorySemantics) -> Self {
286 match value {
287 crate::capability::PhysicalMemorySemantics::Unified => Self::Unified,
288 crate::capability::PhysicalMemorySemantics::SeparateTiers => Self::SeparateTiers,
289 crate::capability::PhysicalMemorySemantics::Unknown => Self::Unknown,
290 }
291 }
292}
293
294#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
296#[serde(rename_all = "snake_case")]
297pub enum PlanExplanationLevel {
298 Decision,
300 Warning,
302 Rejection,
304}
305
306#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
308pub struct PlanExplanationEntry {
309 pub level: PlanExplanationLevel,
311 pub code: String,
313 pub detail: String,
315}
316
317#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
319pub struct PlanExplanation {
320 pub summary: String,
322 pub entries: Vec<PlanExplanationEntry>,
324}
325
326#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
328pub struct ExecutionPlanReport {
329 pub schema_version: u32,
331 pub hardware: HardwareProfile,
333 pub resources: ModelResourceProfile,
335 pub plan: ExecutionPlan,
337 pub explanation: PlanExplanation,
339}
340
341#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
343#[serde(default)]
344#[non_exhaustive]
345pub struct AutomaticPlannerPolicy {
346 pub device_memory_fallback_bytes: u64,
348 pub host_memory_fallback_bytes: u64,
350 pub memory_headroom_percent: u8,
352 pub expert_cache_share_percent: u8,
354 pub device_layer_window: usize,
356 pub max_cached_shards: usize,
358 pub embedded_mtp_draft_tokens: usize,
360 pub minimum_feedback_tokens: usize,
362}
363
364impl Default for AutomaticPlannerPolicy {
365 fn default() -> Self {
366 Self {
367 device_memory_fallback_bytes: 4 << 30,
368 host_memory_fallback_bytes: 16 << 30,
369 memory_headroom_percent: 30,
370 expert_cache_share_percent: 40,
371 device_layer_window: 1,
372 max_cached_shards: DEFAULT_MAX_CACHED_SHARDS,
373 embedded_mtp_draft_tokens: 3,
374 minimum_feedback_tokens: 1,
375 }
376 }
377}
378
379#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
381pub struct TimingTelemetry {
382 pub load_seconds: f64,
384 pub generation_seconds: f64,
386 #[serde(skip_serializing_if = "Option::is_none")]
388 pub time_to_first_token_seconds: Option<f64>,
389 pub total_seconds: f64,
391 pub token_rate: f64,
393 #[serde(skip_serializing_if = "Option::is_none")]
395 pub decode_token_rate: Option<f64>,
396}
397
398impl TimingTelemetry {
399 pub fn new(
401 load: Duration,
402 generation: Duration,
403 time_to_first_token: Option<Duration>,
404 generated_tokens: usize,
405 total: Duration,
406 ) -> Self {
407 fn rate(tokens: usize, elapsed: Duration) -> f64 {
408 if elapsed.is_zero() {
409 0.0
410 } else {
411 tokens as f64 / elapsed.as_secs_f64()
412 }
413 }
414 Self {
415 load_seconds: load.as_secs_f64(),
416 generation_seconds: generation.as_secs_f64(),
417 time_to_first_token_seconds: time_to_first_token.map(|value| value.as_secs_f64()),
418 total_seconds: total.as_secs_f64(),
419 token_rate: rate(generated_tokens, generation),
420 decode_token_rate: time_to_first_token.map(|first| {
421 rate(
422 generated_tokens.saturating_sub(1),
423 generation.saturating_sub(first),
424 )
425 }),
426 }
427 }
428}
429
430#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
432pub struct AllocatorTelemetry {
433 pub peak_bytes: u64,
435 pub active_bytes: u64,
437 pub cache_bytes: u64,
439}
440
441#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
443pub struct ResidencyTelemetry {
444 pub planned_disk_bytes: u64,
446 pub planned_host_bytes: u64,
448 pub planned_device_bytes: u64,
450 pub current_host_bytes: u64,
452 pub current_device_bytes: u64,
454 pub peak_host_bytes: u64,
456 pub peak_device_bytes: u64,
458 pub transfers: Vec<TransferTelemetry>,
460}
461
462#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
464pub struct TransferTelemetry {
465 pub direction: String,
467 pub count: u64,
469 pub bytes: u64,
471 pub seconds: DurationSeconds,
473}
474
475#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
477#[serde(transparent)]
478pub struct DurationSeconds(pub f64);
479
480impl PartialEq for DurationSeconds {
481 fn eq(&self, other: &Self) -> bool {
482 self.0.to_bits() == other.0.to_bits()
483 }
484}
485impl Eq for DurationSeconds {}
486
487#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
489pub struct ExpertCacheTelemetry {
490 pub owned_experts: usize,
492 pub owned_bytes: u64,
494 pub host_resident_experts: usize,
496 pub device_resident_experts: usize,
498 pub host_resident_bytes: u64,
500 pub device_resident_bytes: u64,
502 pub peak_host_resident_bytes: u64,
504 pub peak_device_resident_bytes: u64,
506}
507
508#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
510pub struct SpeculativeDecodingTelemetry {
511 pub execution_topology: String,
513 pub target_tokens: usize,
515 pub draft_tokens: usize,
517 pub accepted_tokens: usize,
519 pub accept_rate: f64,
521 pub rounds: usize,
523 pub accept_lens: Vec<usize>,
525 pub emitted_tokens: usize,
527 pub optimistic_draft_tokens: usize,
529 pub reused_optimistic_tokens: usize,
531 pub discarded_optimistic_tokens: usize,
533 pub adaptive_lookahead_disabled: bool,
535 pub optimistic_draft_seconds: f64,
537 pub verification_in_flight_seconds: f64,
539}
540
541pub fn speculative_decoding_telemetry(
543 stats: &crate::speculative::SpeculativeStats,
544) -> SpeculativeDecodingTelemetry {
545 SpeculativeDecodingTelemetry {
546 execution_topology: stats.execution_topology().to_string(),
547 target_tokens: stats.target_tokens(),
548 draft_tokens: stats.draft_tokens(),
549 accepted_tokens: stats.accepted_tokens(),
550 accept_rate: stats.accept_rate(),
551 rounds: stats.rounds(),
552 accept_lens: stats.accept_lens().to_vec(),
553 emitted_tokens: stats.emitted_tokens(),
554 optimistic_draft_tokens: stats.optimistic_draft_tokens(),
555 reused_optimistic_tokens: stats.reused_optimistic_tokens(),
556 discarded_optimistic_tokens: stats.discarded_optimistic_tokens(),
557 adaptive_lookahead_disabled: stats.adaptive_lookahead_disabled(),
558 optimistic_draft_seconds: stats.optimistic_draft_time().as_secs_f64(),
559 verification_in_flight_seconds: stats.verification_in_flight_time().as_secs_f64(),
560 }
561}
562
563#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
565pub struct ExecutionTelemetry {
566 pub schema_version: u32,
568 pub effective_model_type: String,
570 #[serde(skip_serializing_if = "Option::is_none")]
572 pub plan: Option<ExecutionPlan>,
573 #[serde(skip_serializing_if = "Option::is_none")]
575 pub plan_explanation: Option<PlanExplanation>,
576 #[serde(skip_serializing_if = "Option::is_none")]
578 pub hardware: Option<HardwareProfile>,
579 #[serde(skip_serializing_if = "Option::is_none")]
581 pub resources: Option<ModelResourceProfile>,
582 pub prompt_tokens: usize,
584 pub generated_tokens: usize,
586 pub stop_reason: String,
588 pub timing: TimingTelemetry,
590 #[serde(skip_serializing_if = "Option::is_none")]
592 pub allocator: Option<AllocatorTelemetry>,
593 #[serde(skip_serializing_if = "Option::is_none")]
595 pub residency: Option<ResidencyTelemetry>,
596 #[serde(skip_serializing_if = "Option::is_none")]
598 pub expert_cache: Option<ExpertCacheTelemetry>,
599 #[serde(skip_serializing_if = "Option::is_none")]
601 pub speculative: Option<SpeculativeDecodingTelemetry>,
602}
603
604#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
606#[non_exhaustive]
607pub struct AutomaticPlanRequest {
608 pub schema_version: u32,
610 pub model_path: PathBuf,
612 pub device: DevicePlan,
614 #[serde(default, skip_serializing_if = "Vec::is_empty")]
616 pub prior_telemetry: Vec<ExecutionTelemetry>,
617}
618
619impl AutomaticPlanRequest {
620 pub fn new(model_path: impl Into<PathBuf>, device: DevicePlan) -> Self {
622 Self {
623 schema_version: AUTOMATIC_SCHEMA_VERSION,
624 model_path: model_path.into(),
625 device,
626 prior_telemetry: Vec::new(),
627 }
628 }
629
630 pub fn with_prior_telemetry(
632 mut self,
633 telemetry: impl IntoIterator<Item = ExecutionTelemetry>,
634 ) -> Self {
635 self.prior_telemetry.extend(telemetry);
636 self
637 }
638}
639
640#[derive(Debug, Clone, Eq, PartialEq)]
642pub struct CandidateAdmission {
643 pub supported: bool,
645 pub rejection: Option<String>,
647}
648
649#[derive(Debug, Clone, Copy, Eq, PartialEq)]
651pub struct BoundedResidencyRequirement {
652 pub static_bytes: u64,
654 pub window_bytes: u64,
656 pub required_bytes: u64,
658 pub depth: usize,
660}
661
662pub trait AutomaticPlanningBackend {
664 type Inspection;
666 fn backend_id(&self) -> crate::execution::BackendId;
668 fn discover_hardware(&self) -> Result<HardwareProfile, AutomaticPlanningError>;
670 fn inspect_resources(
672 &self,
673 model_path: &std::path::Path,
674 ) -> Result<(ModelResourceProfile, Self::Inspection), AutomaticPlanningError>;
675 fn admit_candidate(
677 &self,
678 inspection: &Self::Inspection,
679 plan: &ExecutionPlan,
680 ) -> Result<CandidateAdmission, AutomaticPlanningError>;
681 fn bounded_residency_requirement(
683 &self,
684 inspection: &Self::Inspection,
685 plan: &ExecutionPlan,
686 ) -> Result<BoundedResidencyRequirement, AutomaticPlanningError>;
687}
688
689pub struct RetainedAutomaticPlan<I> {
691 report: ExecutionPlanReport,
692 inspection: I,
693}
694
695impl<I> RetainedAutomaticPlan<I> {
696 pub const fn report(&self) -> &ExecutionPlanReport {
698 &self.report
699 }
700
701 pub fn into_parts(self) -> (ExecutionPlanReport, I) {
703 (self.report, self.inspection)
704 }
705}
706
707pub struct ExecutionPlanTargetSelection<B: ModelLoadingBackend> {
709 policy: PreparationPolicy,
710 selected: B::SelectedPreparation,
711 capabilities: SessionCapabilities,
712}
713
714impl<B: ModelLoadingBackend> ExecutionPlanTargetSelection<B> {
715 pub fn new(
717 policy: PreparationPolicy,
718 selected: B::SelectedPreparation,
719 capabilities: SessionCapabilities,
720 ) -> Self {
721 Self {
722 policy,
723 selected,
724 capabilities,
725 }
726 }
727}
728
729pub struct SelectedExecutionPlanTarget<B: ModelLoadingBackend> {
731 execution_plan: ExecutionPlan,
732 preparation: crate::backend::SelectedModelPreparation<B>,
733 target_id: u64,
734}
735
736impl<B: ModelLoadingBackend> SelectedExecutionPlanTarget<B> {
737 fn into_preparation(self) -> crate::backend::SelectedModelPreparation<B> {
738 self.preparation
739 }
740
741 pub fn inspection(
743 &self,
744 ) -> &ArtifactInspection<<B::ConfigurationResolver as ModelConfigurationResolver>::ArtifactPlan>
745 {
746 self.preparation.plan().inspection()
747 }
748
749 pub const fn execution_plan(&self) -> &ExecutionPlan {
751 &self.execution_plan
752 }
753}
754
755pub struct ExecutionPlanTarget<B: ModelLoadingBackend> {
761 backend: B,
762 selected: SelectedExecutionPlanTarget<B>,
763}
764
765pub type PreparedExecutionPlanTarget<B> = ModelRuntime<B>;
767
768pub type ExecutionPlanTargetLoadError<B> =
770 crate::backend::ModelLoadError<<B as BackendProvider>::Error>;
771
772impl<B: ModelLoadingBackend> ExecutionPlanTarget<B> {
773 pub fn new(backend: B, selected: SelectedExecutionPlanTarget<B>) -> Self {
779 Self { backend, selected }
780 }
781
782 pub const fn backend(&self) -> &B {
784 &self.backend
785 }
786
787 pub fn into_runtime(
789 self,
790 ) -> Result<PreparedExecutionPlanTarget<B>, ExecutionPlanTargetLoadError<B>> {
791 let target_id = self.selected.target_id;
792 let preparation = self.selected.into_preparation();
793 let prepared = crate::backend::prepare_selected_model(&self.backend, preparation)?;
794 ModelRuntime::from_prepared_execution_plan_target(self.backend, prepared, target_id)
795 .map_err(crate::backend::ModelLoadError::Backend)
796 }
797}
798
799#[derive(Debug, Clone, Copy, Eq, PartialEq)]
805pub struct TokenizerCompatibilityProof {
806 fingerprint: [u8; 32],
807}
808
809impl TokenizerCompatibilityProof {
810 pub fn prove(
812 target_fingerprint: [u8; 32],
813 assistant_fingerprint: [u8; 32],
814 ) -> Result<Self, TokenizerCompatibilityError> {
815 if target_fingerprint != assistant_fingerprint {
816 return Err(TokenizerCompatibilityError);
817 }
818 Ok(Self {
819 fingerprint: target_fingerprint,
820 })
821 }
822
823 pub const fn fingerprint(self) -> [u8; 32] {
825 self.fingerprint
826 }
827
828 pub fn validate_target(
830 self,
831 target_fingerprint: [u8; 32],
832 ) -> Result<(), TokenizerCompatibilityError> {
833 if self.fingerprint != target_fingerprint {
834 return Err(TokenizerCompatibilityError);
835 }
836 Ok(())
837 }
838}
839
840#[derive(Debug, Clone, Copy, Eq, PartialEq, thiserror::Error)]
842#[error("assistant token-id vocabulary mapping does not match the target")]
843pub struct TokenizerCompatibilityError;
844
845#[derive(Debug, Clone, Eq, PartialEq)]
847pub struct ExternalDraftArtifact<P> {
848 pub preparation: P,
850 pub tokenizer_compatibility: TokenizerCompatibilityProof,
852}
853
854pub struct SelectedExecutionPlanDrafting<P> {
861 execution_plan: ExecutionPlan,
862 target_id: u64,
863 external_artifact: Option<ExternalDraftArtifact<P>>,
864}
865
866impl<P> std::fmt::Debug for SelectedExecutionPlanDrafting<P> {
867 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
868 formatter
869 .debug_struct("SelectedExecutionPlanDrafting")
870 .field("execution_plan", &self.execution_plan)
871 .field("has_external_artifact", &self.external_artifact.is_some())
872 .finish()
873 }
874}
875
876impl<P> SelectedExecutionPlanDrafting<P> {
877 pub fn into_external_artifact<B: BackendProvider>(
883 self,
884 plan: &ExecutionPlan,
885 target: &ModelRuntime<B>,
886 ) -> Result<Option<ExternalDraftArtifact<P>>, AutomaticPlanningError> {
887 if self.execution_plan != *plan {
888 return Err(AutomaticPlanningError::Invalid(
889 "selected drafting was established for a different execution plan".into(),
890 ));
891 }
892 if target.execution_plan_target_id() != Some(self.target_id) {
893 return Err(AutomaticPlanningError::Invalid(
894 "selected drafting was established for a different realized target".into(),
895 ));
896 }
897 Ok(self.external_artifact)
898 }
899}
900
901pub enum RealizedDrafting<D> {
903 Disabled,
905 Embedded,
907 External(D),
909}
910
911impl<D> RealizedDrafting<D> {
912 pub fn as_speculative_draft(&mut self) -> Option<SpeculativeDraft<'_, D>> {
914 match self {
915 Self::Disabled => None,
916 Self::Embedded => Some(SpeculativeDraft::Embedded),
917 Self::External(drafter) => Some(SpeculativeDraft::External(drafter)),
918 }
919 }
920
921 pub const fn is_external(&self) -> bool {
923 matches!(self, Self::External(_))
924 }
925}
926
927pub trait ExecutionPlanBackendFactory: AutomaticPlanningBackend {
934 type Backend: ModelLoadingBackend;
936 type DrafterPreparation;
938 type SelectedDrafterPreparation;
940 type Drafter;
942
943 fn select_target(
945 &self,
946 inspection: &ArtifactInspection<
947 <<Self::Backend as ModelLoadingBackend>::ConfigurationResolver as ModelConfigurationResolver>::ArtifactPlan,
948 >,
949 plan: &ExecutionPlan,
950 ) -> Result<ExecutionPlanTargetSelection<Self::Backend>, AutomaticPlanningError>;
951
952 fn select_drafting(
957 &self,
958 plan: &ExecutionPlan,
959 target: &SelectedExecutionPlanTarget<Self::Backend>,
960 external_artifact: Option<ExternalDraftArtifact<Self::DrafterPreparation>>,
961 ) -> Result<
962 Option<ExternalDraftArtifact<Self::SelectedDrafterPreparation>>,
963 AutomaticPlanningError,
964 >;
965
966 fn realize_target(
971 &self,
972 selected: SelectedExecutionPlanTarget<Self::Backend>,
973 ) -> Result<ExecutionPlanTarget<Self::Backend>, AutomaticPlanningError>;
974
975 fn realize_drafting(
983 &self,
984 plan: &ExecutionPlan,
985 target: &ModelRuntime<Self::Backend>,
986 selected: SelectedExecutionPlanDrafting<Self::SelectedDrafterPreparation>,
987 ) -> Result<RealizedDrafting<Self::Drafter>, AutomaticPlanningError>;
988}
989
990pub fn select_execution_plan_target<F: ExecutionPlanBackendFactory>(
992 factory: &F,
993 plan: &ExecutionPlan,
994 inspection: ArtifactInspection<
995 <<F::Backend as ModelLoadingBackend>::ConfigurationResolver as ModelConfigurationResolver>::ArtifactPlan,
996 >,
997) -> Result<SelectedExecutionPlanTarget<F::Backend>, AutomaticPlanningError> {
998 let expected_backend = factory.backend_id();
999 if plan.device.backend != expected_backend {
1000 return Err(AutomaticPlanningError::Invalid(format!(
1001 "execution plan selects backend {} but factory owns {}",
1002 plan.device.backend, expected_backend
1003 )));
1004 }
1005 plan.validate_structure()
1006 .map_err(|error| AutomaticPlanningError::Invalid(error.to_string()))?;
1007
1008 let selection = factory.select_target(&inspection, plan)?;
1009 selection
1010 .policy
1011 .validate_session_capabilities(&selection.capabilities)
1012 .map_err(|error| AutomaticPlanningError::Invalid(error.to_string()))?;
1013 let preparation = plan_model_preparation(inspection, selection.policy, selection.capabilities)
1014 .map_err(|error| AutomaticPlanningError::Invalid(error.to_string()))?;
1015 Ok(SelectedExecutionPlanTarget {
1016 execution_plan: plan.clone(),
1017 preparation: crate::backend::SelectedModelPreparation::new(preparation, selection.selected),
1018 target_id: next_execution_plan_target_id()?,
1019 })
1020}
1021
1022pub fn realize_execution_plan_target<F: ExecutionPlanBackendFactory>(
1024 factory: &F,
1025 plan: &ExecutionPlan,
1026 selected: SelectedExecutionPlanTarget<F::Backend>,
1027) -> Result<ExecutionPlanTarget<F::Backend>, AutomaticPlanningError> {
1028 let expected_backend = factory.backend_id();
1029 if plan.device.backend != expected_backend {
1030 return Err(AutomaticPlanningError::Invalid(format!(
1031 "execution plan selects backend {} but factory owns {}",
1032 plan.device.backend, expected_backend
1033 )));
1034 }
1035 plan.validate_structure()
1036 .map_err(|error| AutomaticPlanningError::Invalid(error.to_string()))?;
1037 if selected.execution_plan != *plan {
1038 return Err(AutomaticPlanningError::Invalid(
1039 "selected target was established for a different execution plan".into(),
1040 ));
1041 }
1042 let realization = factory.realize_target(selected)?;
1043 let descriptor = realization.backend().descriptor();
1044 if descriptor.name() != expected_backend.as_str() {
1045 return Err(AutomaticPlanningError::Invalid(format!(
1046 "factory identity {} does not match realized backend {}",
1047 expected_backend,
1048 descriptor.name()
1049 )));
1050 }
1051 let devices =
1052 realization
1053 .backend()
1054 .devices()
1055 .map_err(|error| AutomaticPlanningError::Backend {
1056 operation: "realize_execution_plan_devices",
1057 message: error.to_string(),
1058 })?;
1059 let capabilities = devices
1060 .iter()
1061 .find_map(|(device, capabilities)| {
1062 (device.id() == plan.device.device).then_some(capabilities)
1063 })
1064 .ok_or_else(|| {
1065 AutomaticPlanningError::Invalid(format!(
1066 "realized backend {} does not expose selected device {}",
1067 expected_backend, plan.device.device
1068 ))
1069 })?;
1070 plan.validate_device_capabilities(capabilities)
1071 .map_err(|error| AutomaticPlanningError::Invalid(error.to_string()))?;
1072 Ok(realization)
1073}
1074
1075pub fn realize_execution_plan_drafting<F: ExecutionPlanBackendFactory>(
1077 factory: &F,
1078 plan: &ExecutionPlan,
1079 target: &ModelRuntime<F::Backend>,
1080 selected: SelectedExecutionPlanDrafting<F::SelectedDrafterPreparation>,
1081) -> Result<RealizedDrafting<F::Drafter>, AutomaticPlanningError> {
1082 if selected.execution_plan != *plan {
1083 return Err(AutomaticPlanningError::Invalid(
1084 "selected drafting was established for a different execution plan".into(),
1085 ));
1086 }
1087 if target.execution_plan_target_id() != Some(selected.target_id) {
1088 return Err(AutomaticPlanningError::Invalid(
1089 "selected drafting was established for a different realized target".into(),
1090 ));
1091 }
1092 match (&plan.drafting, selected.external_artifact.as_ref()) {
1093 (DraftingPlan::External { .. }, None) => {
1094 return Err(AutomaticPlanningError::Invalid(
1095 "external drafting requires proven tokenizer compatibility".into(),
1096 ));
1097 }
1098 (DraftingPlan::Disabled | DraftingPlan::Embedded { .. }, Some(_)) => {
1099 return Err(AutomaticPlanningError::Invalid(
1100 "tokenizer compatibility was supplied for a plan without an external assistant"
1101 .into(),
1102 ));
1103 }
1104 _ => {}
1105 }
1106 let drafting = factory.realize_drafting(plan, target, selected)?;
1107 let matches_plan = matches!(
1108 (&plan.drafting, &drafting),
1109 (DraftingPlan::Disabled, RealizedDrafting::Disabled)
1110 | (DraftingPlan::Embedded { .. }, RealizedDrafting::Embedded)
1111 | (DraftingPlan::External { .. }, RealizedDrafting::External(_))
1112 );
1113 if !matches_plan {
1114 return Err(AutomaticPlanningError::Invalid(
1115 "backend factory realized a drafting mode different from the execution plan".into(),
1116 ));
1117 }
1118 Ok(drafting)
1119}
1120
1121pub fn select_execution_plan_drafting<F: ExecutionPlanBackendFactory>(
1123 factory: &F,
1124 plan: &ExecutionPlan,
1125 target: &SelectedExecutionPlanTarget<F::Backend>,
1126 external_artifact: Option<ExternalDraftArtifact<F::DrafterPreparation>>,
1127) -> Result<SelectedExecutionPlanDrafting<F::SelectedDrafterPreparation>, AutomaticPlanningError> {
1128 if target.execution_plan != *plan {
1129 return Err(AutomaticPlanningError::Invalid(
1130 "selected target was established for a different execution plan".into(),
1131 ));
1132 }
1133 match (&plan.drafting, external_artifact.as_ref()) {
1134 (DraftingPlan::External { .. }, None) => {
1135 return Err(AutomaticPlanningError::Invalid(
1136 "external drafting requires proven tokenizer compatibility".into(),
1137 ));
1138 }
1139 (DraftingPlan::Disabled | DraftingPlan::Embedded { .. }, Some(_)) => {
1140 return Err(AutomaticPlanningError::Invalid(
1141 "tokenizer compatibility was supplied for a plan without an external assistant"
1142 .into(),
1143 ));
1144 }
1145 _ => {}
1146 }
1147 let external_artifact = factory.select_drafting(plan, target, external_artifact)?;
1148 Ok(SelectedExecutionPlanDrafting {
1149 execution_plan: plan.clone(),
1150 target_id: target.target_id,
1151 external_artifact,
1152 })
1153}
1154
1155#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
1157pub enum AutomaticPlanningError {
1158 #[error("automatic planning error: {0}")]
1160 Invalid(String),
1161 #[error("automatic planning backend failed during {operation}: {message}")]
1163 Backend {
1164 operation: &'static str,
1166 message: String,
1168 },
1169}
1170
1171#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, Default)]
1173pub struct AutomaticPlanner {
1174 policy: AutomaticPlannerPolicy,
1175}
1176
1177impl AutomaticPlanner {
1178 pub fn new(policy: AutomaticPlannerPolicy) -> Self {
1180 Self { policy }
1181 }
1182
1183 pub fn policy(&self) -> &AutomaticPlannerPolicy {
1185 &self.policy
1186 }
1187
1188 pub fn plan<B: AutomaticPlanningBackend>(
1190 &self,
1191 backend: &B,
1192 request: &AutomaticPlanRequest,
1193 ) -> Result<ExecutionPlanReport, AutomaticPlanningError> {
1194 Ok(self.plan_retained(backend, request)?.into_parts().0)
1195 }
1196
1197 pub fn plan_retained<B: AutomaticPlanningBackend>(
1199 &self,
1200 backend: &B,
1201 request: &AutomaticPlanRequest,
1202 ) -> Result<RetainedAutomaticPlan<B::Inspection>, AutomaticPlanningError> {
1203 self.plan_retained_with_overrides(backend, request, |plan, _| Ok(plan.clone()))
1204 }
1205
1206 pub fn plan_retained_with_overrides<B: AutomaticPlanningBackend>(
1212 &self,
1213 backend: &B,
1214 request: &AutomaticPlanRequest,
1215 overrides: impl Fn(
1216 &ExecutionPlan,
1217 &ModelResourceProfile,
1218 ) -> Result<ExecutionPlan, AutomaticPlanningError>,
1219 ) -> Result<RetainedAutomaticPlan<B::Inspection>, AutomaticPlanningError> {
1220 validate_request(request, &self.policy)?;
1221 let backend_id = backend.backend_id();
1222 if request.device.backend != backend_id {
1223 return Err(AutomaticPlanningError::Invalid(format!(
1224 "selected planning backend {} cannot plan device owned by {}",
1225 backend_id, request.device.backend
1226 )));
1227 }
1228 let hardware = backend.discover_hardware()?;
1229 validate_device(&hardware, &request.device)?;
1230 let (mut resources, inspection) = backend.inspect_resources(&request.model_path)?;
1231 let selected_device =
1232 selected_device(&hardware, &request.device).expect("validated device is present");
1233 let device_capacity = memory_basis(
1234 observed_u64(&selected_device.available_memory_bytes),
1235 observed_u64(&selected_device.total_memory_bytes)
1236 .or_else(|| observed_u64(&hardware.physical_memory_bytes)),
1237 hardware.physical_memory_semantics,
1238 );
1239 let host_capacity = memory_basis(
1240 observed_u64(&hardware.available_memory_bytes),
1241 observed_u64(&hardware.physical_memory_bytes),
1242 hardware.physical_memory_semantics,
1243 );
1244 let device_budget = budget(
1245 device_capacity,
1246 self.policy.device_memory_fallback_bytes,
1247 self.policy.memory_headroom_percent,
1248 );
1249 let host_budget = budget(
1250 host_capacity,
1251 self.policy.host_memory_fallback_bytes,
1252 self.policy.memory_headroom_percent,
1253 );
1254 let model_bytes = observed_u64(&resources.materialized_parameter_bytes)
1255 .or_else(|| observed_u64(&resources.stored_tensor_bytes));
1256 let embedded_capacity = resources.embedded_draft_capacity.value().copied();
1257 let drafting = match embedded_capacity.filter(|capacity| *capacity > 0) {
1258 Some(capacity) => DraftingPlan::Embedded {
1259 max_draft_tokens: self.policy.embedded_mtp_draft_tokens.min(capacity),
1260 lookahead: true,
1261 adaptive_lookahead: true,
1262 },
1263 None => DraftingPlan::Disabled,
1264 };
1265 let mut candidates = base_candidates(
1266 request.device.clone(),
1267 device_budget,
1268 host_budget,
1269 &self.policy,
1270 );
1271 for candidate in &mut candidates {
1272 candidate.drafting = drafting.clone();
1273 *candidate = overrides(candidate, &resources)?;
1274 }
1275 let mut entries = vec![PlanExplanationEntry {
1276 level: PlanExplanationLevel::Decision,
1277 code: "single_device_scope".into(),
1278 detail: format!(
1279 "automatic planning is restricted to {}:{} with {}% memory headroom",
1280 request.device.backend, request.device.device, self.policy.memory_headroom_percent
1281 ),
1282 }];
1283 let mut selected = None;
1284 for candidate in candidates {
1285 validate_candidate(&candidate, request)?;
1286 let mut admission = backend.admit_candidate(&inspection, &candidate)?;
1287 let fits = match candidate.residency() {
1288 ResidencyPlan::FullyResident => {
1289 model_bytes.is_some_and(|bytes| bytes <= device_budget)
1290 }
1291 ResidencyPlan::LayerwiseHost {
1292 host_budget_bytes, ..
1293 } => {
1294 let capacity = host_budget_bytes.unwrap_or(host_budget);
1295 model_bytes.is_some_and(|bytes| {
1296 bytes
1297 <= if hardware.physical_memory_semantics
1298 == HardwareMemorySemantics::Unified
1299 {
1300 capacity.saturating_mul(2)
1301 } else {
1302 capacity
1303 }
1304 })
1305 }
1306 ResidencyPlan::DenseDiskStream { .. } => true,
1307 };
1308 if !fits {
1309 admission.supported = false;
1310 admission
1311 .rejection
1312 .get_or_insert_with(|| "the model exceeds the residency memory budget".into());
1313 }
1314 probe_candidate(
1315 backend,
1316 &inspection,
1317 &candidate,
1318 device_budget,
1319 &mut admission,
1320 &mut resources,
1321 )?;
1322 if admission.supported {
1323 selected = Some(candidate);
1324 break;
1325 }
1326 entries.push(PlanExplanationEntry {
1327 level: PlanExplanationLevel::Rejection,
1328 code: match candidate.residency() {
1329 ResidencyPlan::FullyResident => "fully_resident_not_admitted",
1330 ResidencyPlan::LayerwiseHost { .. } => "layerwise_not_admitted",
1331 ResidencyPlan::DenseDiskStream { .. } => "disk_stream_not_admitted",
1332 }
1333 .into(),
1334 detail: format!("{:?}: {}", candidate.residency(), rejection(&admission)),
1335 });
1336 }
1337 let mut plan = selected.ok_or_else(|| {
1338 AutomaticPlanningError::Invalid(format!(
1339 "no loadable single-device policy: {}",
1340 entries
1341 .iter()
1342 .filter(|entry| entry.level == PlanExplanationLevel::Rejection)
1343 .map(|entry| entry.detail.as_str())
1344 .collect::<Vec<_>>()
1345 .join("; ")
1346 ))
1347 })?;
1348 let mut summary = match plan.residency() {
1349 ResidencyPlan::FullyResident => {
1350 "selected fully resident execution for the lowest expected latency"
1351 }
1352 ResidencyPlan::LayerwiseHost { .. } => {
1353 "selected host-backed layerwise execution with a validated bounded device window"
1354 }
1355 ResidencyPlan::DenseDiskStream { .. } => "selected bounded dense disk streaming",
1356 }
1357 .to_string();
1358
1359 if !matches!(plan.residency(), ResidencyPlan::FullyResident)
1360 && plan.expert_cache().is_none()
1361 {
1362 let expert_plan =
1363 overrides(&with_expert_cache(plan.clone(), &self.policy), &resources)?;
1364 validate_candidate(&expert_plan, request)?;
1365 let mut expert = backend.admit_candidate(&inspection, &expert_plan)?;
1366 probe_candidate(
1367 backend,
1368 &inspection,
1369 &expert_plan,
1370 device_budget,
1371 &mut expert,
1372 &mut resources,
1373 )?;
1374 if expert.supported && expert_plan.expert_cache().is_some() {
1375 plan = expert_plan;
1376 entries.push(PlanExplanationEntry {
1377 level: PlanExplanationLevel::Decision,
1378 code: "expert_cache_selected".into(),
1379 detail: "the backend admitted independent routed-expert caching".into(),
1380 });
1381 }
1382 }
1383
1384 if let Some((feedback, samples, median)) = select_feedback_plan(
1385 backend,
1386 &inspection,
1387 request,
1388 &hardware,
1389 &resources,
1390 &self.policy,
1391 &overrides,
1392 )? {
1393 plan = feedback;
1394 summary = format!(
1395 "selected a previously observed plan at {median:.2} median decode tokens/s"
1396 );
1397 entries.push(PlanExplanationEntry {
1398 level: PlanExplanationLevel::Decision,
1399 code: "prior_telemetry_selected".into(),
1400 detail: format!("selected using {samples} matching runtime sample(s)"),
1401 });
1402 }
1403
1404 if let DraftingPlan::Embedded {
1405 max_draft_tokens, ..
1406 } = plan.drafting
1407 {
1408 entries.push(PlanExplanationEntry {
1409 level: PlanExplanationLevel::Decision,
1410 code: "embedded_mtp_selected".into(),
1411 detail: format!("selected {max_draft_tokens} embedded draft token(s); architecture capacity is {}", embedded_capacity.map_or_else(|| "unknown".into(), |capacity| capacity.to_string())),
1412 });
1413 }
1414
1415 let final_admission = backend.admit_candidate(&inspection, &plan)?;
1416 if !final_admission.supported {
1417 return Err(AutomaticPlanningError::Invalid(format!(
1418 "selected final plan is not loadable: {}",
1419 rejection(&final_admission)
1420 )));
1421 }
1422 if !matches!(plan.residency(), ResidencyPlan::FullyResident) {
1423 let final_budget = match plan.residency() {
1424 ResidencyPlan::LayerwiseHost {
1425 device_budget_bytes,
1426 ..
1427 } => device_budget_bytes.unwrap_or(device_budget),
1428 ResidencyPlan::DenseDiskStream {
1429 device_budget_bytes,
1430 ..
1431 } => *device_budget_bytes,
1432 ResidencyPlan::FullyResident => unreachable!(),
1433 };
1434 let mut final_probe = final_admission;
1435 apply_bounded_probe(
1436 backend,
1437 &inspection,
1438 &plan,
1439 final_budget,
1440 &mut final_probe,
1441 &mut resources,
1442 matches!(plan.residency(), ResidencyPlan::DenseDiskStream { .. }),
1443 )?;
1444 if !final_probe.supported {
1445 return Err(AutomaticPlanningError::Invalid(format!(
1446 "selected final plan exceeds its exact bounded residency: {}",
1447 rejection(&final_probe)
1448 )));
1449 }
1450 }
1451 let report = ExecutionPlanReport {
1452 schema_version: AUTOMATIC_SCHEMA_VERSION,
1453 hardware,
1454 resources,
1455 plan,
1456 explanation: PlanExplanation { summary, entries },
1457 };
1458 Ok(RetainedAutomaticPlan { report, inspection })
1459 }
1460}
1461
1462fn observed_u64(value: &Observed<u64>) -> Option<u64> {
1463 value.value().copied()
1464}
1465
1466fn validate_request(
1467 request: &AutomaticPlanRequest,
1468 policy: &AutomaticPlannerPolicy,
1469) -> Result<(), AutomaticPlanningError> {
1470 if request.schema_version != AUTOMATIC_SCHEMA_VERSION {
1471 return Err(AutomaticPlanningError::Invalid(format!(
1472 "automatic request schema {} does not match supported schema {}",
1473 request.schema_version, AUTOMATIC_SCHEMA_VERSION
1474 )));
1475 }
1476 if policy.device_memory_fallback_bytes == 0 || policy.host_memory_fallback_bytes == 0 {
1477 return Err(AutomaticPlanningError::Invalid(
1478 "automatic fallback memory budgets must be greater than zero".into(),
1479 ));
1480 }
1481 if policy.memory_headroom_percent >= 100
1482 || policy.expert_cache_share_percent == 0
1483 || policy.expert_cache_share_percent >= 100
1484 || policy.device_layer_window == 0
1485 || policy.max_cached_shards == 0
1486 || policy.embedded_mtp_draft_tokens == 0
1487 || policy.minimum_feedback_tokens == 0
1488 {
1489 return Err(AutomaticPlanningError::Invalid(
1490 "automatic percentage and count policy values are outside their valid ranges".into(),
1491 ));
1492 }
1493 Ok(())
1494}
1495
1496fn selected_device<'a>(
1497 hardware: &'a HardwareProfile,
1498 device: &DevicePlan,
1499) -> Option<&'a HardwareDeviceProfile> {
1500 hardware
1501 .backends
1502 .iter()
1503 .find(|backend| backend.backend == device.backend && backend.available)
1504 .and_then(|backend| backend.devices.iter().find(|item| item.id == device.device))
1505}
1506
1507fn validate_device(
1508 hardware: &HardwareProfile,
1509 device: &DevicePlan,
1510) -> Result<(), AutomaticPlanningError> {
1511 selected_device(hardware, device)
1512 .map(|_| ())
1513 .ok_or_else(|| {
1514 AutomaticPlanningError::Invalid(format!(
1515 "hardware discovery did not report available {} device {}",
1516 device.backend, device.device
1517 ))
1518 })
1519}
1520
1521fn memory_basis(
1522 available: Option<u64>,
1523 physical: Option<u64>,
1524 semantics: HardwareMemorySemantics,
1525) -> Option<u64> {
1526 available.or_else(|| {
1527 (semantics == HardwareMemorySemantics::Unified)
1528 .then_some(physical)
1529 .flatten()
1530 })
1531}
1532
1533fn budget(available: Option<u64>, fallback: u64, headroom_percent: u8) -> u64 {
1534 available
1535 .map(|bytes| bytes.saturating_mul(u64::from(100 - headroom_percent)) / 100)
1536 .unwrap_or(fallback)
1537 .max(1)
1538}
1539
1540fn base_candidates(
1541 device: DevicePlan,
1542 device_budget: u64,
1543 host_budget: u64,
1544 policy: &AutomaticPlannerPolicy,
1545) -> [ExecutionPlan; 3] {
1546 let mut resident = ExecutionPlan::fully_resident(device);
1547 resident.max_cached_shards = policy.max_cached_shards;
1548 let mut layerwise = resident.clone();
1549 layerwise.residency = ResidencyPlan::LayerwiseHost {
1550 device_layer_window: policy.device_layer_window,
1551 device_budget_bytes: Some(device_budget),
1552 host_budget_bytes: Some(host_budget),
1553 };
1554 let mut disk = resident.clone();
1555 disk.residency = ResidencyPlan::DenseDiskStream {
1556 device_budget_bytes: device_budget,
1557 host_budget_bytes: host_budget,
1558 host_lookahead: usize::from(host_budget > 0) * 2,
1559 background_queue: usize::from(host_budget > 0) * 2,
1560 };
1561 [resident, layerwise, disk]
1562}
1563
1564fn validate_candidate(
1565 plan: &ExecutionPlan,
1566 request: &AutomaticPlanRequest,
1567) -> Result<(), AutomaticPlanningError> {
1568 plan.validate_structure()
1569 .map_err(|error| AutomaticPlanningError::Invalid(error.to_string()))?;
1570 if plan.device != request.device
1571 || plan.topology
1572 != crate::topology::ParallelTopology::new(1, 1, 1, 1).expect("singleton topology")
1573 {
1574 return Err(AutomaticPlanningError::Invalid(
1575 "automatic overrides must preserve the selected device and singleton topology".into(),
1576 ));
1577 }
1578 Ok(())
1579}
1580
1581fn probe_candidate<B: AutomaticPlanningBackend>(
1582 backend: &B,
1583 inspection: &B::Inspection,
1584 plan: &ExecutionPlan,
1585 default_device_budget: u64,
1586 admission: &mut CandidateAdmission,
1587 resources: &mut ModelResourceProfile,
1588) -> Result<(), AutomaticPlanningError> {
1589 let budget = match plan.residency() {
1590 ResidencyPlan::FullyResident => return Ok(()),
1591 ResidencyPlan::LayerwiseHost {
1592 device_budget_bytes,
1593 ..
1594 } => device_budget_bytes.unwrap_or(default_device_budget),
1595 ResidencyPlan::DenseDiskStream {
1596 device_budget_bytes,
1597 ..
1598 } => *device_budget_bytes,
1599 };
1600 apply_bounded_probe(
1601 backend,
1602 inspection,
1603 plan,
1604 budget,
1605 admission,
1606 resources,
1607 matches!(plan.residency(), ResidencyPlan::DenseDiskStream { .. }),
1608 )
1609}
1610
1611fn apply_bounded_probe<B: AutomaticPlanningBackend>(
1612 backend: &B,
1613 inspection: &B::Inspection,
1614 plan: &ExecutionPlan,
1615 budget: u64,
1616 admission: &mut CandidateAdmission,
1617 resources: &mut ModelResourceProfile,
1618 adjacent: bool,
1619) -> Result<(), AutomaticPlanningError> {
1620 if !admission.supported {
1621 return Ok(());
1622 }
1623 let requirement = backend.bounded_residency_requirement(inspection, plan)?;
1624 if requirement.required_bytes > budget {
1625 admission.supported = false;
1626 admission.rejection = Some(format!(
1627 "device budget {budget} bytes cannot contain {} pinned static bytes plus the depth-{} device window ({} bytes, {} total)",
1628 requirement.static_bytes,
1629 requirement.depth,
1630 requirement.window_bytes,
1631 requirement.required_bytes
1632 ));
1633 }
1634 resources.pinned_parameter_bytes =
1635 Observed::exact(requirement.static_bytes, "validated backend parameter plan");
1636 if adjacent {
1637 resources.largest_adjacent_execution_groups_bytes =
1638 Observed::exact(requirement.window_bytes, "validated backend parameter plan");
1639 } else {
1640 resources.largest_execution_group_bytes =
1641 Observed::exact(requirement.window_bytes, "validated backend parameter plan");
1642 }
1643 Ok(())
1644}
1645
1646fn rejection(admission: &CandidateAdmission) -> &str {
1647 admission.rejection.as_deref().unwrap_or("not admitted")
1648}
1649
1650fn with_expert_cache(mut plan: ExecutionPlan, policy: &AutomaticPlannerPolicy) -> ExecutionPlan {
1651 let split = |bytes: u64, percent: u8| bytes.saturating_mul(u64::from(percent)) / 100;
1652 let ordinary_share = 100 - policy.expert_cache_share_percent;
1653 let (device_budget, host_budget) = match &mut plan.residency {
1654 ResidencyPlan::FullyResident => (
1655 policy.device_memory_fallback_bytes,
1656 policy.host_memory_fallback_bytes,
1657 ),
1658 ResidencyPlan::LayerwiseHost {
1659 device_budget_bytes,
1660 host_budget_bytes,
1661 ..
1662 } => {
1663 let device = device_budget_bytes.unwrap_or(policy.device_memory_fallback_bytes);
1664 let host = host_budget_bytes.unwrap_or(policy.host_memory_fallback_bytes);
1665 *device_budget_bytes = Some(split(device, ordinary_share).max(1));
1666 *host_budget_bytes = Some(split(host, ordinary_share).max(1));
1667 (device, host)
1668 }
1669 ResidencyPlan::DenseDiskStream {
1670 device_budget_bytes,
1671 host_budget_bytes,
1672 ..
1673 } => {
1674 let (device, host) = (*device_budget_bytes, *host_budget_bytes);
1675 *device_budget_bytes = split(device, ordinary_share).max(1);
1676 *host_budget_bytes = split(host, ordinary_share).max(1);
1677 (device, host)
1678 }
1679 };
1680 let scratch = (1_u64 << 30).min(device_budget.max(1));
1681 plan.expert_cache = Some(ExpertCachePlan {
1682 device_budget_bytes: Some(split(device_budget, policy.expert_cache_share_percent).max(1)),
1683 host_budget_bytes: Some(split(host_budget, policy.expert_cache_share_percent).max(1)),
1684 scratch_bytes: scratch,
1685 prefill_bank_bytes: scratch,
1686 eviction_policy: crate::residency::CacheEvictionPolicy::LeastRecentlyUsed,
1687 });
1688 plan
1689}
1690
1691fn select_feedback_plan<B: AutomaticPlanningBackend>(
1692 backend: &B,
1693 inspection: &B::Inspection,
1694 request: &AutomaticPlanRequest,
1695 hardware: &HardwareProfile,
1696 resources: &ModelResourceProfile,
1697 policy: &AutomaticPlannerPolicy,
1698 overrides: &impl Fn(
1699 &ExecutionPlan,
1700 &ModelResourceProfile,
1701 ) -> Result<ExecutionPlan, AutomaticPlanningError>,
1702) -> Result<Option<(ExecutionPlan, usize, f64)>, AutomaticPlanningError> {
1703 let mut groups: Vec<(ExecutionPlan, Vec<f64>)> = Vec::new();
1704 for telemetry in &request.prior_telemetry {
1705 let (Some(plan), Some(prior_hardware), Some(prior_resources)) = (
1706 telemetry.plan.as_ref(),
1707 telemetry.hardware.as_ref(),
1708 telemetry.resources.as_ref(),
1709 ) else {
1710 continue;
1711 };
1712 if telemetry.schema_version != AUTOMATIC_SCHEMA_VERSION
1713 || telemetry.generated_tokens < policy.minimum_feedback_tokens
1714 || plan.device != request.device
1715 || prior_resources.path != resources.path
1716 || prior_resources.artifact_format != resources.artifact_format
1717 || prior_resources.model_family != resources.model_family
1718 || prior_hardware.operating_system != hardware.operating_system
1719 || prior_hardware.architecture != hardware.architecture
1720 || matches!(plan.drafting, DraftingPlan::External { .. })
1721 || overrides(plan, resources)? != *plan
1722 || matches!(plan.drafting, DraftingPlan::Embedded { max_draft_tokens, .. }
1723 if !resources.embedded_draft_capacity.value().is_some_and(|capacity| max_draft_tokens <= *capacity))
1724 {
1725 continue;
1726 }
1727 let rate = telemetry
1728 .timing
1729 .decode_token_rate
1730 .filter(|value| value.is_finite() && *value > 0.0)
1731 .or_else(|| {
1732 (telemetry.timing.token_rate.is_finite() && telemetry.timing.token_rate > 0.0)
1733 .then_some(telemetry.timing.token_rate)
1734 });
1735 let Some(rate) = rate else { continue };
1736 if let Some((_, rates)) = groups.iter_mut().find(|(candidate, _)| candidate == plan) {
1737 rates.push(rate);
1738 } else {
1739 groups.push((plan.clone(), vec![rate]));
1740 }
1741 }
1742 let mut accepted = Vec::new();
1743 for (plan, mut rates) in groups {
1744 if validate_candidate(&plan, request).is_err()
1745 || !backend.admit_candidate(inspection, &plan)?.supported
1746 {
1747 continue;
1748 }
1749 rates.sort_by(f64::total_cmp);
1750 let middle = rates.len() / 2;
1751 let median = if rates.len() % 2 == 0 {
1752 (rates[middle - 1] + rates[middle]) / 2.0
1753 } else {
1754 rates[middle]
1755 };
1756 accepted.push((plan, rates.len(), median));
1757 }
1758 Ok(accepted
1759 .into_iter()
1760 .max_by(|left, right| left.2.total_cmp(&right.2)))
1761}
1762
1763#[cfg(test)]
1764mod tests {
1765 use super::*;
1766 use crate::execution::BackendId;
1767
1768 #[test]
1769 fn host_observation_preserves_injected_memory_and_backend_facts() {
1770 let physical = Observed::exact(4096, "foreign memory provider");
1771 let available = Observed::unavailable("not measured");
1772 let backend = HardwareBackendProfile {
1773 backend: BackendId::new("independent").unwrap(),
1774 available: false,
1775 detail: Some("no native context created".into()),
1776 devices: vec![],
1777 };
1778 let profile = HardwareProfile::observe_host(
1779 physical.clone(),
1780 available.clone(),
1781 HardwareMemorySemantics::SeparateTiers,
1782 vec![backend.clone()],
1783 );
1784 assert_eq!(profile.physical_memory_bytes, physical);
1785 assert_eq!(profile.available_memory_bytes, available);
1786 assert_eq!(profile.backends, vec![backend]);
1787 assert_eq!(
1788 profile.physical_memory_semantics,
1789 HardwareMemorySemantics::SeparateTiers
1790 );
1791 assert!(!profile.operating_system.is_empty());
1792 assert!(!profile.architecture.is_empty());
1793 }
1794
1795 #[test]
1796 fn physical_memory_semantics_preserve_unknown_and_separate_capacity() {
1797 use crate::capability::PhysicalMemorySemantics;
1798 for (physical, hardware) in [
1799 (
1800 PhysicalMemorySemantics::Unified,
1801 HardwareMemorySemantics::Unified,
1802 ),
1803 (
1804 PhysicalMemorySemantics::SeparateTiers,
1805 HardwareMemorySemantics::SeparateTiers,
1806 ),
1807 (
1808 PhysicalMemorySemantics::Unknown,
1809 HardwareMemorySemantics::Unknown,
1810 ),
1811 ] {
1812 assert_eq!(HardwareMemorySemantics::from(physical), hardware);
1813 }
1814 }
1815
1816 struct MockPlanningBackend {
1817 model_bytes: u64,
1818 embedded_layers: usize,
1819 embedded_capacity: Option<usize>,
1820 required_cached_shards: Option<usize>,
1821 }
1822
1823 impl Default for MockPlanningBackend {
1824 fn default() -> Self {
1825 Self {
1826 model_bytes: 2 << 30,
1827 embedded_layers: 0,
1828 embedded_capacity: Some(0),
1829 required_cached_shards: None,
1830 }
1831 }
1832 }
1833
1834 impl AutomaticPlanningBackend for MockPlanningBackend {
1835 type Inspection = ();
1836
1837 fn backend_id(&self) -> BackendId {
1838 BackendId::new("mock").unwrap()
1839 }
1840
1841 fn discover_hardware(&self) -> Result<HardwareProfile, AutomaticPlanningError> {
1842 Ok(HardwareProfile {
1843 schema_version: AUTOMATIC_SCHEMA_VERSION,
1844 operating_system: "test".into(),
1845 architecture: "mock".into(),
1846 logical_cpu_count: Observed::exact(8, "fixture"),
1847 physical_memory_bytes: Observed::exact(32 << 30, "fixture"),
1848 available_memory_bytes: Observed::exact(24 << 30, "fixture"),
1849 physical_memory_semantics: HardwareMemorySemantics::SeparateTiers,
1850 backends: vec![HardwareBackendProfile {
1851 backend: BackendId::new("mock").unwrap(),
1852 available: true,
1853 detail: None,
1854 devices: vec![HardwareDeviceProfile {
1855 id: "gpu:0".into(),
1856 family: "gpu".into(),
1857 index: 0,
1858 total_memory_bytes: Observed::exact(16 << 30, "fixture"),
1859 available_memory_bytes: Observed::exact(12 << 30, "fixture"),
1860 }],
1861 }],
1862 })
1863 }
1864
1865 fn inspect_resources(
1866 &self,
1867 path: &std::path::Path,
1868 ) -> Result<(ModelResourceProfile, Self::Inspection), AutomaticPlanningError> {
1869 let mut profile =
1870 ModelResourceProfile::unmeasured(path.into(), ArtifactFormat::SafeTensors);
1871 profile.model_family = Some("llama".into());
1872 profile.embedded_draft_layers =
1873 Observed::exact(self.embedded_layers, "normalized architecture fixture");
1874 profile.embedded_draft_capacity = self.embedded_capacity.map_or_else(
1875 || Observed::unavailable("unknown fixture capacity"),
1876 |capacity| Observed::exact(capacity, "architecture contract fixture"),
1877 );
1878 profile.stored_tensor_bytes = Observed::exact(self.model_bytes, "fixture");
1879 profile.materialized_parameter_bytes = Observed::exact(self.model_bytes, "fixture");
1880 Ok((profile, ()))
1881 }
1882
1883 fn admit_candidate(
1884 &self,
1885 _inspection: &Self::Inspection,
1886 plan: &ExecutionPlan,
1887 ) -> Result<CandidateAdmission, AutomaticPlanningError> {
1888 let rejection = if matches!(plan.drafting(), DraftingPlan::Embedded { max_draft_tokens, .. } if *max_draft_tokens > self.embedded_capacity.unwrap_or(0))
1889 {
1890 Some("draft capacity exceeded".into())
1891 } else if self
1892 .required_cached_shards
1893 .is_some_and(|required| plan.max_cached_shards() != required)
1894 {
1895 Some("cached-shard override missing".into())
1896 } else {
1897 None
1898 };
1899 Ok(CandidateAdmission {
1900 supported: rejection.is_none(),
1901 rejection,
1902 })
1903 }
1904
1905 fn bounded_residency_requirement(
1906 &self,
1907 _inspection: &Self::Inspection,
1908 _plan: &ExecutionPlan,
1909 ) -> Result<BoundedResidencyRequirement, AutomaticPlanningError> {
1910 Ok(BoundedResidencyRequirement {
1911 static_bytes: 1 << 20,
1912 window_bytes: 2 << 20,
1913 required_bytes: 3 << 20,
1914 depth: 1,
1915 })
1916 }
1917 }
1918
1919 #[test]
1920 fn neutral_planner_selects_a_mock_backend_session_plan() {
1921 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
1922 let report = AutomaticPlanner::default()
1923 .plan(&MockPlanningBackend::default(), &request)
1924 .unwrap();
1925 assert_eq!(report.plan.device.backend.as_str(), "mock");
1926 assert_eq!(report.plan.residency, ResidencyPlan::FullyResident);
1927 }
1928
1929 #[test]
1930 fn neutral_planner_selects_bounded_residency_and_embedded_drafting() {
1931 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
1932 let report = AutomaticPlanner::default()
1933 .plan(
1934 &MockPlanningBackend {
1935 model_bytes: 10 << 30,
1936 embedded_layers: 2,
1937 embedded_capacity: Some(2),
1938 ..MockPlanningBackend::default()
1939 },
1940 &request,
1941 )
1942 .unwrap();
1943 assert!(matches!(
1944 report.plan.residency,
1945 ResidencyPlan::LayerwiseHost { .. }
1946 ));
1947 assert!(matches!(
1948 report.plan.drafting,
1949 DraftingPlan::Embedded { .. }
1950 ));
1951 assert_eq!(
1952 observed_u64(&report.resources.pinned_parameter_bytes),
1953 Some(1 << 20)
1954 );
1955 }
1956
1957 #[test]
1958 fn automatic_draft_width_respects_capacity_instead_of_physical_depth() {
1959 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
1960 for (capacity, expected) in [
1961 (Some(1), 1),
1962 (Some(2), 2),
1963 (Some(8), 3),
1964 (Some(0), 0),
1965 (None, 0),
1966 ] {
1967 let backend = MockPlanningBackend {
1968 embedded_layers: 1,
1969 embedded_capacity: capacity,
1970 ..MockPlanningBackend::default()
1971 };
1972 let report = AutomaticPlanner::default()
1973 .plan(&backend, &request)
1974 .unwrap();
1975 match report.plan.drafting() {
1976 DraftingPlan::Embedded {
1977 max_draft_tokens, ..
1978 } => assert_eq!(*max_draft_tokens, expected),
1979 DraftingPlan::Disabled => assert_eq!(expected, 0),
1980 other => panic!("unexpected drafting {other:?}"),
1981 }
1982 }
1983 }
1984
1985 #[test]
1986 fn explicit_draft_width_is_validated_without_silent_clamping() {
1987 let backend = MockPlanningBackend {
1988 embedded_layers: 1,
1989 embedded_capacity: Some(1),
1990 ..MockPlanningBackend::default()
1991 };
1992 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
1993 let error = AutomaticPlanner::default()
1994 .plan_retained_with_overrides(&backend, &request, |plan, _| {
1995 Ok(plan.clone().with_drafting(DraftingPlan::Embedded {
1996 max_draft_tokens: 3,
1997 lookahead: false,
1998 adaptive_lookahead: false,
1999 }))
2000 })
2001 .err()
2002 .unwrap();
2003 assert!(error.to_string().contains("draft capacity exceeded"));
2004 }
2005
2006 #[test]
2007 fn resource_documents_without_capacity_remain_unknown() {
2008 let mut profile =
2009 ModelResourceProfile::unmeasured("model".into(), ArtifactFormat::SafeTensors);
2010 profile.embedded_draft_layers = Observed::exact(1, "legacy depth");
2011 let mut encoded = serde_json::to_value(profile).unwrap();
2012 encoded
2013 .as_object_mut()
2014 .unwrap()
2015 .remove("embedded_draft_capacity");
2016 let decoded: ModelResourceProfile = serde_json::from_value(encoded).unwrap();
2017 assert_eq!(decoded.embedded_draft_layers.value(), Some(&1));
2018 assert_eq!(decoded.embedded_draft_capacity.value(), None);
2019 }
2020
2021 #[test]
2022 fn selected_backend_identity_fails_closed() {
2023 let request =
2024 AutomaticPlanRequest::new("model", DevicePlan::new("other", "gpu:0").unwrap());
2025 assert!(matches!(
2026 AutomaticPlanner::default().plan(&MockPlanningBackend::default(), &request),
2027 Err(AutomaticPlanningError::Invalid(message))
2028 if message.contains("cannot plan device")
2029 ));
2030 }
2031
2032 #[test]
2033 fn documents_round_trip_without_an_accelerator_runtime() {
2034 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
2035 let encoded = serde_json::to_vec(&request).unwrap();
2036 assert_eq!(
2037 serde_json::from_slice::<AutomaticPlanRequest>(&encoded).unwrap(),
2038 request
2039 );
2040 let unavailable = serde_json::to_value(Observed::<u64>::unavailable("unknown")).unwrap();
2041 assert!(unavailable.get("value").is_none());
2042 }
2043
2044 #[test]
2045 fn tokenizer_compatibility_requires_identical_vocabularies() {
2046 let fingerprint = [7; 32];
2047 let proof = TokenizerCompatibilityProof::prove(fingerprint, fingerprint).unwrap();
2048 assert_eq!(proof.fingerprint(), fingerprint);
2049 assert_eq!(proof.validate_target(fingerprint), Ok(()));
2050 assert_eq!(
2051 proof.validate_target([8; 32]),
2052 Err(TokenizerCompatibilityError)
2053 );
2054 assert_eq!(
2055 TokenizerCompatibilityProof::prove(fingerprint, [8; 32]),
2056 Err(TokenizerCompatibilityError)
2057 );
2058 }
2059
2060 struct RetainedPlanningBackend {
2061 inner: MockPlanningBackend,
2062 inspections: std::cell::Cell<usize>,
2063 admissions: std::cell::RefCell<Vec<ExecutionPlan>>,
2064 bounded_probes: std::cell::RefCell<Vec<ExecutionPlan>>,
2065 }
2066
2067 impl AutomaticPlanningBackend for RetainedPlanningBackend {
2068 type Inspection = usize;
2069
2070 fn backend_id(&self) -> BackendId {
2071 self.inner.backend_id()
2072 }
2073
2074 fn discover_hardware(&self) -> Result<HardwareProfile, AutomaticPlanningError> {
2075 self.inner.discover_hardware()
2076 }
2077
2078 fn inspect_resources(
2079 &self,
2080 path: &std::path::Path,
2081 ) -> Result<(ModelResourceProfile, Self::Inspection), AutomaticPlanningError> {
2082 self.inspections.set(self.inspections.get() + 1);
2083 self.inner
2084 .inspect_resources(path)
2085 .map(|(resources, ())| (resources, 7))
2086 }
2087
2088 fn admit_candidate(
2089 &self,
2090 inspection: &Self::Inspection,
2091 plan: &ExecutionPlan,
2092 ) -> Result<CandidateAdmission, AutomaticPlanningError> {
2093 assert_eq!(*inspection, 7, "every admission must reuse one inspection");
2094 self.admissions.borrow_mut().push(plan.clone());
2095 self.inner.admit_candidate(&(), plan)
2096 }
2097
2098 fn bounded_residency_requirement(
2099 &self,
2100 inspection: &Self::Inspection,
2101 plan: &ExecutionPlan,
2102 ) -> Result<BoundedResidencyRequirement, AutomaticPlanningError> {
2103 assert_eq!(
2104 *inspection, 7,
2105 "every bounded probe must reuse one inspection"
2106 );
2107 self.bounded_probes.borrow_mut().push(plan.clone());
2108 self.inner.bounded_residency_requirement(&(), plan)
2109 }
2110 }
2111
2112 #[test]
2113 fn automatic_planning_retains_one_inspection_and_exactly_reprobes_the_final_plan() {
2114 let backend = RetainedPlanningBackend {
2115 inner: MockPlanningBackend {
2116 model_bytes: 10 << 30,
2117 embedded_layers: 2,
2118 embedded_capacity: Some(2),
2119 ..MockPlanningBackend::default()
2120 },
2121 inspections: std::cell::Cell::new(0),
2122 admissions: std::cell::RefCell::new(Vec::new()),
2123 bounded_probes: std::cell::RefCell::new(Vec::new()),
2124 };
2125 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
2126 let retained = AutomaticPlanner::default()
2127 .plan_retained(&backend, &request)
2128 .unwrap();
2129 assert_eq!(backend.inspections.get(), 1);
2130 assert_eq!(
2131 backend.admissions.borrow().last(),
2132 Some(&retained.report().plan)
2133 );
2134 assert_eq!(
2135 backend.bounded_probes.borrow().last(),
2136 Some(&retained.report().plan),
2137 "drafting/expert/feedback mutations must be exact-probed, not only their base candidate"
2138 );
2139 assert!(matches!(
2140 retained.report().plan.drafting(),
2141 DraftingPlan::Embedded { .. }
2142 ));
2143 let (_, inspection) = retained.into_parts();
2144 assert_eq!(inspection, 7);
2145 }
2146
2147 #[test]
2148 fn overrides_precede_every_admission_and_probe_including_expert_cache() {
2149 let backend = RetainedPlanningBackend {
2150 inner: MockPlanningBackend {
2151 model_bytes: 64 << 30,
2152 embedded_layers: 1,
2153 embedded_capacity: Some(1),
2154 required_cached_shards: Some(7),
2155 },
2156 inspections: std::cell::Cell::new(0),
2157 admissions: std::cell::RefCell::new(Vec::new()),
2158 bounded_probes: std::cell::RefCell::new(Vec::new()),
2159 };
2160 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
2161 assert!(AutomaticPlanner::default()
2162 .plan(&backend, &request)
2163 .is_err());
2164 backend.admissions.borrow_mut().clear();
2165 backend.inspections.set(0);
2166 let retained = AutomaticPlanner::default()
2167 .plan_retained_with_overrides(&backend, &request, |plan, _| {
2168 Ok(plan
2169 .clone()
2170 .with_drafting(DraftingPlan::Disabled)
2171 .with_max_cached_shards(7)
2172 .with_residency(ResidencyPlan::DenseDiskStream {
2173 device_budget_bytes: 8 << 20,
2174 host_budget_bytes: 16 << 20,
2175 host_lookahead: 1,
2176 background_queue: 1,
2177 }))
2178 })
2179 .unwrap();
2180 assert_eq!(backend.inspections.get(), 1);
2181 assert!(retained.report().plan.expert_cache().is_some());
2182 for plan in backend
2183 .admissions
2184 .borrow()
2185 .iter()
2186 .chain(backend.bounded_probes.borrow().iter())
2187 {
2188 assert_eq!(plan.max_cached_shards(), 7);
2189 assert_eq!(plan.drafting(), &DraftingPlan::Disabled);
2190 assert!(
2191 matches!(plan.residency(), ResidencyPlan::DenseDiskStream { device_budget_bytes, .. } if *device_budget_bytes == 8 << 20)
2192 );
2193 }
2194 assert_eq!(
2195 backend.bounded_probes.borrow().last(),
2196 Some(&retained.report().plan)
2197 );
2198 }
2199
2200 #[test]
2201 fn overridden_device_budget_is_used_for_candidate_probes() {
2202 let backend = MockPlanningBackend::default();
2203 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
2204 let error = AutomaticPlanner::default()
2205 .plan_retained_with_overrides(&backend, &request, |plan, _| {
2206 Ok(plan.clone().with_residency(ResidencyPlan::DenseDiskStream {
2207 device_budget_bytes: 1 << 20,
2208 host_budget_bytes: 16 << 20,
2209 host_lookahead: 1,
2210 background_queue: 1,
2211 }))
2212 })
2213 .err()
2214 .unwrap();
2215 assert!(error
2216 .to_string()
2217 .contains("device budget 1048576 bytes cannot contain"));
2218 }
2219
2220 #[test]
2221 fn automatic_feedback_cannot_select_an_uninspected_external_assistant() {
2222 let backend = MockPlanningBackend::default();
2223 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
2224 let hardware = backend.discover_hardware().unwrap();
2225 let resources = backend.inspect_resources(&request.model_path).unwrap().0;
2226 let external = ExecutionPlan::fully_resident(request.device.clone()).with_drafting(
2227 DraftingPlan::External {
2228 model: "missing-assistant".into(),
2229 placement: crate::execution::DraftPlacementPlan::Target,
2230 max_draft_tokens: 2,
2231 lookahead: false,
2232 adaptive_lookahead: false,
2233 },
2234 );
2235 let telemetry = ExecutionTelemetry {
2236 schema_version: AUTOMATIC_SCHEMA_VERSION,
2237 effective_model_type: "fixture".into(),
2238 plan: Some(external),
2239 plan_explanation: None,
2240 hardware: Some(hardware),
2241 resources: Some(resources),
2242 prompt_tokens: 1,
2243 generated_tokens: 1,
2244 stop_reason: "length".into(),
2245 timing: TimingTelemetry::new(
2246 Duration::from_secs(1),
2247 Duration::from_secs(1),
2248 None,
2249 100,
2250 Duration::from_secs(2),
2251 ),
2252 allocator: None,
2253 residency: None,
2254 expert_cache: None,
2255 speculative: None,
2256 };
2257
2258 let report = AutomaticPlanner::default()
2259 .plan(&backend, &request.with_prior_telemetry([telemetry]))
2260 .unwrap();
2261
2262 assert!(matches!(report.plan.drafting(), DraftingPlan::Disabled));
2263 assert!(!report
2264 .explanation
2265 .entries
2266 .iter()
2267 .any(|entry| entry.code == "prior_telemetry_selected"));
2268 }
2269
2270 #[test]
2271 fn feedback_must_satisfy_current_capacity_and_explicit_overrides() {
2272 let backend = MockPlanningBackend {
2273 embedded_layers: 1,
2274 embedded_capacity: Some(1),
2275 ..MockPlanningBackend::default()
2276 };
2277 let request = AutomaticPlanRequest::new("model", DevicePlan::new("mock", "gpu:0").unwrap());
2278 let measured = |width| ExecutionTelemetry {
2279 schema_version: AUTOMATIC_SCHEMA_VERSION,
2280 effective_model_type: "fixture".into(),
2281 plan: Some(
2282 ExecutionPlan::fully_resident(request.device.clone()).with_drafting(
2283 DraftingPlan::Embedded {
2284 max_draft_tokens: width,
2285 lookahead: false,
2286 adaptive_lookahead: false,
2287 },
2288 ),
2289 ),
2290 plan_explanation: None,
2291 hardware: Some(backend.discover_hardware().unwrap()),
2292 resources: Some(backend.inspect_resources(&request.model_path).unwrap().0),
2293 prompt_tokens: 1,
2294 generated_tokens: 10,
2295 stop_reason: "length".into(),
2296 timing: TimingTelemetry::new(
2297 Duration::from_secs(1),
2298 Duration::from_secs(1),
2299 None,
2300 100,
2301 Duration::from_secs(2),
2302 ),
2303 allocator: None,
2304 residency: None,
2305 expert_cache: None,
2306 speculative: None,
2307 };
2308 let request = request
2309 .clone()
2310 .with_prior_telemetry([measured(1), measured(8)]);
2311 let report = AutomaticPlanner::default()
2312 .plan(&backend, &request)
2313 .unwrap();
2314 assert!(report
2315 .explanation
2316 .entries
2317 .iter()
2318 .any(|entry| entry.code == "prior_telemetry_selected"));
2319 assert!(matches!(
2320 report.plan.drafting(),
2321 DraftingPlan::Embedded {
2322 max_draft_tokens: 1,
2323 lookahead: false,
2324 ..
2325 }
2326 ));
2327 let retained = AutomaticPlanner::default()
2328 .plan_retained_with_overrides(&backend, &request, |plan, _| {
2329 Ok(plan.clone().with_drafting(DraftingPlan::Disabled))
2330 })
2331 .unwrap();
2332 assert_eq!(retained.report().plan.drafting(), &DraftingPlan::Disabled);
2333 assert!(!retained
2334 .report()
2335 .explanation
2336 .entries
2337 .iter()
2338 .any(|entry| entry.code == "prior_telemetry_selected"));
2339 }
2340
2341 #[test]
2342 fn zero_duration_rates_are_finite() {
2343 let timing = TimingTelemetry::new(
2344 Duration::ZERO,
2345 Duration::ZERO,
2346 Some(Duration::ZERO),
2347 3,
2348 Duration::ZERO,
2349 );
2350 assert_eq!(timing.token_rate, 0.0);
2351 assert_eq!(timing.decode_token_rate, Some(0.0));
2352 }
2353}