roder_core/
conversation_forks.rs

1//! Conversation forks (roadmap phases 90 + 81).
2//!
3//! Forks an existing thread into a child thread backed by a workspace fork
4//! from any registered `ForkProvider` (default: `git-worktree`): the child
5//! starts from the parent transcript (no side-effectful tool replay — only
6//! conversation history records are copied) and all subsequent tool
7//! execution resolves against the fork workspace because the child's
8//! `ThreadMetadata.workspace` points at it. Cleanup is explicit and
9//! path-confirmed; the parent workspace is never modified.
10
11use std::path::PathBuf;
12use std::sync::Arc;
13
14use roder_api::events::{
15    EventEnvelope, RoderEvent, ThreadCreated, ThreadForkFailed, ThreadForkRemoved,
16    ThreadForkRequested, ThreadForked, ThreadId, TurnId,
17};
18use roder_api::forks::{
19    ForkPolicy, ForkReason, ForkRequest, ForkStatus, RemoveForkPolicy, WorkspaceFork,
20};
21use roder_api::thread::{ThreadMetadata, ThreadStore};
22use time::OffsetDateTime;
23
24use crate::Runtime;
25use crate::forks::DEFAULT_FORK_PROVIDER;
26
27#[derive(Debug, Clone)]
28pub struct ForkThreadRequest {
29    pub parent_thread_id: ThreadId,
30    /// User-facing fork name; the provider sanitizes it into its naming
31    /// scheme (directories, branches, snapshot names).
32    pub name: String,
33    /// Fork at a specific parent turn; `None` forks at the latest turn.
34    pub from_turn_id: Option<TurnId>,
35    /// Fork provider id; `None` uses [`DEFAULT_FORK_PROVIDER`].
36    pub provider_id: Option<String>,
37    /// Provider-specific options (never secrets).
38    pub provider_config: serde_json::Value,
39}
40
41impl ForkThreadRequest {
42    pub fn new(parent_thread_id: ThreadId, name: impl Into<String>) -> Self {
43        Self {
44            parent_thread_id,
45            name: name.into(),
46            from_turn_id: None,
47            provider_id: None,
48            provider_config: serde_json::json!({}),
49        }
50    }
51}
52
53#[derive(Debug, Clone)]
54pub struct ForkThreadOutcome {
55    pub child: ThreadMetadata,
56    pub warnings: Vec<String>,
57}
58
59impl Runtime {
60    /// Forks `parent_thread_id` into a new child thread backed by a fresh
61    /// workspace fork of the parent workspace.
62    pub async fn fork_thread(
63        &self,
64        request: ForkThreadRequest,
65    ) -> anyhow::Result<ForkThreadOutcome> {
66        self.emit(RoderEvent::ThreadForkRequested(ThreadForkRequested {
67            parent_thread_id: request.parent_thread_id.clone(),
68            name: request.name.clone(),
69            timestamp: OffsetDateTime::now_utc(),
70        }))
71        .await;
72        match self.fork_thread_inner(&request).await {
73            Ok(outcome) => Ok(outcome),
74            Err(error) => {
75                self.emit(RoderEvent::ThreadForkFailed(ThreadForkFailed {
76                    parent_thread_id: request.parent_thread_id.clone(),
77                    name: request.name.clone(),
78                    message: error.to_string(),
79                    timestamp: OffsetDateTime::now_utc(),
80                }))
81                .await;
82                Err(error)
83            }
84        }
85    }
86
87    async fn fork_thread_inner(
88        &self,
89        request: &ForkThreadRequest,
90    ) -> anyhow::Result<ForkThreadOutcome> {
91        let store = self
92            .thread_store
93            .clone()
94            .ok_or_else(|| anyhow::anyhow!("conversation forks require a thread store"))?;
95        let parent = store
96            .load_thread(&request.parent_thread_id)
97            .await?
98            .ok_or_else(|| {
99                anyhow::anyhow!("parent thread {} was not found", request.parent_thread_id)
100            })?;
101        let parent_metadata = parent.metadata.clone().ok_or_else(|| {
102            anyhow::anyhow!(
103                "parent thread {} has no metadata to fork from",
104                request.parent_thread_id
105            )
106        })?;
107
108        // Materialize the workspace fork first; thread creation only
109        // proceeds once an isolated workspace exists.
110        let provider_id = request
111            .provider_id
112            .clone()
113            .unwrap_or_else(|| DEFAULT_FORK_PROVIDER.to_string());
114        let fork = self
115            .create_workspace_fork(
116                &provider_id,
117                ForkRequest {
118                    source_workspace: PathBuf::from(&parent_metadata.workspace),
119                    name: Some(request.name.clone()),
120                    reason: ForkReason::ConversationFork,
121                    policy: ForkPolicy::default(),
122                    provider_config: request.provider_config.clone(),
123                },
124            )
125            .await?;
126
127        let now = OffsetDateTime::now_utc();
128        let seed_events = seed_events_for_child(&parent.events, request.from_turn_id.as_deref())?;
129        let mut warnings = Vec::new();
130        if request.from_turn_id.is_none() && seed_events.is_empty() && !parent.events.is_empty() {
131            warnings.push(
132                "parent thread has events but none were conversation records; the fork starts \
133                 with an empty transcript"
134                    .to_string(),
135            );
136        }
137
138        let child_id = uuid::Uuid::new_v4().to_string();
139        let child_metadata = ThreadMetadata {
140            thread_id: child_id.clone(),
141            title: Some(match &parent_metadata.title {
142                Some(title) => format!("{title} (fork: {})", request.name),
143                None => format!("fork: {}", request.name),
144            }),
145            workspace: fork.workspace.display().to_string(),
146            // The fork workspace lives outside registered workspace roots.
147            workspace_id: None,
148            root_id: None,
149            provider: parent_metadata.provider.clone(),
150            model: parent_metadata.model.clone(),
151            selection_mode: parent_metadata.selection_mode.clone(),
152            tool_allowlist: parent_metadata.tool_allowlist.clone(),
153            developer_instructions: parent_metadata.developer_instructions.clone(),
154            external_tools: parent_metadata.external_tools.clone(),
155            // Local workspace forks never inherit runner bindings.
156            runner_destination: None,
157            runner_state: None,
158            runner_binding: None,
159            created_at: now,
160            updated_at: now,
161            message_count: 0,
162            usage: None,
163            parent_thread_id: Some(request.parent_thread_id.clone()),
164            forked_from_turn_id: request.from_turn_id.clone(),
165            workspace_fork: Some(fork.clone()),
166        };
167
168        if let Err(error) = self
169            .seed_child_thread(&store, child_metadata.clone(), &child_id, seed_events)
170            .await
171        {
172            // Best-effort cleanup so a failed fork does not leak a workspace.
173            let _ = self
174                .remove_workspace_fork(
175                    &provider_id,
176                    &fork.id,
177                    RemoveForkPolicy {
178                        confirm_workspace: fork.workspace.clone(),
179                    },
180                )
181                .await;
182            return Err(error);
183        }
184
185        self.emit(RoderEvent::ThreadCreated(ThreadCreated {
186            thread_id: child_id.clone(),
187            timestamp: OffsetDateTime::now_utc(),
188        }))
189        .await;
190        self.emit(RoderEvent::ThreadForked(ThreadForked {
191            parent_thread_id: request.parent_thread_id.clone(),
192            child_thread_id: child_id.clone(),
193            fork,
194            timestamp: OffsetDateTime::now_utc(),
195        }))
196        .await;
197
198        let child = store
199            .load_thread_metadata(&child_id)
200            .await?
201            .unwrap_or(child_metadata);
202        Ok(ForkThreadOutcome { child, warnings })
203    }
204
205    async fn seed_child_thread(
206        &self,
207        store: &Arc<dyn ThreadStore>,
208        child_metadata: ThreadMetadata,
209        child_id: &ThreadId,
210        seed_events: Vec<EventEnvelope>,
211    ) -> anyhow::Result<()> {
212        store.create_thread(child_metadata).await?;
213        for envelope in &seed_events {
214            store.append_event(child_id, envelope).await?;
215        }
216        Ok(())
217    }
218
219    /**
220     * Removes the workspace fork behind a forked thread. Destructive and
221     * explicit: `confirm_path` must match the fork workspace exactly. The
222     * thread itself is kept (status flips to `Removed`) so the conversation
223     * stays readable.
224     */
225    pub async fn remove_thread_workspace_fork(
226        &self,
227        thread_id: &ThreadId,
228        confirm_path: &str,
229    ) -> anyhow::Result<WorkspaceFork> {
230        let store = self
231            .thread_store
232            .clone()
233            .ok_or_else(|| anyhow::anyhow!("conversation forks require a thread store"))?;
234        let mut metadata = store
235            .load_thread_metadata(thread_id)
236            .await?
237            .ok_or_else(|| anyhow::anyhow!("thread {thread_id} was not found"))?;
238        let mut fork = metadata
239            .workspace_fork
240            .clone()
241            .ok_or_else(|| anyhow::anyhow!("thread {thread_id} is not a workspace fork"))?;
242        anyhow::ensure!(
243            fork.status == ForkStatus::Active,
244            "fork {} was already removed",
245            fork.id
246        );
247        anyhow::ensure!(
248            std::path::Path::new(confirm_path) == fork.workspace,
249            "confirmation path does not match the fork workspace {}; removal is \
250             path-confirmed to prevent accidental deletion",
251            fork.workspace.display()
252        );
253
254        self.remove_workspace_fork(
255            &fork.provider_id.clone(),
256            &fork.id.clone(),
257            RemoveForkPolicy {
258                confirm_workspace: fork.workspace.clone(),
259            },
260        )
261        .await?;
262
263        fork.status = ForkStatus::Removed;
264        metadata.workspace_fork = Some(fork.clone());
265        metadata.updated_at = OffsetDateTime::now_utc();
266        store.update_thread_metadata(metadata).await?;
267
268        self.emit(RoderEvent::ThreadForkRemoved(ThreadForkRemoved {
269            thread_id: thread_id.clone(),
270            fork_id: fork.id.clone(),
271            worktree_path: fork.workspace.display().to_string(),
272            timestamp: OffsetDateTime::now_utc(),
273        }))
274        .await;
275        Ok(fork)
276    }
277}
278
279/**
280 * Selects the parent events that seed the child transcript: only
281 * conversation-history records (turn lifecycle and transcript items), never
282 * tool/approval/audit events, so nothing side-effectful is replayed. When
283 * `from_turn_id` is set, events after that turn's records are dropped.
284 */
285fn seed_events_for_child(
286    events: &[EventEnvelope],
287    from_turn_id: Option<&str>,
288) -> anyhow::Result<Vec<EventEnvelope>> {
289    let mut ordered: Vec<&EventEnvelope> = events.iter().collect();
290    ordered.sort_by_key(|envelope| envelope.seq);
291
292    let cutoff = match from_turn_id {
293        Some(turn_id) => {
294            let last = ordered
295                .iter()
296                .rposition(|envelope| envelope.turn_id.as_deref() == Some(turn_id))
297                .ok_or_else(|| {
298                    anyhow::anyhow!("turn {turn_id} was not found in the parent thread")
299                })?;
300            last + 1
301        }
302        None => ordered.len(),
303    };
304
305    Ok(ordered[..cutoff]
306        .iter()
307        .filter(|envelope| {
308            matches!(
309                envelope.event,
310                RoderEvent::TurnStarted(_)
311                    | RoderEvent::TranscriptItemAppended(_)
312                    | RoderEvent::TurnCompleted(_)
313                    | RoderEvent::TurnFailed(_)
314                    | RoderEvent::TurnInterrupted(_)
315            )
316        })
317        .map(|envelope| (*envelope).clone())
318        .collect())
319}
320
321#[cfg(test)]
322mod tests {
323    use super::*;
324    use roder_api::events::{EventSource, TranscriptItemAppended, TurnCompleted, TurnStarted};
325    use roder_api::transcript::{TranscriptItem, UserMessage};
326
327    fn envelope(seq: u64, turn_id: &str, event: RoderEvent) -> EventEnvelope {
328        EventEnvelope {
329            event_id: format!("event-{seq}"),
330            seq,
331            timestamp: OffsetDateTime::UNIX_EPOCH,
332            source: EventSource::Core,
333            kind: event.kind().to_string(),
334            thread_id: Some("parent".to_string()),
335            turn_id: Some(turn_id.to_string()),
336            event,
337        }
338    }
339
340    fn turn_events(seq: u64, turn_id: &str, text: &str) -> Vec<EventEnvelope> {
341        vec![
342            envelope(
343                seq,
344                turn_id,
345                RoderEvent::TurnStarted(TurnStarted {
346                    thread_id: "parent".to_string(),
347                    turn_id: turn_id.to_string(),
348                    runtime_profile: Default::default(),
349                    timestamp: OffsetDateTime::UNIX_EPOCH,
350                }),
351            ),
352            envelope(
353                seq + 1,
354                turn_id,
355                RoderEvent::TranscriptItemAppended(TranscriptItemAppended {
356                    thread_id: "parent".to_string(),
357                    turn_id: turn_id.to_string(),
358                    item_type: "user_message".to_string(),
359                    item_index: None,
360                    item: Some(TranscriptItem::UserMessage(UserMessage::text(text))),
361                    timestamp: OffsetDateTime::UNIX_EPOCH,
362                }),
363            ),
364            envelope(
365                seq + 2,
366                turn_id,
367                RoderEvent::TurnCompleted(TurnCompleted {
368                    thread_id: "parent".to_string(),
369                    turn_id: turn_id.to_string(),
370                    usage: None,
371                    finish_reason: Some("stop".to_string()),
372                    timestamp: OffsetDateTime::UNIX_EPOCH,
373                }),
374            ),
375        ]
376    }
377
378    #[test]
379    fn seed_events_keep_conversation_records_only() {
380        let mut events = turn_events(1, "turn-1", "hello");
381        events.push(envelope(
382            4,
383            "turn-1",
384            RoderEvent::ToolCallStarted(roder_api::events::ToolCallStarted {
385                thread_id: "parent".to_string(),
386                turn_id: "turn-1".to_string(),
387                tool_id: "call-1".to_string(),
388                tool_name: Some("shell".to_string()),
389                display_payload: None,
390                timestamp: OffsetDateTime::UNIX_EPOCH,
391            }),
392        ));
393
394        let seeded = seed_events_for_child(&events, None).unwrap();
395
396        assert_eq!(seeded.len(), 3, "tool events must not be replayed");
397        assert!(
398            seeded
399                .iter()
400                .all(|envelope| !matches!(envelope.event, RoderEvent::ToolCallStarted(_)))
401        );
402    }
403
404    #[test]
405    fn seed_events_truncate_at_requested_turn() {
406        let mut events = turn_events(1, "turn-1", "first");
407        events.extend(turn_events(10, "turn-2", "second"));
408
409        let seeded = seed_events_for_child(&events, Some("turn-1")).unwrap();
410        assert_eq!(seeded.len(), 3);
411        assert!(
412            seeded
413                .iter()
414                .all(|envelope| envelope.turn_id.as_deref() == Some("turn-1"))
415        );
416
417        let error = seed_events_for_child(&events, Some("missing-turn")).unwrap_err();
418        assert!(error.to_string().contains("missing-turn"));
419    }
420}
roder_core/conversation_forks.rs

roder_core/
conversation_forks.rs
