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(
134 provider,
135 model,
136 build_compaction_verify_prompt(&draft),
137 )
138 .await?
139 .unwrap_or(draft.clone());
140 if accept_llm_compaction_summary(head, &verified) {
141 Ok(Some(verified))
142 } else if accept_llm_compaction_summary(head, &draft) {
143 Ok(Some(draft))
144 } else {
145 Ok(None)
146 }
147 }
148
149 async fn run_compaction_summary_inference(
150 &self,
151 provider: &str,
152 model: &str,
153 prompt: String,
154 ) -> anyhow::Result<Option<String>> {
155 let engine = self.engine_for(provider)?;
156 let request = AgentInferenceRequest {
157 model: ModelSelection {
158 provider: provider.to_string(),
159 model: model.to_string(),
160 },
161 instructions: InstructionBundle {
162 system: Some(
163 "You compress conversation history into durable state snapshots.".to_string(),
164 ),
165 developer: None,
166 developer_context: None,
167 },
168 transcript: vec![TranscriptItem::UserMessage(UserMessage::text(prompt))],
169 tools: Vec::new(),
170 tool_choice: ToolChoice::None,
171 reasoning: ReasoningConfig::default(),
172 output: OutputConfig::default(),
173 runtime: RuntimeHints {
174 profile: RuntimeProfile::Interactive,
175 ..RuntimeHints::default()
176 },
177 metadata: serde_json::json!({ "roderCompactionSummary": true }),
178 };
179 let ctx = InferenceTurnContext {
180 thread_id: &"compaction-summary".to_string(),
181 turn_id: &"compaction-summary".to_string(),
182 tool_executor: None,
183 };
184 let mut stream = engine.stream_turn(ctx, request).await?;
185 let mut text = String::new();
186 while let Some(event) = stream.next().await {
187 match event? {
188 InferenceEvent::MessageDelta(MessageDelta { text: delta, .. }) => {
189 text.push_str(&delta)
190 }
191 InferenceEvent::Failed(failure) => {
192 anyhow::bail!("compaction summary inference failed: {}", failure.message);
193 }
194 InferenceEvent::Completed(_) => break,
195 _ => {}
196 }
197 }
198 if text.trim().is_empty() {
199 Ok(None)
200 } else {
201 Ok(Some(text.trim().to_string()))
202 }
203 }
204}
205
206#[derive(Debug, Clone)]
207pub struct ForceCompactOutcome {
208 pub compacted: bool,
209 pub reason: Option<String>,
210 pub estimated_tokens_before: u32,
211 pub estimated_tokens_after: u32,
212}
213
214pub(crate) fn compaction_hysteresis_state() -> Mutex<HashMap<ThreadId, u32>> {
215 Mutex::new(HashMap::new())
216}