1#[cfg(any(test, not(unix)))]
2use std::fs;
3use std::fs::{File, OpenOptions};
4use std::io::{self, BufRead, BufReader, Write};
5use std::path::{Path, PathBuf};
6use std::sync::atomic::{AtomicU64, Ordering};
7
8use serde::{Deserialize, Serialize};
9
10use super::Session;
11use crate::config::{ensure_not_symlink, ensure_private_dir};
12
13#[cfg(unix)]
14use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
15
16static TURN_COUNTER: AtomicU64 = AtomicU64::new(0);
17const TURN_JOURNAL_VERSION: u8 = 1;
18
19#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
20#[serde(rename_all = "snake_case")]
21pub enum TurnKind {
22 User,
23 Background,
24}
25
26#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
27#[serde(rename_all = "snake_case")]
28pub enum TurnEvent {
29 Started,
30 PhaseChanged,
31 Finished,
32}
33
34#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
35#[serde(rename_all = "snake_case")]
36pub enum TurnPhase {
37 Accepted,
38 Compacting,
39 ProviderStream,
40 ExecutingTools,
41}
42
43#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
44#[serde(rename_all = "snake_case")]
45pub enum TurnOutcome {
46 Completed,
47 RetryableFailure,
48 TerminalFailure,
49 Interrupted,
50 Abandoned,
51}
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub enum TurnStatus {
55 Active,
56 Retryable,
57 Completed,
58 TerminalFailure,
59 Interrupted,
60 Abandoned,
61}
62
63#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
64pub struct TurnLifecycleRecord {
65 pub timestamp: u64,
66 pub turn_id: String,
67 pub event: TurnEvent,
68 #[serde(default, skip_serializing_if = "Option::is_none")]
69 pub kind: Option<TurnKind>,
70 #[serde(default, skip_serializing_if = "Option::is_none")]
71 pub phase: Option<TurnPhase>,
72 #[serde(default, skip_serializing_if = "Option::is_none")]
73 pub outcome: Option<TurnOutcome>,
74 #[serde(default, skip_serializing_if = "Option::is_none")]
75 pub first_message: Option<usize>,
76 #[serde(default, skip_serializing_if = "Option::is_none")]
77 pub last_message: Option<usize>,
78 #[serde(default, skip_serializing_if = "Option::is_none")]
79 pub error: Option<String>,
80}
81
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct TurnState {
84 pub turn_id: String,
85 pub kind: TurnKind,
86 pub phase: TurnPhase,
87 pub status: TurnStatus,
88 pub first_message: usize,
90 pub last_message: Option<usize>,
92 pub error: Option<String>,
93}
94
95impl TurnState {
96 pub fn is_pending(&self) -> bool {
97 matches!(self.status, TurnStatus::Active | TurnStatus::Retryable)
98 }
99}
100
101#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
102struct StoredTurnRecord {
103 record: String,
104 version: u8,
105 #[serde(flatten)]
106 lifecycle: TurnLifecycleRecord,
107}
108
109impl StoredTurnRecord {
110 fn new(lifecycle: TurnLifecycleRecord) -> Self {
111 Self {
112 record: "turn".to_owned(),
113 version: TURN_JOURNAL_VERSION,
114 lifecycle,
115 }
116 }
117}
118
119impl Session {
120 pub fn start_turn(&mut self, kind: TurnKind) -> Result<String, String> {
121 if self.latest_turn()?.is_some_and(|turn| turn.is_pending()) {
122 return Err("session already has a pending turn".to_owned());
123 }
124 let turn_id = new_turn_id(&self.id);
125 self.append_turn_lifecycle(TurnLifecycleRecord {
126 timestamp: now(),
127 turn_id: turn_id.clone(),
128 event: TurnEvent::Started,
129 kind: Some(kind),
130 phase: Some(TurnPhase::Accepted),
131 outcome: None,
132 first_message: Some(self.messages.len()),
133 last_message: None,
134 error: None,
135 })?;
136 Ok(turn_id)
137 }
138
139 pub fn set_turn_phase(&mut self, turn_id: &str, phase: TurnPhase) -> Result<(), String> {
140 let state = self.require_pending_turn(turn_id)?;
141 if state.status != TurnStatus::Active {
142 return Err("retryable turn must be resumed before changing phase".to_owned());
143 }
144 self.append_turn_lifecycle(TurnLifecycleRecord {
145 timestamp: now(),
146 turn_id: turn_id.to_owned(),
147 event: TurnEvent::PhaseChanged,
148 kind: None,
149 phase: Some(phase),
150 outcome: None,
151 first_message: None,
152 last_message: None,
153 error: None,
154 })
155 }
156
157 pub fn resume_retryable_turn(&mut self, turn_id: &str) -> Result<(), String> {
158 let state = self.require_pending_turn(turn_id)?;
159 if !matches!(state.status, TurnStatus::Active | TurnStatus::Retryable) {
160 return Err("turn is not retryable".to_owned());
161 }
162 self.append_turn_lifecycle(TurnLifecycleRecord {
163 timestamp: now(),
164 turn_id: turn_id.to_owned(),
165 event: TurnEvent::PhaseChanged,
166 kind: None,
167 phase: Some(TurnPhase::Accepted),
168 outcome: None,
169 first_message: None,
170 last_message: None,
171 error: None,
172 })
173 }
174
175 pub fn finish_turn(
176 &mut self,
177 turn_id: &str,
178 outcome: TurnOutcome,
179 error: Option<String>,
180 ) -> Result<(), String> {
181 let state = self.require_pending_turn(turn_id)?;
182 if outcome == TurnOutcome::RetryableFailure {
183 return Err("use fail_turn_retryably for a retryable failure".to_owned());
184 }
185 if state.status == TurnStatus::Retryable && outcome != TurnOutcome::Abandoned {
186 return Err("retryable turn must be resumed or abandoned before finishing".to_owned());
187 }
188 self.append_turn_lifecycle(TurnLifecycleRecord {
189 timestamp: now(),
190 turn_id: turn_id.to_owned(),
191 event: TurnEvent::Finished,
192 kind: None,
193 phase: None,
194 outcome: Some(outcome),
195 first_message: None,
196 last_message: Some(self.messages.len()),
197 error,
198 })
199 }
200
201 pub fn fail_turn_retryably(&mut self, turn_id: &str, error: String) -> Result<(), String> {
202 let state = self.require_pending_turn(turn_id)?;
203 if state.status != TurnStatus::Active {
204 return Err("turn is already retryable".to_owned());
205 }
206 self.append_turn_lifecycle(TurnLifecycleRecord {
207 timestamp: now(),
208 turn_id: turn_id.to_owned(),
209 event: TurnEvent::Finished,
210 kind: None,
211 phase: None,
212 outcome: Some(TurnOutcome::RetryableFailure),
213 first_message: None,
214 last_message: Some(self.messages.len()),
215 error: Some(error),
216 })
217 }
218
219 pub fn latest_turn(&self) -> Result<Option<TurnState>, String> {
220 reduce_turn_records(&load_turn_records(&turn_journal_path(&self.path))?)
221 }
222
223 pub fn turn_lifecycle(&self) -> Result<Vec<TurnLifecycleRecord>, String> {
224 load_turn_records(&turn_journal_path(&self.path))
225 }
226
227 fn require_pending_turn(&self, turn_id: &str) -> Result<TurnState, String> {
228 self.latest_turn()?
229 .filter(|turn| turn.turn_id == turn_id && turn.is_pending())
230 .ok_or_else(|| "turn is not pending".to_owned())
231 }
232
233 fn append_turn_lifecycle(&mut self, lifecycle: TurnLifecycleRecord) -> Result<(), String> {
234 let path = turn_journal_path(&self.path);
235 let mut records = load_turn_records(&path)?;
236 records.push(lifecycle.clone());
237 reduce_turn_records(&records)?;
238
239 let stored = StoredTurnRecord::new(lifecycle);
240 let mut encoded = serde_json::to_string(&stored)
241 .map_err(|error| format!("unable to encode turn lifecycle record: {error}"))?;
242 let secret = std::env::var(&self.llm.api_key_env).ok();
243 if secret
244 .as_deref()
245 .is_some_and(|secret| !secret.is_empty() && encoded.contains(secret))
246 {
247 return Err("turn lifecycle record rejected".to_owned());
248 }
249 encoded.push('\n');
250
251 let mut file = open_turn_journal_for_append(&path)?;
252 file.write_all(encoded.as_bytes())
253 .map_err(|_| "unable to write turn lifecycle journal".to_owned())?;
254 file.sync_data()
255 .map_err(|_| "unable to checkpoint turn lifecycle journal".to_owned())?;
256 Ok(())
257 }
258}
259
260fn turn_journal_path(session_path: &Path) -> PathBuf {
261 let sessions_directory = session_path
262 .parent()
263 .expect("session transcript has a parent directory");
264 let lucy_directory = sessions_directory
265 .parent()
266 .expect("sessions directory has a Lucy parent");
267 let file_name = session_path
268 .file_name()
269 .expect("session transcript has a file name");
270 lucy_directory.join("turns").join(file_name)
271}
272
273fn secure_turn_journal_directory(path: &Path, create: bool) -> Result<bool, String> {
274 let directory = path
275 .parent()
276 .ok_or_else(|| "turn lifecycle journal has no parent".to_owned())?;
277 ensure_not_symlink(directory)
278 .map_err(|_| "turn lifecycle journal directory is unsafe".to_owned())?;
279 if !directory.exists() && !create {
280 return Ok(false);
281 }
282 ensure_private_dir(directory)
283 .map_err(|_| "unable to secure turn lifecycle journal directory".to_owned())?;
284 Ok(true)
285}
286
287fn open_turn_journal_for_append(path: &Path) -> Result<File, String> {
288 secure_turn_journal_directory(path, true)?;
289 #[cfg(not(unix))]
290 reject_symlink(path)?;
291
292 let mut options = OpenOptions::new();
293 options.write(true).append(true).create(true);
294 #[cfg(unix)]
295 {
296 options.mode(0o600);
297 options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
298 }
299 let file = options
300 .open(path)
301 .map_err(|_| "unable to open turn lifecycle journal".to_owned())?;
302 validate_private_regular_file(&file)?;
303 Ok(file)
304}
305
306fn open_turn_journal_for_read(path: &Path) -> Result<Option<File>, String> {
307 if !secure_turn_journal_directory(path, false)? {
308 return Ok(None);
309 }
310 #[cfg(not(unix))]
311 reject_symlink(path)?;
312
313 let mut options = OpenOptions::new();
314 options.read(true);
315 #[cfg(unix)]
316 options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
317 match options.open(path) {
318 Ok(file) => {
319 validate_private_regular_file(&file)?;
320 Ok(Some(file))
321 }
322 Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
323 Err(_) => Err("unable to read turn lifecycle journal".to_owned()),
324 }
325}
326
327#[cfg(not(unix))]
328fn reject_symlink(path: &Path) -> Result<(), String> {
329 if let Ok(metadata) = fs::symlink_metadata(path) {
330 if metadata.file_type().is_symlink() {
331 return Err("turn lifecycle journal is unsafe".to_owned());
332 }
333 }
334 Ok(())
335}
336
337fn validate_private_regular_file(file: &File) -> Result<(), String> {
338 let metadata = file
339 .metadata()
340 .map_err(|_| "unable to inspect turn lifecycle journal".to_owned())?;
341 if !metadata.is_file() {
342 return Err("turn lifecycle journal is unsafe".to_owned());
343 }
344 #[cfg(unix)]
345 if metadata.permissions().mode() & 0o777 != 0o600 {
346 return Err("turn lifecycle journal is not private".to_owned());
347 }
348 Ok(())
349}
350
351fn load_turn_records(path: &Path) -> Result<Vec<TurnLifecycleRecord>, String> {
352 let Some(file) = open_turn_journal_for_read(path)? else {
353 return Ok(Vec::new());
354 };
355 let mut records = Vec::new();
356 for (line_number, line) in BufReader::new(file).lines().enumerate() {
357 let line = line.map_err(|_| "unable to read turn lifecycle journal".to_owned())?;
358 if line.trim().is_empty() {
359 continue;
360 }
361 let stored: StoredTurnRecord = serde_json::from_str(&line)
362 .map_err(|_| format!("invalid turn lifecycle record at line {}", line_number + 1))?;
363 if stored.record != "turn" || stored.version != TURN_JOURNAL_VERSION {
364 return Err(format!(
365 "unsupported turn lifecycle record at line {}",
366 line_number + 1
367 ));
368 }
369 records.push(stored.lifecycle);
370 }
371 Ok(records)
372}
373
374fn reduce_turn_records(records: &[TurnLifecycleRecord]) -> Result<Option<TurnState>, String> {
375 let mut latest: Option<TurnState> = None;
376 for lifecycle in records {
377 match lifecycle.event {
378 TurnEvent::Started => {
379 if latest.as_ref().is_some_and(TurnState::is_pending) {
380 return Err("invalid turn lifecycle sequence".to_owned());
381 }
382 let (Some(kind), Some(phase), Some(first_message)) =
383 (lifecycle.kind, lifecycle.phase, lifecycle.first_message)
384 else {
385 return Err("invalid turn start record".to_owned());
386 };
387 if lifecycle.outcome.is_some()
388 || lifecycle.last_message.is_some()
389 || lifecycle.error.is_some()
390 {
391 return Err("invalid turn start record".to_owned());
392 }
393 latest = Some(TurnState {
394 turn_id: lifecycle.turn_id.clone(),
395 kind,
396 phase,
397 status: TurnStatus::Active,
398 first_message,
399 last_message: None,
400 error: None,
401 });
402 }
403 TurnEvent::PhaseChanged => {
404 let Some(state) = latest
405 .as_mut()
406 .filter(|state| state.turn_id == lifecycle.turn_id && state.is_pending())
407 else {
408 return Err("invalid turn phase record".to_owned());
409 };
410 let Some(phase) = lifecycle.phase else {
411 return Err("invalid turn phase record".to_owned());
412 };
413 if lifecycle.kind.is_some()
414 || lifecycle.outcome.is_some()
415 || lifecycle.first_message.is_some()
416 || lifecycle.last_message.is_some()
417 || lifecycle.error.is_some()
418 {
419 return Err("invalid turn phase record".to_owned());
420 }
421 state.phase = phase;
422 state.status = TurnStatus::Active;
423 state.error = None;
424 }
425 TurnEvent::Finished => {
426 let Some(state) = latest
427 .as_mut()
428 .filter(|state| state.turn_id == lifecycle.turn_id && state.is_pending())
429 else {
430 return Err("invalid turn finish record".to_owned());
431 };
432 let Some(outcome) = lifecycle.outcome else {
433 return Err("invalid turn finish record".to_owned());
434 };
435 if lifecycle.kind.is_some()
436 || lifecycle.phase.is_some()
437 || lifecycle.first_message.is_some()
438 {
439 return Err("invalid turn finish record".to_owned());
440 }
441 if state.status == TurnStatus::Retryable && outcome != TurnOutcome::Abandoned {
442 return Err("invalid retryable turn transition".to_owned());
443 }
444 state.status = status_for_outcome(outcome);
445 state.last_message = lifecycle.last_message;
446 state.error = lifecycle.error.clone();
447 }
448 }
449 }
450 Ok(latest)
451}
452
453fn status_for_outcome(outcome: TurnOutcome) -> TurnStatus {
454 match outcome {
455 TurnOutcome::Completed => TurnStatus::Completed,
456 TurnOutcome::RetryableFailure => TurnStatus::Retryable,
457 TurnOutcome::TerminalFailure => TurnStatus::TerminalFailure,
458 TurnOutcome::Interrupted => TurnStatus::Interrupted,
459 TurnOutcome::Abandoned => TurnStatus::Abandoned,
460 }
461}
462
463fn new_turn_id(session_id: &str) -> String {
464 let counter = TURN_COUNTER.fetch_add(1, Ordering::Relaxed);
465 format!("{session_id}-turn-{}-{counter}", now())
466}
467
468fn now() -> u64 {
469 std::time::SystemTime::now()
470 .duration_since(std::time::UNIX_EPOCH)
471 .map(|duration| duration.as_millis().min(u64::MAX as u128) as u64)
472 .unwrap_or(0)
473}
474
475#[cfg(test)]
476mod tests {
477 use std::path::PathBuf;
478 use std::sync::atomic::{AtomicU64, Ordering};
479
480 use crate::config::LlmSettings;
481
482 use super::*;
483
484 static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
485
486 fn temporary_home() -> PathBuf {
487 loop {
488 let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
489 let path = std::env::temp_dir().join(format!(
490 "lucy-turn-{}-{}-{counter}",
491 now(),
492 std::process::id()
493 ));
494 match fs::create_dir(&path) {
495 Ok(()) => return path,
496 Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
497 Err(error) => panic!("temp home: {error}"),
498 }
499 }
500 }
501
502 fn create_session(home: &Path) -> Session {
503 Session::create_with_secret(
504 home,
505 &std::env::current_dir().expect("cwd"),
506 "prompt".to_owned(),
507 LlmSettings {
508 base_url: "http://localhost".to_owned(),
509 model: "model".to_owned(),
510 api_key_env: "LUCY_TURN_TEST_KEY".to_owned(),
511 effort: None,
512 },
513 None,
514 )
515 .expect("session")
516 }
517
518 #[test]
519 fn lifecycle_round_trips_and_reconstructs_retryable_state() {
520 let home = temporary_home();
521 let mut session = create_session(&home);
522 let turn_id = session.start_turn(TurnKind::User).expect("start");
523 session
524 .set_turn_phase(&turn_id, TurnPhase::Compacting)
525 .expect("phase");
526 session
527 .fail_turn_retryably(&turn_id, "provider unavailable".to_owned())
528 .expect("failure");
529
530 let expected = session.latest_turn().expect("read").expect("turn");
531 assert_eq!(expected.status, TurnStatus::Retryable);
532 assert_eq!(expected.error.as_deref(), Some("provider unavailable"));
533 assert_eq!(
534 fs::read_dir(home.join(".lucy/sessions"))
535 .expect("sessions")
536 .count(),
537 1
538 );
539 assert_eq!(
540 fs::read_dir(home.join(".lucy/turns"))
541 .expect("turn journals")
542 .count(),
543 1
544 );
545
546 let session_id = session.id.clone();
547
548 drop(session);
549
550 let resumed = Session::resume(&home, &session_id).expect("resume");
551 assert_eq!(resumed.latest_turn().expect("read"), Some(expected));
552 fs::remove_dir_all(home).expect("cleanup");
553 }
554
555 #[test]
556 fn pending_turn_blocks_a_second_turn_until_resolved() {
557 let home = temporary_home();
558 let mut session = create_session(&home);
559 let turn_id = session.start_turn(TurnKind::User).expect("start");
560 assert!(session.start_turn(TurnKind::User).is_err());
561 session
562 .finish_turn(&turn_id, TurnOutcome::Completed, None)
563 .expect("finish");
564 assert!(session.start_turn(TurnKind::User).is_ok());
565 fs::remove_dir_all(home).expect("cleanup");
566 }
567
568 #[test]
569 fn legacy_session_without_lifecycle_journal_remains_valid() {
570 let home = temporary_home();
571 let session = create_session(&home);
572 let session_id = session.id.clone();
573 drop(session);
574 let resumed = Session::resume(&home, &session_id).expect("resume");
575 assert_eq!(resumed.latest_turn().expect("read"), None);
576 fs::remove_dir_all(home).expect("cleanup");
577 }
578}