feat(events)!: cull stale annotated LLM histories

crates/cli/src/adapters/mod.rs

@@ -6,6 +6,7 @@ pub(crate) mod codex;
 pub(crate) mod hermes;
 
 use axum::http::HeaderMap;
+use nemo_relay::api::scope::COMPACTION_EVENT_NAME;
 use serde_json::{Map, Value, json};
 use uuid::Uuid;
 
@@ -895,7 +896,7 @@ fn classify_primary(
                     fallback_session_id,
                 ))
             }
-            "precompact" | "compaction" => {
+            "precompact" | "postcompact" | COMPACTION_EVENT_NAME => {
                 NormalizedEvent::Compaction(common_session_event_with_fallback(
                     payload,
                     headers,

crates/cli/tests/coverage/adapters_tests.rs

@@ -930,6 +930,8 @@ fn normalizes_mark_style_events_and_header_session_ids() {
         ("UserPromptSubmit", "prompt"),
         ("afterAgentResponse", "response"),
         ("PreCompact", "compact"),
+        ("PostCompact", "compact"),
+        (COMPACTION_EVENT_NAME, "compact"),
         ("Notification", "notification"),
         ("Unrecognized.Event", "hook"),
     ] {

crates/core/src/api/llm.rs

@@ -29,7 +29,7 @@ use crate::api::shared::{
     resolve_parent_uuid, run_request_intercepts_with_codec_and_recorder,
     sanitize_event_with_scope_stack, snapshot_event_subscribers,
 };
-use crate::codec::request::AnnotatedLlmRequest;
+use crate::codec::request::{AnnotatedLlmRequest, Message};
 use crate::codec::response::{AnnotatedLlmResponse, attach_estimated_cost_for_provider};
 use crate::codec::traits::{LlmCodec, LlmResponseCodec};
 use crate::error::{FlowError, Result};
@@ -298,6 +298,89 @@ fn create_llm_handle(params: CreateLlmHandleParams<'_>) -> Result<LlmHandle> {
     Ok(state.create_llm_handle(params))
 }
 
+fn request_turn_projection_needed<T>(
+    items: &[T],
+    is_user: &impl Fn(&T) -> bool,
+    is_instruction: &impl Fn(&T) -> bool,
+) -> bool {
+    let Some(last_index) = items.len().checked_sub(1) else {
+        return false;
+    };
+    match items.iter().rposition(is_user) {
+        Some(start) => items[..start].iter().any(|item| !is_instruction(item)),
+        None => items
+            .iter()
+            .enumerate()
+            .any(|(index, item)| index != last_index && !is_instruction(item)),
+    }
+}
+
+fn retain_current_request_turn<T>(
+    items: &mut Vec<T>,
+    is_user: impl Fn(&T) -> bool,
+    is_instruction: impl Fn(&T) -> bool,
+) -> bool {
+    if !request_turn_projection_needed(items, &is_user, &is_instruction) {
+        return false;
+    }
+    let last_index = items.len() - 1;
+    let Some(start) = items.iter().rposition(is_user) else {
+        let mut index = 0;
+        items.retain(|item| {
+            let retain = index == last_index || is_instruction(item);
+            index += 1;
+            retain
+        });
+        return true;
+    };
+    let mut current_turn = items.split_off(start);
+    items.retain(is_instruction);
+    items.append(&mut current_turn);
+    true
+}
+
+fn project_llm_request_to_current_user_turn(
+    request: &mut LlmRequest,
+    annotated_request: &mut Option<Arc<AnnotatedLlmRequest>>,
+    request_codec: Option<&dyn LlmCodec>,
+) {
+    let Some(annotation) = annotated_request.as_mut() else {
+        return;
+    };
+    if !request_turn_projection_needed(
+        &annotation.messages,
+        &|message| matches!(message, Message::User { .. }),
+        &|message| matches!(message, Message::System { .. }),
+    ) {
+        return;
+    }
+    let original_annotation = request_codec.map(|_| Arc::clone(annotation));
+    let projected = limit_annotated_request_history_to_current_user_turn(Arc::make_mut(annotation));
+    debug_assert!(projected);
+    if let Some(codec) = request_codec {
+        match codec.encode(annotation, request) {
+            Ok(encoded) => *request = encoded,
+            Err(error) => {
+                eprintln!(
+                    "nemo_relay: LLM request projection encode failed; preserving full event history: {error}"
+                );
+                *annotation = original_annotation
+                    .expect("codec-backed projection should preserve the original annotation")
+            }
+        }
+    }
+}
+
+fn limit_annotated_request_history_to_current_user_turn(
+    annotated_request: &mut AnnotatedLlmRequest,
+) -> bool {
+    retain_current_request_turn(
+        &mut annotated_request.messages,
+        |message| matches!(message, Message::User { .. }),
+        |message| matches!(message, Message::System { .. }),
+    )
+}
+
 fn emit_llm_start(
     handle: &LlmHandle,
     request: &LlmRequest,
@@ -339,9 +422,9 @@ fn emit_llm_start_with_subscribers(
             .map_err(|error| FlowError::Internal(error.to_string()))?;
         state.llm_sanitize_request_entries(&scope_locals)
     };
-    let sanitized_request =
+    let mut sanitized_request =
         NemoRelayContextState::llm_sanitize_request_snapshot_chain(request.clone(), &entries);
-    let annotated_request = match request_codec {
+    let mut annotated_request = match request_codec {
         Some(codec)
             if sanitized_request.headers != request.headers
                 || sanitized_request.content != request.content =>
@@ -350,6 +433,18 @@ fn emit_llm_start_with_subscribers(
         }
         _ => annotated_request,
     };
+    let scope_stack = handle.captured_scope_stack();
+    let agent_is_fresh = {
+        let mut scope_guard = scope_stack.write().expect("scope stack lock poisoned");
+        scope_guard.take_agent_freshness(handle.parent_uuid)
+    };
+    if !agent_is_fresh {
+        project_llm_request_to_current_user_turn(
+            &mut sanitized_request,
+            &mut annotated_request,
+            request_codec,
+        );
+    }
     let input = serde_json::to_value(&sanitized_request).unwrap_or(Json::Null);
     let event = {
         let context = global_context();
@@ -358,7 +453,7 @@ fn emit_llm_start_with_subscribers(
             .map_err(|error| FlowError::Internal(error.to_string()))?;
         state.build_llm_start_event(handle, Some(input), annotated_request)
     };
-    if let Some(event) = sanitize_event_with_scope_stack(event, handle.captured_scope_stack()) {
+    if let Some(event) = sanitize_event_with_scope_stack(event, scope_stack) {
         NemoRelayContextState::emit_event(&event, subscribers);
     }
     Ok(())
@@ -486,7 +581,10 @@ fn emit_optimization_marks_with(
 ///
 /// # Notes
 /// Sanitize-request guardrails affect only the emitted start-event payload, not
-/// the caller-owned [`LlmRequest`].
+/// the caller-owned [`LlmRequest`]. When the owning agent is not fresh, the
+/// emitted request annotation is limited to the current user turn. Managed
+/// calls with a request codec also apply that projection to the event input,
+/// without changing the request used for provider execution.
 pub fn llm_call(params: LlmCallParams<'_>) -> Result<LlmHandle> {
     let handle_params = CreateLlmHandleParams::builder()
         .name(params.name)

crates/core/src/api/runtime/scope_stack.rs

@@ -9,6 +9,7 @@
 //! context into worker threads.
 
 use std::cell::RefCell;
+use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, RwLock};
 
 use uuid::Uuid;
@@ -21,12 +22,15 @@ use crate::registry::{RegistryEntry, SortedRegistry};
 
 /// Mutable stack of active scopes plus their scope-local registries.
 ///
-/// The stack always contains an implicit root scope. Additional scopes are
-/// pushed as the public API opens lifecycle spans and removed when those spans
-/// close.
+/// The stack always contains an implicit root agent scope. It owns freshness
+/// for work that is not nested under an explicit agent; non-agent scopes inherit
+/// their nearest agent's freshness instead of creating a separate budget.
+/// Additional scopes are pushed as the public API opens lifecycle spans and
+/// removed when those spans close.
 pub struct ScopeStack {
     stack: Vec<ScopeHandle>,
-    scope_registries: std::collections::HashMap<Uuid, ScopeLocalRegistries>,
+    scope_registries: HashMap<Uuid, ScopeLocalRegistries>,
+    fresh_agents: HashSet<Uuid>,
 }
 
 impl ScopeStack {
@@ -40,9 +44,11 @@ impl ScopeStack {
             .name("root")
             .scope_type(ScopeType::Agent)
             .build();
+        let root_uuid = root.uuid;
         Self {
             stack: vec![root],
-            scope_registries: std::collections::HashMap::new(),
+            scope_registries: HashMap::new(),
+            fresh_agents: HashSet::from([root_uuid]),
         }
     }
 
@@ -51,6 +57,9 @@ impl ScopeStack {
     /// # Parameters
     /// - `handle`: Scope handle to make the new top-most active scope.
     pub fn push(&mut self, handle: ScopeHandle) {
+        if matches!(handle.scope_type, ScopeType::Agent) {
+            self.fresh_agents.insert(handle.uuid);
+        }
         self.stack.push(handle);
     }
 
@@ -134,6 +143,7 @@ impl ScopeStack {
                 ));
             }
             self.scope_registries.remove(uuid);
+            self.fresh_agents.remove(uuid);
             return Ok(self
                 .stack
                 .pop()
@@ -149,6 +159,34 @@ impl ScopeStack {
         Err(FlowError::NotFound("scope handle not found".into()))
     }
 
+    fn owning_agent_uuid(&self, parent_uuid: Option<Uuid>) -> Uuid {
+        let search_end = parent_uuid
+            .and_then(|parent_uuid| {
+                self.stack
+                    .iter()
+                    .position(|scope| scope.uuid == parent_uuid)
+            })
+            .map_or(self.stack.len(), |index| index + 1);
+        self.stack[..search_end]
+            .iter()
+            .rev()
+            .find(|scope| matches!(scope.scope_type, ScopeType::Agent))
+            .map(|scope| scope.uuid)
+            .expect("scope stack should always contain an owning agent")
+    }
+
+    /// Return whether the owning agent is fresh, then mark it stale.
+    pub(crate) fn take_agent_freshness(&mut self, parent_uuid: Option<Uuid>) -> bool {
+        let uuid = self.owning_agent_uuid(parent_uuid);
+        self.fresh_agents.remove(&uuid)
+    }
+
+    /// Mark the agent that owns a compaction event as fresh.
+    pub(crate) fn mark_agent_fresh(&mut self, parent_uuid: Option<Uuid>) {
+        let uuid = self.owning_agent_uuid(parent_uuid);
+        self.fresh_agents.insert(uuid);
+    }
+
     /// Get or create the scope-local registries for an active scope.
     ///
     /// # Parameters
@@ -210,6 +248,7 @@ impl std::fmt::Debug for ScopeStack {
         f.debug_struct("ScopeStack")
             .field("stack", &self.stack)
             .field("scope_registries_count", &self.scope_registries.len())
+            .field("fresh_agent_count", &self.fresh_agents.len())
             .finish()
     }
 }

crates/core/src/api/scope.rs

@@ -19,6 +19,9 @@ use uuid::Uuid;
 
 pub use nemo_relay_types::api::scope::{HandleAttributes, ScopeAttributes, ScopeType};
 
+/// Canonical mark-event name used to indicate agent context compaction.
+pub const COMPACTION_EVENT_NAME: &str = "compaction";
+
 /// Runtime-owned handle identifying an active or completed scope.
 #[derive(Debug, Clone, Serialize, Deserialize, TypedBuilder)]
 #[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
@@ -328,11 +331,18 @@ pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> {
 pub fn event(params: EmitMarkEventParams<'_>) -> Result<()> {
     ensure_runtime_owner()?;
     let parent_uuid = resolve_parent_uuid(params.parent);
+    let scope_stack = current_scope_stack();
     let (event, subscribers) = {
-        let scope_stack = current_scope_stack();
-        let scope_guard = scope_stack.read().expect("scope stack lock poisoned");
-        let scope_subscribers = scope_guard.collect_scope_local_subscribers();
-        let subscribers = snapshot_event_subscribers(scope_subscribers)?;
+        let subscribers = if params.name == COMPACTION_EVENT_NAME {
+            let mut scope_guard = scope_stack.write().expect("scope stack lock poisoned");
+            let subscribers =
+                snapshot_event_subscribers(scope_guard.collect_scope_local_subscribers())?;
+            scope_guard.mark_agent_fresh(parent_uuid);
+            subscribers
+        } else {
+            let scope_guard = scope_stack.read().expect("scope stack lock poisoned");
+            snapshot_event_subscribers(scope_guard.collect_scope_local_subscribers())?
+        };
         let context = global_context();
         let state = context
             .read()

crates/core/tests/unit/context_tests.rs

@@ -95,6 +95,46 @@ fn scope_stack_rejects_removing_non_top_or_root_scopes() {
     assert!(err.to_string().contains("root scope cannot be removed"));
 }
 
+#[test]
+fn scope_stack_tracks_freshness_per_owning_agent() {
+    let mut stack = ScopeStack::new();
+    let root_uuid = stack.root_uuid();
+    assert!(stack.take_agent_freshness(Some(root_uuid)));
+    assert!(!stack.take_agent_freshness(Some(root_uuid)));
+
+    let parent = ScopeHandle::builder()
+        .name("parent-agent")
+        .scope_type(ScopeType::Agent)
+        .parent_uuid(root_uuid)
+        .build();
+    let parent_uuid = parent.uuid;
+    stack.push(parent);
+    let turn = ScopeHandle::builder()
+        .name("turn")
+        .scope_type(ScopeType::Custom)
+        .parent_uuid(parent_uuid)
+        .build();
+    let turn_uuid = turn.uuid;
+    stack.push(turn);
+    assert!(stack.take_agent_freshness(Some(turn_uuid)));
+
+    let child = ScopeHandle::builder()
+        .name("child-agent")
+        .scope_type(ScopeType::Agent)
+        .parent_uuid(turn_uuid)
+        .build();
+    let child_uuid = child.uuid;
+    stack.push(child);
+    assert!(stack.take_agent_freshness(Some(child_uuid)));
+    stack.mark_agent_fresh(Some(turn_uuid));
+    assert!(!stack.take_agent_freshness(Some(child_uuid)));
+
+    stack.remove(&child_uuid).unwrap();
+    assert!(stack.take_agent_freshness(Some(turn_uuid)));
+    stack.remove(&turn_uuid).unwrap();
+    stack.remove(&parent_uuid).unwrap();
+}
+
 #[test]
 fn merge_helpers_preserve_global_and_scope_local_priority_order() {
     let mut global_guardrails = SortedRegistry::new();