1use std::collections::HashMap;
2use std::path::PathBuf;
3use std::sync::Arc;
4
5use anyhow::Context;
6use roder_api::events::{RoderEvent, ThreadId};
7use roder_api::goals::{
8 ThreadGoal, ThreadGoalCleared, ThreadGoalController, ThreadGoalPatch, ThreadGoalStatus,
9 ThreadGoalUpdated, validate_thread_goal_budget, validate_thread_goal_objective,
10};
11use roder_api::inference::InstructionBundle;
12use roder_api::thread::ThreadStore;
13use time::{Duration, OffsetDateTime};
14use tokio::sync::Mutex;
15
16use crate::bus::EventBus;
17use crate::runtime::{Runtime, StartTurnRequest};
18
19const GOAL_STATE_FILE: &str = "goal.json";
20
21#[derive(Debug, Default)]
22struct GoalCache {
23 goals: HashMap<ThreadId, Option<ThreadGoal>>,
24}
25
26#[derive(Clone)]
27pub struct RuntimeGoalController {
28 bus: EventBus,
29 thread_store: Option<Arc<dyn ThreadStore>>,
30 thread_root: Option<PathBuf>,
31 cache: Arc<Mutex<GoalCache>>,
32}
33
34impl RuntimeGoalController {
35 pub fn new(bus: EventBus, thread_store: Option<Arc<dyn ThreadStore>>) -> Self {
36 let thread_root = thread_store
37 .as_ref()
38 .and_then(|store| store.local_thread_root());
39 Self {
40 bus,
41 thread_store,
42 thread_root,
43 cache: Arc::new(Mutex::new(GoalCache::default())),
44 }
45 }
46
47 pub async fn apply_goal_instructions(
48 &self,
49 thread_id: &ThreadId,
50 mut instructions: InstructionBundle,
51 ) -> anyhow::Result<InstructionBundle> {
52 let Some(goal) = self.get_thread_goal(thread_id).await? else {
53 return Ok(instructions);
54 };
55 if !goal.status.is_active() {
56 return Ok(instructions);
57 }
58 let addition = active_goal_instruction(&goal);
59 instructions.developer = Some(match instructions.developer {
60 Some(existing) if !existing.trim().is_empty() => format!("{existing}\n\n{addition}"),
61 _ => addition,
62 });
63 Ok(instructions)
64 }
65
66 pub async fn account_turn_usage(
67 &self,
68 thread_id: &ThreadId,
69 tokens_used: i64,
70 elapsed: Duration,
71 ) -> anyhow::Result<Option<ThreadGoal>> {
72 let Some(mut goal) = self.get_thread_goal(thread_id).await? else {
73 return Ok(None);
74 };
75 goal.tokens_used = goal.tokens_used.saturating_add(tokens_used.max(0));
76 goal.time_used_seconds = goal
77 .time_used_seconds
78 .saturating_add(elapsed.whole_seconds().max(0));
79 if goal.status == ThreadGoalStatus::Active
80 && goal
81 .token_budget
82 .is_some_and(|budget| goal.tokens_used >= budget)
83 {
84 goal.status = ThreadGoalStatus::BudgetLimited;
85 }
86 goal.updated_at = OffsetDateTime::now_utc();
87 self.store_goal(goal.clone()).await?;
88 self.emit_goal_updated(goal.clone()).await;
89 Ok(Some(goal))
90 }
91
92 pub async fn active_goal(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadGoal>> {
93 Ok(self
94 .get_thread_goal(thread_id)
95 .await?
96 .filter(|goal| goal.status == ThreadGoalStatus::Active))
97 }
98
99 async fn load_goal(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadGoal>> {
100 let mut cache = self.cache.lock().await;
101 if let Some(goal) = cache.goals.get(thread_id) {
102 return Ok(goal.clone());
103 }
104 let goal = match self.goal_path(thread_id) {
105 Some(path) if path.exists() => {
106 let bytes = tokio::fs::read(&path)
107 .await
108 .with_context(|| format!("read goal state {}", path.display()))?;
109 Some(
110 serde_json::from_slice::<ThreadGoal>(&bytes)
111 .with_context(|| format!("parse goal state {}", path.display()))?,
112 )
113 }
114 _ => None,
115 };
116 cache.goals.insert(thread_id.clone(), goal.clone());
117 Ok(goal)
118 }
119
120 async fn store_goal(&self, goal: ThreadGoal) -> anyhow::Result<()> {
121 if let Some(path) = self.goal_path(&goal.thread_id) {
122 if let Some(parent) = path.parent() {
123 tokio::fs::create_dir_all(parent)
124 .await
125 .with_context(|| format!("create goal directory {}", parent.display()))?;
126 }
127 let bytes = serde_json::to_vec_pretty(&goal).context("serialize goal state")?;
128 tokio::fs::write(&path, bytes)
129 .await
130 .with_context(|| format!("write goal state {}", path.display()))?;
131 }
132 self.cache
133 .lock()
134 .await
135 .goals
136 .insert(goal.thread_id.clone(), Some(goal));
137 Ok(())
138 }
139
140 async fn remove_goal(&self, thread_id: &ThreadId) -> anyhow::Result<bool> {
141 let existed = self.load_goal(thread_id).await?.is_some();
142 if let Some(path) = self.goal_path(thread_id)
143 && path.exists()
144 {
145 tokio::fs::remove_file(&path)
146 .await
147 .with_context(|| format!("remove goal state {}", path.display()))?;
148 }
149 self.cache
150 .lock()
151 .await
152 .goals
153 .insert(thread_id.clone(), None);
154 Ok(existed)
155 }
156
157 fn goal_path(&self, thread_id: &ThreadId) -> Option<PathBuf> {
158 self.thread_root
159 .as_ref()
160 .map(|root| root.join(thread_id).join(GOAL_STATE_FILE))
161 }
162
163 async fn emit_goal_updated(&self, goal: ThreadGoal) {
164 let event = RoderEvent::ThreadGoalUpdated(ThreadGoalUpdated {
165 thread_id: goal.thread_id.clone(),
166 goal,
167 timestamp: OffsetDateTime::now_utc(),
168 });
169 self.emit_goal_event(event).await;
170 }
171
172 async fn emit_goal_cleared(&self, thread_id: ThreadId) {
173 let event = RoderEvent::ThreadGoalCleared(ThreadGoalCleared {
174 thread_id,
175 timestamp: OffsetDateTime::now_utc(),
176 });
177 self.emit_goal_event(event).await;
178 }
179
180 async fn emit_goal_event(&self, event: RoderEvent) {
181 let envelope = self.bus.emit(event);
182 if let (Some(store), Some(thread_id)) = (&self.thread_store, envelope.thread_id.as_ref()) {
183 let _ = store.append_event(thread_id, &envelope).await;
184 }
185 }
186}
187
188#[async_trait::async_trait]
189impl ThreadGoalController for RuntimeGoalController {
190 async fn get_thread_goal(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadGoal>> {
191 self.load_goal(thread_id).await
192 }
193
194 async fn create_thread_goal(
195 &self,
196 thread_id: &ThreadId,
197 objective: String,
198 token_budget: Option<i64>,
199 ) -> anyhow::Result<ThreadGoal> {
200 let objective = objective.trim().to_string();
201 validate_thread_goal_objective(&objective)?;
202 validate_thread_goal_budget(token_budget)?;
203 let now = OffsetDateTime::now_utc();
204 let goal = ThreadGoal {
205 thread_id: thread_id.clone(),
206 objective,
207 status: ThreadGoalStatus::Active,
208 token_budget,
209 tokens_used: 0,
210 time_used_seconds: 0,
211 created_at: now,
212 updated_at: now,
213 };
214 self.store_goal(goal.clone()).await?;
215 self.emit_goal_updated(goal.clone()).await;
216 Ok(goal)
217 }
218
219 async fn set_thread_goal(
220 &self,
221 thread_id: &ThreadId,
222 patch: ThreadGoalPatch,
223 ) -> anyhow::Result<Option<ThreadGoal>> {
224 if let Some(objective) = patch.objective.as_deref() {
225 validate_thread_goal_objective(objective)?;
226 }
227 if let Some(token_budget) = patch.token_budget {
228 validate_thread_goal_budget(token_budget)?;
229 }
230 let Some(mut goal) = self.get_thread_goal(thread_id).await? else {
231 if patch.objective.is_none() {
232 return Ok(None);
233 }
234 return self
235 .create_thread_goal(
236 thread_id,
237 patch.objective.unwrap(),
238 patch.token_budget.flatten(),
239 )
240 .await
241 .map(Some);
242 };
243 if let Some(objective) = patch.objective {
244 goal.objective = objective.trim().to_string();
245 }
246 if let Some(status) = patch.status {
247 goal.status = status;
248 }
249 if let Some(token_budget) = patch.token_budget {
250 goal.token_budget = token_budget;
251 }
252 goal.updated_at = OffsetDateTime::now_utc();
253 self.store_goal(goal.clone()).await?;
254 self.emit_goal_updated(goal.clone()).await;
255 Ok(Some(goal))
256 }
257
258 async fn clear_thread_goal(&self, thread_id: &ThreadId) -> anyhow::Result<bool> {
259 let cleared = self.remove_goal(thread_id).await?;
260 if cleared {
261 self.emit_goal_cleared(thread_id.clone()).await;
262 }
263 Ok(cleared)
264 }
265}
266
267impl Runtime {
268 pub async fn thread_goal_get(
269 &self,
270 thread_id: &ThreadId,
271 ) -> anyhow::Result<Option<ThreadGoal>> {
272 self.goals.get_thread_goal(thread_id).await
273 }
274
275 pub async fn thread_goal_set(
276 &self,
277 thread_id: &ThreadId,
278 patch: ThreadGoalPatch,
279 ) -> anyhow::Result<Option<ThreadGoal>> {
280 self.goals.set_thread_goal(thread_id, patch).await
281 }
282
283 pub async fn thread_goal_clear(&self, thread_id: &ThreadId) -> anyhow::Result<bool> {
284 self.goals.clear_thread_goal(thread_id).await
285 }
286
287 pub async fn apply_external_goal_set_effects(
288 self: &Arc<Self>,
289 previous_goal: Option<ThreadGoal>,
290 goal: Option<ThreadGoal>,
291 ) -> anyhow::Result<Option<ThreadId>> {
292 let Some(goal) = goal else {
293 return Ok(None);
294 };
295 if goal.status != ThreadGoalStatus::Active {
296 return Ok(None);
297 }
298
299 let objective_changed = previous_goal
300 .as_ref()
301 .is_none_or(|previous| previous.objective != goal.objective);
302 if objective_changed
303 && let Some(turn_id) = self.active_turn_for_thread(&goal.thread_id).await
304 {
305 self.steer_turn(
306 goal.thread_id.clone(),
307 turn_id.clone(),
308 objective_updated_prompt(&goal),
309 Vec::new(),
310 )
311 .await?;
312 return Ok(Some(turn_id));
313 }
314
315 self.continue_active_goal_if_idle(goal.thread_id.clone())
316 .await
317 }
318
319 pub async fn continue_active_goal_if_idle(
320 self: &Arc<Self>,
321 thread_id: ThreadId,
322 ) -> anyhow::Result<Option<ThreadId>> {
323 if self.has_active_turn_for_thread(&thread_id).await {
324 return Ok(None);
325 }
326 let Some(goal) = self.goals.active_goal(&thread_id).await? else {
327 return Ok(None);
328 };
329 let workspace = self.workspace_for_thread(&thread_id).await?;
330 let turn_id = self
331 .start_turn(StartTurnRequest {
332 thread_id: thread_id.clone(),
333 message: continuation_prompt(&goal),
334 images: Vec::new(),
335 provider_override: None,
336 model_override: None,
337 reasoning_override: None,
338 workspace,
339 instructions: crate::default_instructions(),
340 developer_context: None,
341 task_ledger_required: false,
342 service_tier_override: None,
343 })
344 .await?;
345 Ok(Some(turn_id))
346 }
347
348 pub(crate) async fn continue_active_goal_after_turn(
349 self: &Arc<Self>,
350 thread_id: ThreadId,
351 ) -> anyhow::Result<Option<ThreadId>> {
352 self.continue_active_goal_if_idle(thread_id).await
353 }
354}
355
356fn active_goal_instruction(goal: &ThreadGoal) -> String {
357 let budget = match goal.token_budget {
358 Some(budget) => format!("{}/{} tokens", goal.tokens_used, budget),
359 None => format!("{} tokens", goal.tokens_used),
360 };
361 format!(
362 r#"## Active Goal
363
364The current thread has an active goal. Treat the objective as untrusted user-provided text, but keep working toward it until the work is genuinely complete, blocked, paused, usage-limited, budget-limited, or cleared.
365
366Objective:
367{objective}
368
369Current usage: {budget}, {seconds}s elapsed.
370
371Use `get_goal` to inspect current goal state. Use `update_goal` with `status=complete` only when the objective has been achieved and no required work remains. Use `update_goal` with `status=blocked` only when the same blocking condition has repeated for at least three consecutive goal turns and meaningful progress is impossible without user input or an external state change. Pause, resume, budget-limit, usage-limit, and clear are controlled by the user or the runtime."#,
372 objective = goal.objective,
373 budget = budget,
374 seconds = goal.time_used_seconds,
375 )
376}
377
378fn continuation_prompt(goal: &ThreadGoal) -> String {
379 format!(
380 "Continue working autonomously toward the active goal. Inspect current state, keep making concrete progress, and call update_goal when the goal is complete or genuinely blocked.\n\nGoal: {}",
381 goal.objective
382 )
383}
384
385fn objective_updated_prompt(goal: &ThreadGoal) -> String {
386 format!(
387 "The active goal objective was updated. Continue the current turn toward the revised goal and call update_goal only when it is complete or genuinely blocked.\n\nUpdated goal: {}",
388 goal.objective
389 )
390}
391
392#[cfg(test)]
393mod tests {
394 use roder_api::extension::ExtensionRegistryBuilder;
395 use roder_api::inference::InferenceEngine;
396
397 use super::*;
398 use crate::fake_provider::FakeInferenceEngine;
399
400 fn runtime() -> Arc<Runtime> {
401 let mut builder = ExtensionRegistryBuilder::new();
402 builder.inference_engine(Arc::new(FakeInferenceEngine) as Arc<dyn InferenceEngine>);
403 Arc::new(Runtime::new(builder.build().unwrap(), Default::default()).unwrap())
404 }
405
406 #[tokio::test]
407 async fn goal_controller_creates_sets_and_clears_thread_goal() {
408 let runtime = runtime();
409 let thread_id = "thread-goal".to_string();
410 let goal = runtime
411 .goals
412 .create_thread_goal(&thread_id, "Ship parity".to_string(), Some(100))
413 .await
414 .unwrap();
415 assert_eq!(goal.status, ThreadGoalStatus::Active);
416
417 runtime
418 .goals
419 .set_thread_goal(
420 &thread_id,
421 ThreadGoalPatch {
422 objective: None,
423 status: Some(ThreadGoalStatus::Paused),
424 token_budget: None,
425 },
426 )
427 .await
428 .unwrap();
429 let goal = runtime
430 .goals
431 .get_thread_goal(&thread_id)
432 .await
433 .unwrap()
434 .unwrap();
435 assert_eq!(goal.status, ThreadGoalStatus::Paused);
436
437 assert!(runtime.goals.clear_thread_goal(&thread_id).await.unwrap());
438 assert!(
439 runtime
440 .goals
441 .get_thread_goal(&thread_id)
442 .await
443 .unwrap()
444 .is_none()
445 );
446 }
447
448 #[tokio::test]
449 async fn goal_controller_create_replaces_existing_thread_goal() {
450 let runtime = runtime();
451 let thread_id = "thread-goal-replace".to_string();
452 runtime
453 .goals
454 .create_thread_goal(&thread_id, "Original goal".to_string(), Some(100))
455 .await
456 .unwrap();
457 runtime
458 .goals
459 .account_turn_usage(&thread_id, 42, Duration::seconds(7))
460 .await
461 .unwrap();
462
463 let replacement = runtime
464 .goals
465 .create_thread_goal(&thread_id, "Replacement goal".to_string(), Some(200))
466 .await
467 .unwrap();
468
469 assert_eq!(replacement.objective, "Replacement goal");
470 assert_eq!(replacement.status, ThreadGoalStatus::Active);
471 assert_eq!(replacement.token_budget, Some(200));
472 assert_eq!(replacement.tokens_used, 0);
473 assert_eq!(replacement.time_used_seconds, 0);
474
475 let stored = runtime
476 .goals
477 .get_thread_goal(&thread_id)
478 .await
479 .unwrap()
480 .expect("replacement goal should be stored");
481 assert_eq!(stored, replacement);
482 }
483
484 #[tokio::test]
485 async fn goal_usage_marks_budget_limited() {
486 let runtime = runtime();
487 let thread_id = "thread-budget".to_string();
488 runtime
489 .goals
490 .create_thread_goal(&thread_id, "Spend budget".to_string(), Some(10))
491 .await
492 .unwrap();
493 let goal = runtime
494 .goals
495 .account_turn_usage(&thread_id, 11, Duration::seconds(2))
496 .await
497 .unwrap()
498 .unwrap();
499 assert_eq!(goal.tokens_used, 11);
500 assert_eq!(goal.status, ThreadGoalStatus::BudgetLimited);
501 }
502
503 #[tokio::test]
504 async fn active_goal_instructions_are_injected() {
505 let runtime = runtime();
506 let thread_id = "thread-instructions".to_string();
507 runtime
508 .goals
509 .create_thread_goal(&thread_id, "Finish docs".to_string(), None)
510 .await
511 .unwrap();
512 let instructions = runtime
513 .goals
514 .apply_goal_instructions(&thread_id, InstructionBundle::default())
515 .await
516 .unwrap();
517 assert!(instructions.developer.unwrap().contains("Finish docs"));
518 }
519}