Skip to main content

xlog_runtime/executor/
resident.rs

1use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
2use std::marker::PhantomData;
3use std::sync::Arc;
4use std::time::Instant;
5
6use cudarc::driver::{sys::CUevent_flags, CudaEvent};
7use xlog_core::{symbol, RelId, Result, ScalarType, Schema, XlogError};
8use xlog_cuda::cuda_graph::{
9    CapturedCudaGraph, ConditionalCudaGraphSequenceBuilder, CudaConditionalGraphUnavailable,
10    CudaGraphNodeKind,
11};
12use xlog_cuda::device_runtime::{BlockId, ResidentCompletionEvent, XlogDeviceRuntime};
13use xlog_cuda::launch::LaunchRecorder;
14use xlog_cuda::memory::GpuMemoryReservation;
15use xlog_cuda::provider::resident_filter_project::{
16    resident_filter_scratch_device_bytes, ResidentFilterScratch,
17};
18use xlog_cuda::provider::resident_relational::{
19    resident_control_device_bytes, resident_device_trace_bytes,
20    resident_join_workspace_device_bytes, resident_packed_receipt_with_schema_winners_device_bytes,
21    resident_relation_device_bytes, resident_schema_winners_device_bytes,
22    resident_set_workspace_device_bytes, ResidentConvergenceControl, ResidentDeviceTrace,
23    ResidentJoinKind, ResidentJoinWorkspace, ResidentPackedReceipt, ResidentPinnedReceipt,
24    ResidentRelation, ResidentResourceCode, ResidentSchemaWinners, ResidentSetWorkspace,
25    ResidentTerminalCode, ResidentTerminalStatus,
26};
27use xlog_cuda::provider::resident_schedule::{
28    resident_schedule_metadata_device_bytes, ResidentExecutionDomain,
29    ResidentFilterComparisonDescriptor, ResidentOpDescriptor, ResidentProjectExpressionDescriptor,
30    ResidentRegionDescriptor, ResidentScheduleDeviceProgram, ResidentScheduleExternalBindings,
31    ResidentScheduleOpKind, ResidentScheduleSlotBinding, ResidentWaveDescriptor,
32    RESIDENT_SCHEDULE_OP_MARK_NOVELTY, RESIDENT_SCHEDULE_OP_MARK_SCHEMA_WINNER,
33    RESIDENT_SCHEDULE_REGION_FINALIZE, RESIDENT_SCHEDULE_REGION_INITIALIZE,
34    RESIDENT_SCHEDULE_REGION_RECURSIVE, RESIDENT_SCHEDULE_REGION_SCC_BEGIN,
35};
36use xlog_cuda::{
37    CompareOp as CudaCompareOp, CudaBuffer, CudaColumn, CudaKernelProvider, CudaStream,
38};
39use xlog_ir::{
40    CompareOp as RirCompareOp, CompiledRule, ConstValue, ExecutionPlan, Expr, JoinType,
41    ProjectExpr, RirNode,
42};
43
44use super::Executor;
45use crate::resident_graph::{
46    ResidentGraphCertifiedPlan, ResidentGraphDeclineReason, ResidentGraphDeviceStatus,
47    ResidentGraphExecutionError, ResidentGraphPrepareDiagnosticSnapshot,
48    ResidentGraphPrepareOptions, ResidentGraphRouteCertificate,
49};
50
51const MAX_RESIDENT_CAPACITY: u32 = 65_536;
52const RESIDENT_DYNAMIC_SCHEMA_ID: u32 = u32::MAX;
53
54fn resident_schemas_type_compatible(left: &Schema, right: &Schema) -> bool {
55    left.arity() == right.arity()
56        && (0..left.arity()).all(|column| left.column_type(column) == right.column_type(column))
57}
58
59fn resident_intern_schema(candidates: &mut Vec<Schema>, schema: Schema) -> Result<u32> {
60    let index = candidates
61        .iter()
62        .position(|candidate| candidate == &schema)
63        .unwrap_or_else(|| {
64            candidates.push(schema);
65            candidates.len() - 1
66        });
67    u32::try_from(index)
68        .map_err(|_| XlogError::Execution("resident schema candidate count exceeds u32".into()))
69}
70#[derive(Debug, Clone)]
71pub(super) struct ResidentWorkspaceAdmission {
72    pub(super) relation_capacity: u32,
73    pub(super) head_schemas: BTreeMap<String, Schema>,
74    head_schema_choices: BTreeMap<String, Vec<Schema>>,
75    rule_schema_ids: Vec<Vec<Option<u32>>>,
76    head_schema_selections: BTreeMap<String, ResidentHeadSchemaSelection>,
77}
78
79#[derive(Debug, Clone)]
80struct ResidentHeadSchemaSelection {
81    source_head: String,
82    output_schemas_by_source_winner: Vec<Schema>,
83}
84
85#[derive(Debug, Clone)]
86enum ResidentSchemaVariants {
87    Fixed(Schema),
88    Dynamic {
89        source_head: String,
90        schemas: Vec<Schema>,
91    },
92}
93
94fn resident_schema_variants_from_source(
95    source_head: String,
96    schemas: Vec<Schema>,
97) -> std::result::Result<ResidentSchemaVariants, ResidentGraphDeclineReason> {
98    let Some(first) = schemas.first() else {
99        return Err(resident_workspace_decline(format!(
100            "resident schema source {source_head} has no admitted candidate"
101        )));
102    };
103    if schemas.iter().all(|schema| schema == first) {
104        Ok(ResidentSchemaVariants::Fixed(first.clone()))
105    } else {
106        Ok(ResidentSchemaVariants::Dynamic {
107            source_head,
108            schemas,
109        })
110    }
111}
112
113fn resident_register_schema_selection(
114    target_head: &str,
115    selection: ResidentHeadSchemaSelection,
116    head_schema_choices: &BTreeMap<String, Vec<Schema>>,
117    head_schema_selections: &mut BTreeMap<String, ResidentHeadSchemaSelection>,
118) -> std::result::Result<(), ResidentGraphDeclineReason> {
119    if selection.source_head == target_head {
120        return Err(resident_workspace_decline(format!(
121            "resident schema lineage for {target_head} contains a cycle"
122        )));
123    }
124    let source_candidates = head_schema_choices
125        .get(&selection.source_head)
126        .filter(|candidates| !candidates.is_empty())
127        .ok_or_else(|| {
128            resident_workspace_decline(format!(
129                "resident schema source {} for {target_head} has no admitted candidate",
130                selection.source_head
131            ))
132        })?;
133    if source_candidates.len() != selection.output_schemas_by_source_winner.len() {
134        return Err(resident_workspace_decline(format!(
135            "resident schema source {} for {target_head} has {} candidates but {} mappings",
136            selection.source_head,
137            source_candidates.len(),
138            selection.output_schemas_by_source_winner.len()
139        )));
140    }
141
142    let mut source = selection.source_head.as_str();
143    let mut seen = HashSet::new();
144    while let Some(parent) = head_schema_selections.get(source) {
145        if source == target_head || !seen.insert(source) {
146            return Err(resident_workspace_decline(format!(
147                "resident schema lineage for {target_head} contains a cycle"
148            )));
149        }
150        source = parent.source_head.as_str();
151    }
152    if source == target_head {
153        return Err(resident_workspace_decline(format!(
154            "resident schema lineage for {target_head} contains a cycle"
155        )));
156    }
157
158    if let Some(existing) = head_schema_selections.get(target_head) {
159        if existing.source_head != selection.source_head
160            || existing.output_schemas_by_source_winner != selection.output_schemas_by_source_winner
161        {
162            return Err(resident_workspace_decline(format!(
163                "resident head {target_head} has multiple schema sources or mappings"
164            )));
165        }
166        return Ok(());
167    }
168    head_schema_selections.insert(target_head.to_string(), selection);
169    Ok(())
170}
171
172#[derive(Debug, Clone)]
173enum ResidentOutputSchemaSelection {
174    OwnWinner,
175    SourceWinner {
176        source_output: usize,
177        schemas: Vec<Schema>,
178    },
179}
180
181#[derive(Debug, Clone)]
182struct ResidentOutputSchemaPlan {
183    candidates: Vec<Schema>,
184    selection: ResidentOutputSchemaSelection,
185}
186
187fn resident_resolve_output_schemas(
188    plans: &[ResidentOutputSchemaPlan],
189    winner_ids: &[u32],
190) -> Result<Vec<Schema>> {
191    if plans.len() != winner_ids.len() {
192        return Err(XlogError::Execution(format!(
193            "resident schema plan count {} does not match winner count {}",
194            plans.len(),
195            winner_ids.len()
196        )));
197    }
198    (0..plans.len())
199        .map(|output| {
200            resident_resolve_output_schema(plans, winner_ids, output, &mut HashSet::new())
201        })
202        .collect()
203}
204
205fn resident_resolve_output_schema(
206    plans: &[ResidentOutputSchemaPlan],
207    winner_ids: &[u32],
208    output: usize,
209    resolving: &mut HashSet<usize>,
210) -> Result<Schema> {
211    if !resolving.insert(output) {
212        return Err(XlogError::Execution(
213            "resident schema source lineage contains a cycle".into(),
214        ));
215    }
216    let result = (|| {
217        let plan = plans.get(output).ok_or_else(|| {
218            XlogError::Execution(format!("resident schema output {output} is missing"))
219        })?;
220        let winner = *winner_ids.get(output).ok_or_else(|| {
221            XlogError::Execution(format!("resident schema winner {output} is missing"))
222        })?;
223        if winner != RESIDENT_DYNAMIC_SCHEMA_ID {
224            return plan
225                .candidates
226                .get(winner as usize)
227                .cloned()
228                .ok_or_else(|| {
229                    XlogError::Execution(format!(
230                        "resident schema winner {winner} exceeds {} candidates",
231                        plan.candidates.len()
232                    ))
233                });
234        }
235        let ResidentOutputSchemaSelection::SourceWinner {
236            source_output,
237            schemas,
238        } = &plan.selection
239        else {
240            return Err(XlogError::Execution(
241                "resident dynamic schema winner has no source lineage".into(),
242            ));
243        };
244        let source_schema =
245            resident_resolve_output_schema(plans, winner_ids, *source_output, resolving)?;
246        let source_plan = plans
247            .get(*source_output)
248            .ok_or_else(|| XlogError::Execution("resident schema source plan is missing".into()))?;
249        let source_winner = source_plan
250            .candidates
251            .iter()
252            .position(|candidate| candidate == &source_schema)
253            .ok_or_else(|| {
254                XlogError::Execution(
255                    "resident resolved source schema has no candidate index".into(),
256                )
257            })?;
258        schemas.get(source_winner).cloned().ok_or_else(|| {
259            XlogError::Execution(format!(
260                "resident schema source winner {source_winner} exceeds {} mappings",
261                schemas.len()
262            ))
263        })
264    })();
265    resolving.remove(&output);
266    result
267}
268
269struct ResidentRunOwners {
270    provider: Arc<CudaKernelProvider>,
271    runtime: Arc<XlogDeviceRuntime>,
272    stream: Arc<CudaStream>,
273    // Field drop order is intentional: destroy the graph before its compact
274    // program/domain and every externally owned device allocation.
275    graph: CapturedCudaGraph,
276    execution_domain: ResidentExecutionDomain,
277    schedule_program: ResidentScheduleDeviceProgram,
278    recorder: LaunchRecorder,
279    relations: Vec<Option<ResidentRelation>>,
280    output_indices: Vec<(String, usize)>,
281    filter_scratch: Option<ResidentFilterScratch>,
282    set_workspace: ResidentSetWorkspace,
283    join_workspace: ResidentJoinWorkspace,
284    control: ResidentConvergenceControl,
285    device_trace: ResidentDeviceTrace,
286    schema_winners: ResidentSchemaWinners,
287    receipt: ResidentPackedReceipt,
288    pinned_receipt: ResidentPinnedReceipt,
289    source_epoch: u64,
290    relation_registration: Vec<(RelId, String)>,
291    transaction_identity: Arc<()>,
292    output_schema_plans: Vec<ResidentOutputSchemaPlan>,
293}
294
295#[derive(Clone, Debug, PartialEq, Eq)]
296struct ResidentSourceSetSnapshot {
297    name: String,
298    version: u64,
299    schema: Schema,
300    row_capacity: u64,
301    column_blocks: Vec<Option<BlockId>>,
302    row_count_block: BlockId,
303}
304
305#[derive(Clone)]
306enum ResidentBufferRef {
307    Source(String),
308    Private(usize),
309}
310
311enum ResidentRecordedOp {
312    Unit {
313        output: usize,
314        op_id: u32,
315    },
316    Scan {
317        relation: ResidentBufferRef,
318        op_id: u32,
319    },
320    TraceDelta {
321        scan_delta: u32,
322        filter_delta: u32,
323        semantic_guard: Option<ResidentBufferRef>,
324    },
325    Clear {
326        output: usize,
327    },
328    Filter {
329        input: ResidentBufferRef,
330        output: usize,
331        workspace: usize,
332        op_id: u32,
333    },
334    Project {
335        input: ResidentBufferRef,
336        output: usize,
337        workspace: usize,
338        op_id: u32,
339    },
340    Union {
341        left: ResidentBufferRef,
342        right: ResidentBufferRef,
343        output: usize,
344        op_id: u32,
345    },
346    Diff {
347        left: ResidentBufferRef,
348        right: ResidentBufferRef,
349        output: usize,
350        op_id: u32,
351    },
352    Join {
353        kind: ResidentJoinKind,
354        left: ResidentBufferRef,
355        left_key: usize,
356        right: ResidentBufferRef,
357        right_key: usize,
358        output: usize,
359        op_id: u32,
360    },
361    ChangedReset,
362    ChangedMark {
363        relation: usize,
364    },
365    TestStatus(ResidentTerminalStatus),
366    SchemaWinnerMark {
367        contribution: ResidentBufferRef,
368        head_index: u32,
369        schema_id: u32,
370    },
371}
372
373fn resident_record_unit_leaf(
374    output: usize,
375    op_id: u32,
376    ops: &mut Vec<ResidentRecordedOp>,
377    push_physical: impl FnOnce(&mut Vec<ResidentRecordedOp>, ResidentRecordedOp, u32),
378) -> ResidentBufferRef {
379    push_physical(ops, ResidentRecordedOp::Unit { output, op_id }, op_id);
380    ResidentBufferRef::Private(output)
381}
382
383fn resident_new_phase_unit(
384    relations: &mut Vec<ResidentLogicalRelation>,
385    op_id: u32,
386) -> Result<(ResidentBufferRef, ResidentRecordedOp)> {
387    let output = relations.len();
388    relations.push(ResidentLogicalRelation {
389        schema: Schema::new(Vec::new()),
390        initial_count: 0,
391        permanent: false,
392    });
393    Ok((
394        ResidentBufferRef::Private(output),
395        ResidentRecordedOp::Unit { output, op_id },
396    ))
397}
398
399fn resident_record_scan_leaf(
400    relation: ResidentBufferRef,
401    op_id: u32,
402    semantic_guard: Option<ResidentBufferRef>,
403    ops: &mut Vec<ResidentRecordedOp>,
404    push_physical: impl FnOnce(&mut Vec<ResidentRecordedOp>, ResidentRecordedOp, u32),
405) -> ResidentBufferRef {
406    push_physical(
407        ops,
408        ResidentRecordedOp::Scan {
409            relation: relation.clone(),
410            op_id,
411        },
412        op_id,
413    );
414    ops.push(ResidentRecordedOp::TraceDelta {
415        scan_delta: 1,
416        filter_delta: 0,
417        semantic_guard,
418    });
419    relation
420}
421
422fn resident_semantic_trace_guard(
423    override_scan: Option<(RelId, usize, usize)>,
424) -> Option<ResidentBufferRef> {
425    override_scan.map(|(_, _, delta)| ResidentBufferRef::Private(delta))
426}
427
428fn resident_record_schema_winner_mark(
429    ops: &mut Vec<ResidentRecordedOp>,
430    contribution: ResidentBufferRef,
431    head_index: u32,
432    schema_id: u32,
433) {
434    ops.push(ResidentRecordedOp::SchemaWinnerMark {
435        contribution,
436        head_index,
437        schema_id,
438    });
439}
440
441#[derive(Debug, Clone, Copy, PartialEq, Eq)]
442enum ResidentUnionFoldMode {
443    SelfUnion,
444    LeftAssociated,
445}
446
447fn resident_union_fold_mode(input_count: usize) -> Result<ResidentUnionFoldMode> {
448    match input_count {
449        0 => Err(XlogError::Execution("resident union has no inputs".into())),
450        1 => Ok(ResidentUnionFoldMode::SelfUnion),
451        _ => Ok(ResidentUnionFoldMode::LeftAssociated),
452    }
453}
454
455enum ResidentPhaseMergeStep<T> {
456    Deduplicate(T),
457    Union(T, T),
458}
459
460fn resident_phase_merge<T, E>(
461    current: Option<T>,
462    contribution: T,
463    mut execute: impl FnMut(ResidentPhaseMergeStep<T>) -> std::result::Result<T, E>,
464) -> std::result::Result<T, E> {
465    let contribution = execute(ResidentPhaseMergeStep::Deduplicate(contribution))?;
466    match current {
467        Some(current) => execute(ResidentPhaseMergeStep::Union(current, contribution)),
468        None => Ok(contribution),
469    }
470}
471
472enum ResidentCapturePhase {
473    Segment {
474        ops: Vec<ResidentRecordedOp>,
475        scc_begin: Option<(u32, u32)>,
476    },
477    ConditionalWhile {
478        ops: Vec<ResidentRecordedOp>,
479        iteration_limit: u32,
480        convergence_op_id: u32,
481    },
482}
483
484struct ResidentCompactLogicalRegion {
485    ops: Vec<ResidentRecordedOp>,
486    iteration_limit: u32,
487    op_id: u32,
488    flags: u32,
489}
490
491struct ResidentCompactSchedulePlan {
492    source_slots: BTreeMap<String, u32>,
493    ops: Vec<ResidentOpDescriptor>,
494    waves: Vec<ResidentWaveDescriptor>,
495    regions: Vec<ResidentRegionDescriptor>,
496    generation_bases: Vec<u32>,
497    filter_comparisons: Vec<ResidentFilterComparisonDescriptor>,
498    project_expressions: Vec<ResidentProjectExpressionDescriptor>,
499}
500
501#[derive(Debug, Clone, Copy, PartialEq, Eq)]
502enum ResidentCaptureParentKind {
503    Kernel,
504    Conditional,
505}
506
507struct ResidentCompactTopology {
508    parent_kinds: Vec<ResidentCaptureParentKind>,
509    conditional_body_kernel_counts: Vec<usize>,
510    hierarchical_node_count: usize,
511}
512
513trait ResidentRegionFlags {
514    fn region_flags(&self) -> u32;
515}
516
517impl ResidentRegionFlags for ResidentCompactLogicalRegion {
518    fn region_flags(&self) -> u32 {
519        self.flags
520    }
521}
522
523impl ResidentRegionFlags for ResidentRegionDescriptor {
524    fn region_flags(&self) -> u32 {
525        self.flags
526    }
527}
528
529fn resident_compact_topology<R: ResidentRegionFlags>(
530    regions: &[R],
531) -> Result<ResidentCompactTopology> {
532    let parent_kinds = regions
533        .iter()
534        .map(|region| {
535            if region.region_flags() == RESIDENT_SCHEDULE_REGION_RECURSIVE {
536                ResidentCaptureParentKind::Conditional
537            } else {
538                ResidentCaptureParentKind::Kernel
539            }
540        })
541        .collect::<Vec<_>>();
542    let conditional_count = parent_kinds
543        .iter()
544        .filter(|kind| **kind == ResidentCaptureParentKind::Conditional)
545        .count();
546    let hierarchical_node_count = parent_kinds
547        .len()
548        .checked_add(conditional_count)
549        .ok_or_else(|| XlogError::Execution("resident topology node count overflow".into()))?;
550    Ok(ResidentCompactTopology {
551        parent_kinds,
552        conditional_body_kernel_counts: vec![1; conditional_count],
553        hierarchical_node_count,
554    })
555}
556
557fn resident_validate_parent_graph_kinds(
558    actual: &[CudaGraphNodeKind],
559    expected: &[ResidentCaptureParentKind],
560) -> Result<()> {
561    if actual.len() != expected.len()
562        || actual.iter().zip(expected).any(|(actual, expected)| {
563            !matches!(
564                (actual, expected),
565                (CudaGraphNodeKind::Kernel, ResidentCaptureParentKind::Kernel)
566                    | (
567                        CudaGraphNodeKind::Conditional,
568                        ResidentCaptureParentKind::Conditional
569                    )
570            )
571        })
572    {
573        return Err(XlogError::Execution(
574            "resident parent graph topology differs from the compact schedule".into(),
575        ));
576    }
577    Ok(())
578}
579
580fn resident_validate_conditional_body_node_kinds(
581    actual: &[Vec<CudaGraphNodeKind>],
582    expected_conditional_count: usize,
583) -> Result<Vec<usize>> {
584    if actual.len() != expected_conditional_count {
585        return Err(XlogError::Execution(format!(
586            "resident conditional body inventory has {} bodies, expected {expected_conditional_count}",
587            actual.len()
588        )));
589    }
590    let mut kernel_counts = Vec::with_capacity(actual.len());
591    for (index, kinds) in actual.iter().enumerate() {
592        if kinds.as_slice() != [CudaGraphNodeKind::Kernel] {
593            return Err(XlogError::Execution(format!(
594                "resident conditional body {index} must contain exactly one kernel node, found {kinds:?}"
595            )));
596        }
597        kernel_counts.push(1);
598    }
599    Ok(kernel_counts)
600}
601
602#[derive(Default)]
603struct ResidentCompactDescriptorTables {
604    filter_comparisons: Vec<ResidentFilterComparisonDescriptor>,
605    filter_ranges: Vec<(u32, u32)>,
606    project_expressions: Vec<ResidentProjectExpressionDescriptor>,
607    project_ranges: Vec<(u32, u32)>,
608}
609
610#[allow(clippy::too_many_arguments)]
611fn resident_compact_schedule_metadata_bytes(
612    slot_count: usize,
613    op_count: usize,
614    wave_count: usize,
615    region_count: usize,
616    generation_base_count: usize,
617    head_count: usize,
618    filter_comparison_count: usize,
619    project_expression_count: usize,
620) -> Result<u64> {
621    let generation_metadata_count =
622        generation_base_count
623            .checked_add(head_count)
624            .ok_or_else(|| {
625                XlogError::Execution("resident generation metadata count overflow".into())
626            })?;
627    resident_schedule_metadata_device_bytes(
628        slot_count,
629        op_count,
630        wave_count,
631        region_count,
632        generation_metadata_count,
633        filter_comparison_count,
634        project_expression_count,
635    )
636}
637
638fn resident_compact_schema_defaults(
639    ops: &[ResidentOpDescriptor],
640    head_count: usize,
641) -> Result<Vec<u32>> {
642    let mut defaults = vec![None; head_count];
643    for op in ops {
644        if op.flags & RESIDENT_SCHEDULE_OP_MARK_SCHEMA_WINNER == 0 {
645            continue;
646        }
647        let head = usize::try_from(op.schema_winner_head).map_err(|_| {
648            XlogError::Execution("resident schema winner head index overflow".into())
649        })?;
650        let default = defaults.get_mut(head).ok_or_else(|| {
651            XlogError::Execution("resident schema winner head is out of range".into())
652        })?;
653        if default.is_none() {
654            *default = Some(op.schema_winner_id);
655        }
656    }
657    defaults
658        .into_iter()
659        .enumerate()
660        .map(|(head, default)| {
661            default.ok_or_else(|| {
662                XlogError::Execution(format!(
663                    "resident head {head} has no schema winner candidate"
664                ))
665            })
666        })
667        .collect()
668}
669
670fn resident_compact_allocation_bytes(
671    relation_bytes: u64,
672    filter_scratch_bytes: u64,
673    fixed_workspace_bytes: u64,
674    private_slot_count: usize,
675    head_count: usize,
676    plan: &ResidentCompactSchedulePlan,
677) -> Result<(u64, u64)> {
678    let slot_count = private_slot_count
679        .checked_add(plan.source_slots.len())
680        .ok_or_else(|| XlogError::Execution("resident compact slot count overflow".into()))?;
681    let metadata_bytes = resident_compact_schedule_metadata_bytes(
682        slot_count,
683        plan.ops.len(),
684        plan.waves.len(),
685        plan.regions.len(),
686        plan.generation_bases.len(),
687        head_count,
688        plan.filter_comparisons.len(),
689        plan.project_expressions.len(),
690    )?;
691    let required_bytes = relation_bytes
692        .checked_add(filter_scratch_bytes)
693        .and_then(|bytes| bytes.checked_add(fixed_workspace_bytes))
694        .and_then(|bytes| bytes.checked_add(metadata_bytes))
695        .ok_or_else(|| XlogError::Execution("resident compact manifest overflow".into()))?;
696    Ok((required_bytes, metadata_bytes))
697}
698
699fn resident_compact_preflight_device_bytes(
700    manifest: &ResidentAllocationManifest,
701    plan: &ResidentCompactSchedulePlan,
702) -> Result<(u64, u64, u64)> {
703    let filter_count = u64::try_from(plan.filter_comparisons.len().max(1)).map_err(|_| {
704        XlogError::Execution("resident compact filter descriptor count exceeds u64".into())
705    })?;
706    let project_count = u64::try_from(plan.project_expressions.len().max(1)).map_err(|_| {
707        XlogError::Execution("resident compact project descriptor count exceeds u64".into())
708    })?;
709    let filter_bytes = u64::try_from(std::mem::size_of::<ResidentFilterComparisonDescriptor>())
710        .ok()
711        .and_then(|size| size.checked_mul(filter_count))
712        .ok_or_else(|| {
713            XlogError::Execution("resident compact filter descriptor bytes overflow".into())
714        })?;
715    let project_bytes = u64::try_from(std::mem::size_of::<ResidentProjectExpressionDescriptor>())
716        .ok()
717        .and_then(|size| size.checked_mul(project_count))
718        .ok_or_else(|| {
719            XlogError::Execution("resident compact project descriptor bytes overflow".into())
720        })?;
721    let descriptor_bytes = filter_bytes
722        .checked_add(project_bytes)
723        .ok_or_else(|| XlogError::Execution("resident compact descriptor bytes overflow".into()))?;
724    let remaining_metadata_bytes = manifest
725        .schedule_metadata_bytes
726        .checked_sub(descriptor_bytes)
727        .ok_or_else(|| {
728            XlogError::Execution(
729                "resident compact descriptor bytes exceed schedule metadata".into(),
730            )
731        })?;
732    let fixed_bytes = manifest
733        .fixed_workspace_bytes
734        .checked_add(remaining_metadata_bytes)
735        .ok_or_else(|| XlogError::Execution("resident compact fixed bytes overflow".into()))?;
736    let reported_total = manifest
737        .relation_bytes
738        .checked_add(manifest.filter_scratch_bytes)
739        .and_then(|bytes| bytes.checked_add(filter_bytes))
740        .and_then(|bytes| bytes.checked_add(project_bytes))
741        .and_then(|bytes| bytes.checked_add(fixed_bytes))
742        .ok_or_else(|| XlogError::Execution("resident compact preflight bytes overflow".into()))?;
743    if reported_total != manifest.required_bytes {
744        return Err(XlogError::Execution(format!(
745            "resident compact preflight component mismatch: reported {reported_total}, required {}",
746            manifest.required_bytes
747        )));
748    }
749    Ok((filter_bytes, project_bytes, fixed_bytes))
750}
751
752fn resident_compact_tables(
753    filter_workspaces: &[ResidentFilterPlan],
754    project_workspaces: &[ResidentProjectPlan],
755) -> Result<ResidentCompactDescriptorTables> {
756    let mut tables = ResidentCompactDescriptorTables::default();
757    for workspace in filter_workspaces {
758        let offset = u32::try_from(tables.filter_comparisons.len()).map_err(|_| {
759            XlogError::Execution("resident filter comparison offset exceeds u32".into())
760        })?;
761        let count = u32::try_from(workspace.compact_comparisons.len()).map_err(|_| {
762            XlogError::Execution("resident filter comparison count exceeds u32".into())
763        })?;
764        tables
765            .filter_comparisons
766            .extend_from_slice(&workspace.compact_comparisons);
767        tables.filter_ranges.push((offset, count));
768    }
769    for workspace in project_workspaces {
770        let offset = u32::try_from(tables.project_expressions.len()).map_err(|_| {
771            XlogError::Execution("resident project expression offset exceeds u32".into())
772        })?;
773        let count = u32::try_from(workspace.compact_expressions.len()).map_err(|_| {
774            XlogError::Execution("resident project expression count exceeds u32".into())
775        })?;
776        tables
777            .project_expressions
778            .extend_from_slice(&workspace.compact_expressions);
779        tables.project_ranges.push((offset, count));
780    }
781    Ok(tables)
782}
783
784#[derive(Debug, Clone, Copy, PartialEq, Eq)]
785struct ResidentCompactSlotRef {
786    slot: u32,
787    generation: u32,
788}
789
790fn resident_compact_slot_ref(
791    reference: &ResidentBufferRef,
792    assignments: &[ResidentSlotAssignment],
793    source_slots: &BTreeMap<String, u32>,
794) -> Result<ResidentCompactSlotRef> {
795    match reference {
796        ResidentBufferRef::Private(logical) => {
797            let assignment = assignments.get(*logical).ok_or_else(|| {
798                XlogError::Execution(format!(
799                    "resident logical relation {logical} has no compact slot assignment"
800                ))
801            })?;
802            Ok(ResidentCompactSlotRef {
803                slot: u32::try_from(assignment.slot).map_err(|_| {
804                    XlogError::Execution("resident compact slot index exceeds u32".into())
805                })?,
806                generation: assignment.generation,
807            })
808        }
809        ResidentBufferRef::Source(name) => Ok(ResidentCompactSlotRef {
810            slot: *source_slots.get(name).ok_or_else(|| {
811                XlogError::Execution(format!(
812                    "resident source {name} has no compact slot assignment"
813                ))
814            })?,
815            generation: 0,
816        }),
817    }
818}
819
820#[allow(clippy::too_many_arguments)]
821fn resident_lower_compact_regions<'a>(
822    logical_regions: Vec<ResidentCompactLogicalRegion>,
823    physical_slots: &[ResidentPhysicalSlotPlan],
824    assignments: &[ResidentSlotAssignment],
825    source_names: impl IntoIterator<Item = &'a str>,
826    tables: ResidentCompactDescriptorTables,
827) -> Result<ResidentCompactSchedulePlan> {
828    let source_slots = resident_source_slot_map(physical_slots.len(), source_names)?;
829    let slot_count = physical_slots
830        .len()
831        .checked_add(source_slots.len())
832        .ok_or_else(|| XlogError::Execution("resident compact slot count overflow".into()))?;
833    let slot_count_u32 = u32::try_from(slot_count)
834        .map_err(|_| XlogError::Execution("resident compact slot count exceeds u32".into()))?;
835    let mut ops = Vec::new();
836    let mut waves = Vec::new();
837    let mut regions = Vec::with_capacity(logical_regions.len());
838    let filter_total = u32::try_from(tables.filter_comparisons.len())
839        .map_err(|_| XlogError::Execution("resident filter comparison count exceeds u32".into()))?;
840    let project_total = u32::try_from(tables.project_expressions.len()).map_err(|_| {
841        XlogError::Execution("resident project expression count exceeds u32".into())
842    })?;
843    let mut filter_cursor = 0_u32;
844    let mut project_cursor = 0_u32;
845    let mut generation_bases = Vec::with_capacity(
846        logical_regions
847            .len()
848            .checked_mul(slot_count)
849            .ok_or_else(|| {
850                XlogError::Execution("resident generation baseline count overflow".into())
851            })?,
852    );
853
854    for logical_region in logical_regions {
855        let region_flags = logical_region.flags;
856        let first_wave = u32::try_from(waves.len())
857            .map_err(|_| XlogError::Execution("resident wave count exceeds u32".into()))?;
858        let generation_offset = u32::try_from(generation_bases.len()).map_err(|_| {
859            XlogError::Execution("resident generation baseline offset exceeds u32".into())
860        })?;
861        let mut first_generations = vec![None; slot_count];
862        let mut last_relation_descriptor = None::<(usize, ResidentCompactSlotRef)>;
863
864        for (logical_op_index, logical_op) in logical_region.ops.into_iter().enumerate() {
865            let mut emit = |descriptor: ResidentOpDescriptor,
866                            output: ResidentCompactSlotRef,
867                            references: Vec<ResidentCompactSlotRef>|
868             -> Result<()> {
869                for reference in &references {
870                    let generation = first_generations
871                        .get_mut(reference.slot as usize)
872                        .ok_or_else(|| {
873                            XlogError::Execution(
874                                "resident compact operation slot is out of range".into(),
875                            )
876                        })?;
877                    if generation.is_none() {
878                        *generation = Some(reference.generation);
879                    }
880                }
881                let op_index = ops.len();
882                let first_op = u32::try_from(op_index).map_err(|_| {
883                    XlogError::Execution("resident compact op count exceeds u32".into())
884                })?;
885                ops.push(descriptor);
886                waves.push(ResidentWaveDescriptor {
887                    first_op,
888                    op_count: 1,
889                    flags: 0,
890                    reserved: 0,
891                });
892                last_relation_descriptor = Some((op_index, output));
893                Ok(())
894            };
895
896            match logical_op {
897                ResidentRecordedOp::Unit { output, op_id } => {
898                    let output = resident_compact_slot_ref(
899                        &ResidentBufferRef::Private(output),
900                        assignments,
901                        &source_slots,
902                    )?;
903                    emit(
904                        ResidentOpDescriptor::unit(op_id, output.slot, output.generation),
905                        output,
906                        vec![output],
907                    )?;
908                }
909                ResidentRecordedOp::Scan { relation, op_id } => {
910                    let source = resident_compact_slot_ref(&relation, assignments, &source_slots)?;
911                    emit(
912                        ResidentOpDescriptor::scan(op_id, source.slot, source.generation),
913                        source,
914                        vec![source],
915                    )?;
916                }
917                op @ (ResidentRecordedOp::Filter { .. } | ResidentRecordedOp::Project { .. }) => {
918                    let (is_filter, input, output, workspace, op_id) = match op {
919                        ResidentRecordedOp::Filter {
920                            input,
921                            output,
922                            workspace,
923                            op_id,
924                        } => (true, input, output, workspace, op_id),
925                        ResidentRecordedOp::Project {
926                            input,
927                            output,
928                            workspace,
929                            op_id,
930                        } => (false, input, output, workspace, op_id),
931                        _ => unreachable!("compact filter/project arm"),
932                    };
933                    let input = resident_compact_slot_ref(&input, assignments, &source_slots)?;
934                    let output = resident_compact_slot_ref(
935                        &ResidentBufferRef::Private(output),
936                        assignments,
937                        &source_slots,
938                    )?;
939                    let (aux_offset, aux_count) = if is_filter {
940                        *tables.filter_ranges.get(workspace).ok_or_else(|| {
941                            XlogError::Execution(
942                                "resident compact filter workspace is out of range".into(),
943                            )
944                        })?
945                    } else {
946                        *tables.project_ranges.get(workspace).ok_or_else(|| {
947                            XlogError::Execution(
948                                "resident compact project workspace is out of range".into(),
949                            )
950                        })?
951                    };
952                    let (cursor, total, table_name) = if is_filter {
953                        (&mut filter_cursor, filter_total, "filter")
954                    } else {
955                        (&mut project_cursor, project_total, "project")
956                    };
957                    if aux_offset != *cursor || aux_offset > total || aux_count > total - aux_offset
958                    {
959                        return Err(XlogError::Execution(format!(
960                            "resident compact {table_name} descriptor range is not contiguous"
961                        )));
962                    }
963                    *cursor = aux_offset + aux_count;
964                    emit(
965                        ResidentOpDescriptor {
966                            kind: if is_filter {
967                                ResidentScheduleOpKind::Filter
968                            } else {
969                                ResidentScheduleOpKind::Project
970                            },
971                            op_id,
972                            out: output.slot,
973                            in0: input.slot,
974                            in0_generation: input.generation,
975                            out_generation: output.generation,
976                            aux_offset,
977                            aux_count,
978                            ..Default::default()
979                        },
980                        output,
981                        vec![input, output],
982                    )?;
983                }
984                op @ (ResidentRecordedOp::Union { .. } | ResidentRecordedOp::Diff { .. }) => {
985                    let (is_union, left, right, output, op_id) = match op {
986                        ResidentRecordedOp::Union {
987                            left,
988                            right,
989                            output,
990                            op_id,
991                        } => (true, left, right, output, op_id),
992                        ResidentRecordedOp::Diff {
993                            left,
994                            right,
995                            output,
996                            op_id,
997                        } => (false, left, right, output, op_id),
998                        _ => unreachable!("compact set arm"),
999                    };
1000                    let left = resident_compact_slot_ref(&left, assignments, &source_slots)?;
1001                    let right = resident_compact_slot_ref(&right, assignments, &source_slots)?;
1002                    let output = resident_compact_slot_ref(
1003                        &ResidentBufferRef::Private(output),
1004                        assignments,
1005                        &source_slots,
1006                    )?;
1007                    emit(
1008                        ResidentOpDescriptor {
1009                            kind: if is_union {
1010                                ResidentScheduleOpKind::Union
1011                            } else {
1012                                ResidentScheduleOpKind::Diff
1013                            },
1014                            op_id,
1015                            out: output.slot,
1016                            in0: left.slot,
1017                            in1: right.slot,
1018                            in0_generation: left.generation,
1019                            in1_generation: right.generation,
1020                            out_generation: output.generation,
1021                            ..Default::default()
1022                        },
1023                        output,
1024                        vec![left, right, output],
1025                    )?;
1026                }
1027                ResidentRecordedOp::Join {
1028                    kind,
1029                    left,
1030                    left_key,
1031                    right,
1032                    right_key,
1033                    output,
1034                    op_id,
1035                } => {
1036                    let left = resident_compact_slot_ref(&left, assignments, &source_slots)?;
1037                    let right = resident_compact_slot_ref(&right, assignments, &source_slots)?;
1038                    let output = resident_compact_slot_ref(
1039                        &ResidentBufferRef::Private(output),
1040                        assignments,
1041                        &source_slots,
1042                    )?;
1043                    emit(
1044                        ResidentOpDescriptor {
1045                            kind: match kind {
1046                                ResidentJoinKind::Inner => ResidentScheduleOpKind::JoinInner,
1047                                ResidentJoinKind::Semi => ResidentScheduleOpKind::JoinSemi,
1048                            },
1049                            op_id,
1050                            out: output.slot,
1051                            in0: left.slot,
1052                            in1: right.slot,
1053                            in0_generation: left.generation,
1054                            in1_generation: right.generation,
1055                            out_generation: output.generation,
1056                            left_key: u32::try_from(left_key).map_err(|_| {
1057                                XlogError::Execution(
1058                                    "resident compact left join key exceeds u32".into(),
1059                                )
1060                            })?,
1061                            right_key: u32::try_from(right_key).map_err(|_| {
1062                                XlogError::Execution(
1063                                    "resident compact right join key exceeds u32".into(),
1064                                )
1065                            })?,
1066                            ..Default::default()
1067                        },
1068                        output,
1069                        vec![left, right, output],
1070                    )?;
1071                }
1072                ResidentRecordedOp::TraceDelta {
1073                    scan_delta,
1074                    filter_delta,
1075                    semantic_guard,
1076                } => {
1077                    let semantic_guard = semantic_guard
1078                        .as_ref()
1079                        .map(|reference| {
1080                            resident_compact_slot_ref(reference, assignments, &source_slots)
1081                        })
1082                        .transpose()?;
1083                    if let Some(reference) = semantic_guard {
1084                        let generation = first_generations
1085                            .get_mut(reference.slot as usize)
1086                            .ok_or_else(|| {
1087                                XlogError::Execution(
1088                                    "resident compact trace guard slot is out of range".into(),
1089                                )
1090                            })?;
1091                        if generation.is_none() {
1092                            *generation = Some(reference.generation);
1093                        }
1094                    }
1095                    let first_op = u32::try_from(ops.len()).map_err(|_| {
1096                        XlogError::Execution("resident compact op count exceeds u32".into())
1097                    })?;
1098                    ops.push(ResidentOpDescriptor::trace_delta(
1099                        scan_delta,
1100                        filter_delta,
1101                        semantic_guard.map(|reference| (reference.slot, reference.generation)),
1102                    ));
1103                    waves.push(ResidentWaveDescriptor {
1104                        first_op,
1105                        op_count: 1,
1106                        flags: 0,
1107                        reserved: 0,
1108                    });
1109                }
1110                ResidentRecordedOp::TestStatus(status) => {
1111                    let first_op = u32::try_from(ops.len()).map_err(|_| {
1112                        XlogError::Execution("resident compact op count exceeds u32".into())
1113                    })?;
1114                    ops.push(ResidentOpDescriptor::test_status(status)?);
1115                    waves.push(ResidentWaveDescriptor {
1116                        first_op,
1117                        op_count: 1,
1118                        flags: 0,
1119                        reserved: 0,
1120                    });
1121                }
1122                ResidentRecordedOp::SchemaWinnerMark {
1123                    contribution,
1124                    head_index,
1125                    schema_id,
1126                } => {
1127                    let contribution =
1128                        resident_compact_slot_ref(&contribution, assignments, &source_slots)?;
1129                    let (index, output) = last_relation_descriptor.as_ref().ok_or_else(|| {
1130                        XlogError::Execution(
1131                            "resident schema marker has no contribution operation".into(),
1132                        )
1133                    })?;
1134                    if *output != contribution {
1135                        return Err(XlogError::Execution(
1136                            "resident schema marker does not match its contribution operation"
1137                                .into(),
1138                        ));
1139                    }
1140                    ops[*index] = ops[*index].with_schema_winner(head_index, schema_id);
1141                }
1142                ResidentRecordedOp::ChangedMark { relation } => {
1143                    let relation = resident_compact_slot_ref(
1144                        &ResidentBufferRef::Private(relation),
1145                        assignments,
1146                        &source_slots,
1147                    )?;
1148                    let (index, output) = last_relation_descriptor.as_ref().ok_or_else(|| {
1149                        XlogError::Execution(
1150                            "resident novelty marker has no completed delta copy".into(),
1151                        )
1152                    })?;
1153                    if *output != relation || ops[*index].kind != ResidentScheduleOpKind::Project {
1154                        return Err(XlogError::Execution(
1155                            "resident novelty marker does not follow its completed delta copy"
1156                                .into(),
1157                        ));
1158                    }
1159                    ops[*index].flags |= RESIDENT_SCHEDULE_OP_MARK_NOVELTY;
1160                }
1161                ResidentRecordedOp::ChangedReset => {
1162                    if region_flags != RESIDENT_SCHEDULE_REGION_RECURSIVE || logical_op_index != 0 {
1163                        return Err(XlogError::Execution(
1164                            "resident changed reset is not the recursive region entry".into(),
1165                        ));
1166                    }
1167                }
1168                ResidentRecordedOp::Clear { .. } => {
1169                    return Err(XlogError::Execution(
1170                        "resident compact SSA still contains a Clear operation".into(),
1171                    ));
1172                }
1173            }
1174        }
1175
1176        for (slot, generation) in first_generations.into_iter().enumerate() {
1177            let baseline = if slot < physical_slots.len() && physical_slots[slot].permanent {
1178                0
1179            } else {
1180                generation.unwrap_or(0)
1181            };
1182            generation_bases.push(baseline);
1183        }
1184        let wave_count = u32::try_from(waves.len())
1185            .map_err(|_| XlogError::Execution("resident wave count exceeds u32".into()))?
1186            .checked_sub(first_wave)
1187            .ok_or_else(|| XlogError::Execution("resident wave range underflow".into()))?;
1188        regions.push(ResidentRegionDescriptor {
1189            first_wave,
1190            wave_count,
1191            iteration_limit: logical_region.iteration_limit,
1192            op_id: logical_region.op_id,
1193            flags: region_flags,
1194            first_slot: 0,
1195            slot_count: slot_count_u32,
1196            generation_offset,
1197        });
1198    }
1199
1200    if filter_cursor != filter_total || project_cursor != project_total {
1201        return Err(XlogError::Execution(
1202            "resident compact descriptor tables are not exactly covered".into(),
1203        ));
1204    }
1205
1206    Ok(ResidentCompactSchedulePlan {
1207        source_slots,
1208        ops,
1209        waves,
1210        regions,
1211        generation_bases,
1212        filter_comparisons: tables.filter_comparisons,
1213        project_expressions: tables.project_expressions,
1214    })
1215}
1216
1217impl ResidentCompactLogicalRegion {
1218    fn initializes(&self) -> bool {
1219        self.flags & RESIDENT_SCHEDULE_REGION_INITIALIZE != 0
1220    }
1221
1222    fn begins_scc(&self) -> bool {
1223        self.flags & RESIDENT_SCHEDULE_REGION_SCC_BEGIN != 0
1224    }
1225
1226    fn recursive(&self) -> bool {
1227        self.flags == RESIDENT_SCHEDULE_REGION_RECURSIVE
1228    }
1229
1230    fn finalizes(&self) -> bool {
1231        self.flags & RESIDENT_SCHEDULE_REGION_FINALIZE != 0
1232    }
1233}
1234
1235fn resident_compact_regions(
1236    initial_ops: Vec<ResidentRecordedOp>,
1237    phases: Vec<ResidentCapturePhase>,
1238    success_op_id: u32,
1239) -> Result<Vec<ResidentCompactLogicalRegion>> {
1240    let mut regions = Vec::new();
1241    let mut pending = initial_ops;
1242    let mut pending_seed_region = None;
1243
1244    for phase in phases {
1245        match phase {
1246            ResidentCapturePhase::Segment {
1247                mut ops,
1248                scc_begin: None,
1249            } => {
1250                if pending_seed_region.is_some() {
1251                    return Err(XlogError::Execution(
1252                        "resident SCC seed is not followed by its recursive body".into(),
1253                    ));
1254                }
1255                pending.append(&mut ops);
1256            }
1257            ResidentCapturePhase::Segment {
1258                mut ops,
1259                scc_begin: Some((iteration_limit, op_id)),
1260            } => {
1261                if pending_seed_region.is_some() {
1262                    return Err(XlogError::Execution(
1263                        "resident SCC seed is not followed by its recursive body".into(),
1264                    ));
1265                }
1266                pending.append(&mut ops);
1267                let mut flags = RESIDENT_SCHEDULE_REGION_SCC_BEGIN;
1268                if regions.is_empty() {
1269                    flags |= RESIDENT_SCHEDULE_REGION_INITIALIZE;
1270                }
1271                regions.push(ResidentCompactLogicalRegion {
1272                    ops: std::mem::take(&mut pending),
1273                    iteration_limit,
1274                    op_id,
1275                    flags,
1276                });
1277                pending_seed_region = Some(regions.len() - 1);
1278            }
1279            ResidentCapturePhase::ConditionalWhile {
1280                ops,
1281                iteration_limit,
1282                convergence_op_id,
1283            } => {
1284                let seed_region = pending_seed_region.take().ok_or_else(|| {
1285                    XlogError::Execution("resident recursive body has no preceding SCC seed".into())
1286                })?;
1287                if regions[seed_region].iteration_limit != iteration_limit {
1288                    return Err(XlogError::Execution(
1289                        "resident SCC seed and recursive body iteration limits differ".into(),
1290                    ));
1291                }
1292                regions[seed_region].op_id = convergence_op_id;
1293                regions.push(ResidentCompactLogicalRegion {
1294                    ops,
1295                    iteration_limit,
1296                    op_id: convergence_op_id,
1297                    flags: RESIDENT_SCHEDULE_REGION_RECURSIVE,
1298                });
1299            }
1300        }
1301    }
1302    if pending_seed_region.is_some() {
1303        return Err(XlogError::Execution(
1304            "resident SCC seed is missing its recursive body".into(),
1305        ));
1306    }
1307
1308    let mut flags = RESIDENT_SCHEDULE_REGION_FINALIZE;
1309    if regions.is_empty() {
1310        flags |= RESIDENT_SCHEDULE_REGION_INITIALIZE;
1311    }
1312    regions.push(ResidentCompactLogicalRegion {
1313        ops: pending,
1314        iteration_limit: 1,
1315        op_id: success_op_id,
1316        flags,
1317    });
1318    Ok(regions)
1319}
1320
1321fn coalesce_resident_capture_phases(
1322    phases: Vec<ResidentCapturePhase>,
1323) -> Vec<ResidentCapturePhase> {
1324    let mut coalesced = Vec::new();
1325    let mut ordinary_ops = Vec::new();
1326    let flush_ordinary = |coalesced: &mut Vec<ResidentCapturePhase>,
1327                          ordinary_ops: &mut Vec<ResidentRecordedOp>| {
1328        if !ordinary_ops.is_empty() {
1329            coalesced.push(ResidentCapturePhase::Segment {
1330                ops: std::mem::take(ordinary_ops),
1331                scc_begin: None,
1332            });
1333        }
1334    };
1335
1336    for phase in phases {
1337        match phase {
1338            ResidentCapturePhase::Segment {
1339                mut ops,
1340                scc_begin: None,
1341            } => ordinary_ops.append(&mut ops),
1342            boundary => {
1343                flush_ordinary(&mut coalesced, &mut ordinary_ops);
1344                coalesced.push(boundary);
1345            }
1346        }
1347    }
1348    flush_ordinary(&mut coalesced, &mut ordinary_ops);
1349    coalesced
1350}
1351
1352#[cfg(test)]
1353mod capture_phase_tests {
1354    use super::{coalesce_resident_capture_phases, ResidentCapturePhase, ResidentRecordedOp};
1355
1356    fn segment(output: usize, scc_begin: Option<(u32, u32)>) -> ResidentCapturePhase {
1357        ResidentCapturePhase::Segment {
1358            ops: vec![ResidentRecordedOp::Clear { output }],
1359            scc_begin,
1360        }
1361    }
1362
1363    fn assert_clear_range(phase: &ResidentCapturePhase, expected: std::ops::Range<usize>) {
1364        let ResidentCapturePhase::Segment {
1365            ops,
1366            scc_begin: None,
1367        } = phase
1368        else {
1369            panic!("expected an ordinary capture segment");
1370        };
1371        assert_eq!(ops.len(), expected.len());
1372        for (op, output) in ops.iter().zip(expected) {
1373            assert!(
1374                matches!(op, ResidentRecordedOp::Clear { output: actual } if *actual == output)
1375            );
1376        }
1377    }
1378
1379    #[test]
1380    fn coalesces_only_maximal_ordinary_phase_runs() {
1381        let mut phases = (0..1_000)
1382            .map(|output| segment(output, None))
1383            .collect::<Vec<_>>();
1384        phases.push(segment(10_000, Some((64, 41))));
1385        phases.extend((1_000..1_500).map(|output| segment(output, None)));
1386        phases.push(ResidentCapturePhase::ConditionalWhile {
1387            ops: vec![ResidentRecordedOp::Clear { output: 20_000 }],
1388            iteration_limit: 64,
1389            convergence_op_id: 42,
1390        });
1391        phases.extend((1_500..1_800).map(|output| segment(output, None)));
1392
1393        let coalesced = coalesce_resident_capture_phases(phases);
1394        assert_eq!(coalesced.len(), 5);
1395        assert_clear_range(&coalesced[0], 0..1_000);
1396        assert!(matches!(
1397            &coalesced[1],
1398            ResidentCapturePhase::Segment {
1399                ops,
1400                scc_begin: Some((64, 41)),
1401            } if matches!(ops.as_slice(), [ResidentRecordedOp::Clear { output: 10_000 }])
1402        ));
1403        assert_clear_range(&coalesced[2], 1_000..1_500);
1404        assert!(matches!(
1405            &coalesced[3],
1406            ResidentCapturePhase::ConditionalWhile {
1407                ops,
1408                iteration_limit: 64,
1409                convergence_op_id: 42,
1410            } if matches!(ops.as_slice(), [ResidentRecordedOp::Clear { output: 20_000 }])
1411        ));
1412        assert_clear_range(&coalesced[4], 1_500..1_800);
1413    }
1414}
1415
1416struct ResidentBuild<'executor> {
1417    executor: &'executor Executor,
1418    certificate: &'executor ResidentGraphRouteCertificate,
1419    capacity: u32,
1420    relations: Vec<ResidentLogicalRelation>,
1421    filter_workspaces: Vec<ResidentFilterPlan>,
1422    project_workspaces: Vec<ResidentProjectPlan>,
1423    heads: BTreeMap<String, usize>,
1424    head_winner_indices: BTreeMap<String, u32>,
1425    source_names: HashSet<String>,
1426    source_aliases: HashMap<String, usize>,
1427    next_op_id: u32,
1428    injection: Option<crate::resident_graph::ResidentGraphDeviceStatusTestInjection>,
1429    injection_recorded: bool,
1430}
1431
1432#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1433enum ResidentSourceBindingRoute {
1434    Direct,
1435    NormalizeEmpty,
1436}
1437
1438fn resident_source_logical_count(cached: Option<u32>) -> Result<u64> {
1439    cached.map(u64::from).ok_or_else(|| {
1440        XlogError::Execution("resident source requires a cold-path cached logical row count".into())
1441    })
1442}
1443
1444fn resident_source_binding_route(
1445    row_count: u64,
1446    count_tracked: bool,
1447    columns_tracked: bool,
1448) -> Result<ResidentSourceBindingRoute> {
1449    if count_tracked && columns_tracked {
1450        return Ok(ResidentSourceBindingRoute::Direct);
1451    }
1452    if row_count == 0 {
1453        return Ok(ResidentSourceBindingRoute::NormalizeEmpty);
1454    }
1455    Err(XlogError::Execution(
1456        "nonempty resident source is not fully runtime tracked".into(),
1457    ))
1458}
1459
1460#[derive(Debug, Clone)]
1461struct ResidentLogicalRelation {
1462    schema: Schema,
1463    initial_count: u32,
1464    permanent: bool,
1465}
1466
1467#[derive(Debug, Clone)]
1468struct ResidentFilterPlan {
1469    compact_comparisons: Vec<ResidentFilterComparisonDescriptor>,
1470}
1471
1472#[derive(Debug, Clone)]
1473struct ResidentProjectPlan {
1474    compact_expressions: Vec<ResidentProjectExpressionDescriptor>,
1475}
1476
1477#[derive(Debug, Clone, Copy)]
1478struct ResidentSlotAssignment {
1479    slot: usize,
1480    generation: u32,
1481}
1482
1483#[derive(Debug, Clone)]
1484struct ResidentPhysicalSlotPlan {
1485    schema: Schema,
1486    initial_count: u32,
1487    permanent: bool,
1488}
1489
1490#[derive(Debug, Clone)]
1491struct ResidentAllocationManifest {
1492    slots: Vec<ResidentPhysicalSlotPlan>,
1493    logical_to_slot: Vec<ResidentSlotAssignment>,
1494    required_bytes: u64,
1495    relation_bytes: u64,
1496    filter_scratch_bytes: u64,
1497    schedule_metadata_bytes: u64,
1498    fixed_workspace_bytes: u64,
1499    logical_relation_values: usize,
1500    permanent_slots: u32,
1501    scratch_slots: u32,
1502    filter_scratch_allocations: u32,
1503    max_row_bytes: u64,
1504}
1505
1506struct ResidentPhysicalBuild {
1507    relations: Vec<Option<ResidentRelation>>,
1508    filter_scratch: Option<ResidentFilterScratch>,
1509    set_workspace: ResidentSetWorkspace,
1510    join_workspace: ResidentJoinWorkspace,
1511    control: ResidentConvergenceControl,
1512}
1513
1514/// Raw fixed-cost counters retained until outer latency sampling is complete.
1515#[doc(hidden)]
1516#[derive(Default)]
1517pub struct ResidentPrepareDiagnostics {
1518    sample: u64,
1519    total_ns: u64,
1520    required_reservation_bytes: u64,
1521    logical_relation_values: usize,
1522    physical_relation_slots: usize,
1523    relation_device_allocation_calls: u64,
1524    compact_ops: usize,
1525    compact_waves: usize,
1526    compact_regions: usize,
1527    conditional_regions: usize,
1528    parent_graph_nodes: usize,
1529    conditional_body_nodes: usize,
1530    admission_and_source_snapshot_ns: u64,
1531    execution_domain_and_build_setup_ns: u64,
1532    logical_schedule_planning_ns: u64,
1533    manifest_compact_construction_ns: u64,
1534    schedule_lowering_ns: u64,
1535    reservation_ns: u64,
1536    relation_slot_allocation_ns: u64,
1537    relation_slot_allocation_ns_max: u64,
1538    relation_slot_allocation_bytes: u64,
1539    count_initialization_ns: u64,
1540    count_initialization_ns_max: u64,
1541    count_memset_calls: u64,
1542    workspace_provider_calls: u64,
1543    filter_scratch_allocation_ns: u64,
1544    filter_scratch_allocation_bytes: u64,
1545    set_workspace_allocation_ns: u64,
1546    set_workspace_allocation_bytes: u64,
1547    join_workspace_allocation_ns: u64,
1548    join_workspace_allocation_bytes: u64,
1549    control_allocation_ns: u64,
1550    control_allocation_bytes: u64,
1551    metadata_binding_construction_ns: u64,
1552    metadata_provider_calls: u64,
1553    device_trace_preparation_ns: u64,
1554    device_trace_reserved_bytes: u64,
1555    device_trace_initial_htod_calls: u64,
1556    device_trace_initial_htod_bytes: u64,
1557    schema_winners_preparation_ns: u64,
1558    schema_winners_reserved_bytes: u64,
1559    schema_winners_initial_htod_calls: u64,
1560    schema_winners_initial_htod_bytes: u64,
1561    receipt_preparation_ns: u64,
1562    receipt_reserved_bytes: u64,
1563    receipt_initial_htod_calls: u64,
1564    receipt_initial_htod_bytes: u64,
1565    schedule_program_preparation_ns: u64,
1566    schedule_program_reserved_bytes: u64,
1567    schedule_program_initial_htod_calls: u64,
1568    schedule_program_initial_htod_bytes: u64,
1569    reservation_validation_and_release_ns: u64,
1570    pinned_receipt_ns: u64,
1571    graph_body_capture_ns: u64,
1572    graph_instantiate_ns: u64,
1573    validation_owner_assembly_ns: u64,
1574}
1575
1576fn resident_prepare_diagnostics_for_sample(
1577    sample: Option<u64>,
1578) -> Option<ResidentPrepareDiagnostics> {
1579    sample.map(|sample| ResidentPrepareDiagnostics {
1580        sample,
1581        ..ResidentPrepareDiagnostics::default()
1582    })
1583}
1584
1585fn resident_prepare_elapsed_ns(started: std::time::Instant) -> u64 {
1586    u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX)
1587}
1588
1589fn resident_schema_winners_initial_htod(default_count: usize) -> (u64, u64) {
1590    if default_count == 0 {
1591        return (0, 0);
1592    }
1593    (
1594        1,
1595        u64::try_from(default_count)
1596            .unwrap_or(u64::MAX)
1597            .saturating_mul(std::mem::size_of::<u32>() as u64),
1598    )
1599}
1600
1601fn resident_receipt_initial_htod(output_count: usize) -> (u64, u64) {
1602    let pointee_count = output_count.saturating_mul(2).saturating_add(4);
1603    if pointee_count == 0 {
1604        return (0, 0);
1605    }
1606    (
1607        1,
1608        u64::try_from(pointee_count)
1609            .unwrap_or(u64::MAX)
1610            .saturating_mul(std::mem::size_of::<u64>() as u64),
1611    )
1612}
1613
1614fn resident_schedule_initial_htod(reserved_bytes: u64) -> (u64, u64) {
1615    (8, reserved_bytes)
1616}
1617
1618impl ResidentPrepareDiagnostics {
1619    pub fn into_snapshot(self) -> ResidentGraphPrepareDiagnosticSnapshot {
1620        let workspace_allocation_ns = self
1621            .filter_scratch_allocation_ns
1622            .saturating_add(self.set_workspace_allocation_ns)
1623            .saturating_add(self.join_workspace_allocation_ns)
1624            .saturating_add(self.control_allocation_ns);
1625        let workspace_allocation_bytes = self
1626            .filter_scratch_allocation_bytes
1627            .saturating_add(self.set_workspace_allocation_bytes)
1628            .saturating_add(self.join_workspace_allocation_bytes)
1629            .saturating_add(self.control_allocation_bytes);
1630        let metadata_preparation_ns = self
1631            .device_trace_preparation_ns
1632            .saturating_add(self.schema_winners_preparation_ns)
1633            .saturating_add(self.receipt_preparation_ns)
1634            .saturating_add(self.schedule_program_preparation_ns);
1635        let metadata_reserved_bytes = self
1636            .device_trace_reserved_bytes
1637            .saturating_add(self.schema_winners_reserved_bytes)
1638            .saturating_add(self.receipt_reserved_bytes)
1639            .saturating_add(self.schedule_program_reserved_bytes);
1640        let additive = [
1641            self.admission_and_source_snapshot_ns,
1642            self.execution_domain_and_build_setup_ns,
1643            self.logical_schedule_planning_ns,
1644            self.manifest_compact_construction_ns,
1645            self.schedule_lowering_ns,
1646            self.reservation_ns,
1647            self.relation_slot_allocation_ns,
1648            self.count_initialization_ns,
1649            workspace_allocation_ns,
1650            self.metadata_binding_construction_ns,
1651            metadata_preparation_ns,
1652            self.reservation_validation_and_release_ns,
1653            self.pinned_receipt_ns,
1654            self.graph_body_capture_ns,
1655            self.graph_instantiate_ns,
1656            self.validation_owner_assembly_ns,
1657        ];
1658        let attributed_ns = additive.iter().copied().fold(0u64, u64::saturating_add);
1659        let unattributed_ns = self.total_ns.saturating_sub(attributed_ns);
1660        let metadata_initial_htod_calls = self
1661            .device_trace_initial_htod_calls
1662            .saturating_add(self.schema_winners_initial_htod_calls)
1663            .saturating_add(self.receipt_initial_htod_calls)
1664            .saturating_add(self.schedule_program_initial_htod_calls);
1665        let metadata_initial_htod_bytes = self
1666            .device_trace_initial_htod_bytes
1667            .saturating_add(self.schema_winners_initial_htod_bytes)
1668            .saturating_add(self.receipt_initial_htod_bytes)
1669            .saturating_add(self.schedule_program_initial_htod_bytes);
1670        let to_u64 = |value: usize| u64::try_from(value).unwrap_or(u64::MAX);
1671        ResidentGraphPrepareDiagnosticSnapshot {
1672            sample: self.sample,
1673            total_ns: self.total_ns,
1674            admission_and_source_snapshot_ns: self.admission_and_source_snapshot_ns,
1675            execution_domain_and_build_setup_ns: self.execution_domain_and_build_setup_ns,
1676            logical_schedule_planning_ns: self.logical_schedule_planning_ns,
1677            manifest_compact_construction_ns: self.manifest_compact_construction_ns,
1678            schedule_lowering_ns: self.schedule_lowering_ns,
1679            reservation_ns: self.reservation_ns,
1680            relation_preparation_ns: self.relation_slot_allocation_ns,
1681            count_initialization_ns: self.count_initialization_ns,
1682            workspace_preparation_ns: workspace_allocation_ns,
1683            metadata_binding_construction_ns: self.metadata_binding_construction_ns,
1684            metadata_preparation_ns,
1685            reservation_validation_and_release_ns: self.reservation_validation_and_release_ns,
1686            pinned_receipt_ns: self.pinned_receipt_ns,
1687            graph_body_capture_ns: self.graph_body_capture_ns,
1688            graph_instantiate_ns: self.graph_instantiate_ns,
1689            validation_owner_assembly_ns: self.validation_owner_assembly_ns,
1690            unattributed_ns,
1691            required_reservation_bytes: self.required_reservation_bytes,
1692            logical_relation_values: to_u64(self.logical_relation_values),
1693            physical_relation_slots: to_u64(self.physical_relation_slots),
1694            relation_device_allocation_calls: self.relation_device_allocation_calls,
1695            relation_reserved_bytes: self.relation_slot_allocation_bytes,
1696            relation_slot_preparation_ns_max: self.relation_slot_allocation_ns_max,
1697            count_memset_calls: self.count_memset_calls,
1698            count_memset_bytes: self
1699                .count_memset_calls
1700                .saturating_mul(std::mem::size_of::<u32>() as u64),
1701            count_initialization_ns_max: self.count_initialization_ns_max,
1702            workspace_provider_calls: self.workspace_provider_calls,
1703            workspace_reserved_bytes: workspace_allocation_bytes,
1704            filter_scratch_preparation_ns: self.filter_scratch_allocation_ns,
1705            filter_scratch_reserved_bytes: self.filter_scratch_allocation_bytes,
1706            set_workspace_preparation_ns: self.set_workspace_allocation_ns,
1707            set_workspace_reserved_bytes: self.set_workspace_allocation_bytes,
1708            join_workspace_preparation_ns: self.join_workspace_allocation_ns,
1709            join_workspace_reserved_bytes: self.join_workspace_allocation_bytes,
1710            control_preparation_ns: self.control_allocation_ns,
1711            control_reserved_bytes: self.control_allocation_bytes,
1712            metadata_provider_calls: self.metadata_provider_calls,
1713            metadata_reserved_bytes,
1714            metadata_initial_htod_calls,
1715            metadata_initial_htod_bytes,
1716            device_trace_preparation_ns: self.device_trace_preparation_ns,
1717            device_trace_reserved_bytes: self.device_trace_reserved_bytes,
1718            device_trace_initial_htod_calls: self.device_trace_initial_htod_calls,
1719            device_trace_initial_htod_bytes: self.device_trace_initial_htod_bytes,
1720            schema_winners_preparation_ns: self.schema_winners_preparation_ns,
1721            schema_winners_reserved_bytes: self.schema_winners_reserved_bytes,
1722            schema_winners_initial_htod_calls: self.schema_winners_initial_htod_calls,
1723            schema_winners_initial_htod_bytes: self.schema_winners_initial_htod_bytes,
1724            receipt_preparation_ns: self.receipt_preparation_ns,
1725            receipt_reserved_bytes: self.receipt_reserved_bytes,
1726            receipt_initial_htod_calls: self.receipt_initial_htod_calls,
1727            receipt_initial_htod_bytes: self.receipt_initial_htod_bytes,
1728            schedule_program_preparation_ns: self.schedule_program_preparation_ns,
1729            schedule_program_reserved_bytes: self.schedule_program_reserved_bytes,
1730            schedule_program_initial_htod_calls: self.schedule_program_initial_htod_calls,
1731            schedule_program_initial_htod_bytes: self.schedule_program_initial_htod_bytes,
1732            compact_ops: to_u64(self.compact_ops),
1733            compact_waves: to_u64(self.compact_waves),
1734            compact_regions: to_u64(self.compact_regions),
1735            conditional_regions: to_u64(self.conditional_regions),
1736            parent_graph_nodes: to_u64(self.parent_graph_nodes),
1737            conditional_body_nodes: to_u64(self.conditional_body_nodes),
1738        }
1739    }
1740}
1741
1742#[cfg(test)]
1743mod prepare_diagnostic_tests {
1744    use super::{
1745        resident_prepare_diagnostics_for_sample, resident_receipt_initial_htod,
1746        resident_schedule_initial_htod, resident_schema_winners_initial_htod,
1747        ResidentPrepareDiagnostics,
1748    };
1749
1750    #[test]
1751    fn prepare_diagnostic_counts_only_source_proven_initial_transfers() {
1752        assert_eq!(resident_schema_winners_initial_htod(0), (0, 0));
1753        assert_eq!(resident_schema_winners_initial_htod(3), (1, 12));
1754        assert_eq!(resident_receipt_initial_htod(0), (1, 32));
1755        assert_eq!(resident_receipt_initial_htod(3), (1, 80));
1756        assert_eq!(resident_schedule_initial_htod(4096), (8, 4096));
1757
1758        let (schema_calls, schema_bytes) = resident_schema_winners_initial_htod(3);
1759        let (receipt_calls, receipt_bytes) = resident_receipt_initial_htod(0);
1760        let (schedule_calls, schedule_bytes) = resident_schedule_initial_htod(4096);
1761        let diagnostics = ResidentPrepareDiagnostics {
1762            sample: 9,
1763            total_ns: 100,
1764            relation_slot_allocation_ns: 11,
1765            relation_slot_allocation_ns_max: 7,
1766            relation_slot_allocation_bytes: 2048,
1767            workspace_provider_calls: 4,
1768            count_initialization_ns: 10,
1769            count_initialization_ns_max: 5,
1770            count_memset_calls: 3,
1771            metadata_provider_calls: 4,
1772            schema_winners_initial_htod_calls: schema_calls,
1773            schema_winners_initial_htod_bytes: schema_bytes,
1774            receipt_initial_htod_calls: receipt_calls,
1775            receipt_initial_htod_bytes: receipt_bytes,
1776            schedule_program_initial_htod_calls: schedule_calls,
1777            schedule_program_initial_htod_bytes: schedule_bytes,
1778            ..ResidentPrepareDiagnostics::default()
1779        };
1780        let snapshot = diagnostics.into_snapshot();
1781        assert_eq!(snapshot.sample, 9);
1782        assert_eq!(snapshot.total_ns, 100);
1783        assert_eq!(snapshot.relation_preparation_ns, 11);
1784        assert_eq!(snapshot.relation_reserved_bytes, 2048);
1785        assert_eq!(snapshot.relation_slot_preparation_ns_max, 7);
1786        assert_eq!(snapshot.workspace_provider_calls, 4);
1787        assert_eq!(snapshot.count_memset_calls, 3);
1788        assert_eq!(snapshot.count_initialization_ns, 10);
1789        assert_eq!(snapshot.count_initialization_ns_max, 5);
1790        assert_eq!(
1791            snapshot.count_memset_bytes,
1792            3 * std::mem::size_of::<u32>() as u64
1793        );
1794        assert_eq!(snapshot.metadata_provider_calls, 4);
1795        assert_eq!(snapshot.device_trace_initial_htod_calls, 0);
1796        assert_eq!(snapshot.device_trace_initial_htod_bytes, 0);
1797        assert_eq!(snapshot.schema_winners_initial_htod_calls, 1);
1798        assert_eq!(snapshot.schema_winners_initial_htod_bytes, 12);
1799        assert_eq!(snapshot.receipt_initial_htod_calls, 1);
1800        assert_eq!(snapshot.receipt_initial_htod_bytes, 32);
1801        assert_eq!(snapshot.schedule_program_initial_htod_calls, 8);
1802        assert_eq!(snapshot.schedule_program_initial_htod_bytes, 4096);
1803        assert_eq!(snapshot.metadata_initial_htod_calls, 10);
1804        assert_eq!(snapshot.metadata_initial_htod_bytes, 4140);
1805    }
1806
1807    #[test]
1808    fn prepare_diagnostic_is_absent_when_sampling_is_disabled() {
1809        assert!(resident_prepare_diagnostics_for_sample(None).is_none());
1810        assert_eq!(
1811            resident_prepare_diagnostics_for_sample(Some(23))
1812                .expect("enabled diagnostics create raw timing state")
1813                .sample,
1814            23
1815        );
1816    }
1817
1818    #[test]
1819    fn prepare_diagnostic_keeps_setup_and_reservation_validation_separate() {
1820        let snapshot = ResidentPrepareDiagnostics {
1821            sample: 5,
1822            total_ns: 100,
1823            admission_and_source_snapshot_ns: 11,
1824            execution_domain_and_build_setup_ns: 13,
1825            metadata_binding_construction_ns: 17,
1826            reservation_validation_and_release_ns: 19,
1827            ..ResidentPrepareDiagnostics::default()
1828        }
1829        .into_snapshot();
1830
1831        assert_eq!(snapshot.admission_and_source_snapshot_ns, 11);
1832        assert_eq!(snapshot.execution_domain_and_build_setup_ns, 13);
1833        assert_eq!(snapshot.metadata_binding_construction_ns, 17);
1834        assert_eq!(snapshot.reservation_validation_and_release_ns, 19);
1835        assert_eq!(snapshot.unattributed_ns, 40);
1836    }
1837}
1838
1839impl ResidentAllocationManifest {
1840    fn finalize_compact_schedule(
1841        &mut self,
1842        schedule: &ResidentCompactSchedulePlan,
1843        head_count: usize,
1844    ) -> Result<()> {
1845        let (required_bytes, schedule_metadata_bytes) = resident_compact_allocation_bytes(
1846            self.relation_bytes,
1847            self.filter_scratch_bytes,
1848            self.fixed_workspace_bytes,
1849            self.slots.len(),
1850            head_count,
1851            schedule,
1852        )?;
1853        self.required_bytes = required_bytes;
1854        self.schedule_metadata_bytes = schedule_metadata_bytes;
1855        Ok(())
1856    }
1857}
1858
1859fn resident_output_indices(
1860    heads: &BTreeMap<String, usize>,
1861    assignments: &[ResidentSlotAssignment],
1862    slots: &[ResidentPhysicalSlotPlan],
1863) -> Result<Vec<(String, usize)>> {
1864    heads
1865        .iter()
1866        .map(|(name, logical)| {
1867            let assignment = assignments.get(*logical).ok_or_else(|| {
1868                XlogError::Execution("resident head has no physical slot assignment".into())
1869            })?;
1870            if !slots
1871                .get(assignment.slot)
1872                .is_some_and(|slot| slot.permanent)
1873            {
1874                return Err(XlogError::Execution(
1875                    "resident head is not assigned to a permanent physical slot".into(),
1876                ));
1877            }
1878            Ok((name.clone(), assignment.slot))
1879        })
1880        .collect()
1881}
1882
1883fn resident_validate_exact_reservation(required: u64, used: u64, remaining: u64) -> Result<()> {
1884    if used == required && remaining == 0 {
1885        return Ok(());
1886    }
1887    Err(XlogError::Execution(format!(
1888        "resident allocation manifest accounted for {required} manager-tracked bytes but materialization consumed {used} and left {remaining} reserved bytes"
1889    )))
1890}
1891
1892#[derive(Debug)]
1893struct ResidentScratchSlotState {
1894    slot: usize,
1895    layout: Vec<ScalarType>,
1896    last_use: usize,
1897    generation: u32,
1898}
1899
1900fn resident_schema_layout(schema: &Schema) -> Vec<ScalarType> {
1901    schema
1902        .columns
1903        .iter()
1904        .map(|(_, scalar)| scalar.clone())
1905        .collect()
1906}
1907
1908fn resident_private_inputs(op: &ResidentRecordedOp, mut visit: impl FnMut(usize)) {
1909    let mut visit_ref = |reference: &ResidentBufferRef| {
1910        if let ResidentBufferRef::Private(index) = reference {
1911            visit(*index);
1912        }
1913    };
1914    match op {
1915        ResidentRecordedOp::Scan {
1916            relation: input, ..
1917        }
1918        | ResidentRecordedOp::Filter { input, .. }
1919        | ResidentRecordedOp::Project { input, .. } => visit_ref(input),
1920        ResidentRecordedOp::Union { left, right, .. }
1921        | ResidentRecordedOp::Diff { left, right, .. }
1922        | ResidentRecordedOp::Join { left, right, .. } => {
1923            visit_ref(left);
1924            visit_ref(right);
1925        }
1926        ResidentRecordedOp::ChangedMark { relation } => visit(*relation),
1927        ResidentRecordedOp::TraceDelta { semantic_guard, .. } => {
1928            if let Some(semantic_guard) = semantic_guard {
1929                visit_ref(semantic_guard);
1930            }
1931        }
1932        ResidentRecordedOp::SchemaWinnerMark { contribution, .. } => visit_ref(contribution),
1933        ResidentRecordedOp::Unit { .. }
1934        | ResidentRecordedOp::Clear { .. }
1935        | ResidentRecordedOp::ChangedReset
1936        | ResidentRecordedOp::TestStatus(_) => {}
1937    }
1938}
1939
1940fn resident_private_output(op: &ResidentRecordedOp) -> Option<usize> {
1941    match op {
1942        ResidentRecordedOp::Unit { output, .. }
1943        | ResidentRecordedOp::Clear { output }
1944        | ResidentRecordedOp::Filter { output, .. }
1945        | ResidentRecordedOp::Project { output, .. }
1946        | ResidentRecordedOp::Union { output, .. }
1947        | ResidentRecordedOp::Diff { output, .. }
1948        | ResidentRecordedOp::Join { output, .. } => Some(*output),
1949        ResidentRecordedOp::Scan { .. }
1950        | ResidentRecordedOp::TraceDelta { .. }
1951        | ResidentRecordedOp::ChangedReset
1952        | ResidentRecordedOp::ChangedMark { .. }
1953        | ResidentRecordedOp::SchemaWinnerMark { .. }
1954        | ResidentRecordedOp::TestStatus(_) => None,
1955    }
1956}
1957
1958fn resident_source_slot_map<'a>(
1959    private_slot_count: usize,
1960    sources: impl IntoIterator<Item = &'a str>,
1961) -> Result<BTreeMap<String, u32>> {
1962    let first_source_slot = u32::try_from(private_slot_count)
1963        .map_err(|_| XlogError::Execution("resident private slot count exceeds u32".into()))?;
1964    let names = sources
1965        .into_iter()
1966        .map(str::to_owned)
1967        .collect::<BTreeSet<_>>();
1968    names
1969        .into_iter()
1970        .enumerate()
1971        .map(|(offset, name)| {
1972            let offset = u32::try_from(offset).map_err(|_| {
1973                XlogError::Execution("resident source slot count exceeds u32".into())
1974            })?;
1975            let slot = first_source_slot.checked_add(offset).ok_or_else(|| {
1976                XlogError::Execution("resident source slot index overflow".into())
1977            })?;
1978            Ok((name, slot))
1979        })
1980        .collect()
1981}
1982
1983fn resident_record_lifetimes(
1984    ops: &[ResidentRecordedOp],
1985    relations: &[ResidentLogicalRelation],
1986    definitions: &mut [Option<usize>],
1987    last_uses: &mut [Option<usize>],
1988    ordinal: &mut usize,
1989) -> Result<(usize, usize)> {
1990    let start = *ordinal;
1991    for op in ops {
1992        let current = *ordinal;
1993        let mut invalid_input = None;
1994        let mut input_before_definition = None;
1995        resident_private_inputs(op, |logical| {
1996            let Some(relation) = relations.get(logical) else {
1997                invalid_input.get_or_insert(logical);
1998                return;
1999            };
2000            if !relation.permanent {
2001                if definitions[logical].is_none() {
2002                    input_before_definition.get_or_insert(logical);
2003                } else {
2004                    last_uses[logical] = Some(last_uses[logical].unwrap_or(current).max(current));
2005                }
2006            }
2007        });
2008        if let Some(logical) = invalid_input {
2009            return Err(XlogError::Execution(format!(
2010                "resident logical input relation {logical} is missing"
2011            )));
2012        }
2013        if let Some(logical) = input_before_definition {
2014            return Err(XlogError::Execution(format!(
2015                "resident scratch relation {logical} is used before its definition"
2016            )));
2017        }
2018        if let Some(logical) = resident_private_output(op) {
2019            let relation = relations.get(logical).ok_or_else(|| {
2020                XlogError::Execution(format!(
2021                    "resident logical output relation {logical} is missing"
2022                ))
2023            })?;
2024            if !relation.permanent {
2025                if definitions[logical].replace(current).is_some() {
2026                    return Err(XlogError::Execution(format!(
2027                        "resident scratch relation {logical} has multiple definitions"
2028                    )));
2029                }
2030                last_uses[logical] = Some(last_uses[logical].unwrap_or(current).max(current));
2031            }
2032        }
2033        *ordinal = ordinal
2034            .checked_add(1)
2035            .ok_or_else(|| XlogError::Execution("resident operation ordinal overflow".into()))?;
2036    }
2037    Ok((start, *ordinal))
2038}
2039
2040fn resident_validate_slot_assignments(
2041    relations: &[ResidentLogicalRelation],
2042    definitions: &[Option<usize>],
2043    last_uses: &[Option<usize>],
2044    slots: &[ResidentPhysicalSlotPlan],
2045    assignments: &[ResidentSlotAssignment],
2046) -> Result<()> {
2047    if definitions.len() != relations.len()
2048        || last_uses.len() != relations.len()
2049        || assignments.len() != relations.len()
2050    {
2051        return Err(XlogError::Execution(
2052            "resident slot validation vector length mismatch".into(),
2053        ));
2054    }
2055
2056    let mut occupied_permanent_slots = HashSet::new();
2057    let mut scratch_by_slot = BTreeMap::<usize, Vec<(usize, usize, usize, u32)>>::new();
2058    let mut assigned_slots = HashSet::new();
2059    for (logical, relation) in relations.iter().enumerate() {
2060        let assignment = assignments[logical];
2061        let slot = slots.get(assignment.slot).ok_or_else(|| {
2062            XlogError::Execution(format!(
2063                "resident logical relation {logical} maps to missing physical slot {}",
2064                assignment.slot
2065            ))
2066        })?;
2067        assigned_slots.insert(assignment.slot);
2068        if relation.permanent {
2069            if assignment.generation != 0 {
2070                return Err(XlogError::Execution(format!(
2071                    "resident permanent relation {logical} has nonzero generation {}",
2072                    assignment.generation
2073                )));
2074            }
2075            if !slot.permanent
2076                || slot.schema != relation.schema
2077                || slot.initial_count != relation.initial_count
2078            {
2079                return Err(XlogError::Execution(format!(
2080                    "resident permanent relation {logical} has an incompatible physical slot"
2081                )));
2082            }
2083            if !occupied_permanent_slots.insert(assignment.slot) {
2084                return Err(XlogError::Execution(format!(
2085                    "resident permanent physical slot {} has multiple owners",
2086                    assignment.slot
2087                )));
2088            }
2089            continue;
2090        }
2091
2092        if slot.permanent
2093            || resident_schema_layout(&slot.schema) != resident_schema_layout(&relation.schema)
2094        {
2095            return Err(XlogError::Execution(format!(
2096                "resident scratch relation {logical} has an incompatible physical slot"
2097            )));
2098        }
2099        let definition = definitions[logical].ok_or_else(|| {
2100            XlogError::Execution(format!(
2101                "resident scratch relation {logical} is never defined"
2102            ))
2103        })?;
2104        let last_use = last_uses[logical].unwrap_or(definition);
2105        if last_use < definition {
2106            return Err(XlogError::Execution(format!(
2107                "resident scratch relation {logical} ends before its definition"
2108            )));
2109        }
2110        scratch_by_slot.entry(assignment.slot).or_default().push((
2111            definition,
2112            last_use,
2113            logical,
2114            assignment.generation,
2115        ));
2116    }
2117
2118    if assigned_slots.len() != slots.len() {
2119        return Err(XlogError::Execution(
2120            "resident allocation manifest contains an unassigned physical slot".into(),
2121        ));
2122    }
2123    for (slot, mut generations) in scratch_by_slot {
2124        generations.sort_by_key(|(definition, _, logical, _)| (*definition, *logical));
2125        let mut previous_last_use = None;
2126        for (expected_generation, (definition, last_use, logical, generation)) in
2127            generations.into_iter().enumerate()
2128        {
2129            let expected_generation = u32::try_from(expected_generation).map_err(|_| {
2130                XlogError::Execution("resident scratch generation exceeds u32".into())
2131            })?;
2132            if generation != expected_generation {
2133                return Err(XlogError::Execution(format!(
2134                    "resident scratch relation {logical} in slot {slot} has generation {generation} but expected {expected_generation}"
2135                )));
2136            }
2137            if previous_last_use.is_some_and(|previous| previous >= definition) {
2138                return Err(XlogError::Execution(format!(
2139                    "resident scratch generations overlap in physical slot {slot}"
2140                )));
2141            }
2142            previous_last_use = Some(last_use);
2143        }
2144    }
2145    Ok(())
2146}
2147
2148fn resident_checked_add(total: &mut u64, bytes: u64, label: &str) -> Result<()> {
2149    *total = total.checked_add(bytes).ok_or_else(|| {
2150        XlogError::Execution(format!("resident {label} allocation byte overflow"))
2151    })?;
2152    Ok(())
2153}
2154
2155/// A fully allocated, instantiated resident transaction that has not launched.
2156pub struct PreparedResidentGraph<'executor> {
2157    owners: ResidentRunOwners,
2158    preflight_report: ResidentGraphPreflightReport,
2159    has_device_status_writer: bool,
2160    source_guard: &'executor Executor,
2161    source_set_snapshots: Vec<ResidentSourceSetSnapshot>,
2162    prepare_diagnostic: Option<ResidentPrepareDiagnostics>,
2163}
2164
2165/// Read-only topology and memory facts captured after preparation and before launch.
2166#[derive(Debug, Clone, PartialEq, Eq)]
2167pub struct ResidentGraphPreflightReport {
2168    /// Fixed logical row capacity used for every staged relation.
2169    pub relation_capacity: u32,
2170    /// Exact manager-tracked bytes reserved before physical materialization.
2171    pub estimated_required_bytes: u64,
2172    /// Caller budget remaining when admission ran.
2173    pub available_bytes_at_admission: u64,
2174    /// Manager-tracked live allocation growth caused by this prepared transaction.
2175    pub tracked_device_allocation_bytes: u64,
2176    /// Exact manager-tracked relation data and logical-count bytes.
2177    pub relation_device_bytes: u64,
2178    /// Exact immutable filter descriptor bytes retained by the graph.
2179    pub filter_descriptor_device_bytes: u64,
2180    /// Exact mutable filter scratch bytes shared by all sequential filters.
2181    pub filter_scratch_device_bytes: u64,
2182    /// Exact immutable project and copy descriptor bytes.
2183    pub project_descriptor_device_bytes: u64,
2184    /// Exact set, join, control, trace, packed-receipt, and remaining compact schedule metadata bytes.
2185    pub fixed_workspace_device_bytes: u64,
2186    /// Nodes in the instantiated parent graph, excluding conditional body children.
2187    pub parent_graph_nodes: usize,
2188    /// Conditional-WHILE nodes in the instantiated parent graph.
2189    pub conditional_while_nodes: usize,
2190    /// Actual node kinds in root-to-leaf dependency-chain order.
2191    pub parent_graph_node_kinds: Vec<CudaGraphNodeKind>,
2192    /// Actual node kinds in each conditional-WHILE body, in parent order.
2193    pub conditional_body_node_kinds: Vec<Vec<CudaGraphNodeKind>>,
2194    /// Actual kernel-node count in each conditional-WHILE body.
2195    pub conditional_body_kernel_counts: Vec<usize>,
2196    /// Parent nodes plus every conditional-body child node.
2197    pub hierarchical_graph_nodes: usize,
2198    /// Private fixed-capacity relation slots retained by the graph.
2199    pub private_relation_slots: usize,
2200    /// Logical relation values before interval coloring.
2201    pub logical_relation_values: usize,
2202    /// Permanent unit, staged-head, raw, and delta slots.
2203    pub permanent_relation_slots: u32,
2204    /// Final staged relation count included in the terminal receipt.
2205    pub staged_output_relations: usize,
2206    /// Layout-compatible scratch slots selected by interval coloring.
2207    pub scratch_slots: u32,
2208    /// Number of mutable filter scratch allocations retained by this graph.
2209    pub filter_scratch_allocations: u32,
2210    /// Widest allocated permanent or intermediate row in bytes.
2211    pub max_row_bytes: u64,
2212}
2213
2214/// A resident transaction whose one graph launch is in flight.
2215pub struct ResidentGraphInFlight<'executor> {
2216    // Field order is intentional: abandoned execution waits before graph and
2217    // workspace owners are destroyed.
2218    completion: ResidentCompletionEvent,
2219    timing_start: CudaEvent,
2220    timing_end: CudaEvent,
2221    owners: ResidentRunOwners,
2222    _executor: PhantomData<&'executor Executor>,
2223}
2224
2225/// A resident transaction after its one terminal completion wait.
2226pub struct ResidentGraphSynchronized<'executor> {
2227    device_elapsed_ns: u64,
2228    owners: ResidentRunOwners,
2229    _executor: PhantomData<&'executor Executor>,
2230}
2231
2232struct StagedResidentOutput {
2233    name: String,
2234    buffer: CudaBuffer,
2235}
2236
2237/// Borrow-free decoded receipt. Store mutation happens only in [`Self::commit`].
2238pub struct ObservedResidentGraphReceipt {
2239    encoded_len: usize,
2240    device_elapsed_ns: u64,
2241    device_scan_invocations: u64,
2242    device_filter_invocations: u64,
2243    semantic_scan_invocations: u64,
2244    semantic_filter_invocations: u64,
2245    iterations: u32,
2246    terminal: std::result::Result<(), ResidentGraphExecutionError>,
2247    outputs: Vec<StagedResidentOutput>,
2248    source_epoch: u64,
2249    relation_registration: Vec<(RelId, String)>,
2250    transaction_identity: Arc<()>,
2251    provider: Arc<CudaKernelProvider>,
2252    phase_timings: Option<ResidentFinalObservationPhaseTimings>,
2253}
2254
2255/// Opt-in wall-clock breakdown of final resident receipt observation.
2256#[doc(hidden)]
2257#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2258pub struct ResidentFinalObservationPhaseTimings {
2259    /// Time spent in the sole pinned device-to-host receipt transfer.
2260    pub receipt_d2h_ns: u64,
2261    /// Time spent decoding the receipt, resolving schemas, and staging outputs.
2262    pub decode_schema_staging_ns: u64,
2263}
2264
2265fn resident_source_set_snapshot(
2266    provider: &CudaKernelProvider,
2267    name: &str,
2268    version: u64,
2269    buffer: &CudaBuffer,
2270) -> std::result::Result<ResidentSourceSetSnapshot, ResidentGraphDeclineReason> {
2271    if !buffer.canonical_full_row_set_certified() {
2272        return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2273            relation: name.to_owned(),
2274        });
2275    }
2276    let manager_ptr = Arc::as_ptr(provider.memory()) as usize;
2277    let runtime = provider.memory().runtime().ok_or_else(|| {
2278        ResidentGraphDeclineReason::SourceSetUncertified {
2279            relation: format!("{name} memory manager has no device runtime"),
2280        }
2281    })?;
2282    if !Arc::ptr_eq(provider.device(), provider.memory().device())
2283        || !Arc::ptr_eq(provider.device(), runtime.device())
2284        || u32::try_from(provider.device().ordinal()).ok() != Some(runtime.device_ordinal())
2285    {
2286        return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2287            relation: format!("{name} memory manager does not match the resident provider"),
2288        });
2289    }
2290    let column_blocks = buffer
2291        .columns()
2292        .iter()
2293        .enumerate()
2294        .map(|(column_index, column)| {
2295            let CudaColumn::Owned(column) = column else {
2296                return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2297                    relation: format!("{name} column {column_index} is externally owned"),
2298                });
2299            };
2300            if column.memory_manager_ptr_value() != manager_ptr {
2301                return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2302                    relation: format!(
2303                        "{name} column {column_index} belongs to another memory manager"
2304                    ),
2305                });
2306            }
2307            let block = column.runtime_block().map(BlockId::from_block);
2308            if block.is_none() && buffer.num_rows() != 0 {
2309                return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2310                    relation: format!("{name} column {column_index} is not runtime tracked"),
2311                });
2312            }
2313            Ok(block)
2314        })
2315        .collect::<std::result::Result<Vec<_>, _>>()?;
2316    let row_count = buffer.num_rows_device();
2317    if row_count.memory_manager_ptr_value() != manager_ptr {
2318        return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2319            relation: format!("{name} logical count belongs to another memory manager"),
2320        });
2321    }
2322    let row_count_block = row_count
2323        .runtime_block()
2324        .map(BlockId::from_block)
2325        .ok_or_else(|| ResidentGraphDeclineReason::SourceSetUncertified {
2326            relation: format!("{name} logical count is not runtime tracked"),
2327        })?;
2328    Ok(ResidentSourceSetSnapshot {
2329        name: name.to_owned(),
2330        version,
2331        schema: buffer.schema().clone(),
2332        row_capacity: buffer.num_rows(),
2333        column_blocks,
2334        row_count_block,
2335    })
2336}
2337
2338fn validate_resident_source_set_snapshots(
2339    executor: &Executor,
2340    source_epoch: u64,
2341    snapshots: &[ResidentSourceSetSnapshot],
2342) -> std::result::Result<(), ResidentGraphExecutionError> {
2343    if executor.store.mutation_epoch() != source_epoch {
2344        return Err(resident_decline_error(
2345            ResidentGraphDeclineReason::SourceSetUncertified {
2346                relation: "relation store changed after resident preparation".to_owned(),
2347            },
2348        ));
2349    }
2350    for snapshot in snapshots {
2351        let Some((buffer, version)) = executor.store.get_with_version(&snapshot.name) else {
2352            return Err(resident_decline_error(
2353                ResidentGraphDeclineReason::SourceSetUncertified {
2354                    relation: snapshot.name.clone(),
2355                },
2356            ));
2357        };
2358        let current =
2359            resident_source_set_snapshot(&executor.provider, &snapshot.name, version, buffer)
2360                .map_err(resident_decline_error)?;
2361        if &current != snapshot {
2362            return Err(resident_decline_error(
2363                ResidentGraphDeclineReason::SourceSetUncertified {
2364                    relation: snapshot.name.clone(),
2365                },
2366            ));
2367        }
2368    }
2369    Ok(())
2370}
2371
2372impl<'executor> PreparedResidentGraph<'executor> {
2373    /// Return immutable prelaunch topology and memory diagnostics.
2374    pub fn preflight_report(&self) -> &ResidentGraphPreflightReport {
2375        &self.preflight_report
2376    }
2377
2378    /// Move out the opt-in raw prepare diagnostic without deriving or emitting it.
2379    #[doc(hidden)]
2380    pub fn take_prepare_diagnostic(&mut self) -> Option<ResidentPrepareDiagnostics> {
2381        self.prepare_diagnostic.take()
2382    }
2383
2384    #[cfg(feature = "resident-graph-tests")]
2385    pub(crate) fn invalidate_expected_source_epoch(&mut self) {
2386        self.owners.source_epoch = self.owners.source_epoch.wrapping_add(1);
2387    }
2388
2389    /// Launch the already-instantiated graph exactly once.
2390    pub fn launch(
2391        mut self,
2392    ) -> std::result::Result<ResidentGraphInFlight<'executor>, ResidentGraphExecutionError> {
2393        validate_resident_source_set_snapshots(
2394            self.source_guard,
2395            self.owners.source_epoch,
2396            &self.source_set_snapshots,
2397        )?;
2398        self.owners
2399            .execution_domain
2400            .preflight(&mut self.owners.recorder)
2401            .map_err(runtime_error)?;
2402        let timing_start = self
2403            .owners
2404            .stream
2405            .record_event(Some(CUevent_flags::CU_EVENT_DEFAULT))
2406            .map_err(|error| {
2407                runtime_error(format!("resident timing-start event failed: {error}"))
2408            })?;
2409        self.owners
2410            .graph
2411            .launch(&self.owners.stream)
2412            .map_err(runtime_error)?;
2413        let timing_end = match self
2414            .owners
2415            .stream
2416            .record_event(Some(CUevent_flags::CU_EVENT_DEFAULT))
2417        {
2418            Ok(event) => event,
2419            Err(error) => {
2420                let _ = self.owners.stream.synchronize();
2421                return Err(runtime_error(format!(
2422                    "resident timing-end event failed after graph launch: {error}"
2423                )));
2424            }
2425        };
2426        self.owners
2427            .runtime
2428            .record_conditional_graph_launch(self.has_device_status_writer);
2429        let recorder = std::mem::replace(
2430            &mut self.owners.recorder,
2431            self.owners.execution_domain.new_strict_recorder(),
2432        );
2433        if let Err(error) = self.owners.execution_domain.commit(recorder) {
2434            let _ = self.owners.stream.synchronize();
2435            return Err(runtime_error(error));
2436        }
2437        let completion = match self
2438            .owners
2439            .runtime
2440            .record_resident_completion_event(&self.owners.stream)
2441        {
2442            Ok(completion) => completion,
2443            Err(error) => {
2444                let _ = self.owners.stream.synchronize();
2445                return Err(runtime_error(error));
2446            }
2447        };
2448        Ok(ResidentGraphInFlight {
2449            completion,
2450            timing_start,
2451            timing_end,
2452            owners: self.owners,
2453            _executor: PhantomData,
2454        })
2455    }
2456}
2457
2458impl<'executor> ResidentGraphInFlight<'executor> {
2459    /// Wait for the one real completion event and retain every graph owner.
2460    pub fn synchronize_core(
2461        mut self,
2462    ) -> std::result::Result<ResidentGraphSynchronized<'executor>, ResidentGraphExecutionError>
2463    {
2464        self.completion.synchronize().map_err(runtime_error)?;
2465        let elapsed_ms = self
2466            .timing_start
2467            .elapsed_ms(&self.timing_end)
2468            .map_err(|error| {
2469                runtime_error(format!("resident CUDA-event timing failed: {error}"))
2470            })?;
2471        if !elapsed_ms.is_finite() || elapsed_ms < 0.0 {
2472            return Err(runtime_error("resident CUDA-event timing was not finite"));
2473        }
2474        let device_elapsed_ns = (f64::from(elapsed_ms) * 1_000_000.0)
2475            .round()
2476            .clamp(0.0, u64::MAX as f64) as u64;
2477        Ok(ResidentGraphSynchronized {
2478            device_elapsed_ns,
2479            owners: self.owners,
2480            _executor: PhantomData,
2481        })
2482    }
2483}
2484
2485impl<'executor> ResidentGraphSynchronized<'executor> {
2486    /// Perform the transaction's only device-to-host observation and decode it.
2487    pub fn observe_final_receipt(
2488        mut self,
2489    ) -> std::result::Result<ObservedResidentGraphReceipt, ResidentGraphExecutionError> {
2490        let phase_diagnostics =
2491            std::env::var("XLOG_RESIDENT_LATENCY_DIAGNOSTICS").as_deref() == Ok("1");
2492        let receipt_d2h_started = phase_diagnostics.then(Instant::now);
2493        let encoded_len = self.owners.receipt.len_bytes();
2494        let bytes = self
2495            .owners
2496            .provider
2497            .observe_resident_packed_receipt(
2498                &self.owners.receipt,
2499                &mut self.owners.pinned_receipt,
2500                &self.owners.stream,
2501            )
2502            .map_err(runtime_error)?;
2503        let receipt_d2h_ns = receipt_d2h_started
2504            .map(|started| u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX))
2505            .unwrap_or(0);
2506        let decode_started = phase_diagnostics.then(Instant::now);
2507        let relation_count_len = self.owners.output_indices.len();
2508        let schema_winner_count = self.owners.output_schema_plans.len();
2509        let expected_count_fields = relation_count_len + 4 + schema_winner_count;
2510        if self.owners.receipt.relation_count_len() as usize != relation_count_len
2511            || self.owners.receipt.device_trace_field_count() != 4
2512            || self.owners.receipt.schema_winner_count() as usize != schema_winner_count
2513            || self.owners.receipt.total_count_field_len() as usize != expected_count_fields
2514            || bytes.len() != encoded_len
2515            || encoded_len != 44 + 4 * expected_count_fields
2516        {
2517            return Err(runtime_error(format!(
2518                "malformed resident receipt length: got {}, expected {}",
2519                bytes.len(),
2520                44 + 4 * expected_count_fields
2521            )));
2522        }
2523        let field_u32 = |offset| {
2524            read_receipt_u32(&bytes, offset)
2525                .ok_or_else(|| runtime_error("truncated resident receipt u32 field"))
2526        };
2527        let field_u64 = |offset| {
2528            read_receipt_u64(&bytes, offset)
2529                .ok_or_else(|| runtime_error("truncated resident receipt u64 field"))
2530        };
2531        let code = field_u32(0)?;
2532        let op_id = field_u32(4)?;
2533        let resource_code = field_u32(8)?;
2534        let iterations = field_u32(12)?;
2535        let limit = field_u32(16)?;
2536        let reserved = field_u32(20)?;
2537        let required = field_u64(24)?;
2538        let capacity = field_u64(32)?;
2539        let changed = field_u32(40)?;
2540        if reserved != 0 || changed > 1 {
2541            return Err(runtime_error(
2542                "malformed resident terminal status reserved/changed field",
2543            ));
2544        }
2545        let terminal = match code {
2546            value if value == ResidentTerminalCode::Success as u32 => {
2547                if resource_code != ResidentResourceCode::None as u32
2548                    || limit != 0
2549                    || required != 0
2550                    || capacity != 0
2551                {
2552                    return Err(runtime_error("malformed resident success payload"));
2553                }
2554                Ok(())
2555            }
2556            value if value == ResidentTerminalCode::IterationLimit as u32 => {
2557                if resource_code != ResidentResourceCode::None as u32
2558                    || required != 0
2559                    || capacity != 0
2560                    || iterations > limit
2561                {
2562                    return Err(runtime_error("malformed resident iteration-limit payload"));
2563                }
2564                Err(ResidentGraphExecutionError::IterationLimit {
2565                    limit,
2566                    completed: iterations,
2567                })
2568            }
2569            value if value == ResidentTerminalCode::CapacityOverflow as u32 => {
2570                if resource_code != ResidentResourceCode::OutputRows as u32 || required <= capacity
2571                {
2572                    return Err(runtime_error(
2573                        "malformed resident capacity-overflow payload",
2574                    ));
2575                }
2576                Err(ResidentGraphExecutionError::CapacityOverflow {
2577                    op_id,
2578                    required,
2579                    capacity,
2580                })
2581            }
2582            value if value == ResidentTerminalCode::ResourceExhausted as u32 => {
2583                let resource = match resource_code {
2584                    value
2585                        if value == ResidentResourceCode::SetHashSlots as u32
2586                            || value == ResidentResourceCode::JoinBuckets as u32
2587                            || value == ResidentResourceCode::JoinChains as u32 =>
2588                    {
2589                        "workspace_slots"
2590                    }
2591                    value if value == ResidentResourceCode::InputRows as u32 => "input_rows",
2592                    value if value == ResidentResourceCode::OutputRows as u32 => "output_rows",
2593                    _ => return Err(runtime_error("unknown resident resource code")),
2594                };
2595                if required <= capacity {
2596                    return Err(runtime_error("malformed resident resource payload"));
2597                }
2598                Err(ResidentGraphExecutionError::ResourceExhausted {
2599                    op_id,
2600                    resource,
2601                    required,
2602                    capacity,
2603                })
2604            }
2605            _ => {
2606                return Err(runtime_error(format!(
2607                    "unknown resident terminal code {code}"
2608                )))
2609            }
2610        };
2611
2612        let mut counts = Vec::with_capacity(self.owners.output_indices.len());
2613        for index in 0..self.owners.output_indices.len() {
2614            counts.push(field_u32(44 + index * 4)?);
2615        }
2616        let device_scan_invocations = u64::from(field_u32(44 + relation_count_len * 4)?);
2617        let device_filter_invocations = u64::from(field_u32(48 + relation_count_len * 4)?);
2618        let semantic_scan_invocations = u64::from(field_u32(52 + relation_count_len * 4)?);
2619        let semantic_filter_invocations = u64::from(field_u32(56 + relation_count_len * 4)?);
2620        let schema_winner_offset = 60 + relation_count_len * 4;
2621        let mut schema_winner_ids = Vec::with_capacity(schema_winner_count);
2622        for index in 0..schema_winner_count {
2623            schema_winner_ids.push(field_u32(schema_winner_offset + index * 4)?);
2624        }
2625        let mut outputs = Vec::new();
2626        if terminal.is_ok() {
2627            let selected_schemas = resident_resolve_output_schemas(
2628                &self.owners.output_schema_plans,
2629                &schema_winner_ids,
2630            )
2631            .map_err(runtime_error)?;
2632            let mut cache_entries = Vec::with_capacity(counts.len());
2633            for ((_, relation_index), count) in self
2634                .owners
2635                .output_indices
2636                .iter()
2637                .zip(counts.iter().copied())
2638            {
2639                let relation = private_relation(&self.owners.relations, *relation_index)
2640                    .map_err(runtime_error)?;
2641                if count > relation.capacity() {
2642                    return Err(runtime_error(format!(
2643                        "resident receipt count {count} exceeds output capacity {}",
2644                        relation.capacity()
2645                    )));
2646                }
2647                cache_entries.push((relation.buffer(), count));
2648            }
2649            self.owners
2650                .provider
2651                .finalize_resident_logical_counts(&cache_entries)
2652                .map_err(runtime_error)?;
2653            for ((name, relation_index), schema) in
2654                self.owners.output_indices.iter().zip(selected_schemas)
2655            {
2656                let relation = self
2657                    .owners
2658                    .relations
2659                    .get_mut(*relation_index)
2660                    .and_then(Option::take)
2661                    .ok_or_else(|| runtime_error("resident output relation owner missing"))?;
2662                outputs.push(StagedResidentOutput {
2663                    name: name.clone(),
2664                    buffer: relation
2665                        .into_buffer_with_observed_schema(schema)
2666                        .map_err(runtime_error)?,
2667                });
2668            }
2669        }
2670
2671        let decode_schema_staging_ns = decode_started
2672            .map(|started| u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX))
2673            .unwrap_or(0);
2674        Ok(ObservedResidentGraphReceipt {
2675            encoded_len,
2676            device_elapsed_ns: self.device_elapsed_ns,
2677            device_scan_invocations,
2678            device_filter_invocations,
2679            semantic_scan_invocations,
2680            semantic_filter_invocations,
2681            iterations,
2682            terminal,
2683            outputs,
2684            source_epoch: self.owners.source_epoch,
2685            relation_registration: self.owners.relation_registration.clone(),
2686            transaction_identity: Arc::clone(&self.owners.transaction_identity),
2687            provider: Arc::clone(&self.owners.provider),
2688            phase_timings: phase_diagnostics.then_some(ResidentFinalObservationPhaseTimings {
2689                receipt_d2h_ns,
2690                decode_schema_staging_ns,
2691            }),
2692        })
2693    }
2694}
2695
2696impl ObservedResidentGraphReceipt {
2697    /// Exact bytes transferred in the sole final observation.
2698    pub fn encoded_len(&self) -> usize {
2699        self.encoded_len
2700    }
2701
2702    /// Return opt-in final-observation phase timings when diagnostics were enabled.
2703    #[doc(hidden)]
2704    pub fn phase_timings(&self) -> Option<ResidentFinalObservationPhaseTimings> {
2705        self.phase_timings
2706    }
2707
2708    /// Number of relation registrations validated by commit.
2709    #[doc(hidden)]
2710    pub fn relation_registration_count(&self) -> usize {
2711        self.relation_registration.len()
2712    }
2713
2714    /// Device execution duration measured between CUDA events around the graph launch.
2715    pub fn device_elapsed_ns(&self) -> u64 {
2716        self.device_elapsed_ns
2717    }
2718
2719    /// Actual scan invocations counted by device kernels, including WHILE replays.
2720    pub fn device_scan_invocations(&self) -> u64 {
2721        self.device_scan_invocations
2722    }
2723
2724    /// Actual filter invocations counted by device kernels, including WHILE replays.
2725    pub fn device_filter_invocations(&self) -> u64 {
2726        self.device_filter_invocations
2727    }
2728
2729    /// Legacy-semantic scan count, excluding recursive variants whose selected delta was empty.
2730    pub fn semantic_scan_invocations(&self) -> u64 {
2731        self.semantic_scan_invocations
2732    }
2733
2734    /// Legacy-semantic filter count, excluding recursive variants whose selected delta was empty.
2735    pub fn semantic_filter_invocations(&self) -> u64 {
2736        self.semantic_filter_invocations
2737    }
2738
2739    /// Number of staged relation heads published by a successful commit.
2740    pub fn staged_output_count(&self) -> u64 {
2741        self.outputs.len() as u64
2742    }
2743
2744    /// Aggregate recursive iterations reported by the device.
2745    pub fn iterations(&self) -> u32 {
2746        self.iterations
2747    }
2748
2749    /// Atomically publish staged outputs after optimistic validation.
2750    pub fn commit(
2751        mut self,
2752        executor: &mut Executor,
2753    ) -> std::result::Result<(), ResidentGraphExecutionError> {
2754        self.terminal?;
2755        if !Arc::ptr_eq(&self.transaction_identity, &executor.transaction_identity)
2756            || !Arc::ptr_eq(&self.provider, &executor.provider)
2757            || self.source_epoch != executor.store.mutation_epoch()
2758        {
2759            return Err(ResidentGraphExecutionError::Runtime(
2760                "resident transaction became stale before commit".into(),
2761            ));
2762        }
2763        let mut current_registration = executor
2764            .rel_names
2765            .iter()
2766            .map(|(rel, name)| (*rel, name.clone()))
2767            .collect::<Vec<_>>();
2768        current_registration.sort_by_key(|(rel, name)| (rel.0, name.clone()));
2769        if current_registration != self.relation_registration {
2770            return Err(ResidentGraphExecutionError::Runtime(
2771                "resident relation registration changed before commit".into(),
2772            ));
2773        }
2774        let additional = self
2775            .outputs
2776            .iter()
2777            .filter(|output| !executor.store.contains(&output.name))
2778            .count();
2779        executor
2780            .store
2781            .try_reserve_relations(additional)
2782            .map_err(runtime_error)?;
2783        executor.common_subexpression_cache.clear();
2784        for output in self.outputs.drain(..) {
2785            if let Some(&rel) = executor.name_to_rel.get(&output.name) {
2786                executor.join_index_cache.invalidate_rel(rel);
2787            }
2788            executor.store.put_owned(output.name, output.buffer);
2789        }
2790        Ok(())
2791    }
2792}
2793
2794fn runtime_error(error: impl std::fmt::Display) -> ResidentGraphExecutionError {
2795    ResidentGraphExecutionError::Runtime(error.to_string())
2796}
2797
2798impl ResidentBuild<'_> {
2799    fn source_reference(&mut self, name: &str) -> Result<ResidentBufferRef> {
2800        if let Some(index) = self.source_aliases.get(name) {
2801            return Ok(ResidentBufferRef::Private(*index));
2802        }
2803        let (row_count, count_tracked, columns_tracked, schema) = {
2804            let source = self.executor.store.get(name).ok_or_else(|| {
2805                XlogError::Execution(format!(
2806                    "resident source {name} disappeared during planning"
2807                ))
2808            })?;
2809            (
2810                resident_source_logical_count(source.cached_row_count())?,
2811                source.num_rows_device().runtime_block().is_some(),
2812                source
2813                    .columns()
2814                    .iter()
2815                    .all(|column| column.runtime_block().is_some()),
2816                source.schema().clone(),
2817            )
2818        };
2819        match resident_source_binding_route(row_count, count_tracked, columns_tracked)? {
2820            ResidentSourceBindingRoute::Direct => {
2821                self.source_names.insert(name.to_owned());
2822                Ok(ResidentBufferRef::Source(name.to_owned()))
2823            }
2824            ResidentSourceBindingRoute::NormalizeEmpty => {
2825                let index = self.allocate_permanent_relation(schema, 0)?;
2826                self.source_aliases.insert(name.to_owned(), index);
2827                Ok(ResidentBufferRef::Private(index))
2828            }
2829        }
2830    }
2831
2832    fn private(&self, index: usize) -> &ResidentLogicalRelation {
2833        &self.relations[index]
2834    }
2835
2836    fn schema(&self, reference: &ResidentBufferRef) -> Result<&Schema> {
2837        match reference {
2838            ResidentBufferRef::Source(name) => self
2839                .executor
2840                .store
2841                .get(name)
2842                .map(CudaBuffer::schema)
2843                .ok_or_else(|| {
2844                    XlogError::Execution(format!("missing resident source relation {name}"))
2845                }),
2846            ResidentBufferRef::Private(index) => Ok(&self.private(*index).schema),
2847        }
2848    }
2849
2850    fn allocate_relation(
2851        &mut self,
2852        schema: Schema,
2853        initial_count: u32,
2854        permanent: bool,
2855    ) -> Result<usize> {
2856        if initial_count > 1 {
2857            return Err(XlogError::Execution(
2858                "resident logical initial count must be zero or one".into(),
2859            ));
2860        }
2861        let index = self.relations.len();
2862        self.relations.push(ResidentLogicalRelation {
2863            schema,
2864            initial_count,
2865            permanent,
2866        });
2867        Ok(index)
2868    }
2869
2870    fn allocate_permanent_relation(&mut self, schema: Schema, initial_count: u32) -> Result<usize> {
2871        self.allocate_relation(schema, initial_count, true)
2872    }
2873
2874    fn allocate_scratch_relation(&mut self, schema: Schema) -> Result<usize> {
2875        self.allocate_relation(schema, 0, false)
2876    }
2877
2878    fn allocation_manifest(
2879        &self,
2880        initial_ops: &[ResidentRecordedOp],
2881        phases: &[ResidentCapturePhase],
2882    ) -> Result<ResidentAllocationManifest> {
2883        let mut definitions = vec![None; self.relations.len()];
2884        let mut last_uses = vec![None; self.relations.len()];
2885        let mut ordinal = 0usize;
2886        let mut phase_ranges = Vec::with_capacity(phases.len() + 1);
2887        phase_ranges.push(resident_record_lifetimes(
2888            initial_ops,
2889            &self.relations,
2890            &mut definitions,
2891            &mut last_uses,
2892            &mut ordinal,
2893        )?);
2894        for phase in phases {
2895            let ops = match phase {
2896                ResidentCapturePhase::Segment { ops, .. }
2897                | ResidentCapturePhase::ConditionalWhile { ops, .. } => ops,
2898            };
2899            phase_ranges.push(resident_record_lifetimes(
2900                ops,
2901                &self.relations,
2902                &mut definitions,
2903                &mut last_uses,
2904                &mut ordinal,
2905            )?);
2906        }
2907
2908        let mut slots = Vec::<ResidentPhysicalSlotPlan>::new();
2909        let mut logical_to_slot = vec![
2910            ResidentSlotAssignment {
2911                slot: usize::MAX,
2912                generation: 0,
2913            };
2914            self.relations.len()
2915        ];
2916        for (logical, relation) in self.relations.iter().enumerate() {
2917            if relation.permanent {
2918                let slot = slots.len();
2919                slots.push(ResidentPhysicalSlotPlan {
2920                    schema: relation.schema.clone(),
2921                    initial_count: relation.initial_count,
2922                    permanent: true,
2923                });
2924                logical_to_slot[logical] = ResidentSlotAssignment {
2925                    slot,
2926                    generation: 0,
2927                };
2928            }
2929        }
2930
2931        let mut scratch_relations = self
2932            .relations
2933            .iter()
2934            .enumerate()
2935            .filter(|(_, relation)| !relation.permanent)
2936            .map(|(logical, relation)| {
2937                let definition = definitions[logical].ok_or_else(|| {
2938                    XlogError::Execution(format!(
2939                        "resident scratch relation {logical} is never defined"
2940                    ))
2941                })?;
2942                let last_use = last_uses[logical].unwrap_or(definition);
2943                if !phase_ranges
2944                    .iter()
2945                    .any(|(start, end)| *start <= definition && last_use < *end)
2946                {
2947                    return Err(XlogError::Execution(format!(
2948                        "resident scratch relation {logical} crosses a capture phase boundary"
2949                    )));
2950                }
2951                Ok((
2952                    logical,
2953                    definition,
2954                    last_use,
2955                    resident_schema_layout(&relation.schema),
2956                ))
2957            })
2958            .collect::<Result<Vec<_>>>()?;
2959        scratch_relations.sort_by_key(|(logical, definition, _, _)| (*definition, *logical));
2960
2961        let mut scratch_slots = Vec::<ResidentScratchSlotState>::new();
2962        for (logical, definition, last_use, layout) in scratch_relations {
2963            let reusable = scratch_slots
2964                .iter_mut()
2965                .filter(|slot| slot.layout == layout && slot.last_use < definition)
2966                .min_by_key(|slot| slot.last_use);
2967            let assignment = if let Some(slot) = reusable {
2968                slot.generation = slot.generation.checked_add(1).ok_or_else(|| {
2969                    XlogError::Execution("resident scratch generation overflow".into())
2970                })?;
2971                slot.last_use = last_use;
2972                ResidentSlotAssignment {
2973                    slot: slot.slot,
2974                    generation: slot.generation,
2975                }
2976            } else {
2977                let slot_index = slots.len();
2978                slots.push(ResidentPhysicalSlotPlan {
2979                    schema: self.relations[logical].schema.clone(),
2980                    initial_count: 0,
2981                    permanent: false,
2982                });
2983                scratch_slots.push(ResidentScratchSlotState {
2984                    slot: slot_index,
2985                    layout,
2986                    last_use,
2987                    generation: 0,
2988                });
2989                ResidentSlotAssignment {
2990                    slot: slot_index,
2991                    generation: 0,
2992                }
2993            };
2994            logical_to_slot[logical] = assignment;
2995        }
2996        if logical_to_slot
2997            .iter()
2998            .any(|assignment| assignment.slot == usize::MAX)
2999        {
3000            return Err(XlogError::Execution(
3001                "resident logical relation has no physical slot assignment".into(),
3002            ));
3003        }
3004        resident_validate_slot_assignments(
3005            &self.relations,
3006            &definitions,
3007            &last_uses,
3008            &slots,
3009            &logical_to_slot,
3010        )?;
3011
3012        let capacity = u64::from(self.capacity);
3013        let mut relation_bytes = 0u64;
3014        let mut max_row_bytes = 1u64;
3015        for slot in &slots {
3016            max_row_bytes = max_row_bytes.max(slot.schema.row_size_bytes() as u64);
3017            resident_checked_add(
3018                &mut relation_bytes,
3019                resident_relation_device_bytes(&slot.schema, capacity)?,
3020                "relation",
3021            )?;
3022        }
3023        let filter_scratch_bytes = if self.filter_workspaces.is_empty() {
3024            0
3025        } else {
3026            resident_filter_scratch_device_bytes(capacity)?
3027        };
3028        let set_candidate_capacity = capacity.checked_mul(2).ok_or_else(|| {
3029            XlogError::Execution("resident set candidate capacity overflow".into())
3030        })?;
3031        let mut fixed_workspace_bytes = 0u64;
3032        for bytes in [
3033            resident_set_workspace_device_bytes(set_candidate_capacity)?,
3034            resident_join_workspace_device_bytes(capacity)?,
3035            resident_control_device_bytes(),
3036            resident_device_trace_bytes(),
3037            resident_schema_winners_device_bytes(self.heads.len())?,
3038            resident_packed_receipt_with_schema_winners_device_bytes(self.heads.len())?,
3039        ] {
3040            resident_checked_add(&mut fixed_workspace_bytes, bytes, "fixed workspace")?;
3041        }
3042        let mut required_bytes = 0u64;
3043        for bytes in [relation_bytes, filter_scratch_bytes, fixed_workspace_bytes] {
3044            resident_checked_add(&mut required_bytes, bytes, "manifest")?;
3045        }
3046
3047        let permanent_slots = u32::try_from(slots.iter().filter(|slot| slot.permanent).count())
3048            .map_err(|_| {
3049                XlogError::Execution("resident permanent slot count exceeds u32".into())
3050            })?;
3051        Ok(ResidentAllocationManifest {
3052            slots,
3053            logical_to_slot,
3054            required_bytes,
3055            relation_bytes,
3056            filter_scratch_bytes,
3057            schedule_metadata_bytes: 0,
3058            fixed_workspace_bytes,
3059            logical_relation_values: self.relations.len(),
3060            permanent_slots,
3061            scratch_slots: u32::try_from(scratch_slots.len()).map_err(|_| {
3062                XlogError::Execution("resident scratch slot count exceeds u32".into())
3063            })?,
3064            filter_scratch_allocations: u32::from(!self.filter_workspaces.is_empty()),
3065            max_row_bytes,
3066        })
3067    }
3068
3069    fn materialize(
3070        &self,
3071        manifest: &ResidentAllocationManifest,
3072        reservation: &mut GpuMemoryReservation,
3073        mut diagnostics: Option<&mut ResidentPrepareDiagnostics>,
3074    ) -> Result<ResidentPhysicalBuild> {
3075        let capacity = u64::from(self.capacity);
3076        let mut relations = Vec::with_capacity(manifest.slots.len());
3077        for slot in &manifest.slots {
3078            let allocation_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3079            let allocation_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3080            let mut relation = self
3081                .executor
3082                .provider
3083                .prepare_resident_relation_in_reservation(
3084                    slot.schema.clone(),
3085                    capacity,
3086                    reservation,
3087                )?;
3088            if let Some(diagnostics) = diagnostics.as_deref_mut() {
3089                let allocation_ns = resident_prepare_elapsed_ns(
3090                    allocation_started.expect("diagnostic timer exists when enabled"),
3091                );
3092                let allocation_bytes = reservation.used_bytes().saturating_sub(
3093                    allocation_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3094                );
3095                diagnostics.relation_slot_allocation_ns = diagnostics
3096                    .relation_slot_allocation_ns
3097                    .saturating_add(allocation_ns);
3098                diagnostics.relation_slot_allocation_ns_max = diagnostics
3099                    .relation_slot_allocation_ns_max
3100                    .max(allocation_ns);
3101                diagnostics.relation_slot_allocation_bytes = diagnostics
3102                    .relation_slot_allocation_bytes
3103                    .saturating_add(allocation_bytes);
3104                diagnostics.relation_device_allocation_calls =
3105                    diagnostics.relation_device_allocation_calls.saturating_add(
3106                        u64::try_from(slot.schema.arity())
3107                            .unwrap_or(u64::MAX)
3108                            .saturating_add(1),
3109                    );
3110            }
3111            let count_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3112            self.executor
3113                .provider
3114                .initialize_resident_relation_count(&mut relation, slot.initial_count)?;
3115            if let Some(diagnostics) = diagnostics.as_deref_mut() {
3116                let count_ns = resident_prepare_elapsed_ns(
3117                    count_started.expect("diagnostic timer exists when enabled"),
3118                );
3119                diagnostics.count_initialization_ns =
3120                    diagnostics.count_initialization_ns.saturating_add(count_ns);
3121                diagnostics.count_initialization_ns_max =
3122                    diagnostics.count_initialization_ns_max.max(count_ns);
3123                diagnostics.count_memset_calls = diagnostics.count_memset_calls.saturating_add(1);
3124            }
3125            relations.push(Some(relation));
3126        }
3127        let filter_scratch = if self.filter_workspaces.is_empty() {
3128            None
3129        } else {
3130            let bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3131            let started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3132            let scratch = self
3133                .executor
3134                .provider
3135                .prepare_resident_filter_scratch_in_reservation(capacity, reservation)?;
3136            if let Some(diagnostics) = diagnostics.as_deref_mut() {
3137                let elapsed_ns = resident_prepare_elapsed_ns(
3138                    started.expect("diagnostic timer exists when enabled"),
3139                );
3140                let reserved_bytes = reservation.used_bytes().saturating_sub(
3141                    bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3142                );
3143                diagnostics.workspace_provider_calls =
3144                    diagnostics.workspace_provider_calls.saturating_add(1);
3145                diagnostics.filter_scratch_allocation_ns = elapsed_ns;
3146                diagnostics.filter_scratch_allocation_bytes = reserved_bytes;
3147            }
3148            Some(scratch)
3149        };
3150        let set_candidate_capacity = capacity.checked_mul(2).ok_or_else(|| {
3151            XlogError::Execution("resident set candidate capacity overflow".into())
3152        })?;
3153        let set_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3154        let set_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3155        let set_workspace = self
3156            .executor
3157            .provider
3158            .prepare_resident_set_workspace_in_reservation(set_candidate_capacity, reservation)?;
3159        if let Some(diagnostics) = diagnostics.as_deref_mut() {
3160            let set_ns = resident_prepare_elapsed_ns(
3161                set_started.expect("diagnostic timer exists when enabled"),
3162            );
3163            let set_reserved_bytes = reservation.used_bytes().saturating_sub(
3164                set_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3165            );
3166            diagnostics.workspace_provider_calls =
3167                diagnostics.workspace_provider_calls.saturating_add(1);
3168            diagnostics.set_workspace_allocation_ns = set_ns;
3169            diagnostics.set_workspace_allocation_bytes = set_reserved_bytes;
3170        }
3171        let join_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3172        let join_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3173        let join_workspace = self
3174            .executor
3175            .provider
3176            .prepare_resident_join_workspace_in_reservation(capacity, reservation)?;
3177        if let Some(diagnostics) = diagnostics.as_deref_mut() {
3178            let join_ns = resident_prepare_elapsed_ns(
3179                join_started.expect("diagnostic timer exists when enabled"),
3180            );
3181            let join_reserved_bytes = reservation.used_bytes().saturating_sub(
3182                join_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3183            );
3184            diagnostics.workspace_provider_calls =
3185                diagnostics.workspace_provider_calls.saturating_add(1);
3186            diagnostics.join_workspace_allocation_ns = join_ns;
3187            diagnostics.join_workspace_allocation_bytes = join_reserved_bytes;
3188        }
3189        let control_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3190        let control_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3191        let control = self
3192            .executor
3193            .provider
3194            .prepare_resident_convergence_control_in_reservation(reservation)?;
3195        if let Some(diagnostics) = diagnostics.as_deref_mut() {
3196            let control_ns = resident_prepare_elapsed_ns(
3197                control_started.expect("diagnostic timer exists when enabled"),
3198            );
3199            let control_reserved_bytes = reservation.used_bytes().saturating_sub(
3200                control_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3201            );
3202            diagnostics.workspace_provider_calls =
3203                diagnostics.workspace_provider_calls.saturating_add(1);
3204            diagnostics.control_allocation_ns = control_ns;
3205            diagnostics.control_allocation_bytes = control_reserved_bytes;
3206        }
3207        Ok(ResidentPhysicalBuild {
3208            relations,
3209            filter_scratch,
3210            set_workspace,
3211            join_workspace,
3212            control,
3213        })
3214    }
3215
3216    fn next_op_id(&mut self) -> Result<u32> {
3217        let op_id = self.next_op_id;
3218        self.next_op_id = self
3219            .next_op_id
3220            .checked_add(1)
3221            .ok_or_else(|| XlogError::Execution("resident physical op id overflow".into()))?;
3222        Ok(op_id)
3223    }
3224
3225    fn push_physical_op(
3226        &mut self,
3227        ops: &mut Vec<ResidentRecordedOp>,
3228        op: ResidentRecordedOp,
3229        op_id: u32,
3230    ) {
3231        ops.push(op);
3232        if !self.injection_recorded
3233            && self
3234                .injection
3235                .as_ref()
3236                .is_some_and(|injection| injection.after_op == op_id)
3237        {
3238            let status =
3239                terminal_status_for_injection(&self.injection.as_ref().expect("checked").status);
3240            ops.push(ResidentRecordedOp::TestStatus(status));
3241            self.injection_recorded = true;
3242        }
3243    }
3244
3245    fn scan_reference(
3246        &mut self,
3247        rel: RelId,
3248        override_scan: Option<(RelId, usize, usize)>,
3249        occurrences: &mut HashMap<RelId, usize>,
3250    ) -> Result<ResidentBufferRef> {
3251        let occurrence = occurrences.entry(rel).or_insert(0);
3252        let current = *occurrence;
3253        *occurrence += 1;
3254        if let Some((target, target_occurrence, delta)) = override_scan {
3255            if rel == target && current == target_occurrence {
3256                return Ok(ResidentBufferRef::Private(delta));
3257            }
3258        }
3259        let name = self
3260            .executor
3261            .rel_names
3262            .get(&rel)
3263            .ok_or_else(|| XlogError::Execution(format!("unknown resident relation id {rel:?}")))?
3264            .clone();
3265        if let Some(index) = self.heads.get(&name) {
3266            Ok(ResidentBufferRef::Private(*index))
3267        } else {
3268            self.source_reference(&name)
3269        }
3270    }
3271
3272    fn plan_node(
3273        &mut self,
3274        node: &RirNode,
3275        override_scan: Option<(RelId, usize, usize)>,
3276        occurrences: &mut HashMap<RelId, usize>,
3277        ops: &mut Vec<ResidentRecordedOp>,
3278    ) -> Result<ResidentBufferRef> {
3279        match node {
3280            RirNode::Unit => {
3281                let op_id = self.next_op_id()?;
3282                let (reference, op) = resident_new_phase_unit(&mut self.relations, op_id)?;
3283                self.push_physical_op(ops, op, op_id);
3284                Ok(reference)
3285            }
3286            RirNode::Scan { rel } => {
3287                let reference = self.scan_reference(*rel, override_scan, occurrences)?;
3288                let op_id = self.next_op_id()?;
3289                Ok(resident_record_scan_leaf(
3290                    reference,
3291                    op_id,
3292                    resident_semantic_trace_guard(override_scan),
3293                    ops,
3294                    |ops, op, op_id| self.push_physical_op(ops, op, op_id),
3295                ))
3296            }
3297            RirNode::Filter { input, predicate } => {
3298                let input_ref = self.plan_node(input, override_scan, occurrences, ops)?;
3299                let input_schema = self.schema(&input_ref)?.clone();
3300                let compact_comparisons =
3301                    resident_compact_filter_descriptors(predicate, &input_schema)?;
3302                let workspace = ResidentFilterPlan {
3303                    compact_comparisons,
3304                };
3305                let workspace_index = self.filter_workspaces.len();
3306                self.filter_workspaces.push(workspace);
3307                let schema = self
3308                    .certificate
3309                    .node_schema(node)
3310                    .ok_or_else(|| XlogError::Execution("resident filter schema missing".into()))?;
3311                let output = self.allocate_scratch_relation(schema)?;
3312                let op_id = self.next_op_id()?;
3313                self.push_physical_op(
3314                    ops,
3315                    ResidentRecordedOp::Filter {
3316                        input: input_ref,
3317                        output,
3318                        workspace: workspace_index,
3319                        op_id,
3320                    },
3321                    op_id,
3322                );
3323                ops.push(ResidentRecordedOp::TraceDelta {
3324                    scan_delta: 0,
3325                    filter_delta: 1,
3326                    semantic_guard: resident_semantic_trace_guard(override_scan),
3327                });
3328                Ok(ResidentBufferRef::Private(output))
3329            }
3330            RirNode::Project { input, columns } => {
3331                let input_ref = self.plan_node(input, override_scan, occurrences, ops)?;
3332                let input_schema = self.schema(&input_ref)?.clone();
3333                let certified_schema = self.certificate.node_schema(node).ok_or_else(|| {
3334                    XlogError::Execution("resident project schema missing".into())
3335                })?;
3336                let schema = self.executor.project_schema(&input_schema, columns)?;
3337                if schema.arity() != certified_schema.arity()
3338                    || (0..schema.arity()).any(|column| {
3339                        schema.column_type(column) != certified_schema.column_type(column)
3340                    })
3341                {
3342                    return Err(XlogError::Execution(
3343                        "resident project physical schema differs from certificate".into(),
3344                    ));
3345                }
3346                let compact_expressions =
3347                    resident_compact_project_descriptors(columns, &input_schema, &schema)?;
3348                let workspace = ResidentProjectPlan {
3349                    compact_expressions,
3350                };
3351                let workspace_index = self.project_workspaces.len();
3352                self.project_workspaces.push(workspace);
3353                let output = self.allocate_scratch_relation(schema)?;
3354                let op_id = self.next_op_id()?;
3355                self.push_physical_op(
3356                    ops,
3357                    ResidentRecordedOp::Project {
3358                        input: input_ref,
3359                        output,
3360                        workspace: workspace_index,
3361                        op_id,
3362                    },
3363                    op_id,
3364                );
3365                Ok(ResidentBufferRef::Private(output))
3366            }
3367            RirNode::Join {
3368                left,
3369                right,
3370                left_keys,
3371                right_keys,
3372                join_type,
3373            } => {
3374                let left_ref = self.plan_node(left, override_scan, occurrences, ops)?;
3375                let right_ref = self.plan_node(right, override_scan, occurrences, ops)?;
3376                let certified_schema = self
3377                    .certificate
3378                    .node_schema(node)
3379                    .ok_or_else(|| XlogError::Execution("resident join schema missing".into()))?;
3380                let kind = match join_type {
3381                    JoinType::Inner => ResidentJoinKind::Inner,
3382                    JoinType::Semi => ResidentJoinKind::Semi,
3383                    _ => return Err(XlogError::Execution("uncertified resident join".into())),
3384                };
3385                // The provider preserves operand column names. The route
3386                // certificate canonicalizes those names and binds the same
3387                // physical column types, so construct the provider's exact
3388                // schema here and normalize names at the next projection.
3389                let left_schema = self.schema(&left_ref)?.clone();
3390                let right_schema = self.schema(&right_ref)?.clone();
3391                let schema = match kind {
3392                    ResidentJoinKind::Semi => left_schema,
3393                    ResidentJoinKind::Inner => {
3394                        let mut columns = left_schema.columns;
3395                        columns.extend(right_schema.columns);
3396                        Schema::new(columns)
3397                    }
3398                };
3399                if schema.arity() != certified_schema.arity()
3400                    || (0..schema.arity()).any(|column| {
3401                        schema.column_type(column) != certified_schema.column_type(column)
3402                    })
3403                {
3404                    return Err(XlogError::Execution(
3405                        "resident join physical schema differs from certificate".into(),
3406                    ));
3407                }
3408                let output = self.allocate_scratch_relation(schema).map_err(|error| {
3409                    XlogError::Execution(format!(
3410                        "resident join intermediate arity {} at {node:?} could not be allocated: {error}",
3411                        certified_schema.arity()
3412                    ))
3413                })?;
3414                let op_id = self.next_op_id()?;
3415                self.push_physical_op(
3416                    ops,
3417                    ResidentRecordedOp::Join {
3418                        kind,
3419                        left: left_ref,
3420                        left_key: left_keys[0],
3421                        right: right_ref,
3422                        right_key: right_keys[0],
3423                        output,
3424                        op_id,
3425                    },
3426                    op_id,
3427                );
3428                Ok(ResidentBufferRef::Private(output))
3429            }
3430            RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
3431                self.plan_node(fallback, override_scan, occurrences, ops)
3432            }
3433            RirNode::Distinct { input, .. } => {
3434                let input_ref = self.plan_node(input, override_scan, occurrences, ops)?;
3435                self.plan_dedup(input_ref, node, ops)
3436            }
3437            RirNode::Diff { left, right } => {
3438                let left_ref = self.plan_node(left, override_scan, occurrences, ops)?;
3439                let right_ref = self.plan_node(right, override_scan, occurrences, ops)?;
3440                self.plan_diff(left_ref, right_ref, node, ops)
3441            }
3442            RirNode::Union { inputs } => {
3443                let fold_mode = resident_union_fold_mode(inputs.len())?;
3444                let mut iter = inputs.iter();
3445                let first = iter
3446                    .next()
3447                    .expect("union fold mode rejected the empty case");
3448                let mut result = self.plan_node(first, override_scan, occurrences, ops)?;
3449                if fold_mode == ResidentUnionFoldMode::SelfUnion {
3450                    return self.plan_union(result.clone(), result, node, ops);
3451                }
3452                for input in iter {
3453                    let right = self.plan_node(input, override_scan, occurrences, ops)?;
3454                    result = self.plan_union(result, right, node, ops)?;
3455                }
3456                Ok(result)
3457            }
3458            RirNode::Fixpoint { .. }
3459            | RirNode::GroupBy { .. }
3460            | RirNode::TensorMaskedJoin { .. } => Err(XlogError::Execution(
3461                "uncertified node reached resident planner".into(),
3462            )),
3463        }
3464    }
3465
3466    fn plan_dedup(
3467        &mut self,
3468        input: ResidentBufferRef,
3469        node: &RirNode,
3470        ops: &mut Vec<ResidentRecordedOp>,
3471    ) -> Result<ResidentBufferRef> {
3472        let schema = self
3473            .certificate
3474            .node_schema(node)
3475            .ok_or_else(|| XlogError::Execution("resident dedup schema missing".into()))?;
3476        let output = self.allocate_scratch_relation(schema)?;
3477        // Dedup is union against itself in the exact full-row set primitive.
3478        let op_id = self.next_op_id()?;
3479        self.push_physical_op(
3480            ops,
3481            ResidentRecordedOp::Union {
3482                left: input.clone(),
3483                right: input,
3484                output,
3485                op_id,
3486            },
3487            op_id,
3488        );
3489        Ok(ResidentBufferRef::Private(output))
3490    }
3491
3492    fn plan_union(
3493        &mut self,
3494        left: ResidentBufferRef,
3495        right: ResidentBufferRef,
3496        schema_node: &RirNode,
3497        ops: &mut Vec<ResidentRecordedOp>,
3498    ) -> Result<ResidentBufferRef> {
3499        let schema = self
3500            .certificate
3501            .node_schema(schema_node)
3502            .ok_or_else(|| XlogError::Execution("resident union schema missing".into()))?;
3503        self.plan_union_schema(left, right, schema, ops)
3504    }
3505
3506    fn plan_union_schema(
3507        &mut self,
3508        left: ResidentBufferRef,
3509        right: ResidentBufferRef,
3510        schema: Schema,
3511        ops: &mut Vec<ResidentRecordedOp>,
3512    ) -> Result<ResidentBufferRef> {
3513        let output = self.allocate_scratch_relation(schema)?;
3514        let op_id = self.next_op_id()?;
3515        self.push_physical_op(
3516            ops,
3517            ResidentRecordedOp::Union {
3518                left,
3519                right,
3520                output,
3521                op_id,
3522            },
3523            op_id,
3524        );
3525        Ok(ResidentBufferRef::Private(output))
3526    }
3527
3528    fn plan_diff(
3529        &mut self,
3530        left: ResidentBufferRef,
3531        right: ResidentBufferRef,
3532        schema_node: &RirNode,
3533        ops: &mut Vec<ResidentRecordedOp>,
3534    ) -> Result<ResidentBufferRef> {
3535        let schema = self
3536            .certificate
3537            .node_schema(schema_node)
3538            .ok_or_else(|| XlogError::Execution("resident diff schema missing".into()))?;
3539        self.plan_diff_schema(left, right, schema, ops)
3540    }
3541
3542    fn plan_diff_schema(
3543        &mut self,
3544        left: ResidentBufferRef,
3545        right: ResidentBufferRef,
3546        schema: Schema,
3547        ops: &mut Vec<ResidentRecordedOp>,
3548    ) -> Result<ResidentBufferRef> {
3549        let output = self.allocate_scratch_relation(schema)?;
3550        let op_id = self.next_op_id()?;
3551        self.push_physical_op(
3552            ops,
3553            ResidentRecordedOp::Diff {
3554                left,
3555                right,
3556                output,
3557                op_id,
3558            },
3559            op_id,
3560        );
3561        Ok(ResidentBufferRef::Private(output))
3562    }
3563
3564    fn plan_copy(
3565        &mut self,
3566        input: ResidentBufferRef,
3567        output: usize,
3568        schema: &Schema,
3569        ops: &mut Vec<ResidentRecordedOp>,
3570    ) -> Result<()> {
3571        let compact_expressions = (0..schema.arity())
3572            .map(|column| {
3573                let scalar = schema.column_type(column).ok_or_else(|| {
3574                    XlogError::Execution("resident identity project column is missing".into())
3575                })?;
3576                Ok(ResidentProjectExpressionDescriptor::column(
3577                    u32::try_from(column).map_err(|_| {
3578                        XlogError::Execution("resident project column exceeds u32".into())
3579                    })?,
3580                    u32::try_from(scalar.size_bytes()).map_err(|_| {
3581                        XlogError::Execution("resident project scalar width exceeds u32".into())
3582                    })?,
3583                ))
3584            })
3585            .collect::<Result<Vec<_>>>()?;
3586        let workspace = ResidentProjectPlan {
3587            compact_expressions,
3588        };
3589        let workspace_index = self.project_workspaces.len();
3590        self.project_workspaces.push(workspace);
3591        let op_id = self.next_op_id()?;
3592        self.push_physical_op(
3593            ops,
3594            ResidentRecordedOp::Project {
3595                input,
3596                output,
3597                workspace: workspace_index,
3598                op_id,
3599            },
3600            op_id,
3601        );
3602        Ok(())
3603    }
3604
3605    fn normalize_to_schema(
3606        &mut self,
3607        input: ResidentBufferRef,
3608        schema: &Schema,
3609        ops: &mut Vec<ResidentRecordedOp>,
3610    ) -> Result<ResidentBufferRef> {
3611        let input_schema = self.schema(&input)?;
3612        if input_schema == schema {
3613            return Ok(input);
3614        }
3615        if input_schema.arity() != schema.arity()
3616            || input_schema
3617                .columns
3618                .iter()
3619                .zip(&schema.columns)
3620                .any(|(left, right)| left.1 != right.1)
3621        {
3622            return Err(XlogError::Execution(format!(
3623                "resident rule output schema {:?} cannot be normalized to head schema {:?}",
3624                input_schema, schema
3625            )));
3626        }
3627        let output = self.allocate_scratch_relation(schema.clone())?;
3628        self.plan_copy(input, output, schema, ops)?;
3629        Ok(ResidentBufferRef::Private(output))
3630    }
3631
3632    fn merge_phase_contribution(
3633        &mut self,
3634        current: Option<ResidentBufferRef>,
3635        contribution: ResidentBufferRef,
3636        schema: &Schema,
3637        ops: &mut Vec<ResidentRecordedOp>,
3638    ) -> Result<ResidentBufferRef> {
3639        let contribution = self.normalize_to_schema(contribution, schema, ops)?;
3640        resident_phase_merge(current, contribution, |step| match step {
3641            ResidentPhaseMergeStep::Deduplicate(input) => {
3642                self.plan_union_schema(input.clone(), input, schema.clone(), ops)
3643            }
3644            ResidentPhaseMergeStep::Union(left, right) => {
3645                self.plan_union_schema(left, right, schema.clone(), ops)
3646            }
3647        })
3648    }
3649}
3650
3651fn private_relation(
3652    relations: &[Option<ResidentRelation>],
3653    index: usize,
3654) -> Result<&ResidentRelation> {
3655    relations
3656        .get(index)
3657        .and_then(Option::as_ref)
3658        .ok_or_else(|| XlogError::Execution(format!("resident relation slot {index} is missing")))
3659}
3660
3661fn collect_recorded_scans(node: &RirNode, output: &mut Vec<RelId>) {
3662    match node {
3663        RirNode::Unit | RirNode::TensorMaskedJoin { .. } => {}
3664        RirNode::Scan { rel } => output.push(*rel),
3665        RirNode::Filter { input, .. }
3666        | RirNode::Project { input, .. }
3667        | RirNode::GroupBy { input, .. }
3668        | RirNode::Distinct { input, .. } => collect_recorded_scans(input, output),
3669        RirNode::Join { left, right, .. } | RirNode::Diff { left, right } => {
3670            collect_recorded_scans(left, output);
3671            collect_recorded_scans(right, output);
3672        }
3673        RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
3674            collect_recorded_scans(fallback, output);
3675        }
3676        RirNode::Union { inputs } => {
3677            for input in inputs {
3678                collect_recorded_scans(input, output);
3679            }
3680        }
3681        RirNode::Fixpoint {
3682            base, recursive, ..
3683        } => {
3684            collect_recorded_scans(base, output);
3685            collect_recorded_scans(recursive, output);
3686        }
3687    }
3688}
3689
3690fn resident_decline_error(reason: ResidentGraphDeclineReason) -> ResidentGraphExecutionError {
3691    ResidentGraphExecutionError::Declined(reason)
3692}
3693
3694fn conditional_graph_error(error: CudaConditionalGraphUnavailable) -> ResidentGraphExecutionError {
3695    if error.is_unsupported() {
3696        resident_decline_error(ResidentGraphDeclineReason::ConditionalGraphUnavailable {
3697            detail: error.decline_detail(),
3698        })
3699    } else {
3700        runtime_error(error.decline_detail())
3701    }
3702}
3703
3704fn read_receipt_u32(bytes: &[u8], offset: usize) -> Option<u32> {
3705    bytes
3706        .get(offset..offset.checked_add(4)?)?
3707        .try_into()
3708        .ok()
3709        .map(u32::from_le_bytes)
3710}
3711
3712fn read_receipt_u64(bytes: &[u8], offset: usize) -> Option<u64> {
3713    bytes
3714        .get(offset..offset.checked_add(8)?)?
3715        .try_into()
3716        .ok()
3717        .map(u64::from_le_bytes)
3718}
3719
3720fn resident_compact_scalar(value: &ConstValue) -> Result<(ScalarType, u64)> {
3721    match value {
3722        ConstValue::Symbol(value) => Ok((ScalarType::Symbol, u64::from(symbol::intern(value)))),
3723        ConstValue::U32(value) => Ok((ScalarType::U32, u64::from(*value))),
3724        ConstValue::U64(value) => Ok((ScalarType::U64, *value)),
3725        _ => Err(XlogError::Execution(
3726            "uncertified resident constant type".into(),
3727        )),
3728    }
3729}
3730
3731fn resident_compact_filter_descriptors(
3732    expression: &Expr,
3733    input_schema: &Schema,
3734) -> Result<Vec<ResidentFilterComparisonDescriptor>> {
3735    fn operand(expression: &Expr, input_schema: &Schema) -> Result<(u32, u32, ScalarType, u64)> {
3736        match expression {
3737            Expr::Column(column) => {
3738                let scalar = input_schema.column_type(*column).ok_or_else(|| {
3739                    XlogError::Execution("resident filter column is out of range".into())
3740                })?;
3741                Ok((
3742                    0,
3743                    u32::try_from(*column).map_err(|_| {
3744                        XlogError::Execution("resident filter column exceeds u32".into())
3745                    })?,
3746                    scalar,
3747                    0,
3748                ))
3749            }
3750            Expr::Const(value) => {
3751                let (scalar, bits) = resident_compact_scalar(value)?;
3752                Ok((1, 0, scalar, bits))
3753            }
3754            _ => Err(XlogError::Execution(
3755                "uncertified resident filter operand".into(),
3756            )),
3757        }
3758    }
3759
3760    fn append(
3761        expression: &Expr,
3762        input_schema: &Schema,
3763        output: &mut Vec<ResidentFilterComparisonDescriptor>,
3764    ) -> Result<()> {
3765        match expression {
3766            Expr::And(parts) => {
3767                for part in parts {
3768                    append(part, input_schema, output)?;
3769                }
3770                Ok(())
3771            }
3772            Expr::Compare { left, op, right } => {
3773                let (left_kind, left_column, left_type, left_constant) =
3774                    operand(left, input_schema)?;
3775                let (right_kind, right_column, right_type, right_constant) =
3776                    operand(right, input_schema)?;
3777                if left_type != right_type {
3778                    return Err(XlogError::Execution(
3779                        "resident filter operand scalar types differ".into(),
3780                    ));
3781                }
3782                let width = u32::try_from(left_type.size_bytes()).map_err(|_| {
3783                    XlogError::Execution("resident filter scalar width exceeds u32".into())
3784                })?;
3785                if !matches!(width, 4 | 8) {
3786                    return Err(XlogError::Execution(
3787                        "resident filter scalar width is unsupported".into(),
3788                    ));
3789                }
3790                output.push(ResidentFilterComparisonDescriptor {
3791                    left_kind,
3792                    left_column,
3793                    right_kind,
3794                    right_column,
3795                    op: u32::from(resident_compare_op(*op) as u8),
3796                    width,
3797                    reserved_zero: 0,
3798                    reserved_one: 0,
3799                    left_constant,
3800                    right_constant,
3801                });
3802                Ok(())
3803            }
3804            _ => Err(XlogError::Execution(
3805                "uncertified resident filter expression".into(),
3806            )),
3807        }
3808    }
3809
3810    let mut output = Vec::new();
3811    append(expression, input_schema, &mut output)?;
3812    Ok(output)
3813}
3814
3815fn resident_compact_project_descriptors(
3816    columns: &[ProjectExpr],
3817    input_schema: &Schema,
3818    output_schema: &Schema,
3819) -> Result<Vec<ResidentProjectExpressionDescriptor>> {
3820    if columns.len() != output_schema.arity() {
3821        return Err(XlogError::Execution(
3822            "resident project expression count differs from output arity".into(),
3823        ));
3824    }
3825    columns
3826        .iter()
3827        .enumerate()
3828        .map(|(output_column, expression)| {
3829            let output_type = output_schema.column_type(output_column).ok_or_else(|| {
3830                XlogError::Execution("resident project output column is out of range".into())
3831            })?;
3832            let width = u32::try_from(output_type.size_bytes()).map_err(|_| {
3833                XlogError::Execution("resident project scalar width exceeds u32".into())
3834            })?;
3835            if !matches!(width, 4 | 8) {
3836                return Err(XlogError::Execution(
3837                    "resident project scalar width is unsupported".into(),
3838                ));
3839            }
3840            match expression {
3841                ProjectExpr::Column(input_column) => {
3842                    if input_schema.column_type(*input_column) != Some(output_type) {
3843                        return Err(XlogError::Execution(
3844                            "resident project column scalar type differs from output".into(),
3845                        ));
3846                    }
3847                    Ok(ResidentProjectExpressionDescriptor::column(
3848                        u32::try_from(*input_column).map_err(|_| {
3849                            XlogError::Execution("resident project column exceeds u32".into())
3850                        })?,
3851                        width,
3852                    ))
3853                }
3854                ProjectExpr::Computed(Expr::Const(value), declared_type) => {
3855                    let (constant_type, bits) = resident_compact_scalar(value)?;
3856                    if constant_type != *declared_type || constant_type != output_type {
3857                        return Err(XlogError::Execution(
3858                            "resident project constant scalar type differs from output".into(),
3859                        ));
3860                    }
3861                    Ok(ResidentProjectExpressionDescriptor::constant(width, bits))
3862                }
3863                _ => Err(XlogError::Execution(
3864                    "uncertified resident projection expression".into(),
3865                )),
3866            }
3867        })
3868        .collect()
3869}
3870
3871fn resident_compare_op(op: RirCompareOp) -> CudaCompareOp {
3872    match op {
3873        RirCompareOp::Eq => CudaCompareOp::Eq,
3874        RirCompareOp::Ne => CudaCompareOp::Ne,
3875        RirCompareOp::Lt => CudaCompareOp::Lt,
3876        RirCompareOp::Le => CudaCompareOp::Le,
3877        RirCompareOp::Gt => CudaCompareOp::Gt,
3878        RirCompareOp::Ge => CudaCompareOp::Ge,
3879    }
3880}
3881
3882fn terminal_status_for_injection(status: &ResidentGraphDeviceStatus) -> ResidentTerminalStatus {
3883    match status {
3884        ResidentGraphDeviceStatus::Success { iterations } => ResidentTerminalStatus {
3885            code: ResidentTerminalCode::Success as u32,
3886            iterations: *iterations,
3887            ..ResidentTerminalStatus::default()
3888        },
3889        ResidentGraphDeviceStatus::IterationLimit { limit, completed } => ResidentTerminalStatus {
3890            code: ResidentTerminalCode::IterationLimit as u32,
3891            iterations: *completed,
3892            limit: *limit,
3893            ..ResidentTerminalStatus::default()
3894        },
3895        ResidentGraphDeviceStatus::CapacityOverflow {
3896            op_id,
3897            required,
3898            capacity,
3899        } => ResidentTerminalStatus {
3900            code: ResidentTerminalCode::CapacityOverflow as u32,
3901            op_id: *op_id,
3902            resource_code: ResidentResourceCode::OutputRows as u32,
3903            required: *required,
3904            capacity: *capacity,
3905            ..ResidentTerminalStatus::default()
3906        },
3907        ResidentGraphDeviceStatus::ResourceExhausted {
3908            op_id,
3909            required,
3910            capacity,
3911            ..
3912        } => ResidentTerminalStatus {
3913            code: ResidentTerminalCode::ResourceExhausted as u32,
3914            op_id: *op_id,
3915            resource_code: ResidentResourceCode::SetHashSlots as u32,
3916            required: *required,
3917            capacity: *capacity,
3918            ..ResidentTerminalStatus::default()
3919        },
3920    }
3921}
3922
3923fn validate_compact_resident_node_envelope(
3924    node: &RirNode,
3925    schema_for: &impl Fn(&RirNode) -> std::result::Result<Schema, ResidentGraphDeclineReason>,
3926) -> std::result::Result<(), ResidentGraphDeclineReason> {
3927    match node {
3928        RirNode::Unit | RirNode::Scan { .. } => Ok(()),
3929        RirNode::Filter { input, .. } | RirNode::Project { input, .. } => {
3930            validate_compact_resident_node_envelope(input, schema_for)
3931        }
3932        RirNode::Distinct { input, key_cols } => {
3933            validate_compact_resident_node_envelope(input, schema_for)?;
3934            let arity = schema_for(input)?.arity();
3935            if key_cols.len() != arity
3936                || key_cols
3937                    .iter()
3938                    .copied()
3939                    .enumerate()
3940                    .any(|(column, key)| column != key)
3941            {
3942                return Err(resident_workspace_decline(format!(
3943                    "compact resident Distinct requires canonical full-row key columns 0..{arity}"
3944                )));
3945            }
3946            Ok(())
3947        }
3948        RirNode::Union { inputs } => {
3949            if inputs.is_empty() {
3950                return Err(resident_workspace_decline(
3951                    "compact resident Union requires at least one input",
3952                ));
3953            }
3954            for input in inputs {
3955                validate_compact_resident_node_envelope(input, schema_for)?;
3956            }
3957            Ok(())
3958        }
3959        RirNode::Diff { left, right } => {
3960            validate_compact_resident_node_envelope(left, schema_for)?;
3961            validate_compact_resident_node_envelope(right, schema_for)
3962        }
3963        RirNode::Join {
3964            left,
3965            right,
3966            left_keys,
3967            right_keys,
3968            join_type,
3969        } => {
3970            if !matches!(join_type, JoinType::Inner | JoinType::Semi) {
3971                return Err(resident_workspace_decline(format!(
3972                    "compact resident Join does not support {join_type:?}"
3973                )));
3974            }
3975            if left_keys.len() != 1 || right_keys.len() != 1 {
3976                return Err(resident_workspace_decline(
3977                    "compact resident Join requires exactly one key per input",
3978                ));
3979            }
3980            validate_compact_resident_node_envelope(left, schema_for)?;
3981            validate_compact_resident_node_envelope(right, schema_for)?;
3982            let left_schema = schema_for(left)?;
3983            let right_schema = schema_for(right)?;
3984            let left_key = left_keys[0];
3985            let right_key = right_keys[0];
3986            if left_key >= left_schema.arity() || right_key >= right_schema.arity() {
3987                return Err(resident_workspace_decline(
3988                    "compact resident Join key is outside its input schema",
3989                ));
3990            }
3991            let left_key_type = left_schema
3992                .column_type(left_key)
3993                .expect("left key bounds checked");
3994            let right_key_type = right_schema
3995                .column_type(right_key)
3996                .expect("right key bounds checked");
3997            if left_key_type != right_key_type
3998                || !matches!(
3999                    left_key_type,
4000                    ScalarType::U32 | ScalarType::U64 | ScalarType::Symbol
4001                )
4002            {
4003                return Err(resident_workspace_decline(
4004                    "compact resident Join requires matching U32, U64, or Symbol key types",
4005                ));
4006            }
4007            if left_key_type.size_bytes() != right_key_type.size_bytes() {
4008                return Err(resident_workspace_decline(
4009                    "compact resident Join key widths differ",
4010                ));
4011            }
4012            let output_arity = match join_type {
4013                JoinType::Inner => left_schema
4014                    .arity()
4015                    .checked_add(right_schema.arity())
4016                    .ok_or_else(|| {
4017                        resident_workspace_decline("compact resident Join arity overflow")
4018                    })?,
4019                JoinType::Semi => left_schema.arity(),
4020                _ => unreachable!("join kind checked above"),
4021            };
4022            if output_arity > 17 {
4023                return Err(resident_workspace_decline(
4024                    "compact resident Join output arity exceeds 17",
4025                ));
4026            }
4027            Ok(())
4028        }
4029        RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
4030            validate_compact_resident_node_envelope(fallback, schema_for)
4031        }
4032        RirNode::Fixpoint { .. } | RirNode::GroupBy { .. } | RirNode::TensorMaskedJoin { .. } => {
4033            Err(resident_workspace_decline(
4034                "compact resident route contains an unsupported node",
4035            ))
4036        }
4037    }
4038}
4039
4040fn resident_generated_query_heads(
4041    plan: &ExecutionPlan,
4042) -> std::result::Result<BTreeSet<String>, ResidentGraphDeclineReason> {
4043    let mut heads = BTreeSet::new();
4044    let mut rule_positions = HashSet::new();
4045    for (position, provenance) in plan.generated_query_rules.iter().enumerate() {
4046        if provenance.query_index != position {
4047            return Err(resident_workspace_decline(format!(
4048                "generated query provenance position {position} carries query index {}",
4049                provenance.query_index
4050            )));
4051        }
4052        if !rule_positions.insert((provenance.scc_index, provenance.rule_index)) {
4053            return Err(resident_workspace_decline(format!(
4054                "generated query provenance {} reuses compiled rule scc={} rule={}",
4055                provenance.query_index, provenance.scc_index, provenance.rule_index
4056            )));
4057        }
4058        let expected_head = format!("__xlog_query_{}", provenance.query_index);
4059        let rule = plan
4060            .rules_by_scc
4061            .get(provenance.scc_index)
4062            .and_then(|rules| rules.get(provenance.rule_index))
4063            .ok_or_else(|| {
4064                resident_workspace_decline(format!(
4065                    "generated query provenance {} references missing compiled rule scc={} rule={}",
4066                    provenance.query_index, provenance.scc_index, provenance.rule_index
4067                ))
4068            })?;
4069        if rule.head != expected_head {
4070            return Err(resident_workspace_decline(format!(
4071                "generated query provenance {} expects head {expected_head} but references authored head {}",
4072                provenance.query_index, rule.head
4073            )));
4074        }
4075        let occurrence_count = plan
4076            .rules_by_scc
4077            .iter()
4078            .flatten()
4079            .filter(|candidate| candidate.head == expected_head)
4080            .count();
4081        if occurrence_count != 1 {
4082            return Err(resident_workspace_decline(format!(
4083                "generated query head {expected_head} must have exactly one compiled rule, found {occurrence_count}"
4084            )));
4085        }
4086        heads.insert(expected_head);
4087    }
4088    Ok(heads)
4089}
4090
4091impl Executor {
4092    /// Prepare one immutable, fixed-capacity conditional graph transaction.
4093    pub fn prepare_resident_graph<'executor>(
4094        &'executor self,
4095        plan: &ExecutionPlan,
4096        certificate: &ResidentGraphRouteCertificate,
4097        options: ResidentGraphPrepareOptions,
4098    ) -> std::result::Result<PreparedResidentGraph<'executor>, ResidentGraphExecutionError> {
4099        if !certificate.matches_plan(plan).map_err(runtime_error)? {
4100            return Err(resident_decline_error(
4101                ResidentGraphDeclineReason::WorkspaceUnbounded {
4102                    detail: "route certificate does not match the plan being prepared".into(),
4103                },
4104            ));
4105        }
4106        if !certificate.is_supported() {
4107            return Err(resident_decline_error(
4108                certificate.declines().first().cloned().unwrap_or_else(|| {
4109                    ResidentGraphDeclineReason::WorkspaceUnbounded {
4110                        detail: "route certificate is not resident-capable".into(),
4111                    }
4112                }),
4113            ));
4114        }
4115
4116        self.prepare_resident_graph_after_certification(plan, certificate, options)
4117    }
4118
4119    /// Prepare a transaction from a certificate sealed to its exact immutable plan.
4120    pub fn prepare_certified_resident_graph<'executor>(
4121        &'executor self,
4122        certified: &ResidentGraphCertifiedPlan,
4123        options: ResidentGraphPrepareOptions,
4124    ) -> std::result::Result<PreparedResidentGraph<'executor>, ResidentGraphExecutionError> {
4125        let certificate = certified.certificate();
4126        if !certificate.is_supported() {
4127            return Err(resident_decline_error(
4128                certificate.declines().first().cloned().unwrap_or_else(|| {
4129                    ResidentGraphDeclineReason::WorkspaceUnbounded {
4130                        detail: "route certificate is not resident-capable".into(),
4131                    }
4132                }),
4133            ));
4134        }
4135        self.prepare_resident_graph_after_certification(certified.plan(), certificate, options)
4136    }
4137
4138    fn prepare_resident_graph_after_certification<'executor>(
4139        &'executor self,
4140        plan: &ExecutionPlan,
4141        certificate: &ResidentGraphRouteCertificate,
4142        options: ResidentGraphPrepareOptions,
4143    ) -> std::result::Result<PreparedResidentGraph<'executor>, ResidentGraphExecutionError> {
4144        let mut diagnostics =
4145            resident_prepare_diagnostics_for_sample(options.latency_diagnostic_sample);
4146        let total_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4147        let admission_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4148        let source_epoch = self.store.mutation_epoch();
4149        let resident_head_names = plan
4150            .rules_by_scc
4151            .iter()
4152            .flat_map(|rules| rules.iter().map(|rule| rule.head.clone()))
4153            .collect::<HashSet<_>>();
4154        let mut source_set_snapshots = Vec::new();
4155        let mut scans = Vec::new();
4156        for rules in &plan.rules_by_scc {
4157            for rule in rules {
4158                collect_recorded_scans(&rule.body, &mut scans);
4159            }
4160        }
4161        scans.sort_by_key(|rel| rel.0);
4162        scans.dedup();
4163        for rel in scans {
4164            let name = self
4165                .rel_names
4166                .get(&rel)
4167                .ok_or_else(|| runtime_error(format!("unknown resident relation id {rel:?}")))?;
4168            match self.store.get_with_version(name) {
4169                Some((buffer, version)) => {
4170                    let schema = certificate.schema_for(rel).ok_or_else(|| {
4171                        runtime_error(format!("resident scan {name} has no certified schema"))
4172                    })?;
4173                    if buffer.schema() != schema {
4174                        return Err(runtime_error(format!(
4175                            "resident scan {name} store schema differs from its certificate"
4176                        )));
4177                    }
4178                    source_set_snapshots.push(
4179                        resident_source_set_snapshot(&self.provider, name, version, buffer)
4180                            .map_err(resident_decline_error)?,
4181                    );
4182                }
4183                None if resident_head_names.contains(name) => {}
4184                None => {
4185                    return Err(runtime_error(format!(
4186                        "missing resident source relation {name}"
4187                    )))
4188                }
4189            }
4190        }
4191        let admission = self
4192            .resident_workspace_admission_after_certification(plan, certificate)
4193            .map_err(resident_decline_error)?;
4194        for (name, schema) in &admission.head_schemas {
4195            if let Some(existing) = self.store.get(name) {
4196                if !resident_schemas_type_compatible(existing.schema(), schema) {
4197                    return Err(runtime_error(format!(
4198                        "existing resident head {name} has a physically incompatible schema: existing={:?}, staged={schema:?}",
4199                        existing.schema()
4200                    )));
4201                }
4202            }
4203        }
4204        if let Some(diagnostics) = diagnostics.as_mut() {
4205            diagnostics.admission_and_source_snapshot_ns = resident_prepare_elapsed_ns(
4206                admission_started.expect("diagnostic timer exists when enabled"),
4207            );
4208        }
4209
4210        let execution_setup_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4211        let runtime =
4212            Arc::clone(self.provider.memory().runtime().ok_or_else(|| {
4213                runtime_error("resident graph requires the async device runtime")
4214            })?);
4215        let stream_id = if let Some(stream_id) = self.resident_graph_stream.get() {
4216            *stream_id
4217        } else {
4218            let acquired = runtime.stream_pool().acquire().map_err(runtime_error)?;
4219            match self.resident_graph_stream.set(acquired) {
4220                Ok(()) => acquired,
4221                Err(_) => *self
4222                    .resident_graph_stream
4223                    .get()
4224                    .ok_or_else(|| runtime_error("resident stream initialization raced"))?,
4225            }
4226        };
4227        let stream = runtime
4228            .stream_pool()
4229            .resolve(stream_id)
4230            .ok_or_else(|| runtime_error("resident graph stream is no longer live"))?;
4231        let execution_domain = self
4232            .provider
4233            .bind_resident_execution_domain(Arc::clone(&runtime), stream_id, Arc::clone(&stream))
4234            .map_err(runtime_error)?;
4235
4236        let mut relation_registration = self
4237            .rel_names
4238            .iter()
4239            .map(|(rel, name)| (*rel, name.clone()))
4240            .collect::<Vec<_>>();
4241        relation_registration.sort_by_key(|(rel, name)| (rel.0, name.clone()));
4242
4243        let mut build = ResidentBuild {
4244            executor: self,
4245            certificate,
4246            capacity: admission.relation_capacity,
4247            relations: Vec::new(),
4248            filter_workspaces: Vec::new(),
4249            project_workspaces: Vec::new(),
4250            heads: BTreeMap::new(),
4251            head_winner_indices: BTreeMap::new(),
4252            source_names: HashSet::new(),
4253            source_aliases: HashMap::new(),
4254            next_op_id: 0,
4255            injection: options.test_device_status,
4256            injection_recorded: false,
4257        };
4258        if let Some(diagnostics) = diagnostics.as_mut() {
4259            diagnostics.execution_domain_and_build_setup_ns = resident_prepare_elapsed_ns(
4260                execution_setup_started.expect("diagnostic timer exists when enabled"),
4261            );
4262        }
4263        let logical_planning_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4264        for (winner_index, (name, schema)) in admission.head_schemas.iter().enumerate() {
4265            let index = build
4266                .allocate_permanent_relation(schema.clone(), 0)
4267                .map_err(runtime_error)?;
4268            build.heads.insert(name.clone(), index);
4269            build.head_winner_indices.insert(
4270                name.clone(),
4271                u32::try_from(winner_index)
4272                    .map_err(|_| runtime_error("resident head count exceeds u32"))?,
4273            );
4274        }
4275
4276        let mut initial_ops = Vec::new();
4277        for (name, schema) in &admission.head_schemas {
4278            if self.store.contains(name) {
4279                let source = build.source_reference(name).map_err(runtime_error)?;
4280                build
4281                    .plan_copy(source.clone(), build.heads[name], schema, &mut initial_ops)
4282                    .map_err(runtime_error)?;
4283                resident_record_schema_winner_mark(
4284                    &mut initial_ops,
4285                    ResidentBufferRef::Private(build.heads[name]),
4286                    build.head_winner_indices[name],
4287                    0,
4288                );
4289            }
4290        }
4291
4292        let mut scc_by_id = HashMap::new();
4293        for (index, scc) in plan.sccs.iter().enumerate() {
4294            if scc_by_id.insert(scc.id, index).is_some() {
4295                return Err(runtime_error(format!(
4296                    "duplicate resident SCC id {}",
4297                    scc.id
4298                )));
4299            }
4300        }
4301        let ordered_sccs = if plan.strata.is_empty() {
4302            (0..plan.sccs.len()).collect::<Vec<_>>()
4303        } else {
4304            let mut ordered = Vec::new();
4305            for stratum in &plan.strata {
4306                for scc_id in &stratum.sccs {
4307                    ordered.push(*scc_by_id.get(scc_id).ok_or_else(|| {
4308                        runtime_error(format!(
4309                            "stratum {} references unknown SCC {scc_id}",
4310                            stratum.id
4311                        ))
4312                    })?);
4313                }
4314            }
4315            ordered
4316        };
4317        let mut seen_sccs = HashSet::new();
4318        for index in &ordered_sccs {
4319            if !seen_sccs.insert(*index) {
4320                return Err(runtime_error(format!(
4321                    "resident execution schedule repeats SCC {}",
4322                    plan.sccs[*index].id
4323                )));
4324            }
4325        }
4326        if seen_sccs.len() != plan.sccs.len() {
4327            return Err(runtime_error(
4328                "resident execution schedule omits one or more SCCs",
4329            ));
4330        }
4331
4332        let mut phases = Vec::new();
4333        for scc_index in ordered_sccs {
4334            let scc = &plan.sccs[scc_index];
4335            let rules = plan.rules_by_scc.get(scc_index).ok_or_else(|| {
4336                runtime_error(format!("resident SCC {scc_index} has no rule vector"))
4337            })?;
4338            if rules.is_empty() {
4339                continue;
4340            }
4341            if !scc.is_recursive {
4342                let mut ops = Vec::new();
4343                let mut phase_heads = BTreeMap::<String, ResidentBufferRef>::new();
4344                for (rule_index, rule) in rules.iter().enumerate() {
4345                    let mut occurrences = HashMap::new();
4346                    let contribution = build
4347                        .plan_node(&rule.body, None, &mut occurrences, &mut ops)
4348                        .map_err(runtime_error)?;
4349                    let schema_id =
4350                        admission.rule_schema_ids[scc_index][rule_index].ok_or_else(|| {
4351                            runtime_error(format!(
4352                                "resident rule {} has no schema winner id",
4353                                rule.head
4354                            ))
4355                        })?;
4356                    ops.push(ResidentRecordedOp::SchemaWinnerMark {
4357                        contribution: contribution.clone(),
4358                        head_index: build.head_winner_indices[&rule.head],
4359                        schema_id,
4360                    });
4361                    let target = *build.heads.get(&rule.head).ok_or_else(|| {
4362                        runtime_error(format!("resident head {} was not staged", rule.head))
4363                    })?;
4364                    let target_schema = build
4365                        .schema(&ResidentBufferRef::Private(target))
4366                        .map_err(runtime_error)?
4367                        .clone();
4368                    let current = phase_heads
4369                        .remove(&rule.head)
4370                        .unwrap_or(ResidentBufferRef::Private(target));
4371                    let merged = build
4372                        .merge_phase_contribution(
4373                            Some(current),
4374                            contribution,
4375                            &target_schema,
4376                            &mut ops,
4377                        )
4378                        .map_err(runtime_error)?;
4379                    phase_heads.insert(rule.head.clone(), merged);
4380                }
4381                for (head, value) in phase_heads {
4382                    let target = build.heads[&head];
4383                    let target_schema = build
4384                        .schema(&ResidentBufferRef::Private(target))
4385                        .map_err(runtime_error)?
4386                        .clone();
4387                    build
4388                        .plan_copy(value, target, &target_schema, &mut ops)
4389                        .map_err(runtime_error)?;
4390                }
4391                phases.push(ResidentCapturePhase::Segment {
4392                    ops,
4393                    scc_begin: None,
4394                });
4395                continue;
4396            }
4397
4398            let mut recursive_heads = BTreeMap::new();
4399            for rule in rules {
4400                recursive_heads
4401                    .entry(rule.head.clone())
4402                    .or_insert_with(|| rule.meta.schema.clone());
4403            }
4404            let mut delta_heads = BTreeMap::new();
4405            for (name, schema) in &recursive_heads {
4406                delta_heads.insert(
4407                    name.clone(),
4408                    build
4409                        .allocate_permanent_relation(schema.clone(), 0)
4410                        .map_err(runtime_error)?,
4411                );
4412            }
4413
4414            let mut seed_ops = Vec::new();
4415            let mut seed_values = BTreeMap::<String, ResidentBufferRef>::new();
4416            for (rule_index, rule) in rules.iter().enumerate() {
4417                let mut occurrences = HashMap::new();
4418                let contribution = build
4419                    .plan_node(&rule.body, None, &mut occurrences, &mut seed_ops)
4420                    .map_err(runtime_error)?;
4421                let schema_id =
4422                    admission.rule_schema_ids[scc_index][rule_index].ok_or_else(|| {
4423                        runtime_error(format!(
4424                            "resident rule {} has no schema winner id",
4425                            rule.head
4426                        ))
4427                    })?;
4428                seed_ops.push(ResidentRecordedOp::SchemaWinnerMark {
4429                    contribution: contribution.clone(),
4430                    head_index: build.head_winner_indices[&rule.head],
4431                    schema_id,
4432                });
4433                let current = seed_values.remove(&rule.head);
4434                let merged = build
4435                    .merge_phase_contribution(
4436                        current,
4437                        contribution,
4438                        &rule.meta.schema,
4439                        &mut seed_ops,
4440                    )
4441                    .map_err(runtime_error)?;
4442                seed_values.insert(rule.head.clone(), merged);
4443            }
4444            for (name, schema) in &recursive_heads {
4445                let full = build.heads[name];
4446                let seed = seed_values.remove(name).ok_or_else(|| {
4447                    runtime_error(format!(
4448                        "recursive resident seed has no contribution for {name}"
4449                    ))
4450                })?;
4451                let candidate = build
4452                    .plan_union_schema(
4453                        ResidentBufferRef::Private(full),
4454                        seed,
4455                        schema.clone(),
4456                        &mut seed_ops,
4457                    )
4458                    .map_err(runtime_error)?;
4459                let novel = build
4460                    .plan_diff_schema(
4461                        candidate.clone(),
4462                        ResidentBufferRef::Private(full),
4463                        schema.clone(),
4464                        &mut seed_ops,
4465                    )
4466                    .map_err(runtime_error)?;
4467                build
4468                    .plan_copy(candidate, full, schema, &mut seed_ops)
4469                    .map_err(runtime_error)?;
4470                build
4471                    .plan_copy(novel, delta_heads[name], schema, &mut seed_ops)
4472                    .map_err(runtime_error)?;
4473            }
4474            let begin_op_id = build.next_op_id().map_err(runtime_error)?;
4475            phases.push(ResidentCapturePhase::Segment {
4476                ops: seed_ops,
4477                scc_begin: Some((self.config.max_iterations, begin_op_id)),
4478            });
4479
4480            let mut body_ops = vec![ResidentRecordedOp::ChangedReset];
4481            let mut body_values = BTreeMap::<String, ResidentBufferRef>::new();
4482            let recursive_names = recursive_heads.keys().cloned().collect::<HashSet<_>>();
4483            for (rule_index, rule) in rules.iter().enumerate() {
4484                let mut rule_scans = Vec::new();
4485                collect_recorded_scans(&rule.body, &mut rule_scans);
4486                let mut per_relation_occurrence = HashMap::<RelId, usize>::new();
4487                for rel in rule_scans {
4488                    let occurrence = per_relation_occurrence.entry(rel).or_insert(0);
4489                    let current = *occurrence;
4490                    *occurrence += 1;
4491                    let Some(name) = self.rel_names.get(&rel) else {
4492                        return Err(runtime_error(format!(
4493                            "recursive resident scan has unknown relation id {rel:?}"
4494                        )));
4495                    };
4496                    if !recursive_names.contains(name) {
4497                        continue;
4498                    }
4499                    let mut planning_occurrences = HashMap::new();
4500                    let contribution = build
4501                        .plan_node(
4502                            &rule.body,
4503                            Some((rel, current, delta_heads[name])),
4504                            &mut planning_occurrences,
4505                            &mut body_ops,
4506                        )
4507                        .map_err(runtime_error)?;
4508                    let schema_id =
4509                        admission.rule_schema_ids[scc_index][rule_index].ok_or_else(|| {
4510                            runtime_error(format!(
4511                                "resident rule {} has no schema winner id",
4512                                rule.head
4513                            ))
4514                        })?;
4515                    body_ops.push(ResidentRecordedOp::SchemaWinnerMark {
4516                        contribution: contribution.clone(),
4517                        head_index: build.head_winner_indices[&rule.head],
4518                        schema_id,
4519                    });
4520                    let current = body_values.remove(&rule.head);
4521                    let merged = build
4522                        .merge_phase_contribution(
4523                            current,
4524                            contribution,
4525                            &rule.meta.schema,
4526                            &mut body_ops,
4527                        )
4528                        .map_err(runtime_error)?;
4529                    body_values.insert(rule.head.clone(), merged);
4530                }
4531            }
4532            for (name, schema) in &recursive_heads {
4533                let full = build.heads[name];
4534                let accumulated = match body_values.remove(name) {
4535                    Some(value) => value,
4536                    None => build
4537                        .plan_diff_schema(
4538                            ResidentBufferRef::Private(full),
4539                            ResidentBufferRef::Private(full),
4540                            schema.clone(),
4541                            &mut body_ops,
4542                        )
4543                        .map_err(runtime_error)?,
4544                };
4545                let novel = build
4546                    .plan_diff_schema(
4547                        accumulated,
4548                        ResidentBufferRef::Private(full),
4549                        schema.clone(),
4550                        &mut body_ops,
4551                    )
4552                    .map_err(runtime_error)?;
4553                let next_full = build
4554                    .plan_union_schema(
4555                        ResidentBufferRef::Private(full),
4556                        novel.clone(),
4557                        schema.clone(),
4558                        &mut body_ops,
4559                    )
4560                    .map_err(runtime_error)?;
4561                build
4562                    .plan_copy(next_full, full, schema, &mut body_ops)
4563                    .map_err(runtime_error)?;
4564                build
4565                    .plan_copy(novel, delta_heads[name], schema, &mut body_ops)
4566                    .map_err(runtime_error)?;
4567                // The transient Diff slot can be reused by later body operations. Drive
4568                // convergence from the completed permanent delta instead.
4569                body_ops.push(ResidentRecordedOp::ChangedMark {
4570                    relation: delta_heads[name],
4571                });
4572            }
4573            let convergence_op_id = build.next_op_id().map_err(runtime_error)?;
4574            phases.push(ResidentCapturePhase::ConditionalWhile {
4575                ops: body_ops,
4576                iteration_limit: self.config.max_iterations,
4577                convergence_op_id,
4578            });
4579        }
4580
4581        let success_op_id = build.next_op_id().map_err(runtime_error)?;
4582        if let Some(diagnostics) = diagnostics.as_mut() {
4583            diagnostics.logical_schedule_planning_ns = resident_prepare_elapsed_ns(
4584                logical_planning_started.expect("diagnostic timer exists when enabled"),
4585            );
4586        }
4587        let manifest_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4588        let mut manifest = build
4589            .allocation_manifest(&initial_ops, &phases)
4590            .map_err(runtime_error)?;
4591        let compact_tables =
4592            resident_compact_tables(&build.filter_workspaces, &build.project_workspaces)
4593                .map_err(runtime_error)?;
4594        let compact_regions =
4595            resident_compact_regions(initial_ops, phases, success_op_id).map_err(runtime_error)?;
4596        if let Some(diagnostics) = diagnostics.as_mut() {
4597            diagnostics.manifest_compact_construction_ns = resident_prepare_elapsed_ns(
4598                manifest_started.expect("diagnostic timer exists when enabled"),
4599            );
4600            diagnostics.compact_regions = compact_regions.len();
4601            diagnostics.conditional_regions = compact_regions
4602                .iter()
4603                .filter(|region| region.iteration_limit != 0)
4604                .count();
4605        }
4606        let schedule_lowering_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4607        let compact_schedule = resident_lower_compact_regions(
4608            compact_regions,
4609            &manifest.slots,
4610            &manifest.logical_to_slot,
4611            build.source_names.iter().map(String::as_str),
4612            compact_tables,
4613        )
4614        .map_err(runtime_error)?;
4615        manifest
4616            .finalize_compact_schedule(&compact_schedule, build.heads.len())
4617            .map_err(runtime_error)?;
4618        if let Some(diagnostics) = diagnostics.as_mut() {
4619            diagnostics.schedule_lowering_ns = resident_prepare_elapsed_ns(
4620                schedule_lowering_started.expect("diagnostic timer exists when enabled"),
4621            );
4622            diagnostics.required_reservation_bytes = manifest.required_bytes;
4623            diagnostics.logical_relation_values = manifest.logical_relation_values;
4624            diagnostics.physical_relation_slots = manifest.slots.len();
4625            diagnostics.compact_ops = compact_schedule.ops.len();
4626            diagnostics.compact_waves = compact_schedule.waves.len();
4627        }
4628        let reservation_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4629        let manifest_available_bytes = self.provider.memory().remaining_bytes();
4630        let mut reservation = self
4631            .provider
4632            .memory()
4633            .reserve_bytes(manifest.required_bytes)
4634            .map_err(|error| {
4635                resident_decline_error(ResidentGraphDeclineReason::WorkspaceUnbounded {
4636                    detail: format!(
4637                        "resident allocation manifest reservation of {} bytes failed: {error}",
4638                        manifest.required_bytes
4639                    ),
4640                })
4641            })?;
4642        if let Some(diagnostics) = diagnostics.as_mut() {
4643            diagnostics.reservation_ns = resident_prepare_elapsed_ns(
4644                reservation_started.expect("diagnostic timer exists when enabled"),
4645            );
4646        }
4647        let physical = build
4648            .materialize(&manifest, &mut reservation, diagnostics.as_mut())
4649            .map_err(runtime_error)?;
4650
4651        if build.injection.is_some() && !build.injection_recorded {
4652            return Err(runtime_error(
4653                "test device status requested after a nonexistent physical op",
4654            ));
4655        }
4656        let metadata_binding_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4657        let output_indices =
4658            resident_output_indices(&build.heads, &manifest.logical_to_slot, &manifest.slots)
4659                .map_err(runtime_error)?;
4660        let output_relations = output_indices
4661            .iter()
4662            .map(|(_, index)| private_relation(&physical.relations, *index))
4663            .collect::<Result<Vec<_>>>()
4664            .map_err(runtime_error)?;
4665        let output_schema_plans = output_indices
4666            .iter()
4667            .map(|(name, _)| {
4668                let candidates = admission
4669                    .head_schema_choices
4670                    .get(name)
4671                    .cloned()
4672                    .ok_or_else(|| {
4673                        runtime_error(format!("resident head {name} has no schema plan"))
4674                    })?;
4675                let selection = if let Some(selection) = admission.head_schema_selections.get(name)
4676                {
4677                    let source_output = usize::try_from(
4678                        *build
4679                            .head_winner_indices
4680                            .get(&selection.source_head)
4681                            .ok_or_else(|| {
4682                                runtime_error(format!(
4683                                    "resident schema source {} is not staged",
4684                                    selection.source_head
4685                                ))
4686                            })?,
4687                    )
4688                    .map_err(|_| runtime_error("resident schema source index overflow"))?;
4689                    let source_candidates = admission
4690                        .head_schema_choices
4691                        .get(&selection.source_head)
4692                        .ok_or_else(|| runtime_error("resident schema source is missing"))?;
4693                    if source_candidates.len() != selection.output_schemas_by_source_winner.len() {
4694                        return Err(runtime_error(
4695                            "resident schema source mapping has the wrong length",
4696                        ));
4697                    }
4698                    ResidentOutputSchemaSelection::SourceWinner {
4699                        source_output,
4700                        schemas: selection.output_schemas_by_source_winner.clone(),
4701                    }
4702                } else {
4703                    ResidentOutputSchemaSelection::OwnWinner
4704                };
4705                Ok(ResidentOutputSchemaPlan {
4706                    candidates,
4707                    selection,
4708                })
4709            })
4710            .collect::<std::result::Result<Vec<_>, ResidentGraphExecutionError>>()?;
4711        if let Some(diagnostics) = diagnostics.as_mut() {
4712            let metadata_binding_ns = resident_prepare_elapsed_ns(
4713                metadata_binding_started.expect("diagnostic timer exists when enabled"),
4714            );
4715            diagnostics.metadata_binding_construction_ns = diagnostics
4716                .metadata_binding_construction_ns
4717                .saturating_add(metadata_binding_ns);
4718        }
4719        let device_trace_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4720        let device_trace_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4721        let device_trace = self
4722            .provider
4723            .prepare_resident_device_trace_in_reservation(&mut reservation)
4724            .map_err(runtime_error)?;
4725        if let Some(diagnostics) = diagnostics.as_mut() {
4726            let device_trace_ns = resident_prepare_elapsed_ns(
4727                device_trace_started.expect("diagnostic timer exists when enabled"),
4728            );
4729            let device_trace_reserved_bytes = reservation.used_bytes().saturating_sub(
4730                device_trace_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
4731            );
4732            diagnostics.metadata_provider_calls =
4733                diagnostics.metadata_provider_calls.saturating_add(1);
4734            diagnostics.device_trace_preparation_ns = device_trace_ns;
4735            diagnostics.device_trace_reserved_bytes = device_trace_reserved_bytes;
4736        }
4737        let schema_defaults_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4738        let default_schema_ids =
4739            resident_compact_schema_defaults(&compact_schedule.ops, output_relations.len())
4740                .map_err(runtime_error)?;
4741        if let Some(diagnostics) = diagnostics.as_mut() {
4742            let schema_defaults_ns = resident_prepare_elapsed_ns(
4743                schema_defaults_started.expect("diagnostic timer exists when enabled"),
4744            );
4745            diagnostics.metadata_binding_construction_ns = diagnostics
4746                .metadata_binding_construction_ns
4747                .saturating_add(schema_defaults_ns);
4748        }
4749        let schema_winners_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4750        let schema_winners_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4751        let schema_winners = self
4752            .provider
4753            .prepare_resident_schema_winners_in_reservation(&default_schema_ids, &mut reservation)
4754            .map_err(runtime_error)?;
4755        if let Some(diagnostics) = diagnostics.as_mut() {
4756            let schema_winners_ns = resident_prepare_elapsed_ns(
4757                schema_winners_started.expect("diagnostic timer exists when enabled"),
4758            );
4759            let schema_winners_reserved_bytes = reservation.used_bytes().saturating_sub(
4760                schema_winners_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
4761            );
4762            diagnostics.metadata_provider_calls =
4763                diagnostics.metadata_provider_calls.saturating_add(1);
4764            diagnostics.schema_winners_preparation_ns = schema_winners_ns;
4765            diagnostics.schema_winners_reserved_bytes = schema_winners_reserved_bytes;
4766            (
4767                diagnostics.schema_winners_initial_htod_calls,
4768                diagnostics.schema_winners_initial_htod_bytes,
4769            ) = resident_schema_winners_initial_htod(default_schema_ids.len());
4770        }
4771        let receipt_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4772        let receipt_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4773        let receipt = self
4774            .provider
4775            .prepare_resident_packed_receipt_with_trace_and_schema_winners_in_reservation(
4776                &output_relations,
4777                &device_trace,
4778                &schema_winners,
4779                &mut reservation,
4780            )
4781            .map_err(runtime_error)?;
4782        if let Some(diagnostics) = diagnostics.as_mut() {
4783            let receipt_ns = resident_prepare_elapsed_ns(
4784                receipt_started.expect("diagnostic timer exists when enabled"),
4785            );
4786            let receipt_reserved_bytes = reservation.used_bytes().saturating_sub(
4787                receipt_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
4788            );
4789            diagnostics.metadata_provider_calls =
4790                diagnostics.metadata_provider_calls.saturating_add(1);
4791            diagnostics.receipt_preparation_ns = receipt_ns;
4792            diagnostics.receipt_reserved_bytes = receipt_reserved_bytes;
4793            (
4794                diagnostics.receipt_initial_htod_calls,
4795                diagnostics.receipt_initial_htod_bytes,
4796            ) = resident_receipt_initial_htod(output_relations.len());
4797        }
4798        let schedule_bindings_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4799        let mut schedule_bindings = Vec::with_capacity(
4800            manifest
4801                .slots
4802                .len()
4803                .checked_add(compact_schedule.source_slots.len())
4804                .ok_or_else(|| runtime_error("resident compact slot binding count overflow"))?,
4805        );
4806        for (slot_index, slot) in manifest.slots.iter().enumerate() {
4807            let relation =
4808                private_relation(&physical.relations, slot_index).map_err(runtime_error)?;
4809            schedule_bindings.push(if slot.permanent {
4810                ResidentScheduleSlotBinding::permanent(relation.buffer(), 0)
4811            } else {
4812                ResidentScheduleSlotBinding::scratch(relation.buffer(), 0)
4813            });
4814        }
4815        for (name, slot) in &compact_schedule.source_slots {
4816            let expected_slot = u32::try_from(schedule_bindings.len())
4817                .map_err(|_| runtime_error("resident compact slot binding count exceeds u32"))?;
4818            if *slot != expected_slot {
4819                return Err(runtime_error(
4820                    "resident compact source bindings are not contiguous",
4821                ));
4822            }
4823            let source = self.store.get(name).ok_or_else(|| {
4824                runtime_error(format!("resident source {name} disappeared during prepare"))
4825            })?;
4826            schedule_bindings
4827                .push(ResidentScheduleSlotBinding::source(source, 0).map_err(runtime_error)?);
4828        }
4829        let receipt_slots = output_indices
4830            .iter()
4831            .map(|(_, slot)| {
4832                u32::try_from(*slot)
4833                    .map_err(|_| runtime_error("resident receipt slot index exceeds u32"))
4834            })
4835            .collect::<std::result::Result<Vec<_>, _>>()?;
4836        if let Some(diagnostics) = diagnostics.as_mut() {
4837            let schedule_bindings_ns = resident_prepare_elapsed_ns(
4838                schedule_bindings_started.expect("diagnostic timer exists when enabled"),
4839            );
4840            diagnostics.metadata_binding_construction_ns = diagnostics
4841                .metadata_binding_construction_ns
4842                .saturating_add(schedule_bindings_ns);
4843        }
4844        let schedule_program_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4845        let schedule_program_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4846        let schedule_program = self
4847            .provider
4848            .prepare_resident_schedule_program_in_reservation(
4849                &execution_domain,
4850                &schedule_bindings,
4851                &compact_schedule.ops,
4852                &compact_schedule.waves,
4853                &compact_schedule.regions,
4854                &compact_schedule.generation_bases,
4855                &compact_schedule.filter_comparisons,
4856                &compact_schedule.project_expressions,
4857                &receipt_slots,
4858                ResidentScheduleExternalBindings::new(
4859                    physical.filter_scratch.as_ref(),
4860                    &physical.set_workspace,
4861                    &physical.join_workspace,
4862                    &physical.control,
4863                    &device_trace,
4864                    &schema_winners,
4865                    &receipt,
4866                ),
4867                &mut reservation,
4868            )
4869            .map_err(runtime_error)?;
4870        if let Some(diagnostics) = diagnostics.as_mut() {
4871            let schedule_program_ns = resident_prepare_elapsed_ns(
4872                schedule_program_started.expect("diagnostic timer exists when enabled"),
4873            );
4874            let schedule_program_reserved_bytes = reservation.used_bytes().saturating_sub(
4875                schedule_program_bytes_before
4876                    .expect("diagnostic byte snapshot exists when enabled"),
4877            );
4878            diagnostics.metadata_provider_calls =
4879                diagnostics.metadata_provider_calls.saturating_add(1);
4880            diagnostics.schedule_program_preparation_ns = schedule_program_ns;
4881            diagnostics.schedule_program_reserved_bytes = schedule_program_reserved_bytes;
4882            (
4883                diagnostics.schedule_program_initial_htod_calls,
4884                diagnostics.schedule_program_initial_htod_bytes,
4885            ) = resident_schedule_initial_htod(diagnostics.schedule_program_reserved_bytes);
4886        }
4887        let reservation_validation_started =
4888            diagnostics.as_ref().map(|_| std::time::Instant::now());
4889        let tracked_device_allocation_bytes = reservation.used_bytes();
4890        resident_validate_exact_reservation(
4891            manifest.required_bytes,
4892            tracked_device_allocation_bytes,
4893            reservation.remaining_bytes(),
4894        )
4895        .map_err(runtime_error)?;
4896        drop(reservation);
4897        if let Some(diagnostics) = diagnostics.as_mut() {
4898            diagnostics.reservation_validation_and_release_ns = resident_prepare_elapsed_ns(
4899                reservation_validation_started.expect("diagnostic timer exists when enabled"),
4900            );
4901        }
4902        let pinned_receipt_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4903        let pinned_receipt = self
4904            .provider
4905            .prepare_resident_pinned_receipt(&receipt)
4906            .map_err(runtime_error)?;
4907        if let Some(diagnostics) = diagnostics.as_mut() {
4908            diagnostics.pinned_receipt_ns = resident_prepare_elapsed_ns(
4909                pinned_receipt_started.expect("diagnostic timer exists when enabled"),
4910            );
4911        }
4912
4913        let graph_capture_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4914        let mut sequence =
4915            ConditionalCudaGraphSequenceBuilder::new(&stream).map_err(conditional_graph_error)?;
4916        let capture_topology =
4917            resident_compact_topology(&compact_schedule.regions).map_err(runtime_error)?;
4918        let mut conditional_body_node_kinds =
4919            Vec::with_capacity(capture_topology.conditional_body_kernel_counts.len());
4920        for ((region_index, region), parent_kind) in compact_schedule
4921            .regions
4922            .iter()
4923            .enumerate()
4924            .zip(capture_topology.parent_kinds.iter().copied())
4925        {
4926            let region_index = u32::try_from(region_index)
4927                .map_err(|_| runtime_error("resident compact region index exceeds u32"))?;
4928            if parent_kind == ResidentCaptureParentKind::Conditional {
4929                sequence
4930                    .add_conditional_while(u32::from(region.iteration_limit != 0), true, |body| {
4931                        body.capture_on_stream(&stream, || unsafe {
4932                            self.provider.record_resident_schedule_region_on_stream(
4933                                &schedule_program,
4934                                region_index,
4935                                Some(&body),
4936                                &stream,
4937                            )
4938                        })?;
4939                        conditional_body_node_kinds.push(body.linear_chain_node_kinds()?);
4940                        Ok(())
4941                    })
4942                    .map_err(conditional_graph_error)?;
4943            } else {
4944                sequence
4945                    .capture_segment_on_stream(&stream, || unsafe {
4946                        self.provider.record_resident_schedule_region_on_stream(
4947                            &schedule_program,
4948                            region_index,
4949                            None,
4950                            &stream,
4951                        )
4952                    })
4953                    .map_err(conditional_graph_error)?;
4954            }
4955        }
4956        if let Some(diagnostics) = diagnostics.as_mut() {
4957            diagnostics.graph_body_capture_ns = resident_prepare_elapsed_ns(
4958                graph_capture_started.expect("diagnostic timer exists when enabled"),
4959            );
4960        }
4961        let graph_instantiate_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4962        let graph = sequence
4963            .instantiate()
4964            .map_err(conditional_graph_error)?
4965            .bind_resident_lifecycle(&runtime);
4966        if let Some(diagnostics) = diagnostics.as_mut() {
4967            diagnostics.graph_instantiate_ns = resident_prepare_elapsed_ns(
4968                graph_instantiate_started.expect("diagnostic timer exists when enabled"),
4969            );
4970        }
4971
4972        let validation_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4973        let mut recorder = execution_domain.new_strict_recorder();
4974        for binding in &schedule_bindings {
4975            binding.record_uses(&mut recorder);
4976        }
4977        ResidentScheduleExternalBindings::new(
4978            physical.filter_scratch.as_ref(),
4979            &physical.set_workspace,
4980            &physical.join_workspace,
4981            &physical.control,
4982            &device_trace,
4983            &schema_winners,
4984            &receipt,
4985        )
4986        .record_uses(&mut recorder);
4987        schedule_program.record_uses(&mut recorder);
4988
4989        let graph_node_kinds = graph.linear_chain_node_kinds().map_err(runtime_error)?;
4990        resident_validate_parent_graph_kinds(&graph_node_kinds, &capture_topology.parent_kinds)
4991            .map_err(runtime_error)?;
4992        let conditional_body_kernel_counts = resident_validate_conditional_body_node_kinds(
4993            &conditional_body_node_kinds,
4994            capture_topology.conditional_body_kernel_counts.len(),
4995        )
4996        .map_err(runtime_error)?;
4997        if conditional_body_kernel_counts != capture_topology.conditional_body_kernel_counts {
4998            return Err(runtime_error(
4999                "resident conditional body topology differs from the compact schedule",
5000            ));
5001        }
5002        let body_node_count =
5003            conditional_body_node_kinds
5004                .iter()
5005                .try_fold(0usize, |total, kinds| {
5006                    total.checked_add(kinds.len()).ok_or_else(|| {
5007                        runtime_error("resident hierarchical graph node count overflow")
5008                    })
5009                })?;
5010        let hierarchical_graph_nodes = graph_node_kinds
5011            .len()
5012            .checked_add(body_node_count)
5013            .ok_or_else(|| runtime_error("resident hierarchical graph node count overflow"))?;
5014        if hierarchical_graph_nodes != capture_topology.hierarchical_node_count {
5015            return Err(runtime_error(
5016                "resident hierarchical graph topology differs from the compact schedule",
5017            ));
5018        }
5019        let (
5020            filter_descriptor_device_bytes,
5021            project_descriptor_device_bytes,
5022            fixed_workspace_device_bytes,
5023        ) = resident_compact_preflight_device_bytes(&manifest, &compact_schedule)
5024            .map_err(runtime_error)?;
5025        let parent_graph_node_count = graph_node_kinds.len();
5026        let preflight_report = ResidentGraphPreflightReport {
5027            relation_capacity: admission.relation_capacity,
5028            estimated_required_bytes: manifest.required_bytes,
5029            available_bytes_at_admission: manifest_available_bytes,
5030            tracked_device_allocation_bytes,
5031            relation_device_bytes: manifest.relation_bytes,
5032            filter_descriptor_device_bytes,
5033            filter_scratch_device_bytes: manifest.filter_scratch_bytes,
5034            project_descriptor_device_bytes,
5035            fixed_workspace_device_bytes,
5036            parent_graph_nodes: graph_node_kinds.len(),
5037            conditional_while_nodes: graph_node_kinds
5038                .iter()
5039                .filter(|kind| **kind == CudaGraphNodeKind::Conditional)
5040                .count(),
5041            parent_graph_node_kinds: graph_node_kinds,
5042            conditional_body_node_kinds,
5043            conditional_body_kernel_counts,
5044            hierarchical_graph_nodes,
5045            private_relation_slots: physical.relations.len(),
5046            logical_relation_values: manifest.logical_relation_values,
5047            permanent_relation_slots: manifest.permanent_slots,
5048            staged_output_relations: output_indices.len(),
5049            scratch_slots: manifest.scratch_slots,
5050            filter_scratch_allocations: manifest.filter_scratch_allocations,
5051            max_row_bytes: manifest.max_row_bytes,
5052        };
5053        let has_device_status_writer = build.injection_recorded;
5054        let owners = ResidentRunOwners {
5055            provider: Arc::clone(&self.provider),
5056            runtime,
5057            stream,
5058            graph,
5059            execution_domain,
5060            schedule_program,
5061            recorder,
5062            relations: physical.relations,
5063            output_indices,
5064            filter_scratch: physical.filter_scratch,
5065            set_workspace: physical.set_workspace,
5066            join_workspace: physical.join_workspace,
5067            control: physical.control,
5068            device_trace,
5069            schema_winners,
5070            receipt,
5071            pinned_receipt,
5072            source_epoch,
5073            relation_registration,
5074            transaction_identity: Arc::clone(&self.transaction_identity),
5075            output_schema_plans,
5076        };
5077        let validation_owner_assembly_ns = diagnostics.as_ref().map(|_| {
5078            resident_prepare_elapsed_ns(
5079                validation_started.expect("diagnostic timer exists when enabled"),
5080            )
5081        });
5082        let prepare_total_ns = diagnostics.as_ref().map(|_| {
5083            resident_prepare_elapsed_ns(
5084                total_started.expect("diagnostic total timer exists when enabled"),
5085            )
5086        });
5087        let prepare_diagnostic = diagnostics.map(|mut diagnostics| {
5088            diagnostics.total_ns =
5089                prepare_total_ns.expect("diagnostic total sample exists when enabled");
5090            diagnostics.validation_owner_assembly_ns = validation_owner_assembly_ns
5091                .expect("diagnostic validation sample exists when enabled");
5092            diagnostics.parent_graph_nodes = parent_graph_node_count;
5093            diagnostics.conditional_body_nodes = body_node_count;
5094            diagnostics
5095        });
5096        Ok(PreparedResidentGraph {
5097            owners,
5098            preflight_report,
5099            has_device_status_writer,
5100            source_guard: self,
5101            source_set_snapshots,
5102            prepare_diagnostic,
5103        })
5104    }
5105
5106    fn resident_schema_variants(
5107        &self,
5108        node: &RirNode,
5109        certificate: &ResidentGraphRouteCertificate,
5110        head_schema_choices: &BTreeMap<String, Vec<Schema>>,
5111    ) -> std::result::Result<ResidentSchemaVariants, ResidentGraphDeclineReason> {
5112        match node {
5113            RirNode::Unit => certificate
5114                .node_schema(node)
5115                .map(ResidentSchemaVariants::Fixed)
5116                .ok_or_else(|| resident_workspace_decline("resident node schema missing")),
5117            RirNode::Scan { rel } => {
5118                let name = self.rel_names.get(rel).ok_or_else(|| {
5119                    resident_workspace_decline(format!(
5120                        "resident scan relation {rel:?} is not registered"
5121                    ))
5122                })?;
5123                if let Some(schemas) = head_schema_choices.get(name) {
5124                    return resident_schema_variants_from_source(name.clone(), schemas.clone());
5125                }
5126                let schema = self
5127                    .store
5128                    .get(name)
5129                    .map(|buffer| buffer.schema().clone())
5130                    .or_else(|| certificate.schema_for(*rel).cloned())
5131                    .ok_or_else(|| {
5132                        resident_workspace_decline(format!("resident scan {name} has no schema"))
5133                    })?;
5134                Ok(ResidentSchemaVariants::Fixed(schema))
5135            }
5136            RirNode::Filter { input, .. } | RirNode::Distinct { input, .. } => {
5137                self.resident_schema_variants(input, certificate, head_schema_choices)
5138            }
5139            RirNode::Project { input, columns } => {
5140                match self.resident_schema_variants(input, certificate, head_schema_choices)? {
5141                    ResidentSchemaVariants::Fixed(schema) => self
5142                        .project_schema(&schema, columns)
5143                        .map(ResidentSchemaVariants::Fixed)
5144                        .map_err(|error| resident_workspace_decline(error.to_string())),
5145                    ResidentSchemaVariants::Dynamic {
5146                        source_head,
5147                        schemas,
5148                    } => {
5149                        let schemas = schemas
5150                            .iter()
5151                            .map(|schema| self.project_schema(schema, columns))
5152                            .collect::<Result<Vec<_>>>()
5153                            .map_err(|error| resident_workspace_decline(error.to_string()))?;
5154                        resident_schema_variants_from_source(source_head, schemas)
5155                    }
5156                }
5157            }
5158            RirNode::Join {
5159                left,
5160                right,
5161                join_type,
5162                ..
5163            } => {
5164                let left = self.resident_schema_variants(left, certificate, head_schema_choices)?;
5165                match join_type {
5166                    JoinType::Semi => Ok(left),
5167                    JoinType::Inner => {
5168                        let right =
5169                            self.resident_schema_variants(right, certificate, head_schema_choices)?;
5170                        match (left, right) {
5171                            (
5172                                ResidentSchemaVariants::Fixed(left),
5173                                ResidentSchemaVariants::Fixed(right),
5174                            ) => {
5175                                let mut columns = left.columns;
5176                                columns.extend(right.columns);
5177                                Ok(ResidentSchemaVariants::Fixed(Schema::new(columns)))
5178                            }
5179                            (
5180                                ResidentSchemaVariants::Dynamic {
5181                                    source_head,
5182                                    schemas,
5183                                },
5184                                ResidentSchemaVariants::Fixed(right),
5185                            ) => {
5186                                let schemas = schemas
5187                                    .into_iter()
5188                                    .map(|left| {
5189                                        let mut columns = left.columns;
5190                                        columns.extend(right.columns.iter().cloned());
5191                                        Schema::new(columns)
5192                                    })
5193                                    .collect();
5194                                resident_schema_variants_from_source(source_head, schemas)
5195                            }
5196                            (
5197                                ResidentSchemaVariants::Fixed(left),
5198                                ResidentSchemaVariants::Dynamic {
5199                                    source_head,
5200                                    schemas,
5201                                },
5202                            ) => {
5203                                let schemas = schemas
5204                                    .into_iter()
5205                                    .map(|right| {
5206                                        let mut columns = left.columns.clone();
5207                                        columns.extend(right.columns);
5208                                        Schema::new(columns)
5209                                    })
5210                                    .collect();
5211                                resident_schema_variants_from_source(source_head, schemas)
5212                            }
5213                            (
5214                                ResidentSchemaVariants::Dynamic {
5215                                    source_head: left_source,
5216                                    schemas: left_schemas,
5217                                },
5218                                ResidentSchemaVariants::Dynamic {
5219                                    source_head: right_source,
5220                                    schemas: right_schemas,
5221                                },
5222                            ) => {
5223                                if left_source != right_source
5224                                    || left_schemas.len() != right_schemas.len()
5225                                {
5226                                    return Err(resident_workspace_decline(
5227                                        "resident join has multiple schema sources",
5228                                    ));
5229                                }
5230                                let schemas = left_schemas
5231                                    .into_iter()
5232                                    .zip(right_schemas)
5233                                    .map(|(left, right)| {
5234                                        let mut columns = left.columns;
5235                                        columns.extend(right.columns);
5236                                        Schema::new(columns)
5237                                    })
5238                                    .collect();
5239                                resident_schema_variants_from_source(left_source, schemas)
5240                            }
5241                        }
5242                    }
5243                    _ => Err(resident_workspace_decline(
5244                        "uncertified join reached resident schema planning",
5245                    )),
5246                }
5247            }
5248            RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
5249                self.resident_schema_variants(fallback, certificate, head_schema_choices)
5250            }
5251            RirNode::Diff { left, .. } => {
5252                self.resident_schema_variants(left, certificate, head_schema_choices)
5253            }
5254            RirNode::Union { inputs } => {
5255                if inputs.is_empty() {
5256                    return Err(resident_workspace_decline(
5257                        "resident schema union has no inputs",
5258                    ));
5259                }
5260                let mut fixed = None;
5261                for input in inputs {
5262                    match self.resident_schema_variants(input, certificate, head_schema_choices)? {
5263                        ResidentSchemaVariants::Fixed(schema) => match &fixed {
5264                            Some(existing) if existing != &schema => {
5265                                return Err(resident_workspace_decline(
5266                                    "resident schema union has ambiguous input schemas",
5267                                ));
5268                            }
5269                            Some(_) => {}
5270                            None => fixed = Some(schema),
5271                        },
5272                        ResidentSchemaVariants::Dynamic { .. } => {
5273                            return Err(resident_workspace_decline(
5274                                "resident schema union has ambiguous dynamic lineage",
5275                            ));
5276                        }
5277                    }
5278                }
5279                fixed.map(ResidentSchemaVariants::Fixed).ok_or_else(|| {
5280                    resident_workspace_decline("resident schema union has no inputs")
5281                })
5282            }
5283            RirNode::Fixpoint { .. }
5284            | RirNode::GroupBy { .. }
5285            | RirNode::TensorMaskedJoin { .. } => Err(resident_workspace_decline(
5286                "unsupported node reached resident schema planning",
5287            )),
5288        }
5289    }
5290
5291    fn resident_workspace_admission_after_certification(
5292        &self,
5293        plan: &ExecutionPlan,
5294        certificate: &ResidentGraphRouteCertificate,
5295    ) -> std::result::Result<ResidentWorkspaceAdmission, ResidentGraphDeclineReason> {
5296        for rule in plan.rules_by_scc.iter().flatten() {
5297            validate_compact_resident_node_envelope(&rule.body, &|node| {
5298                certificate.node_schema(node).ok_or_else(|| {
5299                    resident_workspace_decline("compact resident node schema is missing")
5300                })
5301            })?;
5302        }
5303
5304        let generated_query_heads = resident_generated_query_heads(plan)?;
5305
5306        let mut generated_query_occurrences = generated_query_heads
5307            .iter()
5308            .map(|head| (head.as_str(), 0usize))
5309            .collect::<BTreeMap<_, _>>();
5310        for rule in plan.rules_by_scc.iter().flatten() {
5311            if let Some(count) = generated_query_occurrences.get_mut(rule.head.as_str()) {
5312                *count = count.checked_add(1).ok_or_else(|| {
5313                    resident_workspace_decline(format!(
5314                        "generated query head {} rule count overflowed",
5315                        rule.head
5316                    ))
5317                })?;
5318            }
5319        }
5320        for (head, count) in generated_query_occurrences {
5321            if count != 1 {
5322                return Err(resident_workspace_decline(format!(
5323                    "generated query head {head} must have exactly one compiled rule, found {count}"
5324                )));
5325            }
5326        }
5327
5328        let (required_row_capacity, capacity_witness) =
5329            self.resident_required_row_capacity(plan, certificate)?;
5330        let required_row_capacity = required_row_capacity.max(1);
5331        let required_row_capacity = u32::try_from(required_row_capacity).map_err(|_| {
5332            ResidentGraphDeclineReason::WorkspaceUnbounded {
5333                detail: format!(
5334                    "certified resident row bound {required_row_capacity} exceeds the u32 count ABI; witness: {capacity_witness}"
5335                ),
5336            }
5337        })?;
5338        let relation_capacity = checked_capacity_class(required_row_capacity).map_err(|error| {
5339            ResidentGraphDeclineReason::WorkspaceUnbounded {
5340                detail: format!(
5341                    "{error}; certified_required_rows={required_row_capacity}; witness: {capacity_witness}"
5342                ),
5343            }
5344        })?;
5345
5346        let mut head_schemas = BTreeMap::<String, Schema>::new();
5347        for rules in &plan.rules_by_scc {
5348            for rule in rules {
5349                match head_schemas.get(&rule.head) {
5350                    Some(existing) if existing != &rule.meta.schema => {
5351                        return Err(ResidentGraphDeclineReason::WorkspaceUnbounded {
5352                            detail: format!(
5353                                "relation {} has inconsistent compiled head schemas",
5354                                rule.head
5355                            ),
5356                        });
5357                    }
5358                    Some(_) => {}
5359                    None => {
5360                        head_schemas.insert(rule.head.clone(), rule.meta.schema.clone());
5361                    }
5362                }
5363            }
5364        }
5365
5366        let mut head_schema_choices = head_schemas
5367            .keys()
5368            .map(|name| {
5369                let mut candidates = Vec::new();
5370                if let Some(existing) = self.store.get(name) {
5371                    let physical = &head_schemas[name];
5372                    if !resident_schemas_type_compatible(existing.schema(), physical) {
5373                        return Err(resident_workspace_decline(format!(
5374                            "existing resident head {name} is physically incompatible with its compiled schema"
5375                        )));
5376                    }
5377                    if !(generated_query_heads.contains(name)
5378                        && existing.cached_row_count() == Some(0))
5379                    {
5380                        candidates.push(existing.schema().clone());
5381                    }
5382                }
5383                Ok((name.clone(), candidates))
5384            })
5385            .collect::<std::result::Result<BTreeMap<_, _>, _>>()?;
5386        let mut rule_schema_ids = plan
5387            .rules_by_scc
5388            .iter()
5389            .map(|rules| vec![None; rules.len()])
5390            .collect::<Vec<_>>();
5391        let mut head_schema_selections = BTreeMap::new();
5392
5393        for (scc_index, rules) in plan.rules_by_scc.iter().enumerate() {
5394            let recursive = plan.sccs.get(scc_index).is_some_and(|scc| scc.is_recursive);
5395            for (rule_index, rule) in rules.iter().enumerate() {
5396                if generated_query_heads.contains(&rule.head) {
5397                    continue;
5398                }
5399                let variants =
5400                    self.resident_schema_variants(&rule.body, certificate, &head_schema_choices)?;
5401                let physical = &head_schemas[&rule.head];
5402                let validate_schema = |schema: &Schema| {
5403                    if !resident_schemas_type_compatible(schema, physical) {
5404                        return Err(resident_workspace_decline(format!(
5405                            "resident rule contribution for {} is physically incompatible with its compiled head",
5406                            rule.head
5407                        )));
5408                    }
5409                    if recursive {
5410                        let rel = self.name_to_rel.get(&rule.head).ok_or_else(|| {
5411                            resident_workspace_decline(format!(
5412                                "recursive resident head {} is not registered",
5413                                rule.head
5414                            ))
5415                        })?;
5416                        let candidate = certificate.schema_for(*rel).ok_or_else(|| {
5417                            resident_workspace_decline(format!(
5418                                "recursive resident head {} has no catalog schema candidate",
5419                                rule.head
5420                            ))
5421                        })?;
5422                        if !resident_schemas_type_compatible(schema, candidate) {
5423                            return Err(resident_workspace_decline(format!(
5424                                "recursive resident head {} contribution schema is physically incompatible with its catalog candidate: contribution={schema:?}, catalog={candidate:?}",
5425                                rule.head
5426                            )));
5427                        }
5428                        if schema.key_columns != candidate.key_columns {
5429                            return Err(resident_workspace_decline(format!(
5430                                "recursive resident head {} contribution key columns do not equal its catalog candidate: contribution={:?}, catalog={:?}",
5431                                rule.head, schema.key_columns, candidate.key_columns
5432                            )));
5433                        }
5434                    }
5435                    Ok(())
5436                };
5437                match &variants {
5438                    ResidentSchemaVariants::Fixed(schema) => validate_schema(schema)?,
5439                    ResidentSchemaVariants::Dynamic { schemas, .. } => {
5440                        for schema in schemas {
5441                            validate_schema(schema)?;
5442                        }
5443                    }
5444                }
5445                match variants {
5446                    ResidentSchemaVariants::Fixed(schema) => {
5447                        let schema_id = resident_intern_schema(
5448                            head_schema_choices
5449                                .get_mut(&rule.head)
5450                                .expect("compiled head choice exists"),
5451                            schema,
5452                        )
5453                        .map_err(|error| resident_workspace_decline(error.to_string()))?;
5454                        rule_schema_ids[scc_index][rule_index] = Some(schema_id);
5455                    }
5456                    ResidentSchemaVariants::Dynamic {
5457                        source_head,
5458                        schemas,
5459                    } => {
5460                        resident_register_schema_selection(
5461                            &rule.head,
5462                            ResidentHeadSchemaSelection {
5463                                source_head,
5464                                output_schemas_by_source_winner: schemas.clone(),
5465                            },
5466                            &head_schema_choices,
5467                            &mut head_schema_selections,
5468                        )?;
5469                        for schema in schemas {
5470                            resident_intern_schema(
5471                                head_schema_choices
5472                                    .get_mut(&rule.head)
5473                                    .expect("compiled head choice exists"),
5474                                schema,
5475                            )
5476                            .map_err(|error| resident_workspace_decline(error.to_string()))?;
5477                        }
5478                        rule_schema_ids[scc_index][rule_index] = Some(RESIDENT_DYNAMIC_SCHEMA_ID);
5479                    }
5480                }
5481            }
5482        }
5483
5484        let mut seen_query_heads = HashSet::new();
5485        for (scc_index, rules) in plan.rules_by_scc.iter().enumerate() {
5486            for (rule_index, rule) in rules.iter().enumerate() {
5487                if !generated_query_heads.contains(&rule.head) {
5488                    continue;
5489                }
5490                if !seen_query_heads.insert(rule.head.clone()) {
5491                    return Err(resident_workspace_decline(format!(
5492                        "synthetic query head {} has more than one compiled rule",
5493                        rule.head
5494                    )));
5495                }
5496                let variants =
5497                    self.resident_schema_variants(&rule.body, certificate, &head_schema_choices)?;
5498                let physical = &head_schemas[&rule.head];
5499                match variants {
5500                    ResidentSchemaVariants::Fixed(schema) => {
5501                        if !resident_schemas_type_compatible(&schema, physical) {
5502                            return Err(resident_workspace_decline(format!(
5503                                "synthetic query head {} is physically incompatible with its compiled head",
5504                                rule.head
5505                            )));
5506                        }
5507                        let schema_id = resident_intern_schema(
5508                            head_schema_choices
5509                                .get_mut(&rule.head)
5510                                .expect("compiled query head choice exists"),
5511                            schema,
5512                        )
5513                        .map_err(|error| resident_workspace_decline(error.to_string()))?;
5514                        rule_schema_ids[scc_index][rule_index] = Some(schema_id);
5515                    }
5516                    ResidentSchemaVariants::Dynamic {
5517                        source_head,
5518                        schemas,
5519                    } => {
5520                        if schemas
5521                            .iter()
5522                            .any(|schema| !resident_schemas_type_compatible(schema, physical))
5523                        {
5524                            return Err(resident_workspace_decline(format!(
5525                                "synthetic query head {} has a physically incompatible dynamic schema",
5526                                rule.head
5527                            )));
5528                        }
5529                        for schema in &schemas {
5530                            resident_intern_schema(
5531                                head_schema_choices
5532                                    .get_mut(&rule.head)
5533                                    .expect("compiled query head choice exists"),
5534                                schema.clone(),
5535                            )
5536                            .map_err(|error| resident_workspace_decline(error.to_string()))?;
5537                        }
5538                        rule_schema_ids[scc_index][rule_index] = Some(RESIDENT_DYNAMIC_SCHEMA_ID);
5539                        resident_register_schema_selection(
5540                            &rule.head,
5541                            ResidentHeadSchemaSelection {
5542                                source_head,
5543                                output_schemas_by_source_winner: schemas,
5544                            },
5545                            &head_schema_choices,
5546                            &mut head_schema_selections,
5547                        )?;
5548                    }
5549                }
5550            }
5551        }
5552
5553        for (name, choices) in &head_schema_choices {
5554            if choices.is_empty() {
5555                return Err(resident_workspace_decline(format!(
5556                    "resident head {name} has no schema candidate"
5557                )));
5558            }
5559        }
5560        Ok(ResidentWorkspaceAdmission {
5561            relation_capacity,
5562            head_schemas,
5563            head_schema_choices,
5564            rule_schema_ids,
5565            head_schema_selections,
5566        })
5567    }
5568
5569    fn resident_required_row_capacity(
5570        &self,
5571        plan: &ExecutionPlan,
5572        certificate: &ResidentGraphRouteCertificate,
5573    ) -> std::result::Result<(u64, String), ResidentGraphDeclineReason> {
5574        let mut relation_domains = HashMap::<RelId, ResidentColumnDomains>::new();
5575        let mut synthetic_domains = HashMap::<String, ResidentColumnDomains>::new();
5576        let mut source_rows = HashMap::<RelId, u64>::new();
5577        let plan_relations =
5578            resident_plan_relation_ids(plan.rules_by_scc.iter().flatten(), &self.name_to_rel);
5579        for (relation, name) in &self.rel_names {
5580            if !plan_relations.contains(relation) {
5581                continue;
5582            }
5583            let schema = certificate.schema_for(*relation).ok_or_else(|| {
5584                resident_workspace_decline(format!(
5585                    "relation {name} has no schema in the resident certificate"
5586                ))
5587            })?;
5588            let mut columns = vec![BTreeMap::new(); schema.arity()];
5589            let rows = self.store.get(name).map(CudaBuffer::num_rows).unwrap_or(0);
5590            if rows != 0 {
5591                for (column, domain) in columns.iter_mut().enumerate() {
5592                    domain.insert(format!("source:{}:{column}", relation.0), rows);
5593                }
5594            }
5595            relation_domains.insert(*relation, columns);
5596            source_rows.insert(*relation, rows);
5597        }
5598
5599        loop {
5600            let mut changed = false;
5601            for rules in &plan.rules_by_scc {
5602                for rule in rules {
5603                    let contribution = resident_node_domains(&rule.body, &relation_domains)?;
5604                    if contribution.len() != rule.meta.schema.arity() {
5605                        return Err(resident_workspace_decline(format!(
5606                            "resident rule head {} has {} certified domains for schema arity {}",
5607                            rule.head,
5608                            contribution.len(),
5609                            rule.meta.schema.arity()
5610                        )));
5611                    }
5612                    let target = if let Some(head) = self.name_to_rel.get(&rule.head) {
5613                        relation_domains.get_mut(head).ok_or_else(|| {
5614                            resident_workspace_decline(format!(
5615                                "resident rule head {} has no domain slot",
5616                                rule.head
5617                            ))
5618                        })?
5619                    } else {
5620                        synthetic_domains
5621                            .entry(rule.head.clone())
5622                            .or_insert_with(|| vec![BTreeMap::new(); rule.meta.schema.arity()])
5623                    };
5624                    changed |= merge_resident_domains(&rule.head, target, contribution)?;
5625                }
5626            }
5627            if !changed {
5628                break;
5629            }
5630        }
5631
5632        let mut relation_set_bounds = source_rows.clone();
5633        let mut synthetic_set_bounds = HashMap::<String, u64>::new();
5634        let mut required = source_rows.values().copied().max().unwrap_or(1).max(1);
5635        let mut witness = source_rows
5636            .iter()
5637            .max_by_key(|(_, rows)| *rows)
5638            .map(|(relation, rows)| {
5639                let name = self
5640                    .rel_names
5641                    .get(relation)
5642                    .map(String::as_str)
5643                    .unwrap_or("<unregistered>");
5644                format!("source relation {name} ({relation:?}) rows={rows}")
5645            })
5646            .unwrap_or_else(|| "unit relation bound=1".to_string());
5647
5648        let mut scc_by_id = HashMap::new();
5649        for (index, scc) in plan.sccs.iter().enumerate() {
5650            if scc_by_id.insert(scc.id, index).is_some() {
5651                return Err(resident_workspace_decline(format!(
5652                    "resident row proof found duplicate SCC id {}",
5653                    scc.id
5654                )));
5655            }
5656        }
5657        let ordered_sccs = if plan.strata.is_empty() {
5658            (0..plan.sccs.len()).collect::<Vec<_>>()
5659        } else {
5660            let mut ordered = Vec::new();
5661            for stratum in &plan.strata {
5662                for scc_id in &stratum.sccs {
5663                    ordered.push(*scc_by_id.get(scc_id).ok_or_else(|| {
5664                        resident_workspace_decline(format!(
5665                            "resident row proof stratum {} references unknown SCC {scc_id}",
5666                            stratum.id
5667                        ))
5668                    })?);
5669                }
5670            }
5671            ordered
5672        };
5673        let mut seen_sccs = HashSet::new();
5674        for scc_index in ordered_sccs {
5675            if !seen_sccs.insert(scc_index) {
5676                return Err(resident_workspace_decline(format!(
5677                    "resident row proof schedule repeats SCC {}",
5678                    plan.sccs[scc_index].id
5679                )));
5680            }
5681            let scc = &plan.sccs[scc_index];
5682            let rules = plan.rules_by_scc.get(scc_index).ok_or_else(|| {
5683                resident_workspace_decline(format!(
5684                    "resident row proof SCC {scc_index} has no rule vector"
5685                ))
5686            })?;
5687
5688            if scc.is_recursive {
5689                for name in &scc.predicates {
5690                    let relation = self.name_to_rel.get(name).ok_or_else(|| {
5691                        resident_workspace_decline(format!(
5692                            "recursive relation {name} has no registered relation id"
5693                        ))
5694                    })?;
5695                    let columns = relation_domains.get(relation).ok_or_else(|| {
5696                        resident_workspace_decline(format!(
5697                            "recursive relation {relation:?} has no active-domain proof"
5698                        ))
5699                    })?;
5700                    let context = format!("recursive relation {name} ({relation:?})");
5701                    let finite_cap = resident_domain_product(columns, &context)?;
5702                    relation_set_bounds.insert(*relation, finite_cap);
5703                    if finite_cap > required {
5704                        required = finite_cap;
5705                        witness = format!(
5706                            "{context} finite_domain_cap={finite_cap} domains={}",
5707                            resident_domain_description(columns)
5708                        );
5709                    }
5710                }
5711            }
5712
5713            let mut additions = BTreeMap::<String, u64>::new();
5714            let mut raw_additions = BTreeMap::<String, u64>::new();
5715            let mut addition_witnesses = BTreeMap::<String, String>::new();
5716            for (rule_index, rule) in rules.iter().enumerate() {
5717                let path = format!("scc[{scc_index}].rule[{rule_index}] head={}", rule.head);
5718                let proof = resident_node_row_bound(
5719                    &rule.body,
5720                    &source_rows,
5721                    &relation_set_bounds,
5722                    &relation_domains,
5723                    &path,
5724                )?;
5725                let rows = proof.rows;
5726                let transient = proof.peak;
5727                let rule_domains = resident_node_domains(&rule.body, &relation_domains)?;
5728                let set_rows = resident_domain_product_capped(
5729                    &rule_domains,
5730                    rows,
5731                    &format!("{path} projected set"),
5732                )?;
5733                if transient > required {
5734                    required = transient;
5735                    witness = format!(
5736                        "{path} transient_bound={transient} proof={} body={:?}",
5737                        proof.peak_detail, rule.body
5738                    );
5739                }
5740                let total = additions.entry(rule.head.clone()).or_default();
5741                *total = total.checked_add(set_rows).ok_or_else(|| {
5742                    resident_workspace_decline(format!(
5743                        "{path} rule-output addition overflow: partial={total} addition={set_rows} raw_rows={rows}"
5744                    ))
5745                })?;
5746                let raw_total = raw_additions.entry(rule.head.clone()).or_default();
5747                *raw_total = raw_total.checked_add(rows).ok_or_else(|| {
5748                    resident_workspace_decline(format!(
5749                        "{path} raw rule-output addition overflow: partial={raw_total} addition={rows}"
5750                    ))
5751                })?;
5752                addition_witnesses.insert(
5753                    rule.head.clone(),
5754                    format!(
5755                        "{path} body={:?} raw_rows={rows} per_clause_set_cap={set_rows} domains={}",
5756                        rule.body,
5757                        resident_domain_description(&rule_domains)
5758                    ),
5759                );
5760            }
5761
5762            for (head, addition) in additions {
5763                let (current, relation) = if let Some(relation) = self.name_to_rel.get(&head) {
5764                    (
5765                        relation_set_bounds.get(relation).copied().unwrap_or(0),
5766                        Some(*relation),
5767                    )
5768                } else {
5769                    (synthetic_set_bounds.get(&head).copied().unwrap_or(0), None)
5770                };
5771                let path = addition_witnesses
5772                    .get(&head)
5773                    .map(String::as_str)
5774                    .unwrap_or("unknown rule");
5775                let raw_addition = raw_additions.get(&head).copied().unwrap_or(0);
5776                let raw_candidate = current.checked_add(raw_addition).ok_or_else(|| {
5777                    resident_workspace_decline(format!(
5778                        "{path} staged candidate addition overflow: prior={current} raw_rule_rows={raw_addition}"
5779                    ))
5780                })?;
5781                if raw_candidate > required {
5782                    required = raw_candidate;
5783                    witness = format!(
5784                        "{path} staged_candidate_bound={raw_candidate} prior={current} raw_rule_rows={raw_addition}"
5785                    );
5786                }
5787                if scc.is_recursive {
5788                    let candidate = current.checked_add(addition).ok_or_else(|| {
5789                        resident_workspace_decline(format!(
5790                            "{path} recursive set candidate addition overflow: finite_cap={current} set_rule_rows={addition}"
5791                        ))
5792                    })?;
5793                    if candidate > required {
5794                        required = candidate;
5795                        witness = format!(
5796                            "{path} recursive_candidate_bound={candidate} finite_cap={current} raw_rule_rows={addition}"
5797                        );
5798                    }
5799                } else {
5800                    let summed_bound = current.checked_add(addition).ok_or_else(|| {
5801                        resident_workspace_decline(format!(
5802                            "{path} nonrecursive set addition overflow: prior={current} rule_rows={addition}"
5803                        ))
5804                    })?;
5805                    let columns = if let Some(relation) = relation {
5806                        relation_domains.get(&relation)
5807                    } else {
5808                        synthetic_domains.get(&head)
5809                    }
5810                    .ok_or_else(|| {
5811                        resident_workspace_decline(format!(
5812                            "{path} nonrecursive head {head} has no active-domain proof"
5813                        ))
5814                    })?;
5815                    let set_bound = resident_domain_product_capped(
5816                        columns,
5817                        summed_bound,
5818                        &format!("{path} merged set {head}"),
5819                    )?;
5820                    if let Some(relation) = relation {
5821                        relation_set_bounds.insert(relation, set_bound);
5822                    } else {
5823                        synthetic_set_bounds.insert(head.clone(), set_bound);
5824                    }
5825                    if set_bound > required {
5826                        required = set_bound;
5827                        witness = format!(
5828                            "{path} nonrecursive_set_bound={set_bound} prior={current} rule_rows={addition}"
5829                        );
5830                    }
5831                }
5832            }
5833        }
5834        if seen_sccs.len() != plan.sccs.len() {
5835            return Err(resident_workspace_decline(
5836                "resident row proof schedule omits one or more SCCs",
5837            ));
5838        }
5839        Ok((required, witness))
5840    }
5841}
5842
5843type ResidentDomain = BTreeMap<String, u64>;
5844type ResidentColumnDomains = Vec<ResidentDomain>;
5845
5846fn resident_plan_relation_ids<'a>(
5847    rules: impl Iterator<Item = &'a CompiledRule>,
5848    name_to_rel: &HashMap<String, RelId>,
5849) -> HashSet<RelId> {
5850    let mut relations = HashSet::new();
5851    for rule in rules {
5852        relations.extend(rule.body.referenced_relations());
5853        if let Some(head) = name_to_rel.get(&rule.head) {
5854            relations.insert(*head);
5855        }
5856    }
5857    relations
5858}
5859
5860fn resident_workspace_decline(detail: impl Into<String>) -> ResidentGraphDeclineReason {
5861    ResidentGraphDeclineReason::WorkspaceUnbounded {
5862        detail: detail.into(),
5863    }
5864}
5865
5866fn merge_resident_domains(
5867    relation: &str,
5868    target: &mut ResidentColumnDomains,
5869    contribution: ResidentColumnDomains,
5870) -> std::result::Result<bool, ResidentGraphDeclineReason> {
5871    if target.len() != contribution.len() {
5872        return Err(resident_workspace_decline(format!(
5873            "resident rule head {relation} domain arity changed"
5874        )));
5875    }
5876    let mut changed = false;
5877    for (target_domain, contribution_domain) in target.iter_mut().zip(contribution) {
5878        for (source, bound) in contribution_domain {
5879            match target_domain.entry(source) {
5880                std::collections::btree_map::Entry::Vacant(entry) => {
5881                    entry.insert(bound);
5882                    changed = true;
5883                }
5884                std::collections::btree_map::Entry::Occupied(mut entry) if *entry.get() < bound => {
5885                    entry.insert(bound);
5886                    changed = true;
5887                }
5888                std::collections::btree_map::Entry::Occupied(_) => {}
5889            }
5890        }
5891    }
5892    Ok(changed)
5893}
5894
5895fn resident_node_domains(
5896    node: &RirNode,
5897    relations: &HashMap<RelId, ResidentColumnDomains>,
5898) -> std::result::Result<ResidentColumnDomains, ResidentGraphDeclineReason> {
5899    match node {
5900        RirNode::Unit => Ok(Vec::new()),
5901        RirNode::Scan { rel } => relations.get(rel).cloned().ok_or_else(|| {
5902            resident_workspace_decline(format!(
5903                "resident scan {rel:?} has no source-domain certificate"
5904            ))
5905        }),
5906        RirNode::Filter { input, predicate } => {
5907            let mut domains = resident_node_domains(input, relations)?;
5908            resident_refine_filter_domains(&mut domains, predicate)?;
5909            Ok(domains)
5910        }
5911        RirNode::Distinct { input, .. } => resident_node_domains(input, relations),
5912        RirNode::Project { input, columns } => {
5913            let input_domains = resident_node_domains(input, relations)?;
5914            resident_project_domains(&input_domains, columns)
5915        }
5916        RirNode::Join {
5917            left,
5918            right,
5919            join_type,
5920            ..
5921        } => {
5922            let mut left_domains = resident_node_domains(left, relations)?;
5923            match join_type {
5924                JoinType::Inner => {
5925                    left_domains.extend(resident_node_domains(right, relations)?);
5926                    Ok(left_domains)
5927                }
5928                JoinType::Semi => Ok(left_domains),
5929                other => Err(resident_workspace_decline(format!(
5930                    "resident domain proof does not support {other:?} joins"
5931                ))),
5932            }
5933        }
5934        RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
5935            resident_node_domains(fallback, relations)
5936        }
5937        RirNode::Union { inputs } => {
5938            let mut merged: Option<ResidentColumnDomains> = None;
5939            for input in inputs {
5940                let contribution = resident_node_domains(input, relations)?;
5941                if let Some(domains) = merged.as_mut() {
5942                    if domains.len() != contribution.len() {
5943                        return Err(resident_workspace_decline(
5944                            "resident union domain arity mismatch",
5945                        ));
5946                    }
5947                    for (domain, addition) in domains.iter_mut().zip(contribution) {
5948                        for (source, bound) in addition {
5949                            domain
5950                                .entry(source)
5951                                .and_modify(|current| *current = (*current).max(bound))
5952                                .or_insert(bound);
5953                        }
5954                    }
5955                } else {
5956                    merged = Some(contribution);
5957                }
5958            }
5959            merged.ok_or_else(|| resident_workspace_decline("resident union has no inputs"))
5960        }
5961        RirNode::Diff { left, .. } => resident_node_domains(left, relations),
5962        RirNode::GroupBy { .. } | RirNode::Fixpoint { .. } | RirNode::TensorMaskedJoin { .. } => {
5963            Err(resident_workspace_decline(
5964                "resident domain proof encountered an uncertified operator",
5965            ))
5966        }
5967    }
5968}
5969
5970fn resident_refine_filter_domains(
5971    domains: &mut ResidentColumnDomains,
5972    predicate: &Expr,
5973) -> std::result::Result<(), ResidentGraphDeclineReason> {
5974    match predicate {
5975        Expr::And(parts) => {
5976            for part in parts {
5977                resident_refine_filter_domains(domains, part)?;
5978            }
5979        }
5980        Expr::Compare {
5981            left,
5982            op: RirCompareOp::Eq,
5983            right,
5984        } => {
5985            let (column, value) = match (&**left, &**right) {
5986                (Expr::Column(column), Expr::Const(value))
5987                | (Expr::Const(value), Expr::Column(column)) => (*column, value),
5988                _ => return Ok(()),
5989            };
5990            let arity = domains.len();
5991            let domain = domains.get_mut(column).ok_or_else(|| {
5992                resident_workspace_decline(format!(
5993                    "resident equality filter column {column} exceeds input arity {}",
5994                    arity
5995                ))
5996            })?;
5997            *domain = BTreeMap::from([(format!("constant:{value:?}"), 1)]);
5998        }
5999        _ => {}
6000    }
6001    Ok(())
6002}
6003
6004fn resident_project_domains(
6005    input: &[ResidentDomain],
6006    columns: &[ProjectExpr],
6007) -> std::result::Result<ResidentColumnDomains, ResidentGraphDeclineReason> {
6008    columns
6009        .iter()
6010        .map(|column| match column {
6011            ProjectExpr::Column(index) => input.get(*index).cloned().ok_or_else(|| {
6012                resident_workspace_decline(format!(
6013                    "resident projection column {index} exceeds input arity {}",
6014                    input.len()
6015                ))
6016            }),
6017            ProjectExpr::Computed(Expr::Const(value), scalar) => Ok(BTreeMap::from([(
6018                format!("constant:{scalar:?}:{value:?}"),
6019                1,
6020            )])),
6021            _ => Err(resident_workspace_decline(
6022                "resident projection domain is not a column or constant",
6023            )),
6024        })
6025        .collect()
6026}
6027
6028fn resident_domain_product(
6029    columns: &[ResidentDomain],
6030    context: &str,
6031) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6032    if columns.is_empty() {
6033        return Ok(1);
6034    }
6035    columns
6036        .iter()
6037        .enumerate()
6038        .try_fold(1u64, |product, (column_index, domain)| {
6039            let values = domain.values().try_fold(0u64, |total, bound| {
6040                total.checked_add(*bound).ok_or_else(|| {
6041                    resident_workspace_decline(format!(
6042                        "{context} active-domain sum overflow at column {column_index}: partial={total} addition={bound} lineage={domain:?}"
6043                    ))
6044                })
6045            })?;
6046            product.checked_mul(values).ok_or_else(|| {
6047                resident_workspace_decline(format!(
6048                    "{context} active-domain product overflow at column {column_index}: partial={product} factor={values} lineage={domain:?}"
6049                ))
6050            })
6051        })
6052}
6053
6054fn resident_domain_product_capped(
6055    columns: &[ResidentDomain],
6056    cap: u64,
6057    context: &str,
6058) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6059    if cap == 0 {
6060        return Ok(0);
6061    }
6062    if columns.is_empty() {
6063        return Ok(1);
6064    }
6065    let mut product = 1u64;
6066    for (column_index, domain) in columns.iter().enumerate() {
6067        let mut values = 0u64;
6068        for bound in domain.values() {
6069            let addition = (*bound).min(cap.saturating_sub(values));
6070            values = values.checked_add(addition).ok_or_else(|| {
6071                resident_workspace_decline(format!(
6072                    "{context} capped active-domain sum overflow at column {column_index}: partial={values} addition={addition}"
6073                ))
6074            })?;
6075            if values == cap {
6076                break;
6077            }
6078        }
6079        if values == 0 {
6080            return Ok(0);
6081        }
6082        if product > cap / values {
6083            return Ok(cap);
6084        }
6085        product = product.checked_mul(values).ok_or_else(|| {
6086            resident_workspace_decline(format!(
6087                "{context} capped active-domain product overflow at column {column_index}: partial={product} factor={values}"
6088            ))
6089        })?;
6090    }
6091    Ok(product.min(cap))
6092}
6093
6094fn resident_domain_description(columns: &[ResidentDomain]) -> String {
6095    columns
6096        .iter()
6097        .enumerate()
6098        .map(|(column, domain)| {
6099            let sum = domain.values().copied().fold(0u64, u64::saturating_add);
6100            format!("column[{column}] sum={sum} lineage={domain:?}")
6101        })
6102        .collect::<Vec<_>>()
6103        .join("; ")
6104}
6105
6106#[derive(Clone, Debug, PartialEq, Eq)]
6107enum ResidentCapacityBound {
6108    Finite { rows: u64, proof: String },
6109    AboveResidentLimit { proof: String },
6110}
6111
6112fn resident_capacity_product_bound(proof: &'static str, factors: &[u64]) -> ResidentCapacityBound {
6113    if factors.contains(&0) {
6114        return ResidentCapacityBound::Finite {
6115            rows: 0,
6116            proof: proof.to_owned(),
6117        };
6118    }
6119
6120    let limit = u64::from(MAX_RESIDENT_CAPACITY);
6121    let mut rows = 1u64;
6122    for factor in factors {
6123        if rows > limit / factor {
6124            return ResidentCapacityBound::AboveResidentLimit {
6125                proof: format!("{proof} factors={factors:?}"),
6126            };
6127        }
6128        rows *= factor;
6129    }
6130    ResidentCapacityBound::Finite {
6131        rows,
6132        proof: proof.to_owned(),
6133    }
6134}
6135
6136fn resident_join_capacity_bound(
6137    left_rows: u64,
6138    right_rows: u64,
6139    left_fanout: u64,
6140    right_fanout: u64,
6141    matching_key_values: u64,
6142) -> ResidentCapacityBound {
6143    let candidates = [
6144        resident_capacity_product_bound("left_rows*right_rows", &[left_rows, right_rows]),
6145        resident_capacity_product_bound("left_rows*right_fanout", &[left_rows, right_fanout]),
6146        resident_capacity_product_bound("right_rows*left_fanout", &[right_rows, left_fanout]),
6147        resident_capacity_product_bound(
6148            "matching_keys*left_fanout*right_fanout",
6149            &[matching_key_values, left_fanout, right_fanout],
6150        ),
6151    ];
6152
6153    let mut tightest: Option<(u64, String)> = None;
6154    let mut above = Vec::new();
6155    for candidate in candidates {
6156        match candidate {
6157            ResidentCapacityBound::Finite { rows, proof } => {
6158                if tightest.as_ref().is_none_or(|(current, _)| rows < *current) {
6159                    tightest = Some((rows, proof));
6160                }
6161            }
6162            ResidentCapacityBound::AboveResidentLimit { proof } => above.push(proof),
6163        }
6164    }
6165    if let Some((rows, proof)) = tightest {
6166        ResidentCapacityBound::Finite { rows, proof }
6167    } else {
6168        ResidentCapacityBound::AboveResidentLimit {
6169            proof: above.join("; "),
6170        }
6171    }
6172}
6173
6174struct ResidentRowProof {
6175    rows: u64,
6176    peak: u64,
6177    domains: ResidentColumnDomains,
6178    full_row_unique: bool,
6179    peak_detail: String,
6180}
6181
6182fn resident_node_row_bound(
6183    node: &RirNode,
6184    source_rows: &HashMap<RelId, u64>,
6185    relation_set_bounds: &HashMap<RelId, u64>,
6186    relation_domains: &HashMap<RelId, ResidentColumnDomains>,
6187    path: &str,
6188) -> std::result::Result<ResidentRowProof, ResidentGraphDeclineReason> {
6189    match node {
6190        RirNode::Unit => Ok(ResidentRowProof {
6191            rows: 1,
6192            peak: 1,
6193            domains: Vec::new(),
6194            full_row_unique: true,
6195            peak_detail: format!("{path}.unit rows=1"),
6196        }),
6197        RirNode::Scan { rel } => {
6198            let rows = source_rows
6199                .get(rel)
6200                .copied()
6201                .unwrap_or(0)
6202                .max(relation_set_bounds.get(rel).copied().unwrap_or(0));
6203            let domains = relation_domains.get(rel).cloned().ok_or_else(|| {
6204                resident_workspace_decline(format!(
6205                    "{path}.scan {rel:?} has no active-domain proof"
6206                ))
6207            })?;
6208            Ok(ResidentRowProof {
6209                rows,
6210                peak: rows,
6211                domains,
6212                full_row_unique: true,
6213                peak_detail: format!("{path}.scan rel={rel:?} rows={rows}"),
6214            })
6215        }
6216        RirNode::Filter { input, predicate } => {
6217            let mut proof = resident_node_row_bound(
6218                input,
6219                source_rows,
6220                relation_set_bounds,
6221                relation_domains,
6222                &format!("{path}.filter.input"),
6223            )?;
6224            resident_refine_filter_domains(&mut proof.domains, predicate)?;
6225            if proof.full_row_unique {
6226                proof.rows = resident_domain_product_capped(
6227                    &proof.domains,
6228                    proof.rows,
6229                    &format!("{path}.filter"),
6230                )?;
6231            }
6232            Ok(proof)
6233        }
6234        RirNode::Project { input, columns } => {
6235            let input = resident_node_row_bound(
6236                input,
6237                source_rows,
6238                relation_set_bounds,
6239                relation_domains,
6240                &format!("{path}.project.input"),
6241            )?;
6242            let domains = resident_project_domains(&input.domains, columns)?;
6243            let full_row_unique = input.full_row_unique
6244                && resident_projection_is_injective(&input.domains, columns, path)?;
6245            Ok(ResidentRowProof {
6246                rows: input.rows,
6247                peak: input.peak.max(input.rows),
6248                domains,
6249                full_row_unique,
6250                peak_detail: input.peak_detail,
6251            })
6252        }
6253        RirNode::Distinct { input, .. } => {
6254            let mut proof = resident_node_row_bound(
6255                input,
6256                source_rows,
6257                relation_set_bounds,
6258                relation_domains,
6259                &format!("{path}.distinct.input"),
6260            )?;
6261            proof.rows = resident_domain_product_capped(
6262                &proof.domains,
6263                proof.rows,
6264                &format!("{path}.distinct"),
6265            )?;
6266            proof.full_row_unique = true;
6267            Ok(proof)
6268        }
6269        RirNode::Join {
6270            left,
6271            right,
6272            left_keys,
6273            right_keys,
6274            join_type,
6275        } => {
6276            let left = resident_node_row_bound(
6277                left,
6278                source_rows,
6279                relation_set_bounds,
6280                relation_domains,
6281                &format!("{path}.join.left"),
6282            )?;
6283            let right = resident_node_row_bound(
6284                right,
6285                source_rows,
6286                relation_set_bounds,
6287                relation_domains,
6288                &format!("{path}.join.right"),
6289            )?;
6290            match join_type {
6291                JoinType::Semi => Ok(ResidentRowProof {
6292                    rows: left.rows,
6293                    peak: left.peak.max(right.peak).max(left.rows),
6294                    domains: left.domains,
6295                    full_row_unique: left.full_row_unique,
6296                    peak_detail: if left.peak >= right.peak {
6297                        left.peak_detail
6298                    } else {
6299                        right.peak_detail
6300                    },
6301                }),
6302                JoinType::Inner => {
6303                    if left_keys.len() != 1 || right_keys.len() != 1 {
6304                        return Err(resident_workspace_decline(format!(
6305                            "{path} resident row proof requires exactly one join key per side"
6306                        )));
6307                    }
6308                    let left_key = left_keys[0];
6309                    let right_key = right_keys[0];
6310                    let left_fanout = resident_key_fanout_bound(
6311                        &left.domains,
6312                        left_key,
6313                        left.rows,
6314                        left.full_row_unique,
6315                        &format!("{path}.join.left"),
6316                    )?;
6317                    let right_fanout = resident_key_fanout_bound(
6318                        &right.domains,
6319                        right_key,
6320                        right.rows,
6321                        right.full_row_unique,
6322                        &format!("{path}.join.right"),
6323                    )?;
6324                    let left_key_values = resident_domain_cardinality_capped(
6325                        left.domains.get(left_key).ok_or_else(|| {
6326                            resident_workspace_decline(format!(
6327                                "{path} left join key {left_key} exceeds arity {}",
6328                                left.domains.len()
6329                            ))
6330                        })?,
6331                        left.rows,
6332                        &format!("{path}.join.left_key"),
6333                    )?;
6334                    let right_key_values = resident_domain_cardinality_capped(
6335                        right.domains.get(right_key).ok_or_else(|| {
6336                            resident_workspace_decline(format!(
6337                                "{path} right join key {right_key} exceeds arity {}",
6338                                right.domains.len()
6339                            ))
6340                        })?,
6341                        right.rows,
6342                        &format!("{path}.join.right_key"),
6343                    )?;
6344                    let matching_key_values = left_key_values.min(right_key_values);
6345                    let (rows, bound_proof) = match resident_join_capacity_bound(
6346                        left.rows,
6347                        right.rows,
6348                        left_fanout,
6349                        right_fanout,
6350                        matching_key_values,
6351                    ) {
6352                        ResidentCapacityBound::Finite { rows, proof } => (rows, proof),
6353                        ResidentCapacityBound::AboveResidentLimit { proof } => {
6354                            return Err(resident_workspace_decline(format!(
6355                                "{path} inner join exceeds the fixed resident capacity limit: left_rows={} right_rows={} left_fanout={left_fanout} right_fanout={right_fanout} matching_keys={matching_key_values} products={proof}",
6356                                left.rows, right.rows
6357                            )));
6358                        }
6359                    };
6360                    let join_detail = format!(
6361                        "{path}.join left_rows={} right_rows={} left_fanout={left_fanout} right_fanout={right_fanout} matching_keys={matching_key_values} bound={rows} proof={bound_proof}",
6362                        left.rows, right.rows
6363                    );
6364                    let (peak, peak_detail) = if rows >= left.peak && rows >= right.peak {
6365                        (rows, join_detail)
6366                    } else if left.peak >= right.peak {
6367                        (left.peak, left.peak_detail)
6368                    } else {
6369                        (right.peak, right.peak_detail)
6370                    };
6371                    let mut domains = left.domains;
6372                    domains.extend(right.domains);
6373                    Ok(ResidentRowProof {
6374                        rows,
6375                        peak,
6376                        domains,
6377                        full_row_unique: left.full_row_unique && right.full_row_unique,
6378                        peak_detail,
6379                    })
6380                }
6381                other => Err(resident_workspace_decline(format!(
6382                    "resident row proof does not support {other:?} joins"
6383                ))),
6384            }
6385        }
6386        RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
6387            resident_node_row_bound(
6388                fallback,
6389                source_rows,
6390                relation_set_bounds,
6391                relation_domains,
6392                &format!("{path}.fallback"),
6393            )
6394        }
6395        RirNode::Union { inputs } => {
6396            let mut rows = 0u64;
6397            let mut peak = 0u64;
6398            let mut peak_detail = format!("{path}.union empty");
6399            let mut domains: Option<ResidentColumnDomains> = None;
6400            for (index, input) in inputs.iter().enumerate() {
6401                let proof = resident_node_row_bound(
6402                    input,
6403                    source_rows,
6404                    relation_set_bounds,
6405                    relation_domains,
6406                    &format!("{path}.union[{index}]"),
6407                )?;
6408                rows = rows.checked_add(proof.rows).ok_or_else(|| {
6409                    resident_workspace_decline(format!(
6410                        "{path} union addition overflow: partial={rows} addition={}",
6411                        proof.rows
6412                    ))
6413                })?;
6414                if proof.peak > peak {
6415                    peak = proof.peak;
6416                    peak_detail = proof.peak_detail;
6417                }
6418                if let Some(merged) = domains.as_mut() {
6419                    merge_domain_columns(merged, proof.domains, path)?;
6420                } else {
6421                    domains = Some(proof.domains);
6422                }
6423            }
6424            if rows > peak {
6425                peak = rows;
6426                peak_detail = format!("{path}.union raw_rows={rows}");
6427            }
6428            Ok(ResidentRowProof {
6429                rows,
6430                peak,
6431                domains: domains
6432                    .ok_or_else(|| resident_workspace_decline("resident union has no inputs"))?,
6433                full_row_unique: false,
6434                peak_detail,
6435            })
6436        }
6437        RirNode::Diff { left, right } => {
6438            let left = resident_node_row_bound(
6439                left,
6440                source_rows,
6441                relation_set_bounds,
6442                relation_domains,
6443                &format!("{path}.diff.left"),
6444            )?;
6445            let right = resident_node_row_bound(
6446                right,
6447                source_rows,
6448                relation_set_bounds,
6449                relation_domains,
6450                &format!("{path}.diff.right"),
6451            )?;
6452            let (peak, peak_detail) = if left.peak >= right.peak {
6453                (left.peak.max(left.rows), left.peak_detail)
6454            } else {
6455                (right.peak.max(left.rows), right.peak_detail)
6456            };
6457            Ok(ResidentRowProof {
6458                rows: left.rows,
6459                peak,
6460                domains: left.domains,
6461                full_row_unique: left.full_row_unique,
6462                peak_detail,
6463            })
6464        }
6465        RirNode::GroupBy { .. } | RirNode::Fixpoint { .. } | RirNode::TensorMaskedJoin { .. } => {
6466            Err(resident_workspace_decline(
6467                "resident row proof encountered an uncertified operator",
6468            ))
6469        }
6470    }
6471}
6472
6473fn resident_projection_is_injective(
6474    input_domains: &[ResidentDomain],
6475    columns: &[ProjectExpr],
6476    path: &str,
6477) -> std::result::Result<bool, ResidentGraphDeclineReason> {
6478    let mut retained = vec![false; input_domains.len()];
6479    for column in columns {
6480        if let ProjectExpr::Column(index) = column {
6481            let slot = retained.get_mut(*index).ok_or_else(|| {
6482                resident_workspace_decline(format!(
6483                    "{path} projection column {index} exceeds input arity {}",
6484                    input_domains.len()
6485                ))
6486            })?;
6487            *slot = true;
6488        }
6489    }
6490    for (index, (retained, domain)) in retained.iter().zip(input_domains).enumerate() {
6491        if !retained
6492            && resident_domain_cardinality_capped(
6493                domain,
6494                2,
6495                &format!("{path}.project.omitted[{index}]"),
6496            )? > 1
6497        {
6498            return Ok(false);
6499        }
6500    }
6501    Ok(true)
6502}
6503
6504fn resident_domain_cardinality_capped(
6505    domain: &ResidentDomain,
6506    cap: u64,
6507    context: &str,
6508) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6509    let mut total = 0u64;
6510    for bound in domain.values() {
6511        let addition = (*bound).min(cap.saturating_sub(total));
6512        total = total.checked_add(addition).ok_or_else(|| {
6513            resident_workspace_decline(format!(
6514                "{context} domain-cardinality addition overflow: partial={total} addition={addition}"
6515            ))
6516        })?;
6517        if total == cap {
6518            break;
6519        }
6520    }
6521    Ok(total)
6522}
6523
6524fn resident_key_fanout_bound(
6525    domains: &[ResidentDomain],
6526    key: usize,
6527    rows: u64,
6528    full_row_unique: bool,
6529    context: &str,
6530) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6531    if key >= domains.len() {
6532        return Err(resident_workspace_decline(format!(
6533            "{context} join key {key} exceeds input arity {}",
6534            domains.len()
6535        )));
6536    }
6537    if !full_row_unique {
6538        return Ok(rows);
6539    }
6540    let non_key = domains
6541        .iter()
6542        .enumerate()
6543        .filter_map(|(index, domain)| (index != key).then_some(domain.clone()))
6544        .collect::<Vec<_>>();
6545    resident_domain_product_capped(&non_key, rows, &format!("{context}.fanout"))
6546}
6547
6548fn merge_domain_columns(
6549    target: &mut ResidentColumnDomains,
6550    addition: ResidentColumnDomains,
6551    path: &str,
6552) -> std::result::Result<(), ResidentGraphDeclineReason> {
6553    if target.len() != addition.len() {
6554        return Err(resident_workspace_decline(format!(
6555            "{path} union domain arity mismatch: left={} right={}",
6556            target.len(),
6557            addition.len()
6558        )));
6559    }
6560    for (target, addition) in target.iter_mut().zip(addition) {
6561        for (source, bound) in addition {
6562            target
6563                .entry(source)
6564                .and_modify(|current| *current = (*current).max(bound))
6565                .or_insert(bound);
6566        }
6567    }
6568    Ok(())
6569}
6570
6571fn checked_capacity_class(source_capacity: u32) -> Result<u32> {
6572    let capacity = source_capacity
6573        .max(1)
6574        .checked_next_power_of_two()
6575        .ok_or_else(|| XlogError::Execution("resident capacity class overflow".to_string()))?;
6576    if capacity > MAX_RESIDENT_CAPACITY {
6577        return Err(XlogError::Execution(format!(
6578            "resident capacity class {capacity} exceeds the fixed scan envelope {MAX_RESIDENT_CAPACITY}"
6579        )));
6580    }
6581    Ok(capacity)
6582}
6583
6584#[cfg(test)]
6585mod tests {
6586    use super::{
6587        checked_capacity_class, resident_compact_allocation_bytes,
6588        resident_compact_filter_descriptors, resident_compact_preflight_device_bytes,
6589        resident_compact_project_descriptors, resident_compact_regions,
6590        resident_compact_schedule_metadata_bytes, resident_compact_schema_defaults,
6591        resident_compact_tables, resident_compact_topology, resident_join_capacity_bound,
6592        resident_lower_compact_regions, resident_new_phase_unit, resident_output_indices,
6593        resident_phase_merge, resident_plan_relation_ids, resident_record_lifetimes,
6594        resident_record_scan_leaf, resident_record_schema_winner_mark, resident_record_unit_leaf,
6595        resident_register_schema_selection, resident_resolve_output_schema,
6596        resident_semantic_trace_guard, resident_source_binding_route,
6597        resident_source_logical_count, resident_source_slot_map, resident_union_fold_mode,
6598        resident_validate_conditional_body_node_kinds, resident_validate_exact_reservation,
6599        resident_validate_parent_graph_kinds, resident_validate_slot_assignments,
6600        validate_compact_resident_node_envelope, CudaGraphNodeKind, ResidentAllocationManifest,
6601        ResidentBufferRef, ResidentCapacityBound, ResidentCaptureParentKind, ResidentCapturePhase,
6602        ResidentCompactDescriptorTables, ResidentCompactLogicalRegion, ResidentFilterPlan,
6603        ResidentGraphDeclineReason, ResidentHeadSchemaSelection, ResidentLogicalRelation,
6604        ResidentOpDescriptor, ResidentOutputSchemaPlan, ResidentOutputSchemaSelection,
6605        ResidentPhaseMergeStep, ResidentPhysicalSlotPlan, ResidentProjectPlan, ResidentRecordedOp,
6606        ResidentSlotAssignment, ResidentUnionFoldMode, ScalarType, Schema,
6607        RESIDENT_DYNAMIC_SCHEMA_ID,
6608    };
6609    use std::collections::{BTreeMap, HashMap, HashSet};
6610    use xlog_core::RelId;
6611    use xlog_ir::{
6612        CompareOp, CompiledRule, ConstValue, Expr, JoinType, ProjectExpr, RirMeta, RirNode,
6613    };
6614
6615    #[test]
6616    fn resident_plan_relations_exclude_registered_but_unreferenced_relations() {
6617        let source = RelId(1);
6618        let head = RelId(2);
6619        let unrelated = RelId(3);
6620        let rules = [CompiledRule {
6621            head: "reachable".to_owned(),
6622            body: RirNode::Scan { rel: source },
6623            meta: RirMeta::default(),
6624        }];
6625        let names = HashMap::from([
6626            ("source".to_owned(), source),
6627            ("reachable".to_owned(), head),
6628            ("unrelated".to_owned(), unrelated),
6629        ]);
6630
6631        let relations = resident_plan_relation_ids(rules.iter(), &names);
6632
6633        assert_eq!(relations, HashSet::from([source, head]));
6634    }
6635
6636    #[test]
6637    fn capacity_class_is_source_bounded_and_checked() {
6638        assert_eq!(checked_capacity_class(0).unwrap(), 1);
6639        assert_eq!(checked_capacity_class(4_393).unwrap(), 8_192);
6640        assert_eq!(checked_capacity_class(65_536).unwrap(), 65_536);
6641        assert!(checked_capacity_class(65_537).is_err());
6642    }
6643
6644    #[test]
6645    fn schema_lineage_fails_closed_for_cycles_and_missing_sources() {
6646        let schema = Schema::new(vec![("value".to_owned(), ScalarType::U32)]);
6647        let selection = |source: &str| ResidentHeadSchemaSelection {
6648            source_head: source.to_owned(),
6649            output_schemas_by_source_winner: vec![schema.clone()],
6650        };
6651        let choices = BTreeMap::from([
6652            ("a".to_owned(), vec![schema.clone()]),
6653            ("b".to_owned(), vec![schema.clone()]),
6654            ("c".to_owned(), vec![schema.clone()]),
6655            ("empty".to_owned(), Vec::new()),
6656        ]);
6657
6658        let direct_cycle =
6659            resident_register_schema_selection("a", selection("a"), &choices, &mut BTreeMap::new())
6660                .expect_err("direct schema lineage cycle must decline");
6661        assert!(matches!(
6662            direct_cycle,
6663            ResidentGraphDeclineReason::WorkspaceUnbounded { detail }
6664                if detail.contains("contains a cycle")
6665        ));
6666
6667        let mut long_cycle = BTreeMap::new();
6668        resident_register_schema_selection("a", selection("b"), &choices, &mut long_cycle)
6669            .expect("first acyclic lineage edge");
6670        resident_register_schema_selection("b", selection("c"), &choices, &mut long_cycle)
6671            .expect("second acyclic lineage edge");
6672        let long_cycle =
6673            resident_register_schema_selection("c", selection("a"), &choices, &mut long_cycle)
6674                .expect_err("transitive schema lineage cycle must decline");
6675        assert!(matches!(
6676            long_cycle,
6677            ResidentGraphDeclineReason::WorkspaceUnbounded { detail }
6678                if detail.contains("contains a cycle")
6679        ));
6680
6681        for source in ["missing", "empty"] {
6682            let missing = resident_register_schema_selection(
6683                "target",
6684                selection(source),
6685                &choices,
6686                &mut BTreeMap::new(),
6687            )
6688            .expect_err("missing schema source must decline");
6689            assert!(matches!(
6690                missing,
6691                ResidentGraphDeclineReason::WorkspaceUnbounded { detail }
6692                    if detail.contains("has no admitted candidate")
6693            ));
6694        }
6695
6696        let dynamic_plan = |source_output| ResidentOutputSchemaPlan {
6697            candidates: vec![schema.clone()],
6698            selection: ResidentOutputSchemaSelection::SourceWinner {
6699                source_output,
6700                schemas: vec![schema.clone()],
6701            },
6702        };
6703        let resolver_cycle = resident_resolve_output_schema(
6704            &[dynamic_plan(1), dynamic_plan(0)],
6705            &[RESIDENT_DYNAMIC_SCHEMA_ID, RESIDENT_DYNAMIC_SCHEMA_ID],
6706            0,
6707            &mut HashSet::new(),
6708        )
6709        .expect_err("receipt schema resolver cycle must fail");
6710        assert!(resolver_cycle.to_string().contains("contains a cycle"));
6711
6712        let missing_plan = resident_resolve_output_schema(
6713            &[dynamic_plan(1)],
6714            &[RESIDENT_DYNAMIC_SCHEMA_ID],
6715            0,
6716            &mut HashSet::new(),
6717        )
6718        .expect_err("missing receipt schema source plan must fail");
6719        assert!(missing_plan.to_string().contains("output 1 is missing"));
6720    }
6721
6722    #[test]
6723    fn join_capacity_uses_a_finite_tight_bound_when_loose_products_overflow() {
6724        let bound = resident_join_capacity_bound(u64::MAX, u64::MAX, 1, 1, 1);
6725        assert_eq!(
6726            bound,
6727            ResidentCapacityBound::Finite {
6728                rows: 1,
6729                proof: "matching_keys*left_fanout*right_fanout".to_owned(),
6730            }
6731        );
6732    }
6733
6734    #[test]
6735    fn true_cartesian_join_remains_above_the_resident_limit() {
6736        let bound = resident_join_capacity_bound(257, 257, 257, 257, 1);
6737        assert!(matches!(
6738            bound,
6739            ResidentCapacityBound::AboveResidentLimit { .. }
6740        ));
6741    }
6742
6743    #[test]
6744    fn lifetime_scan_rejects_an_out_of_range_logical_input() {
6745        let relations = vec![ResidentLogicalRelation {
6746            schema: Schema::new(vec![("value".to_string(), ScalarType::U32)]),
6747            initial_count: 0,
6748            permanent: false,
6749        }];
6750        let mut definitions = vec![None];
6751        let mut last_uses = vec![None];
6752        let mut ordinal = 0;
6753        let error = resident_record_lifetimes(
6754            &[ResidentRecordedOp::Filter {
6755                input: ResidentBufferRef::Private(1),
6756                output: 0,
6757                workspace: 0,
6758                op_id: 0,
6759            }],
6760            &relations,
6761            &mut definitions,
6762            &mut last_uses,
6763            &mut ordinal,
6764        )
6765        .expect_err("an invalid logical input must fail before allocation");
6766        assert!(error
6767            .to_string()
6768            .contains("logical input relation 1 is missing"));
6769    }
6770
6771    #[test]
6772    fn stale_scratch_generation_is_rejected_before_materialization() {
6773        let schema = Schema::new(vec![("value".to_string(), ScalarType::U32)]);
6774        let relations = vec![
6775            ResidentLogicalRelation {
6776                schema: schema.clone(),
6777                initial_count: 0,
6778                permanent: false,
6779            },
6780            ResidentLogicalRelation {
6781                schema: schema.clone(),
6782                initial_count: 0,
6783                permanent: false,
6784            },
6785        ];
6786        let slots = vec![ResidentPhysicalSlotPlan {
6787            schema,
6788            initial_count: 0,
6789            permanent: false,
6790        }];
6791        let assignments = vec![
6792            ResidentSlotAssignment {
6793                slot: 0,
6794                generation: 0,
6795            },
6796            ResidentSlotAssignment {
6797                slot: 0,
6798                generation: 0,
6799            },
6800        ];
6801        let error = resident_validate_slot_assignments(
6802            &relations,
6803            &[Some(0), Some(2)],
6804            &[Some(0), Some(2)],
6805            &slots,
6806            &assignments,
6807        )
6808        .expect_err("a stale scratch generation must fail before allocation");
6809        assert!(error.to_string().contains("generation 0 but expected 1"));
6810    }
6811
6812    #[test]
6813    fn authored_scan_and_unit_are_explicit_lifetime_operations() {
6814        let relations = vec![ResidentLogicalRelation {
6815            schema: Schema::new(Vec::<(String, ScalarType)>::new()),
6816            initial_count: 0,
6817            permanent: false,
6818        }];
6819        let mut definitions = vec![None];
6820        let mut last_uses = vec![None];
6821        let mut ordinal = 0;
6822        let operations = vec![
6823            ResidentRecordedOp::Unit {
6824                output: 0,
6825                op_id: 17,
6826            },
6827            ResidentRecordedOp::Scan {
6828                relation: ResidentBufferRef::Private(0),
6829                op_id: 18,
6830            },
6831            ResidentRecordedOp::TraceDelta {
6832                scan_delta: 1,
6833                filter_delta: 0,
6834                semantic_guard: None,
6835            },
6836        ];
6837
6838        resident_record_lifetimes(
6839            &operations,
6840            &relations,
6841            &mut definitions,
6842            &mut last_uses,
6843            &mut ordinal,
6844        )
6845        .expect("an emitted Unit defines the relation consumed by the emitted Scan");
6846
6847        assert_eq!(definitions, vec![Some(0)]);
6848        assert_eq!(last_uses, vec![Some(1)]);
6849        assert_eq!(ordinal, operations.len());
6850    }
6851
6852    #[test]
6853    fn authored_leaf_emitters_preserve_scan_identity_and_trace_order() {
6854        let mut operations = Vec::new();
6855        let push = |ops: &mut Vec<_>, op, _op_id| ops.push(op);
6856
6857        let unit = resident_record_unit_leaf(3, 17, &mut operations, push);
6858        let first = resident_record_scan_leaf(
6859            ResidentBufferRef::Private(4),
6860            18,
6861            None,
6862            &mut operations,
6863            push,
6864        );
6865        let second = resident_record_scan_leaf(
6866            ResidentBufferRef::Private(5),
6867            19,
6868            None,
6869            &mut operations,
6870            push,
6871        );
6872
6873        assert!(matches!(unit, ResidentBufferRef::Private(3)));
6874        assert!(matches!(first, ResidentBufferRef::Private(4)));
6875        assert!(matches!(second, ResidentBufferRef::Private(5)));
6876        assert!(matches!(
6877            operations.as_slice(),
6878            [
6879                ResidentRecordedOp::Unit {
6880                    output: 3,
6881                    op_id: 17
6882                },
6883                ResidentRecordedOp::Scan {
6884                    relation: ResidentBufferRef::Private(4),
6885                    op_id: 18
6886                },
6887                ResidentRecordedOp::TraceDelta {
6888                    scan_delta: 1,
6889                    filter_delta: 0,
6890                    semantic_guard: None,
6891                },
6892                ResidentRecordedOp::Scan {
6893                    relation: ResidentBufferRef::Private(5),
6894                    op_id: 19
6895                },
6896                ResidentRecordedOp::TraceDelta {
6897                    scan_delta: 1,
6898                    filter_delta: 0,
6899                    semantic_guard: None,
6900                },
6901            ]
6902        ));
6903    }
6904
6905    #[test]
6906    fn recursive_trace_semantics_use_the_selected_delta_and_leave_seed_unguarded() {
6907        let selected_delta = ResidentBufferRef::Private(9);
6908        assert!(resident_semantic_trace_guard(None).is_none());
6909        assert!(matches!(
6910            resident_semantic_trace_guard(Some((RelId(4), 2, 9))),
6911            Some(ResidentBufferRef::Private(9))
6912        ));
6913
6914        let mut operations = Vec::new();
6915        let push = |ops: &mut Vec<_>, op, _op_id| ops.push(op);
6916        resident_record_scan_leaf(
6917            ResidentBufferRef::Private(4),
6918            18,
6919            None,
6920            &mut operations,
6921            push,
6922        );
6923        resident_record_scan_leaf(
6924            ResidentBufferRef::Private(5),
6925            19,
6926            Some(selected_delta.clone()),
6927            &mut operations,
6928            push,
6929        );
6930
6931        assert!(matches!(
6932            operations.as_slice(),
6933            [
6934                ResidentRecordedOp::Scan { op_id: 18, .. },
6935                ResidentRecordedOp::TraceDelta {
6936                    semantic_guard: None,
6937                    ..
6938                },
6939                ResidentRecordedOp::Scan { op_id: 19, .. },
6940                ResidentRecordedOp::TraceDelta {
6941                    semantic_guard: Some(ResidentBufferRef::Private(9)),
6942                    ..
6943                },
6944            ]
6945        ));
6946    }
6947
6948    #[test]
6949    fn phase_unit_allocator_creates_fresh_scratch_values() {
6950        let mut relations = Vec::new();
6951
6952        let (first, first_op) = resident_new_phase_unit(&mut relations, 17).unwrap();
6953        let (second, second_op) = resident_new_phase_unit(&mut relations, 18).unwrap();
6954
6955        assert!(matches!(first, ResidentBufferRef::Private(0)));
6956        assert!(matches!(second, ResidentBufferRef::Private(1)));
6957        assert!(matches!(
6958            first_op,
6959            ResidentRecordedOp::Unit {
6960                output: 0,
6961                op_id: 17
6962            }
6963        ));
6964        assert!(matches!(
6965            second_op,
6966            ResidentRecordedOp::Unit {
6967                output: 1,
6968                op_id: 18
6969            }
6970        ));
6971        assert!(relations
6972            .iter()
6973            .all(|relation| !relation.permanent && relation.initial_count == 0));
6974    }
6975
6976    #[test]
6977    fn phase_local_merge_deduplicates_each_contribution_before_ordered_union() {
6978        let mut steps = Vec::new();
6979        let first = resident_phase_merge(None, 2_u32, |step| match step {
6980            ResidentPhaseMergeStep::Deduplicate(value) => {
6981                steps.push(("dedup", value, 0));
6982                Ok::<_, &'static str>(value * 10)
6983            }
6984            ResidentPhaseMergeStep::Union(left, right) => {
6985                steps.push(("union", left, right));
6986                Ok(left + right)
6987            }
6988        })
6989        .unwrap();
6990        let second = resident_phase_merge(Some(first), 3_u32, |step| match step {
6991            ResidentPhaseMergeStep::Deduplicate(value) => {
6992                steps.push(("dedup", value, 0));
6993                Ok::<_, &'static str>(value * 10)
6994            }
6995            ResidentPhaseMergeStep::Union(left, right) => {
6996                steps.push(("union", left, right));
6997                Ok(left + right)
6998            }
6999        })
7000        .unwrap();
7001
7002        assert_eq!(first, 20);
7003        assert_eq!(second, 50);
7004        assert_eq!(steps, [("dedup", 2, 0), ("dedup", 3, 0), ("union", 20, 30)]);
7005    }
7006
7007    #[test]
7008    fn initial_copy_schema_marker_is_ordered_after_its_count_producer() {
7009        let source = ResidentBufferRef::Source("head".to_owned());
7010        let mut operations = vec![ResidentRecordedOp::Project {
7011            input: source.clone(),
7012            output: 0,
7013            workspace: 0,
7014            op_id: 7,
7015        }];
7016
7017        resident_record_schema_winner_mark(&mut operations, ResidentBufferRef::Private(0), 0, 3);
7018
7019        assert!(matches!(
7020            operations.as_slice(),
7021            [
7022                ResidentRecordedOp::Project { op_id: 7, .. },
7023                ResidentRecordedOp::SchemaWinnerMark {
7024                    contribution: ResidentBufferRef::Private(0),
7025                    head_index: 0,
7026                    schema_id: 3,
7027                }
7028            ]
7029        ));
7030    }
7031
7032    #[test]
7033    fn source_slots_are_deduplicated_sorted_and_distinct_from_private_targets() {
7034        let slots = resident_source_slot_map(3, ["zeta", "head", "zeta", "head"].into_iter())
7035            .expect("source slots");
7036
7037        assert_eq!(slots.get("head"), Some(&3));
7038        assert_eq!(slots.get("zeta"), Some(&4));
7039        assert_eq!(slots.len(), 2);
7040        assert_ne!(slots["head"], 0, "stored source and staged target differ");
7041    }
7042
7043    #[test]
7044    fn empty_untracked_sources_are_normalized_before_slot_binding() {
7045        let logical_count = resident_source_logical_count(Some(0)).unwrap();
7046        assert_eq!(
7047            resident_source_binding_route(logical_count, false, false).unwrap(),
7048            super::ResidentSourceBindingRoute::NormalizeEmpty
7049        );
7050        assert_eq!(
7051            resident_source_binding_route(0, true, true).unwrap(),
7052            super::ResidentSourceBindingRoute::Direct
7053        );
7054        assert!(resident_source_binding_route(1, false, true).is_err());
7055        assert!(resident_source_binding_route(1, true, false).is_err());
7056        assert!(resident_source_logical_count(None).is_err());
7057    }
7058
7059    #[test]
7060    fn compact_regions_preserve_phase_order_and_form_five_parent_nodes() {
7061        let unit = |output, op_id| ResidentRecordedOp::Unit { output, op_id };
7062        let phases = vec![
7063            ResidentCapturePhase::Segment {
7064                ops: vec![unit(10, 10)],
7065                scc_begin: None,
7066            },
7067            ResidentCapturePhase::Segment {
7068                ops: vec![unit(11, 11)],
7069                scc_begin: Some((64, 101)),
7070            },
7071            ResidentCapturePhase::ConditionalWhile {
7072                ops: vec![unit(12, 12)],
7073                iteration_limit: 64,
7074                convergence_op_id: 102,
7075            },
7076            ResidentCapturePhase::Segment {
7077                ops: vec![unit(20, 20)],
7078                scc_begin: None,
7079            },
7080            ResidentCapturePhase::Segment {
7081                ops: vec![ResidentRecordedOp::Scan {
7082                    relation: ResidentBufferRef::Private(20),
7083                    op_id: 21,
7084                }],
7085                scc_begin: Some((32, 201)),
7086            },
7087            ResidentCapturePhase::ConditionalWhile {
7088                ops: vec![unit(22, 22)],
7089                iteration_limit: 32,
7090                convergence_op_id: 202,
7091            },
7092            ResidentCapturePhase::Segment {
7093                ops: vec![unit(30, 30)],
7094                scc_begin: None,
7095            },
7096        ];
7097
7098        let regions = resident_compact_regions(vec![unit(0, 0)], phases, 999).expect("regions");
7099
7100        assert_eq!(regions.len(), 5);
7101        assert_eq!(
7102            regions.iter().filter(|region| region.recursive()).count(),
7103            2
7104        );
7105        assert_eq!(regions.len() + 2, 7, "hierarchical node inventory");
7106        let topology = resident_compact_topology(&regions).unwrap();
7107        assert_eq!(
7108            topology.parent_kinds,
7109            vec![
7110                ResidentCaptureParentKind::Kernel,
7111                ResidentCaptureParentKind::Conditional,
7112                ResidentCaptureParentKind::Kernel,
7113                ResidentCaptureParentKind::Conditional,
7114                ResidentCaptureParentKind::Kernel,
7115            ]
7116        );
7117        assert_eq!(topology.conditional_body_kernel_counts, vec![1, 1]);
7118        assert_eq!(topology.hierarchical_node_count, 7);
7119        assert!(regions[0].initializes());
7120        assert!(regions[0].begins_scc());
7121        assert_eq!(regions[0].op_id, regions[1].op_id);
7122        assert_eq!(regions[0].iteration_limit, regions[1].iteration_limit);
7123        assert_eq!(regions[2].op_id, regions[3].op_id);
7124        assert_eq!(regions[2].iteration_limit, regions[3].iteration_limit);
7125        assert!(regions[4].finalizes());
7126        assert_eq!(regions[4].op_id, 999);
7127        assert!(matches!(
7128            regions[2].ops.as_slice(),
7129            [
7130                ResidentRecordedOp::Unit { output: 20, .. },
7131                ResidentRecordedOp::Scan {
7132                    relation: ResidentBufferRef::Private(20),
7133                    ..
7134                }
7135            ]
7136        ));
7137    }
7138
7139    #[test]
7140    fn conditional_body_inventory_requires_one_actual_kernel_per_body() {
7141        let exact = vec![
7142            vec![CudaGraphNodeKind::Kernel],
7143            vec![CudaGraphNodeKind::Kernel],
7144        ];
7145        assert_eq!(
7146            resident_validate_conditional_body_node_kinds(&exact, 2).unwrap(),
7147            vec![1, 1]
7148        );
7149        assert!(resident_validate_conditional_body_node_kinds(&exact[..1], 2).is_err());
7150        assert!(resident_validate_conditional_body_node_kinds(&[Vec::new()], 1).is_err());
7151        assert!(resident_validate_conditional_body_node_kinds(
7152            &[vec![CudaGraphNodeKind::Kernel, CudaGraphNodeKind::Kernel]],
7153            1,
7154        )
7155        .is_err());
7156        assert!(resident_validate_conditional_body_node_kinds(
7157            &[vec![CudaGraphNodeKind::Memcpy]],
7158            1,
7159        )
7160        .is_err());
7161    }
7162
7163    #[test]
7164    fn compact_parent_graph_kind_validation_is_exact() {
7165        let expected = vec![
7166            ResidentCaptureParentKind::Kernel,
7167            ResidentCaptureParentKind::Conditional,
7168            ResidentCaptureParentKind::Kernel,
7169        ];
7170        assert!(resident_validate_parent_graph_kinds(
7171            &[
7172                CudaGraphNodeKind::Kernel,
7173                CudaGraphNodeKind::Conditional,
7174                CudaGraphNodeKind::Kernel
7175            ],
7176            &expected,
7177        )
7178        .is_ok());
7179        assert!(resident_validate_parent_graph_kinds(
7180            &[
7181                CudaGraphNodeKind::Kernel,
7182                CudaGraphNodeKind::Kernel,
7183                CudaGraphNodeKind::Conditional
7184            ],
7185            &expected,
7186        )
7187        .is_err());
7188        assert!(resident_validate_parent_graph_kinds(
7189            &[CudaGraphNodeKind::Kernel, CudaGraphNodeKind::Conditional],
7190            &expected,
7191        )
7192        .is_err());
7193    }
7194
7195    #[test]
7196    fn compact_descriptors_preserve_physical_generations_and_source_slots() {
7197        let regions = vec![ResidentCompactLogicalRegion {
7198            ops: vec![
7199                ResidentRecordedOp::Unit {
7200                    output: 0,
7201                    op_id: 7,
7202                },
7203                ResidentRecordedOp::SchemaWinnerMark {
7204                    contribution: ResidentBufferRef::Private(0),
7205                    head_index: 0,
7206                    schema_id: 3,
7207                },
7208                ResidentRecordedOp::TraceDelta {
7209                    scan_delta: 0,
7210                    filter_delta: 1,
7211                    semantic_guard: None,
7212                },
7213                ResidentRecordedOp::Scan {
7214                    relation: ResidentBufferRef::Source("source".to_owned()),
7215                    op_id: 8,
7216                },
7217                ResidentRecordedOp::TraceDelta {
7218                    scan_delta: 1,
7219                    filter_delta: 0,
7220                    semantic_guard: None,
7221                },
7222            ],
7223            iteration_limit: 1,
7224            op_id: 0,
7225            flags: super::RESIDENT_SCHEDULE_REGION_INITIALIZE
7226                | super::RESIDENT_SCHEDULE_REGION_FINALIZE,
7227        }];
7228        let slots = vec![
7229            ResidentPhysicalSlotPlan {
7230                schema: Schema::new(Vec::new()),
7231                initial_count: 0,
7232                permanent: true,
7233            },
7234            ResidentPhysicalSlotPlan {
7235                schema: Schema::new(Vec::new()),
7236                initial_count: 0,
7237                permanent: false,
7238            },
7239        ];
7240        let assignments = vec![ResidentSlotAssignment {
7241            slot: 1,
7242            generation: 4,
7243        }];
7244
7245        let plan = resident_lower_compact_regions(
7246            regions,
7247            &slots,
7248            &assignments,
7249            ["source"].into_iter(),
7250            Default::default(),
7251        )
7252        .expect("compact descriptors");
7253
7254        assert_eq!(plan.source_slots["source"], 2);
7255        assert_eq!(plan.ops.len(), 4);
7256        assert_eq!(plan.ops[0].kind, super::ResidentScheduleOpKind::Unit);
7257        assert_eq!(plan.ops[0].out, 1);
7258        assert_eq!(plan.ops[0].out_generation, 4);
7259        assert_eq!(plan.ops[0].schema_winner_head, 0);
7260        assert_eq!(plan.ops[0].schema_winner_id, 3);
7261        assert_eq!(plan.ops[2].kind, super::ResidentScheduleOpKind::Scan);
7262        assert_eq!(plan.ops[2].out, 2);
7263        assert_eq!(plan.ops[2].in0_generation, 0);
7264        assert_eq!(plan.waves.len(), plan.ops.len());
7265        assert_eq!(plan.regions.len(), 1);
7266        assert_eq!(plan.generation_bases, vec![0, 4, 0]);
7267    }
7268
7269    #[test]
7270    fn compact_novelty_marker_observes_the_completed_delta_copy() {
7271        let regions = vec![ResidentCompactLogicalRegion {
7272            ops: vec![
7273                ResidentRecordedOp::ChangedReset,
7274                ResidentRecordedOp::Diff {
7275                    left: ResidentBufferRef::Private(0),
7276                    right: ResidentBufferRef::Private(1),
7277                    output: 2,
7278                    op_id: 10,
7279                },
7280                ResidentRecordedOp::Project {
7281                    input: ResidentBufferRef::Private(2),
7282                    output: 3,
7283                    workspace: 0,
7284                    op_id: 11,
7285                },
7286                ResidentRecordedOp::ChangedMark { relation: 3 },
7287            ],
7288            iteration_limit: 3,
7289            op_id: 12,
7290            flags: super::RESIDENT_SCHEDULE_REGION_RECURSIVE,
7291        }];
7292        let slots = (0..4)
7293            .map(|slot| ResidentPhysicalSlotPlan {
7294                schema: Schema::new(Vec::new()),
7295                initial_count: 0,
7296                permanent: slot != 2,
7297            })
7298            .collect::<Vec<_>>();
7299        let assignments = (0..4)
7300            .map(|slot| ResidentSlotAssignment {
7301                slot,
7302                generation: 0,
7303            })
7304            .collect::<Vec<_>>();
7305        let tables = ResidentCompactDescriptorTables {
7306            project_expressions: vec![super::ResidentProjectExpressionDescriptor::column(0, 4)],
7307            project_ranges: vec![(0, 1)],
7308            ..Default::default()
7309        };
7310
7311        let plan = resident_lower_compact_regions(
7312            regions,
7313            &slots,
7314            &assignments,
7315            std::iter::empty(),
7316            tables,
7317        )
7318        .expect("the final delta copy must carry the novelty marker");
7319
7320        assert_eq!(plan.ops.len(), 2);
7321        assert_eq!(plan.ops[0].kind, super::ResidentScheduleOpKind::Diff);
7322        assert_eq!(plan.ops[0].flags, 0);
7323        assert_eq!(plan.ops[1].kind, super::ResidentScheduleOpKind::Project);
7324        assert_eq!(plan.ops[1].flags, super::RESIDENT_SCHEDULE_OP_MARK_NOVELTY);
7325        assert_eq!(plan.ops[1].out, 3);
7326    }
7327
7328    #[test]
7329    fn compact_schema_marker_must_name_the_producer_output() {
7330        let regions = vec![ResidentCompactLogicalRegion {
7331            ops: vec![
7332                ResidentRecordedOp::Project {
7333                    input: ResidentBufferRef::Source("source".to_owned()),
7334                    output: 0,
7335                    workspace: 0,
7336                    op_id: 7,
7337                },
7338                ResidentRecordedOp::SchemaWinnerMark {
7339                    contribution: ResidentBufferRef::Source("source".to_owned()),
7340                    head_index: 0,
7341                    schema_id: 0,
7342                },
7343            ],
7344            iteration_limit: 1,
7345            op_id: 9,
7346            flags: super::RESIDENT_SCHEDULE_REGION_INITIALIZE
7347                | super::RESIDENT_SCHEDULE_REGION_FINALIZE,
7348        }];
7349        let slots = vec![ResidentPhysicalSlotPlan {
7350            schema: Schema::new(Vec::new()),
7351            initial_count: 0,
7352            permanent: true,
7353        }];
7354        let assignments = vec![ResidentSlotAssignment {
7355            slot: 0,
7356            generation: 0,
7357        }];
7358        let tables = ResidentCompactDescriptorTables {
7359            project_ranges: vec![(0, 0)],
7360            ..Default::default()
7361        };
7362
7363        assert!(resident_lower_compact_regions(
7364            regions,
7365            &slots,
7366            &assignments,
7367            ["source"].into_iter(),
7368            tables,
7369        )
7370        .is_err());
7371    }
7372
7373    #[test]
7374    fn compact_schema_defaults_follow_first_marker_per_head() {
7375        let ops = vec![
7376            ResidentOpDescriptor::unit(1, 0, 0).with_schema_winner(1, u32::MAX),
7377            ResidentOpDescriptor::unit(2, 1, 0).with_schema_winner(0, 7),
7378            ResidentOpDescriptor::unit(3, 2, 0).with_schema_winner(1, 9),
7379        ];
7380        assert_eq!(
7381            resident_compact_schema_defaults(&ops, 2).unwrap(),
7382            [7, u32::MAX]
7383        );
7384        assert!(resident_compact_schema_defaults(&ops[..1], 2).is_err());
7385    }
7386
7387    #[test]
7388    fn compact_changed_reset_is_only_absorbed_at_recursive_region_entry() {
7389        let slot = ResidentPhysicalSlotPlan {
7390            schema: Schema::new(Vec::new()),
7391            initial_count: 0,
7392            permanent: false,
7393        };
7394        let assignment = ResidentSlotAssignment {
7395            slot: 0,
7396            generation: 0,
7397        };
7398        let region = |ops, flags| ResidentCompactLogicalRegion {
7399            ops,
7400            iteration_limit: 4,
7401            op_id: 9,
7402            flags,
7403        };
7404
7405        assert!(resident_lower_compact_regions(
7406            vec![region(vec![ResidentRecordedOp::ChangedReset], 0)],
7407            std::slice::from_ref(&slot),
7408            std::slice::from_ref(&assignment),
7409            std::iter::empty(),
7410            Default::default(),
7411        )
7412        .is_err());
7413        assert!(resident_lower_compact_regions(
7414            vec![region(
7415                vec![
7416                    ResidentRecordedOp::Unit {
7417                        output: 0,
7418                        op_id: 1,
7419                    },
7420                    ResidentRecordedOp::ChangedReset,
7421                ],
7422                super::RESIDENT_SCHEDULE_REGION_RECURSIVE,
7423            )],
7424            std::slice::from_ref(&slot),
7425            std::slice::from_ref(&assignment),
7426            std::iter::empty(),
7427            Default::default(),
7428        )
7429        .is_err());
7430        assert!(resident_lower_compact_regions(
7431            vec![region(
7432                vec![
7433                    ResidentRecordedOp::ChangedReset,
7434                    ResidentRecordedOp::Unit {
7435                        output: 0,
7436                        op_id: 1,
7437                    },
7438                ],
7439                super::RESIDENT_SCHEDULE_REGION_RECURSIVE,
7440            )],
7441            &[slot],
7442            &[assignment],
7443            std::iter::empty(),
7444            Default::default(),
7445        )
7446        .is_ok());
7447    }
7448
7449    #[test]
7450    fn compact_filter_project_tables_preserve_types_widths_and_order() {
7451        let input = Schema::new(vec![
7452            ("symbol".to_owned(), ScalarType::Symbol),
7453            ("number".to_owned(), ScalarType::U64),
7454        ]);
7455        let predicate = Expr::And(vec![
7456            Expr::Compare {
7457                left: Box::new(Expr::Column(1)),
7458                op: CompareOp::Ge,
7459                right: Box::new(Expr::Const(ConstValue::U64(9))),
7460            },
7461            Expr::Compare {
7462                left: Box::new(Expr::Column(0)),
7463                op: CompareOp::Eq,
7464                right: Box::new(Expr::Const(ConstValue::Symbol("x".to_owned()))),
7465            },
7466        ]);
7467        let comparisons =
7468            resident_compact_filter_descriptors(&predicate, &input).expect("filter table");
7469        assert_eq!(comparisons.len(), 2);
7470        assert_eq!(comparisons[0].left_column, 1);
7471        assert_eq!(comparisons[0].width, 8);
7472        assert_eq!(comparisons[0].right_constant, 9);
7473        assert_eq!(comparisons[1].left_column, 0);
7474        assert_eq!(comparisons[1].width, 4);
7475
7476        let output = Schema::new(vec![
7477            ("number".to_owned(), ScalarType::U64),
7478            ("symbol".to_owned(), ScalarType::Symbol),
7479        ]);
7480        let expressions = resident_compact_project_descriptors(
7481            &[
7482                ProjectExpr::Column(1),
7483                ProjectExpr::Computed(
7484                    Expr::Const(ConstValue::Symbol("y".to_owned())),
7485                    ScalarType::Symbol,
7486                ),
7487            ],
7488            &input,
7489            &output,
7490        )
7491        .expect("project table");
7492        assert_eq!(expressions.len(), output.arity());
7493        assert_eq!(expressions[0].column, 1);
7494        assert_eq!(expressions[0].width, 8);
7495        assert_eq!(expressions[1].kind, 1);
7496        assert_eq!(expressions[1].width, 4);
7497
7498        let mismatch = Expr::Compare {
7499            left: Box::new(Expr::Column(1)),
7500            op: CompareOp::Eq,
7501            right: Box::new(Expr::Const(ConstValue::Symbol("wrong".to_owned()))),
7502        };
7503        assert!(resident_compact_filter_descriptors(&mismatch, &input).is_err());
7504    }
7505
7506    #[test]
7507    fn compact_tables_cover_multiple_empty_occurrences_without_duplicate_payloads() {
7508        let filters = vec![
7509            ResidentFilterPlan {
7510                compact_comparisons: Vec::new(),
7511            },
7512            ResidentFilterPlan {
7513                compact_comparisons: Vec::new(),
7514            },
7515        ];
7516        let projects = vec![ResidentProjectPlan {
7517            compact_expressions: Vec::new(),
7518        }];
7519
7520        let tables = resident_compact_tables(&filters, &projects).expect("compact tables");
7521
7522        assert!(tables.filter_comparisons.is_empty());
7523        assert_eq!(tables.filter_ranges, [(0, 0), (0, 0)]);
7524        assert!(tables.project_expressions.is_empty());
7525        assert_eq!(tables.project_ranges, [(0, 0)]);
7526    }
7527
7528    #[test]
7529    fn compact_descriptor_ranges_follow_operation_order_exactly() {
7530        let regions = vec![ResidentCompactLogicalRegion {
7531            ops: vec![
7532                ResidentRecordedOp::Project {
7533                    input: ResidentBufferRef::Source("source".to_owned()),
7534                    output: 0,
7535                    workspace: 1,
7536                    op_id: 1,
7537                },
7538                ResidentRecordedOp::Project {
7539                    input: ResidentBufferRef::Source("source".to_owned()),
7540                    output: 1,
7541                    workspace: 0,
7542                    op_id: 2,
7543                },
7544            ],
7545            iteration_limit: 1,
7546            op_id: 3,
7547            flags: super::RESIDENT_SCHEDULE_REGION_INITIALIZE
7548                | super::RESIDENT_SCHEDULE_REGION_FINALIZE,
7549        }];
7550        let slots = vec![
7551            ResidentPhysicalSlotPlan {
7552                schema: Schema::new(Vec::new()),
7553                initial_count: 0,
7554                permanent: false,
7555            },
7556            ResidentPhysicalSlotPlan {
7557                schema: Schema::new(Vec::new()),
7558                initial_count: 0,
7559                permanent: false,
7560            },
7561        ];
7562        let assignments = vec![
7563            ResidentSlotAssignment {
7564                slot: 0,
7565                generation: 0,
7566            },
7567            ResidentSlotAssignment {
7568                slot: 1,
7569                generation: 0,
7570            },
7571        ];
7572        let tables = ResidentCompactDescriptorTables {
7573            project_expressions: vec![Default::default(), Default::default()],
7574            project_ranges: vec![(0, 1), (1, 1)],
7575            ..Default::default()
7576        };
7577
7578        assert!(resident_lower_compact_regions(
7579            regions,
7580            &slots,
7581            &assignments,
7582            ["source"].into_iter(),
7583            tables,
7584        )
7585        .is_err());
7586    }
7587
7588    #[test]
7589    fn compact_metadata_reservation_counts_generation_bases_and_schema_defaults() {
7590        let actual = resident_compact_schedule_metadata_bytes(3, 5, 5, 2, 6, 4, 0, 0)
7591            .expect("metadata bytes");
7592        let expected = super::resident_schedule_metadata_device_bytes(3, 5, 5, 2, 10, 0, 0)
7593            .expect("expected metadata bytes");
7594        let missing_defaults = super::resident_schedule_metadata_device_bytes(3, 5, 5, 2, 6, 0, 0)
7595            .expect("generation-only bytes");
7596
7597        assert_eq!(actual, expected);
7598        assert_ne!(actual, missing_defaults);
7599    }
7600
7601    #[test]
7602    fn compact_manifest_replaces_per_occurrence_descriptor_allocations() {
7603        let plan = super::ResidentCompactSchedulePlan {
7604            source_slots: [("source".to_owned(), 2)].into_iter().collect(),
7605            ops: Vec::new(),
7606            waves: Vec::new(),
7607            regions: Vec::new(),
7608            generation_bases: Vec::new(),
7609            filter_comparisons: Vec::new(),
7610            project_expressions: Vec::new(),
7611        };
7612        let (required, metadata) =
7613            resident_compact_allocation_bytes(1_000, 2_000, 3_000, 2, 2, &plan).unwrap();
7614        let expected_metadata =
7615            resident_compact_schedule_metadata_bytes(3, 0, 0, 0, 0, 2, 0, 0).unwrap();
7616        assert_eq!(metadata, expected_metadata);
7617        assert_eq!(required, 1_000 + 2_000 + 3_000 + expected_metadata);
7618        assert_ne!(
7619            required,
7620            1_000 + 48 + 2_000 + 24 + 3_000 + expected_metadata
7621        );
7622    }
7623
7624    #[test]
7625    fn compact_preflight_bytes_report_flattened_tables_without_double_counting() {
7626        let assert_components = |filter_count: usize, project_count: usize| {
7627            let plan = super::ResidentCompactSchedulePlan {
7628                source_slots: [("source".to_owned(), 2)].into_iter().collect(),
7629                ops: Vec::new(),
7630                waves: Vec::new(),
7631                regions: Vec::new(),
7632                generation_bases: Vec::new(),
7633                filter_comparisons: vec![
7634                    super::ResidentFilterComparisonDescriptor::default();
7635                    filter_count
7636                ],
7637                project_expressions: vec![
7638                    super::ResidentProjectExpressionDescriptor::default();
7639                    project_count
7640                ],
7641            };
7642            let mut manifest = ResidentAllocationManifest {
7643                slots: Vec::new(),
7644                logical_to_slot: Vec::new(),
7645                required_bytes: 0,
7646                relation_bytes: 1_000,
7647                filter_scratch_bytes: 2_000,
7648                schedule_metadata_bytes: 0,
7649                fixed_workspace_bytes: 3_000,
7650                logical_relation_values: 0,
7651                permanent_slots: 0,
7652                scratch_slots: 0,
7653                filter_scratch_allocations: 1,
7654                max_row_bytes: 0,
7655            };
7656            manifest.finalize_compact_schedule(&plan, 2).unwrap();
7657
7658            let (filter_bytes, project_bytes, fixed_bytes) =
7659                resident_compact_preflight_device_bytes(&manifest, &plan).unwrap();
7660            assert_eq!(
7661                filter_bytes,
7662                48 * u64::try_from(filter_count.max(1)).unwrap()
7663            );
7664            assert_eq!(
7665                project_bytes,
7666                24 * u64::try_from(project_count.max(1)).unwrap()
7667            );
7668            assert_eq!(
7669                manifest.relation_bytes
7670                    + manifest.filter_scratch_bytes
7671                    + filter_bytes
7672                    + project_bytes
7673                    + fixed_bytes,
7674                manifest.required_bytes
7675            );
7676        };
7677
7678        assert_components(0, 0);
7679        assert_components(3, 2);
7680    }
7681
7682    #[test]
7683    fn compact_manifest_is_exact_and_preserves_permanent_head_mapping() {
7684        let mut manifest = ResidentAllocationManifest {
7685            slots: vec![
7686                ResidentPhysicalSlotPlan {
7687                    schema: Schema::new(Vec::new()),
7688                    initial_count: 0,
7689                    permanent: true,
7690                },
7691                ResidentPhysicalSlotPlan {
7692                    schema: Schema::new(Vec::new()),
7693                    initial_count: 0,
7694                    permanent: true,
7695                },
7696            ],
7697            logical_to_slot: vec![
7698                ResidentSlotAssignment {
7699                    slot: 0,
7700                    generation: 0,
7701                },
7702                ResidentSlotAssignment {
7703                    slot: 1,
7704                    generation: 0,
7705                },
7706            ],
7707            required_bytes: 0,
7708            relation_bytes: 1_000,
7709            filter_scratch_bytes: 2_000,
7710            schedule_metadata_bytes: 0,
7711            fixed_workspace_bytes: 3_000,
7712            logical_relation_values: 2,
7713            permanent_slots: 2,
7714            scratch_slots: 0,
7715            filter_scratch_allocations: 1,
7716            max_row_bytes: 1,
7717        };
7718        let plan = super::ResidentCompactSchedulePlan {
7719            source_slots: [("external".to_owned(), 2)].into_iter().collect(),
7720            ops: Vec::new(),
7721            waves: Vec::new(),
7722            regions: Vec::new(),
7723            generation_bases: Vec::new(),
7724            filter_comparisons: Vec::new(),
7725            project_expressions: Vec::new(),
7726        };
7727        manifest.finalize_compact_schedule(&plan, 1).unwrap();
7728        let without_source =
7729            resident_compact_schedule_metadata_bytes(2, 0, 0, 0, 0, 1, 0, 0).unwrap();
7730        assert_eq!(manifest.schedule_metadata_bytes - without_source, 240);
7731        assert_eq!(
7732            manifest.required_bytes,
7733            manifest.relation_bytes
7734                + manifest.filter_scratch_bytes
7735                + manifest.fixed_workspace_bytes
7736                + manifest.schedule_metadata_bytes
7737        );
7738        resident_validate_exact_reservation(manifest.required_bytes, manifest.required_bytes, 0)
7739            .unwrap();
7740        assert!(resident_validate_exact_reservation(manifest.required_bytes, 5_999, 1).is_err());
7741        let heads = [("head".to_owned(), 0_usize)].into_iter().collect();
7742        assert_eq!(
7743            resident_output_indices(&heads, &manifest.logical_to_slot, &manifest.slots).unwrap(),
7744            vec![("head".to_owned(), 0)]
7745        );
7746        let mut scratch_slots = manifest.slots.clone();
7747        scratch_slots[0].permanent = false;
7748        assert!(
7749            resident_output_indices(&heads, &manifest.logical_to_slot, &scratch_slots,).is_err()
7750        );
7751    }
7752
7753    #[test]
7754    fn compact_distinct_declines_partial_keys_before_planning() {
7755        let node = RirNode::Distinct {
7756            input: Box::new(RirNode::Scan { rel: RelId(1) }),
7757            key_cols: vec![0],
7758        };
7759        let schema = Schema::new(vec![
7760            ("key".to_owned(), ScalarType::U32),
7761            ("value".to_owned(), ScalarType::U32),
7762        ]);
7763
7764        let error = validate_compact_resident_node_envelope(&node, &|_| Ok(schema.clone()))
7765            .expect_err("a partial-key distinct must decline before allocation");
7766
7767        let crate::resident_graph::ResidentGraphDeclineReason::WorkspaceUnbounded { detail } =
7768            error
7769        else {
7770            panic!("expected a workspace-envelope decline");
7771        };
7772        assert!(detail.contains("canonical full-row key columns"));
7773    }
7774
7775    #[test]
7776    fn compact_distinct_accepts_empty_keys_for_nullary_rows() {
7777        let node = RirNode::Distinct {
7778            input: Box::new(RirNode::Unit),
7779            key_cols: vec![],
7780        };
7781
7782        validate_compact_resident_node_envelope(&node, &|_| Ok(Schema::new(Vec::new())))
7783            .expect("the canonical full-row key for an arity-zero relation is empty");
7784    }
7785
7786    #[test]
7787    fn compact_join_declines_same_width_different_key_types() {
7788        let node = RirNode::Join {
7789            left: Box::new(RirNode::Scan { rel: RelId(1) }),
7790            right: Box::new(RirNode::Scan { rel: RelId(2) }),
7791            left_keys: vec![0],
7792            right_keys: vec![0],
7793            join_type: JoinType::Inner,
7794        };
7795
7796        let error = validate_compact_resident_node_envelope(&node, &|node| match node {
7797            RirNode::Scan { rel: RelId(1) } => {
7798                Ok(Schema::new(vec![("key".to_owned(), ScalarType::U32)]))
7799            }
7800            RirNode::Scan { rel: RelId(2) } => {
7801                Ok(Schema::new(vec![("key".to_owned(), ScalarType::Symbol)]))
7802            }
7803            _ => unreachable!("the envelope asks only for operand schemas"),
7804        })
7805        .expect_err("same-width but different key types must decline");
7806
7807        let crate::resident_graph::ResidentGraphDeclineReason::WorkspaceUnbounded { detail } =
7808            error
7809        else {
7810            panic!("expected a workspace-envelope decline");
7811        };
7812        assert!(detail.contains("matching U32, U64, or Symbol key types"));
7813    }
7814
7815    #[test]
7816    fn compact_union_shape_requires_unary_self_union() {
7817        assert!(resident_union_fold_mode(0).is_err());
7818        assert_eq!(
7819            resident_union_fold_mode(1).unwrap(),
7820            ResidentUnionFoldMode::SelfUnion
7821        );
7822        assert_eq!(
7823            resident_union_fold_mode(4).unwrap(),
7824            ResidentUnionFoldMode::LeftAssociated
7825        );
7826    }
7827}