1use std::collections::HashMap;
2
3use futures::StreamExt;
4use roder_api::events::{ThreadId, TurnId};
5use roder_api::inference::{
6 AgentInferenceRequest, InferenceEvent, InferenceTurnContext, InstructionBundle, MessageDelta,
7 ModelSelection, OutputConfig, ReasoningConfig, RuntimeHints, RuntimeProfile,
8};
9use roder_api::tools::ToolChoice;
10use roder_api::transcript::{TranscriptItem, UserMessage};
11use std::sync::Mutex;
12
13use crate::compaction::{
14 CompactionOptions, accept_llm_compaction_summary, build_compaction_summary_prompt,
15 build_compaction_verify_prompt, estimate_prompt_tokens,
16};
17use crate::runtime::Runtime;
18
19impl Runtime {
20 pub(crate) fn record_compaction_hysteresis(&self, thread_id: &ThreadId, trigger_tokens: u32) {
21 if let Ok(mut state) = self.compaction_hysteresis.lock() {
22 state.insert(thread_id.clone(), trigger_tokens);
23 }
24 }
25
26 pub(crate) fn compaction_hysteresis_baseline(&self, thread_id: &ThreadId) -> Option<u32> {
27 self.compaction_hysteresis
28 .lock()
29 .ok()
30 .and_then(|state| state.get(thread_id).copied())
31 }
32
33 pub(crate) fn compaction_options_for_turn(
34 &self,
35 thread_id: &ThreadId,
36 allow_repeat: bool,
37 ) -> CompactionOptions {
38 CompactionOptions {
39 allow_repeat,
40 force: false,
41 hysteresis_baseline: self.compaction_hysteresis_baseline(thread_id),
42 preserve_hint: None,
43 }
44 }
45
46 pub async fn force_compact_thread(
47 &self,
48 thread_id: &ThreadId,
49 turn_id: &TurnId,
50 preserve_hint: Option<String>,
51 ) -> anyhow::Result<ForceCompactOutcome> {
52 let cfg = self.status().await;
53 let provider = cfg.default_provider.clone();
54 let model = cfg.default_model.clone();
55 let transcript = self.transcript_for_force_compact(thread_id).await?;
56 if transcript.is_empty() {
57 return Ok(ForceCompactOutcome {
58 compacted: false,
59 reason: Some("empty_transcript".to_string()),
60 estimated_tokens_before: 0,
61 estimated_tokens_after: 0,
62 });
63 }
64 let estimated_before = estimate_prompt_tokens(&transcript);
65 let compacted = self
66 .compact_transcript_if_needed(
67 thread_id,
68 turn_id,
69 &provider,
70 &model,
71 transcript,
72 CompactionOptions {
73 allow_repeat: true,
74 force: true,
75 hysteresis_baseline: None,
76 preserve_hint: preserve_hint.filter(|text| !text.trim().is_empty()),
77 },
78 )
79 .await?;
80 let estimated_after = estimate_prompt_tokens(&compacted);
81 Ok(ForceCompactOutcome {
82 compacted: estimated_after < estimated_before
83 || compacted
84 .iter()
85 .any(|item| matches!(item, TranscriptItem::ContextCompaction(_))),
86 reason: None,
87 estimated_tokens_before: estimated_before,
88 estimated_tokens_after: estimated_after,
89 })
90 }
91
92 async fn transcript_for_force_compact(
93 &self,
94 thread_id: &ThreadId,
95 ) -> anyhow::Result<Vec<TranscriptItem>> {
96 let Some(store) = &self.thread_store else {
97 return Ok(Vec::new());
98 };
99 let Some(snapshot) = store.load_thread(thread_id).await? else {
100 return Ok(Vec::new());
101 };
102 let mut out = Vec::new();
103 for turn in snapshot.turns {
104 out.extend(turn.items);
105 }
106 Ok(crate::compaction::trim_to_last_compaction_boundary(out))
107 }
108
109 pub(crate) async fn summarize_compaction_head(
110 &self,
111 provider: &str,
112 model: &str,
113 head: &[TranscriptItem],
114 preserve_hint: Option<&str>,
115 ) -> anyhow::Result<Option<String>> {
116 if head.is_empty() {
117 return Ok(None);
118 }
119 let draft = self
120 .run_compaction_summary_inference(
121 provider,
122 model,
123 build_compaction_summary_prompt(head, preserve_hint),
124 )
125 .await?;
126 let Some(draft) = draft else {
127 return Ok(None);
128 };
129 if !accept_llm_compaction_summary(head, &draft) {
130 return Ok(None);
131 }
132 let verified = self
133 .run_compaction_summary_inference(provider, model, build_compaction_verify_prompt(&draft))
134 .await?
135 .unwrap_or(draft.clone());
136 if accept_llm_compaction_summary(head, &verified) {
137 Ok(Some(verified))
138 } else if accept_llm_compaction_summary(head, &draft) {
139 Ok(Some(draft))
140 } else {
141 Ok(None)
142 }
143 }
144
145 async fn run_compaction_summary_inference(
146 &self,
147 provider: &str,
148 model: &str,
149 prompt: String,
150 ) -> anyhow::Result<Option<String>> {
151 let engine = self.engine_for(provider)?;
152 let request = AgentInferenceRequest {
153 model: ModelSelection {
154 provider: provider.to_string(),
155 model: model.to_string(),
156 },
157 instructions: InstructionBundle {
158 system: Some(
159 "You compress conversation history into durable state snapshots.".to_string(),
160 ),
161 developer: None,
162 developer_context: None,
163 },
164 transcript: vec![TranscriptItem::UserMessage(UserMessage::text(prompt))],
165 tools: Vec::new(),
166 tool_choice: ToolChoice::None,
167 reasoning: ReasoningConfig::default(),
168 output: OutputConfig::default(),
169 runtime: RuntimeHints {
170 profile: RuntimeProfile::Interactive,
171 ..RuntimeHints::default()
172 },
173 metadata: serde_json::json!({ "roderCompactionSummary": true }),
174 };
175 let ctx = InferenceTurnContext {
176 thread_id: &"compaction-summary".to_string(),
177 turn_id: &"compaction-summary".to_string(),
178 tool_executor: None,
179 };
180 let mut stream = engine.stream_turn(ctx, request).await?;
181 let mut text = String::new();
182 while let Some(event) = stream.next().await {
183 match event? {
184 InferenceEvent::MessageDelta(MessageDelta { text: delta, .. }) => text.push_str(&delta),
185 InferenceEvent::Failed(failure) => {
186 anyhow::bail!("compaction summary inference failed: {}", failure.message);
187 }
188 InferenceEvent::Completed(_) => break,
189 _ => {}
190 }
191 }
192 if text.trim().is_empty() {
193 Ok(None)
194 } else {
195 Ok(Some(text.trim().to_string()))
196 }
197 }
198}
199
200#[derive(Debug, Clone)]
201pub struct ForceCompactOutcome {
202 pub compacted: bool,
203 pub reason: Option<String>,
204 pub estimated_tokens_before: u32,
205 pub estimated_tokens_after: u32,
206}
207
208pub(crate) fn compaction_hysteresis_state() -> Mutex<HashMap<ThreadId, u32>> {
209 Mutex::new(HashMap::new())
210}