remem/context_bundle/
executor.rs

1//! Deterministic v1 executor: applies a [`ContextPlan`] to caller-provided
2//! candidates by reusing the SessionStart relevance selector, enforces
3//! section and total token budgets, and emits an audited [`ContextBundle`].
4//!
5//! v1 has no DB access; wiring the executor to the SessionStart loaders is
6//! follow-up work on GH-932.
7
8use std::collections::HashMap;
9
10use crate::context::{build_sessionstart_relevance_plan, RelevanceCandidate, RelevanceSection};
11
12use super::audit::AuditBuilder;
13use super::domain::{
14    ChannelKind, ContextBundle, ContextItem, ContextPlan, DegradedMode, ItemValidity, SourceKind,
15    TrustClass, CONTEXT_BUNDLE_SCHEMA_VERSION,
16};
17use super::policy::{
18    estimate_tokens, validate_plan, REASON_BRANCH_SCOPE_MISMATCH, REASON_CANONICAL_ONLY_DEGRADED,
19    REASON_CHANNEL_ITEM_LIMIT, REASON_CHANNEL_TOKEN_BUDGET, REASON_PLAN_BLOCKED,
20    REASON_PROJECT_SCOPE_MISMATCH, REASON_QUARANTINED_TRUST, REASON_SELECTED_CHANNEL,
21    REASON_SELECTED_RELEVANCE, REASON_SUPERSEDED_EXCLUDED, REASON_TOTAL_TOKEN_BUDGET,
22};
23
24/// Candidate inputs for one execution. `enrichment_available = false`
25/// degrades the bundle to `canonical_only`: generated and graph-derived
26/// candidates are dropped instead of being served without their backing
27/// enrichment stack.
28#[derive(Debug, Clone)]
29pub struct ExecutorInputs {
30    pub candidates: Vec<ContextItem>,
31    pub enrichment_available: bool,
32}
33
34/// Execute a plan over the provided candidates.
35///
36/// Deterministic: same plan + same inputs always produce the same bundle.
37/// An invalid plan (schema/policy/scope) produces a `blocked` bundle whose
38/// audit drops every candidate; it never partially executes.
39pub fn execute(plan: &ContextPlan, inputs: &ExecutorInputs) -> ContextBundle {
40    if let Err(error) = validate_plan(plan) {
41        return blocked_bundle(plan, inputs, &error.to_string());
42    }
43    let degraded_mode = if inputs.enrichment_available {
44        DegradedMode::Full
45    } else {
46        DegradedMode::CanonicalOnly
47    };
48
49    let mut audit = AuditBuilder::default();
50    let mut in_scope: Vec<&ContextItem> = Vec::new();
51    for item in &inputs.candidates {
52        match scope_drop_reason(plan, degraded_mode, item) {
53            Some(reason) => audit.dropped(item, reason),
54            None => in_scope.push(item),
55        }
56    }
57
58    let relevance = relevance_decisions(plan, &in_scope, &mut audit);
59    let mut survivors: Vec<&ContextItem> = Vec::new();
60    for item in in_scope {
61        let governed = channel_relevance_governed(plan, item.channel);
62        if governed {
63            match relevance.get(item.stable_key.as_str()) {
64                Some(&(true, _)) | None => survivors.push(item),
65                Some(&(false, drop_reason)) => audit.dropped(item, drop_reason),
66            }
67        } else {
68            survivors.push(item);
69        }
70    }
71
72    let mut bundle = empty_bundle(plan, degraded_mode);
73    apply_budgets(plan, degraded_mode, &survivors, &mut bundle, &mut audit);
74    bundle.audit = audit.finalize(plan, degraded_mode);
75    bundle
76}
77
78fn scope_drop_reason(
79    plan: &ContextPlan,
80    degraded_mode: DegradedMode,
81    item: &ContextItem,
82) -> Option<&'static str> {
83    if item.trust == TrustClass::Quarantined {
84        return Some(REASON_QUARANTINED_TRUST);
85    }
86    if let Some(project) = &item.project {
87        if project != &plan.filters.project {
88            return Some(REASON_PROJECT_SCOPE_MISMATCH);
89        }
90    }
91    if let (Some(item_branch), Some(plan_branch)) = (&item.branch, &plan.filters.branch) {
92        if item_branch != plan_branch {
93            return Some(REASON_BRANCH_SCOPE_MISMATCH);
94        }
95    }
96    if item.validity == ItemValidity::Superseded && !plan.filters.include_superseded {
97        return Some(REASON_SUPERSEDED_EXCLUDED);
98    }
99    if degraded_mode == DegradedMode::CanonicalOnly && item.source_kind != SourceKind::Canonical {
100        return Some(REASON_CANONICAL_ONLY_DEGRADED);
101    }
102    None
103}
104
105fn channel_relevance_governed(plan: &ContextPlan, channel: ChannelKind) -> bool {
106    plan.channels
107        .iter()
108        .find(|planned| planned.channel == channel)
109        .is_some_and(|planned| planned.relevance_governed)
110}
111
112fn relevance_section(channel: ChannelKind) -> Option<RelevanceSection> {
113    match channel {
114        ChannelKind::Lessons => Some(RelevanceSection::Lessons),
115        ChannelKind::MemoryIndex => Some(RelevanceSection::MemoryIndex),
116        ChannelKind::Sessions => Some(RelevanceSection::Sessions),
117        ChannelKind::Preferences | ChannelKind::Core | ChannelKind::Workstreams => None,
118    }
119}
120
121/// Reuse the SessionStart relevance selector for the governed channels.
122/// Returns `stable_key -> (selected, drop_reason)`; drop reasons are the
123/// SessionStart reason strings.
124fn relevance_decisions<'a>(
125    plan: &ContextPlan,
126    in_scope: &[&'a ContextItem],
127    audit: &mut AuditBuilder,
128) -> HashMap<&'a str, (bool, &'static str)> {
129    let candidates: Vec<RelevanceCandidate> = in_scope
130        .iter()
131        .filter(|item| channel_relevance_governed(plan, item.channel))
132        .filter_map(|item| {
133            relevance_section(item.channel).map(|section| RelevanceCandidate {
134                stable_key: item.stable_key.clone(),
135                section,
136                text: format!("{} {}", item.title, item.text),
137            })
138        })
139        .collect();
140    let relevance_plan = build_sessionstart_relevance_plan(
141        plan.relevance_query.as_deref(),
142        plan.relevance_k as usize,
143        &candidates,
144    );
145    let mut decisions = HashMap::new();
146    for item in in_scope {
147        let Some(decision) = relevance_plan.decision(&item.stable_key) else {
148            continue;
149        };
150        audit.record_score(&item.stable_key, decision.score);
151        decisions.insert(
152            item.stable_key.as_str(),
153            (decision.selected, decision.drop_reason.unwrap_or("dropped")),
154        );
155    }
156    decisions
157}
158
159/// Enforce per-channel item limits, per-channel token budgets, and the
160/// total token budget in the fixed [`ChannelKind::ORDERED`] order.
161fn apply_budgets(
162    plan: &ContextPlan,
163    _degraded_mode: DegradedMode,
164    survivors: &[&ContextItem],
165    bundle: &mut ContextBundle,
166    audit: &mut AuditBuilder,
167) {
168    let mut total_tokens: u32 = 0;
169    let total_budget = plan.section_budgets.total_tokens;
170    for channel in ChannelKind::ORDERED {
171        let item_limit = plan
172            .channels
173            .iter()
174            .find(|planned| planned.channel == channel)
175            .map(|planned| planned.item_limit)
176            .unwrap_or(0);
177        let channel_budget = plan.section_budgets.for_channel(channel);
178        let governed = channel_relevance_governed(plan, channel);
179        let mut channel_tokens: u32 = 0;
180        let mut channel_count: u32 = 0;
181        for item in survivors.iter().filter(|item| item.channel == channel) {
182            let tokens = estimate_tokens(&item.text);
183            if channel_count >= item_limit {
184                audit.dropped(item, REASON_CHANNEL_ITEM_LIMIT);
185                continue;
186            }
187            if channel_tokens + tokens > channel_budget {
188                audit.dropped(item, REASON_CHANNEL_TOKEN_BUDGET);
189                continue;
190            }
191            if total_tokens + tokens > total_budget {
192                audit.dropped(item, REASON_TOTAL_TOKEN_BUDGET);
193                audit.set_truncation_reason(REASON_TOTAL_TOKEN_BUDGET);
194                continue;
195            }
196            channel_count += 1;
197            channel_tokens += tokens;
198            total_tokens += tokens;
199            let reason = if governed {
200                REASON_SELECTED_RELEVANCE
201            } else {
202                REASON_SELECTED_CHANNEL
203            };
204            audit.selected(item, reason);
205            bundle.section_mut(channel).push((*item).clone());
206        }
207    }
208}
209
210fn empty_bundle(plan: &ContextPlan, degraded_mode: DegradedMode) -> ContextBundle {
211    ContextBundle {
212        schema_version: CONTEXT_BUNDLE_SCHEMA_VERSION,
213        plan_hash: plan.plan_hash.clone(),
214        degraded_mode,
215        preferences: Vec::new(),
216        failure_lessons: Vec::new(),
217        current_truth: Vec::new(),
218        workstreams: Vec::new(),
219        memory_index: Vec::new(),
220        recent_sessions: Vec::new(),
221        audit: super::domain::ContextAudit {
222            schema_version: CONTEXT_BUNDLE_SCHEMA_VERSION,
223            policy_version: plan.policy_version.clone(),
224            relevance_policy_version: plan.relevance_policy_version.clone(),
225            plan_hash: plan.plan_hash.clone(),
226            degraded_mode,
227            candidates_considered: 0,
228            selected_count: 0,
229            dropped_count: 0,
230            token_estimate: 0,
231            token_budget: plan.section_budgets.total_tokens,
232            truncation_reason: None,
233            entries: Vec::new(),
234        },
235    }
236}
237
238fn blocked_bundle(plan: &ContextPlan, inputs: &ExecutorInputs, error: &str) -> ContextBundle {
239    crate::log::error(
240        "context-bundle",
241        &format!("plan validation failed; emitting blocked bundle: {error}"),
242    );
243    let mut audit = AuditBuilder::default();
244    for item in &inputs.candidates {
245        audit.dropped(item, REASON_PLAN_BLOCKED);
246    }
247    audit.set_truncation_reason(REASON_PLAN_BLOCKED);
248    let mut bundle = empty_bundle(plan, DegradedMode::Blocked);
249    bundle.audit = audit.finalize(plan, DegradedMode::Blocked);
250    bundle
251}
remem/context_bundle/executor.rs

remem/context_bundle/
executor.rs
