vtcode_core/core/
threads.rs

1use crate::exec::events::ThreadEvent;
2use crate::llm::provider::Message;
3use crate::utils::session_archive::{
4    SessionArchive, SessionArchiveMetadata, SessionForkMode, SessionListing,
5    find_session_by_identifier, list_recent_sessions, reserve_session_archive_identifier,
6    session_listing_matches_workspace,
7};
8use anyhow::{Result, anyhow};
9use parking_lot::Mutex;
10use serde::{Deserialize, Serialize};
11use std::collections::VecDeque;
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14use uuid::Uuid;
15use vtcode_macros::StringNewtype;
16
17const DEFAULT_EVENT_BUFFER_CAPACITY: usize = 512;
18
19/// Unique identifier for a thread in the runtime.
20#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, StringNewtype)]
21pub struct ThreadId(String);
22
23/// Unique identifier for a submission within a thread turn.
24#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, StringNewtype)]
25pub struct SubmissionId(String);
26
27impl SubmissionId {
28    /// Generate a new submission identifier with a `sub-` prefix.
29    pub fn generate() -> Self {
30        Self(format!("sub-{}", Uuid::new_v4()))
31    }
32}
33
34impl Default for SubmissionId {
35    fn default() -> Self {
36        Self::generate()
37    }
38}
39
40#[derive(Debug, Clone)]
41pub struct ThreadEventRecord {
42    pub sequence: u64,
43    pub thread_id: ThreadId,
44    pub submission_id: Option<SubmissionId>,
45    pub turn_id: Option<String>,
46    pub event: ThreadEvent,
47}
48
49#[derive(Debug, Clone)]
50pub struct ThreadSnapshot {
51    pub thread_id: ThreadId,
52    pub metadata: Option<SessionArchiveMetadata>,
53    pub archive_listing: Option<SessionListing>,
54    pub messages: Vec<Message>,
55    pub loaded_skills: Vec<String>,
56    pub turn_in_flight: bool,
57}
58
59#[derive(Debug, Clone)]
60pub struct ThreadBootstrap {
61    pub metadata: Option<SessionArchiveMetadata>,
62    pub archive_listing: Option<SessionListing>,
63    pub messages: Vec<Message>,
64    pub loaded_skills: Vec<String>,
65}
66
67impl ThreadBootstrap {
68    pub fn new(metadata: Option<SessionArchiveMetadata>) -> Self {
69        Self {
70            metadata,
71            archive_listing: None,
72            messages: Vec::new(),
73            loaded_skills: Vec::new(),
74        }
75    }
76
77    pub fn from_listing(listing: SessionListing) -> Self {
78        Self {
79            metadata: Some(listing.snapshot.metadata.clone()),
80            messages: messages_from_session_listing(&listing),
81            loaded_skills: loaded_skills_from_session_listing(&listing),
82            archive_listing: Some(listing),
83        }
84    }
85
86    pub fn from_snapshot(snapshot: ThreadSnapshot) -> Self {
87        Self {
88            metadata: snapshot.metadata,
89            archive_listing: snapshot.archive_listing,
90            messages: snapshot.messages,
91            loaded_skills: snapshot.loaded_skills,
92        }
93    }
94
95    pub fn with_messages(mut self, messages: Vec<Message>) -> Self {
96        self.messages = messages;
97        self
98    }
99
100    pub fn with_loaded_skills(mut self, loaded_skills: Vec<String>) -> Self {
101        self.loaded_skills = loaded_skills;
102        self
103    }
104}
105
106#[derive(Debug, Clone, PartialEq, Eq)]
107pub enum SessionQueryScope {
108    CurrentWorkspace(PathBuf),
109    All,
110}
111
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub enum ArchivedSessionIntent {
114    ResumeInPlace,
115    ForkNewArchive {
116        custom_suffix: Option<String>,
117        summarize: bool,
118    },
119}
120
121#[derive(Debug, Clone)]
122pub struct PreparedArchivedSession {
123    pub source: SessionListing,
124    pub workspace: PathBuf,
125    pub bootstrap: ThreadBootstrap,
126    pub thread_id: String,
127    pub archive: SessionArchive,
128}
129
130#[derive(Default)]
131struct ThreadEventStore {
132    capacity: usize,
133    next_sequence: u64,
134    events: VecDeque<ThreadEventRecord>,
135}
136
137impl ThreadEventStore {
138    fn with_capacity(capacity: usize) -> Self {
139        Self {
140            capacity: capacity.max(1),
141            ..Self::default()
142        }
143    }
144
145    fn push(
146        &mut self,
147        thread_id: &ThreadId,
148        submission_id: Option<SubmissionId>,
149        turn_id: Option<String>,
150        event: ThreadEvent,
151    ) {
152        let record = ThreadEventRecord {
153            sequence: self.next_sequence,
154            thread_id: thread_id.clone(),
155            submission_id,
156            turn_id,
157            event,
158        };
159        self.next_sequence = self.next_sequence.saturating_add(1);
160
161        if self.events.len() >= self.capacity {
162            self.events.pop_front();
163        }
164        self.events.push_back(record);
165    }
166
167    fn snapshot(&self) -> Vec<ThreadEventRecord> {
168        self.events.iter().cloned().collect()
169    }
170}
171
172struct ThreadSessionState {
173    thread_id: ThreadId,
174    metadata: Option<SessionArchiveMetadata>,
175    archive_listing: Option<SessionListing>,
176    messages: Vec<Message>,
177    loaded_skills: Vec<String>,
178    turn_in_flight: bool,
179}
180
181impl ThreadSessionState {
182    fn snapshot(&self) -> ThreadSnapshot {
183        ThreadSnapshot {
184            thread_id: self.thread_id.clone(),
185            metadata: self.metadata.clone(),
186            archive_listing: self.archive_listing.clone(),
187            messages: self.messages.clone(),
188            loaded_skills: self.loaded_skills.clone(),
189            turn_in_flight: self.turn_in_flight,
190        }
191    }
192}
193
194#[derive(Clone)]
195pub struct ThreadRuntimeHandle {
196    inner: Arc<ThreadRuntimeInner>,
197}
198
199struct ThreadRuntimeInner {
200    session: Mutex<ThreadSessionState>,
201    event_store: Mutex<ThreadEventStore>,
202}
203
204impl ThreadRuntimeHandle {
205    fn new(thread_id: ThreadId, bootstrap: ThreadBootstrap, event_capacity: usize) -> Self {
206        let session = ThreadSessionState {
207            thread_id,
208            metadata: bootstrap.metadata,
209            archive_listing: bootstrap.archive_listing,
210            messages: bootstrap.messages,
211            loaded_skills: bootstrap.loaded_skills,
212            turn_in_flight: false,
213        };
214
215        Self {
216            inner: Arc::new(ThreadRuntimeInner {
217                session: Mutex::new(session),
218                event_store: Mutex::new(ThreadEventStore::with_capacity(event_capacity)),
219            }),
220        }
221    }
222
223    pub fn thread_id(&self) -> ThreadId {
224        self.inner.session.lock().thread_id.clone()
225    }
226
227    pub fn snapshot(&self) -> ThreadSnapshot {
228        self.inner.session.lock().snapshot()
229    }
230
231    pub fn metadata(&self) -> Option<SessionArchiveMetadata> {
232        self.inner.session.lock().metadata.clone()
233    }
234
235    pub fn replace_metadata(&self, metadata: Option<SessionArchiveMetadata>) {
236        self.inner.session.lock().metadata = metadata;
237    }
238
239    pub fn archive_listing(&self) -> Option<SessionListing> {
240        self.inner.session.lock().archive_listing.clone()
241    }
242
243    pub fn messages(&self) -> Vec<Message> {
244        self.inner.session.lock().messages.clone()
245    }
246
247    pub fn replace_messages(&self, messages: Vec<Message>) {
248        self.inner.session.lock().messages = messages;
249    }
250
251    pub fn append_message(&self, message: Message) {
252        self.inner.session.lock().messages.push(message);
253    }
254
255    pub fn begin_turn(&self) -> Result<SubmissionId> {
256        let mut session = self.inner.session.lock();
257        if session.turn_in_flight {
258            return Err(anyhow!(
259                "thread '{}' already has an in-flight turn",
260                session.thread_id
261            ));
262        }
263
264        session.turn_in_flight = true;
265        Ok(SubmissionId::generate())
266    }
267
268    pub fn finish_turn(&self) {
269        self.inner.session.lock().turn_in_flight = false;
270    }
271
272    pub fn record_event(
273        &self,
274        submission_id: Option<SubmissionId>,
275        turn_id: Option<String>,
276        event: ThreadEvent,
277    ) {
278        let thread_id = self.thread_id();
279        self.inner
280            .event_store
281            .lock()
282            .push(&thread_id, submission_id, turn_id, event);
283    }
284
285    pub fn replay_recent(&self) -> Vec<ThreadEventRecord> {
286        self.inner.event_store.lock().snapshot()
287    }
288
289    pub fn recent_events(&self) -> Vec<ThreadEvent> {
290        self.replay_recent()
291            .into_iter()
292            .map(|record| record.event)
293            .collect()
294    }
295}
296
297#[derive(Clone)]
298pub struct ThreadManager {
299    event_buffer_capacity: usize,
300}
301
302impl Default for ThreadManager {
303    fn default() -> Self {
304        Self::new()
305    }
306}
307
308impl ThreadManager {
309    pub fn new() -> Self {
310        Self {
311            event_buffer_capacity: DEFAULT_EVENT_BUFFER_CAPACITY,
312        }
313    }
314
315    pub fn with_event_buffer_capacity(event_buffer_capacity: usize) -> Self {
316        Self {
317            event_buffer_capacity: event_buffer_capacity.max(1),
318        }
319    }
320
321    pub fn start_thread_with_identifier(
322        &self,
323        identifier: impl Into<String>,
324        bootstrap: ThreadBootstrap,
325    ) -> ThreadRuntimeHandle {
326        ThreadRuntimeHandle::new(
327            ThreadId::new(identifier.into()),
328            bootstrap,
329            self.event_buffer_capacity,
330        )
331    }
332
333    pub async fn start_thread(
334        &self,
335        workspace_label: &str,
336        custom_suffix: Option<String>,
337        bootstrap: ThreadBootstrap,
338    ) -> Result<ThreadRuntimeHandle> {
339        let identifier = reserve_session_archive_identifier(workspace_label, custom_suffix).await?;
340        Ok(self.start_thread_with_identifier(identifier, bootstrap))
341    }
342
343    pub async fn resume_thread(&self, identifier: &str) -> Result<Option<ThreadRuntimeHandle>> {
344        let listing = find_session_by_identifier(identifier).await?;
345        Ok(listing.map(|listing| {
346            self.start_thread_with_identifier(
347                listing.identifier(),
348                ThreadBootstrap::from_listing(listing),
349            )
350        }))
351    }
352}
353
354pub async fn list_recent_sessions_in_scope(
355    limit: usize,
356    scope: &SessionQueryScope,
357) -> Result<Vec<SessionListing>> {
358    let mut listings = list_recent_sessions(limit.saturating_mul(4).max(limit)).await?;
359    if let SessionQueryScope::CurrentWorkspace(workspace) = scope {
360        listings.retain(|listing| session_listing_matches_workspace(listing, workspace));
361    }
362    listings.truncate(limit);
363    Ok(listings)
364}
365
366pub async fn prepare_archived_session(
367    source: SessionListing,
368    workspace: PathBuf,
369    metadata: SessionArchiveMetadata,
370    intent: ArchivedSessionIntent,
371    reserved_identifier: Option<String>,
372) -> Result<PreparedArchivedSession> {
373    let mut metadata =
374        preserve_prompt_cache_lineage_if_compatible(metadata, &source.snapshot.metadata);
375    metadata.continuation_metadata = source.snapshot.metadata.continuation_metadata.clone();
376    let mut bootstrap = ThreadBootstrap::from_listing(source.clone());
377    bootstrap.metadata = Some(metadata.clone());
378
379    let thread_id = match &intent {
380        ArchivedSessionIntent::ResumeInPlace => source.identifier(),
381        ArchivedSessionIntent::ForkNewArchive { custom_suffix, .. } => {
382            if let Some(identifier) = reserved_identifier {
383                identifier
384            } else {
385                reserve_session_archive_identifier(&metadata.workspace_label, custom_suffix.clone())
386                    .await?
387            }
388        }
389    };
390
391    if let ArchivedSessionIntent::ForkNewArchive { summarize, .. } = &intent {
392        metadata.parent_session_id = Some(source.identifier());
393        metadata.fork_mode = Some(if *summarize {
394            SessionForkMode::Summarized
395        } else {
396            SessionForkMode::FullCopy
397        });
398        bootstrap.metadata = Some(metadata.clone());
399    }
400
401    let archive = match intent {
402        ArchivedSessionIntent::ResumeInPlace => {
403            SessionArchive::resume_from_listing(&source, metadata)
404        }
405        ArchivedSessionIntent::ForkNewArchive { .. } => {
406            SessionArchive::new_with_identifier(metadata, thread_id.clone()).await?
407        }
408    };
409
410    Ok(PreparedArchivedSession {
411        source,
412        workspace,
413        bootstrap,
414        thread_id,
415        archive,
416    })
417}
418
419fn preserve_prompt_cache_lineage_if_compatible(
420    mut metadata: SessionArchiveMetadata,
421    source: &SessionArchiveMetadata,
422) -> SessionArchiveMetadata {
423    let is_compatible = metadata.workspace_path == source.workspace_path
424        && metadata.provider == source.provider
425        && metadata.model == source.model;
426    if is_compatible && let Some(lineage_id) = source.prompt_cache_lineage_id.as_ref() {
427        metadata.prompt_cache_lineage_id = Some(lineage_id.clone());
428    }
429    metadata
430}
431
432pub fn messages_from_session_listing(listing: &SessionListing) -> Vec<Message> {
433    if let Some(progress) = &listing.snapshot.progress
434        && !progress.recent_messages.is_empty()
435    {
436        progress.recent_messages.iter().map(Message::from).collect()
437    } else if !listing.snapshot.messages.is_empty() {
438        listing
439            .snapshot
440            .messages
441            .iter()
442            .map(Message::from)
443            .collect()
444    } else {
445        Vec::new()
446    }
447}
448
449pub fn loaded_skills_from_session_listing(listing: &SessionListing) -> Vec<String> {
450    listing
451        .snapshot
452        .progress
453        .as_ref()
454        .map(|progress| progress.loaded_skills.clone())
455        .filter(|skills| !skills.is_empty())
456        .unwrap_or_else(|| listing.snapshot.metadata.loaded_skills.clone())
457}
458
459pub fn build_thread_archive_metadata(
460    workspace: &Path,
461    model: &str,
462    provider: &str,
463    theme: &str,
464    reasoning_effort: &str,
465) -> SessionArchiveMetadata {
466    let workspace_label = workspace
467        .file_name()
468        .and_then(|value| value.to_str())
469        .unwrap_or("workspace");
470
471    SessionArchiveMetadata::new(
472        workspace_label,
473        workspace.to_string_lossy().to_string(),
474        model,
475        provider,
476        theme,
477        reasoning_effort,
478    )
479    .ensure_prompt_cache_lineage_id()
480}
481
482#[cfg(test)]
483mod tests {
484    use super::*;
485    use crate::exec::events::{ThreadEvent, ThreadStartedEvent};
486    use crate::llm::provider::MessageRole;
487    use crate::utils::session_archive::{
488        SessionArchiveMetadata, SessionMessage, SessionProgress, SessionSnapshot,
489        clear_sessions_dir_override_for_tests, override_sessions_dir_for_tests,
490    };
491    use chrono::Utc;
492    use std::sync::{LazyLock, Mutex};
493    use tempfile::TempDir;
494
495    static SESSION_DIR_TEST_GUARD: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
496
497    #[test]
498    fn event_store_evicts_old_records() {
499        let manager = ThreadManager::with_event_buffer_capacity(2);
500        let handle = manager.start_thread_with_identifier("thread-1", ThreadBootstrap::new(None));
501
502        handle.record_event(
503            None,
504            None,
505            ThreadEvent::ThreadStarted(ThreadStartedEvent {
506                thread_id: "thread-1".to_string(),
507            }),
508        );
509        handle.record_event(
510            None,
511            Some("turn-1".to_string()),
512            ThreadEvent::ThreadStarted(ThreadStartedEvent {
513                thread_id: "thread-1-turn-1".to_string(),
514            }),
515        );
516        handle.record_event(
517            None,
518            Some("turn-2".to_string()),
519            ThreadEvent::ThreadStarted(ThreadStartedEvent {
520                thread_id: "thread-1-turn-2".to_string(),
521            }),
522        );
523
524        let records = handle.replay_recent();
525        assert_eq!(records.len(), 2);
526        assert_eq!(records[0].sequence, 1);
527        assert_eq!(records[1].sequence, 2);
528    }
529
530    #[test]
531    fn start_thread_with_identifier_preserves_message_history() {
532        let manager = ThreadManager::new();
533        let bootstrap = ThreadBootstrap::new(None)
534            .with_messages(vec![Message::user("hello".to_string())])
535            .with_loaded_skills(vec!["repo-skill".to_string()]);
536        let handle = manager.start_thread_with_identifier("thread-123", bootstrap);
537
538        assert_eq!(handle.thread_id().as_str(), "thread-123");
539        let snapshot = handle.snapshot();
540        assert_eq!(snapshot.messages.len(), 1);
541        assert_eq!(snapshot.loaded_skills, vec!["repo-skill".to_string()]);
542    }
543
544    #[test]
545    fn submit_enforces_single_in_flight_turn() {
546        let manager = ThreadManager::new();
547        let handle = manager.start_thread_with_identifier("thread-123", ThreadBootstrap::new(None));
548
549        let _first = handle.begin_turn().expect("first turn");
550        let err = handle.begin_turn().expect_err("second turn should fail");
551        assert!(err.to_string().contains("in-flight turn"));
552        handle.finish_turn();
553        handle.begin_turn().expect("turn after finish");
554    }
555
556    #[test]
557    #[serial_test::serial(session_dir_override)]
558    fn list_recent_sessions_in_scope_filters_by_workspace() {
559        let _guard = SESSION_DIR_TEST_GUARD
560            .lock()
561            .expect("session dir test guard");
562        let tmp = TempDir::new().expect("temp dir");
563        override_sessions_dir_for_tests(tmp.path());
564
565        let listing = SessionListing {
566            path: tmp.path().join("session-alpha.json"),
567            snapshot: SessionSnapshot {
568                metadata: SessionArchiveMetadata::new(
569                    "ws",
570                    tmp.path().join("workspace").display().to_string(),
571                    "model",
572                    "provider",
573                    "theme",
574                    "medium",
575                ),
576                started_at: Utc::now(),
577                ended_at: Utc::now(),
578                total_messages: 1,
579                distinct_tools: Vec::new(),
580                transcript: Vec::new(),
581                messages: vec![SessionMessage::new(MessageRole::User, "hello")],
582                progress: None,
583                error_logs: Vec::new(),
584            },
585        };
586        std::fs::write(
587            &listing.path,
588            serde_json::to_string(&listing.snapshot).expect("serialize snapshot"),
589        )
590        .expect("write listing");
591
592        let runtime = tokio::runtime::Runtime::new().expect("runtime");
593        let filtered = runtime
594            .block_on(list_recent_sessions_in_scope(
595                5,
596                &SessionQueryScope::CurrentWorkspace(tmp.path().join("workspace")),
597            ))
598            .expect("filter by workspace");
599        let all = runtime
600            .block_on(list_recent_sessions_in_scope(5, &SessionQueryScope::All))
601            .expect("list all");
602
603        clear_sessions_dir_override_for_tests();
604
605        assert_eq!(filtered.len(), 1);
606        assert_eq!(all.len(), 1);
607    }
608
609    #[test]
610    #[serial_test::serial(session_dir_override)]
611    fn prepare_archived_session_resume_reuses_source_identifier_and_archive() {
612        let _guard = SESSION_DIR_TEST_GUARD
613            .lock()
614            .expect("session dir test guard");
615        let tmp = TempDir::new().expect("temp dir");
616        override_sessions_dir_for_tests(tmp.path());
617
618        let listing = SessionListing {
619            path: tmp.path().join("session-source.json"),
620            snapshot: SessionSnapshot {
621                metadata: SessionArchiveMetadata::new(
622                    "ws",
623                    tmp.path().join("workspace").display().to_string(),
624                    "old-model",
625                    "old-provider",
626                    "old-theme",
627                    "medium",
628                ),
629                started_at: Utc::now(),
630                ended_at: Utc::now(),
631                total_messages: 2,
632                distinct_tools: vec!["tool_a".to_string()],
633                transcript: Vec::new(),
634                messages: vec![SessionMessage::new(MessageRole::User, "hello")],
635                progress: Some(Box::new(SessionProgress {
636                    turn_number: 1,
637                    recent_messages: vec![SessionMessage::new(MessageRole::Assistant, "recent")],
638                    tool_summaries: Vec::new(),
639                    token_usage: None,
640                    max_context_tokens: None,
641                    loaded_skills: vec!["skill_a".to_string()],
642                })),
643                error_logs: Vec::new(),
644            },
645        };
646
647        let runtime = tokio::runtime::Runtime::new().expect("runtime");
648        let prepared = runtime
649            .block_on(prepare_archived_session(
650                listing.clone(),
651                tmp.path().join("workspace"),
652                SessionArchiveMetadata::new(
653                    "ws",
654                    tmp.path().join("workspace").display().to_string(),
655                    "new-model",
656                    "new-provider",
657                    "new-theme",
658                    "high",
659                ),
660                ArchivedSessionIntent::ResumeInPlace,
661                Some("should-not-be-used".to_string()),
662            ))
663            .expect("prepare resume");
664
665        clear_sessions_dir_override_for_tests();
666
667        assert_eq!(prepared.thread_id, listing.identifier());
668        assert_eq!(prepared.archive.path(), listing.path.as_path());
669        assert_eq!(prepared.bootstrap.messages[0].content.as_text(), "recent");
670        assert_eq!(
671            prepared.bootstrap.loaded_skills,
672            vec!["skill_a".to_string()]
673        );
674        assert_eq!(
675            prepared
676                .bootstrap
677                .metadata
678                .as_ref()
679                .expect("metadata")
680                .model,
681            "new-model"
682        );
683    }
684
685    #[test]
686    #[serial_test::serial(session_dir_override)]
687    fn prepare_archived_session_fork_uses_new_identifier_and_preserves_history() {
688        let _guard = SESSION_DIR_TEST_GUARD
689            .lock()
690            .expect("session dir test guard");
691        let tmp = TempDir::new().expect("temp dir");
692        override_sessions_dir_for_tests(tmp.path());
693
694        let listing = SessionListing {
695            path: tmp.path().join("session-source.json"),
696            snapshot: SessionSnapshot {
697                metadata: SessionArchiveMetadata::new(
698                    "ws",
699                    tmp.path().join("workspace").display().to_string(),
700                    "model",
701                    "provider",
702                    "theme",
703                    "medium",
704                ),
705                started_at: Utc::now(),
706                ended_at: Utc::now(),
707                total_messages: 1,
708                distinct_tools: Vec::new(),
709                transcript: Vec::new(),
710                messages: vec![SessionMessage::new(MessageRole::User, "hello")],
711                progress: None,
712                error_logs: Vec::new(),
713            },
714        };
715
716        let runtime = tokio::runtime::Runtime::new().expect("runtime");
717        let prepared = runtime
718            .block_on(prepare_archived_session(
719                listing.clone(),
720                tmp.path().join("workspace"),
721                SessionArchiveMetadata::new(
722                    "ws",
723                    tmp.path().join("workspace").display().to_string(),
724                    "model",
725                    "provider",
726                    "theme",
727                    "medium",
728                ),
729                ArchivedSessionIntent::ForkNewArchive {
730                    custom_suffix: Some("branch".to_string()),
731                    summarize: false,
732                },
733                Some("session-forked".to_string()),
734            ))
735            .expect("prepare fork");
736
737        clear_sessions_dir_override_for_tests();
738
739        assert_eq!(prepared.thread_id, "session-forked");
740        assert_ne!(prepared.archive.path(), listing.path.as_path());
741        assert!(
742            prepared
743                .archive
744                .path()
745                .ends_with(Path::new("session-forked.json"))
746        );
747        assert_eq!(prepared.bootstrap.messages[0].content.as_text(), "hello");
748        assert_eq!(
749            prepared
750                .bootstrap
751                .metadata
752                .as_ref()
753                .and_then(|metadata| metadata.parent_session_id.as_deref()),
754            Some("session-source")
755        );
756        assert_eq!(
757            prepared
758                .bootstrap
759                .metadata
760                .as_ref()
761                .and_then(|metadata| metadata.fork_mode),
762            Some(SessionForkMode::FullCopy)
763        );
764    }
765
766    #[test]
767    #[serial_test::serial(session_dir_override)]
768    fn prepare_archived_session_preserves_prompt_cache_lineage_when_compatible() {
769        let _guard = SESSION_DIR_TEST_GUARD
770            .lock()
771            .expect("session dir test guard");
772        let tmp = TempDir::new().expect("temp dir");
773        override_sessions_dir_for_tests(tmp.path());
774
775        let listing = SessionListing {
776            path: tmp.path().join("session-source.json"),
777            snapshot: SessionSnapshot {
778                metadata: SessionArchiveMetadata::new(
779                    "ws",
780                    tmp.path().join("workspace").display().to_string(),
781                    "model",
782                    "provider",
783                    "theme",
784                    "medium",
785                )
786                .with_prompt_cache_lineage_id("lineage-source"),
787                started_at: Utc::now(),
788                ended_at: Utc::now(),
789                total_messages: 1,
790                distinct_tools: Vec::new(),
791                transcript: Vec::new(),
792                messages: vec![SessionMessage::new(MessageRole::User, "hello")],
793                progress: None,
794                error_logs: Vec::new(),
795            },
796        };
797
798        let runtime = tokio::runtime::Runtime::new().expect("runtime");
799        let prepared = runtime
800            .block_on(prepare_archived_session(
801                listing,
802                tmp.path().join("workspace"),
803                SessionArchiveMetadata::new(
804                    "ws",
805                    tmp.path().join("workspace").display().to_string(),
806                    "model",
807                    "provider",
808                    "theme",
809                    "medium",
810                )
811                .with_prompt_cache_lineage_id("lineage-new"),
812                ArchivedSessionIntent::ResumeInPlace,
813                None,
814            ))
815            .expect("prepare resume");
816
817        clear_sessions_dir_override_for_tests();
818
819        assert_eq!(
820            prepared
821                .bootstrap
822                .metadata
823                .as_ref()
824                .and_then(|metadata| metadata.prompt_cache_lineage_id.as_deref()),
825            Some("lineage-source")
826        );
827    }
828
829    #[test]
830    #[serial_test::serial(session_dir_override)]
831    fn prepare_archived_session_resets_prompt_cache_lineage_on_model_change() {
832        let _guard = SESSION_DIR_TEST_GUARD
833            .lock()
834            .expect("session dir test guard");
835        let tmp = TempDir::new().expect("temp dir");
836        override_sessions_dir_for_tests(tmp.path());
837
838        let listing = SessionListing {
839            path: tmp.path().join("session-source.json"),
840            snapshot: SessionSnapshot {
841                metadata: SessionArchiveMetadata::new(
842                    "ws",
843                    tmp.path().join("workspace").display().to_string(),
844                    "model-a",
845                    "provider",
846                    "theme",
847                    "medium",
848                )
849                .with_prompt_cache_lineage_id("lineage-source"),
850                started_at: Utc::now(),
851                ended_at: Utc::now(),
852                total_messages: 1,
853                distinct_tools: Vec::new(),
854                transcript: Vec::new(),
855                messages: vec![SessionMessage::new(MessageRole::User, "hello")],
856                progress: None,
857                error_logs: Vec::new(),
858            },
859        };
860
861        let runtime = tokio::runtime::Runtime::new().expect("runtime");
862        let prepared = runtime
863            .block_on(prepare_archived_session(
864                listing,
865                tmp.path().join("workspace"),
866                SessionArchiveMetadata::new(
867                    "ws",
868                    tmp.path().join("workspace").display().to_string(),
869                    "model-b",
870                    "provider",
871                    "theme",
872                    "medium",
873                )
874                .with_prompt_cache_lineage_id("lineage-new"),
875                ArchivedSessionIntent::ResumeInPlace,
876                None,
877            ))
878            .expect("prepare resume");
879
880        clear_sessions_dir_override_for_tests();
881
882        assert_eq!(
883            prepared
884                .bootstrap
885                .metadata
886                .as_ref()
887                .and_then(|metadata| metadata.prompt_cache_lineage_id.as_deref()),
888            Some("lineage-new")
889        );
890    }
891
892    #[test]
893    fn messages_from_session_listing_preserves_assistant_phases_from_progress() {
894        let listing = SessionListing {
895            path: PathBuf::from("session.json"),
896            snapshot: SessionSnapshot {
897                metadata: SessionArchiveMetadata::new(
898                    "ws", "/tmp/ws", "gpt-5.4", "openai", "theme", "medium",
899                ),
900                started_at: Utc::now(),
901                ended_at: Utc::now(),
902                total_messages: 2,
903                distinct_tools: Vec::new(),
904                transcript: Vec::new(),
905                messages: Vec::new(),
906                progress: Some(Box::new(SessionProgress {
907                    turn_number: 2,
908                    recent_messages: vec![
909                        SessionMessage::from(
910                            &Message::assistant("Working".to_string())
911                                .with_phase(Some(crate::llm::provider::AssistantPhase::Commentary)),
912                        ),
913                        SessionMessage::from(
914                            &Message::assistant("Done".to_string()).with_phase(Some(
915                                crate::llm::provider::AssistantPhase::FinalAnswer,
916                            )),
917                        ),
918                    ],
919                    tool_summaries: Vec::new(),
920                    token_usage: None,
921                    max_context_tokens: None,
922                    loaded_skills: Vec::new(),
923                })),
924                error_logs: Vec::new(),
925            },
926        };
927
928        let messages = messages_from_session_listing(&listing);
929        assert_eq!(
930            messages
931                .iter()
932                .map(|message| message.phase)
933                .collect::<Vec<_>>(),
934            vec![
935                Some(crate::llm::provider::AssistantPhase::Commentary),
936                Some(crate::llm::provider::AssistantPhase::FinalAnswer),
937            ]
938        );
939    }
940}
vtcode_core/core/threads.rs

vtcode_core/core/
threads.rs
