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    /// Seeds a long-lived collaboration agent with a safe subset of the parent
61    /// conversation. Unlike `fork_thread`, this keeps the same workspace: only
62    /// transcript/lifecycle records are copied, never executable tool or approval
63    /// events.
64    pub(crate) async fn seed_agent_thread_history(
65        &self,
66        parent_thread_id: &ThreadId,
67        child_thread_id: &ThreadId,
68        fork_turns: &str,
69    ) -> anyhow::Result<()> {
70        if fork_turns == "none" {
71            return Ok(());
72        }
73        let Some(store) = self.thread_store.clone() else {
74            return Ok(());
75        };
76        let Some(parent) = store.load_thread(parent_thread_id).await? else {
77            return Ok(());
78        };
79        let mut events = seed_events_for_child(&parent.events, None)?;
80        if fork_turns != "all" {
81            let turn_count = fork_turns.parse::<usize>().map_err(|_| {
82                anyhow::anyhow!("fork_turns must be one of none, all, or a positive integer")
83            })?;
84            anyhow::ensure!(turn_count > 0, "fork_turns integer must be positive");
85            let mut ordered_turns = Vec::<TurnId>::new();
86            for event in &events {
87                if let Some(turn_id) = event.turn_id.as_ref()
88                    && ordered_turns.last() != Some(turn_id)
89                {
90                    ordered_turns.push(turn_id.clone());
91                }
92            }
93            let keep_from = ordered_turns.len().saturating_sub(turn_count);
94            let kept = &ordered_turns[keep_from..];
95            events.retain(|event| {
96                event
97                    .turn_id
98                    .as_ref()
99                    .is_some_and(|turn_id| kept.contains(turn_id))
100            });
101        }
102        for event in &events {
103            store.append_event(child_thread_id, event).await?;
104        }
105        Ok(())
106    }
107
108    /// Forks `parent_thread_id` into a new child thread backed by a fresh
109    /// workspace fork of the parent workspace.
110    pub async fn fork_thread(
111        &self,
112        request: ForkThreadRequest,
113    ) -> anyhow::Result<ForkThreadOutcome> {
114        self.emit(RoderEvent::ThreadForkRequested(ThreadForkRequested {
115            parent_thread_id: request.parent_thread_id.clone(),
116            name: request.name.clone(),
117            timestamp: OffsetDateTime::now_utc(),
118        }))
119        .await;
120        match self.fork_thread_inner(&request).await {
121            Ok(outcome) => Ok(outcome),
122            Err(error) => {
123                self.emit(RoderEvent::ThreadForkFailed(ThreadForkFailed {
124                    parent_thread_id: request.parent_thread_id.clone(),
125                    name: request.name.clone(),
126                    message: error.to_string(),
127                    timestamp: OffsetDateTime::now_utc(),
128                }))
129                .await;
130                Err(error)
131            }
132        }
133    }
134
135    async fn fork_thread_inner(
136        &self,
137        request: &ForkThreadRequest,
138    ) -> anyhow::Result<ForkThreadOutcome> {
139        let store = self
140            .thread_store
141            .clone()
142            .ok_or_else(|| anyhow::anyhow!("conversation forks require a thread store"))?;
143        let parent = store
144            .load_thread(&request.parent_thread_id)
145            .await?
146            .ok_or_else(|| {
147                anyhow::anyhow!("parent thread {} was not found", request.parent_thread_id)
148            })?;
149        let parent_metadata = parent.metadata.clone().ok_or_else(|| {
150            anyhow::anyhow!(
151                "parent thread {} has no metadata to fork from",
152                request.parent_thread_id
153            )
154        })?;
155
156        // Materialize the workspace fork first; thread creation only
157        // proceeds once an isolated workspace exists.
158        let provider_id = request
159            .provider_id
160            .clone()
161            .unwrap_or_else(|| DEFAULT_FORK_PROVIDER.to_string());
162        let fork = self
163            .create_workspace_fork(
164                &provider_id,
165                ForkRequest {
166                    source_workspace: PathBuf::from(&parent_metadata.workspace),
167                    name: Some(request.name.clone()),
168                    reason: ForkReason::ConversationFork,
169                    policy: ForkPolicy::default(),
170                    provider_config: request.provider_config.clone(),
171                },
172            )
173            .await?;
174
175        let now = OffsetDateTime::now_utc();
176        let seed_events = seed_events_for_child(&parent.events, request.from_turn_id.as_deref())?;
177        let mut warnings = Vec::new();
178        if request.from_turn_id.is_none() && seed_events.is_empty() && !parent.events.is_empty() {
179            warnings.push(
180                "parent thread has events but none were conversation records; the fork starts \
181                 with an empty transcript"
182                    .to_string(),
183            );
184        }
185
186        let child_id = uuid::Uuid::new_v4().to_string();
187        let child_metadata = ThreadMetadata {
188            thread_id: child_id.clone(),
189            title: Some(match &parent_metadata.title {
190                Some(title) => format!("{title} (fork: {})", request.name),
191                None => format!("fork: {}", request.name),
192            }),
193            workspace: fork.workspace.display().to_string(),
194            // The fork workspace lives outside registered workspace roots.
195            workspace_id: None,
196            root_id: None,
197            provider: parent_metadata.provider.clone(),
198            model: parent_metadata.model.clone(),
199            selection_mode: parent_metadata.selection_mode.clone(),
200            tool_allowlist: parent_metadata.tool_allowlist.clone(),
201            developer_instructions: parent_metadata.developer_instructions.clone(),
202            external_tools: parent_metadata.external_tools.clone(),
203            // Local workspace forks never inherit runner bindings.
204            runner_destination: None,
205            runner_state: None,
206            runner_binding: None,
207            created_at: now,
208            updated_at: now,
209            message_count: 0,
210            usage: None,
211            parent_thread_id: Some(request.parent_thread_id.clone()),
212            forked_from_turn_id: request.from_turn_id.clone(),
213            workspace_fork: Some(fork.clone()),
214        };
215
216        if let Err(error) = self
217            .seed_child_thread(&store, child_metadata.clone(), &child_id, seed_events)
218            .await
219        {
220            // Best-effort cleanup so a failed fork does not leak a workspace.
221            let _ = self
222                .remove_workspace_fork(
223                    &provider_id,
224                    &fork.id,
225                    RemoveForkPolicy {
226                        confirm_workspace: fork.workspace.clone(),
227                    },
228                )
229                .await;
230            return Err(error);
231        }
232
233        self.emit(RoderEvent::ThreadCreated(ThreadCreated {
234            thread_id: child_id.clone(),
235            timestamp: OffsetDateTime::now_utc(),
236        }))
237        .await;
238        self.emit(RoderEvent::ThreadForked(ThreadForked {
239            parent_thread_id: request.parent_thread_id.clone(),
240            child_thread_id: child_id.clone(),
241            fork,
242            timestamp: OffsetDateTime::now_utc(),
243        }))
244        .await;
245
246        let child = store
247            .load_thread_metadata(&child_id)
248            .await?
249            .unwrap_or(child_metadata);
250        Ok(ForkThreadOutcome { child, warnings })
251    }
252
253    async fn seed_child_thread(
254        &self,
255        store: &Arc<dyn ThreadStore>,
256        child_metadata: ThreadMetadata,
257        child_id: &ThreadId,
258        seed_events: Vec<EventEnvelope>,
259    ) -> anyhow::Result<()> {
260        store.create_thread(child_metadata).await?;
261        for envelope in &seed_events {
262            store.append_event(child_id, envelope).await?;
263        }
264        Ok(())
265    }
266
267    /**
268     * Removes the workspace fork behind a forked thread. Destructive and
269     * explicit: `confirm_path` must match the fork workspace exactly. The
270     * thread itself is kept (status flips to `Removed`) so the conversation
271     * stays readable.
272     */
273    pub async fn remove_thread_workspace_fork(
274        &self,
275        thread_id: &ThreadId,
276        confirm_path: &str,
277    ) -> anyhow::Result<WorkspaceFork> {
278        let store = self
279            .thread_store
280            .clone()
281            .ok_or_else(|| anyhow::anyhow!("conversation forks require a thread store"))?;
282        let mut metadata = store
283            .load_thread_metadata(thread_id)
284            .await?
285            .ok_or_else(|| anyhow::anyhow!("thread {thread_id} was not found"))?;
286        let mut fork = metadata
287            .workspace_fork
288            .clone()
289            .ok_or_else(|| anyhow::anyhow!("thread {thread_id} is not a workspace fork"))?;
290        anyhow::ensure!(
291            fork.status == ForkStatus::Active,
292            "fork {} was already removed",
293            fork.id
294        );
295        anyhow::ensure!(
296            std::path::Path::new(confirm_path) == fork.workspace,
297            "confirmation path does not match the fork workspace {}; removal is \
298             path-confirmed to prevent accidental deletion",
299            fork.workspace.display()
300        );
301
302        self.remove_workspace_fork(
303            &fork.provider_id.clone(),
304            &fork.id.clone(),
305            RemoveForkPolicy {
306                confirm_workspace: fork.workspace.clone(),
307            },
308        )
309        .await?;
310
311        fork.status = ForkStatus::Removed;
312        metadata.workspace_fork = Some(fork.clone());
313        metadata.updated_at = OffsetDateTime::now_utc();
314        store.update_thread_metadata(metadata).await?;
315
316        self.emit(RoderEvent::ThreadForkRemoved(ThreadForkRemoved {
317            thread_id: thread_id.clone(),
318            fork_id: fork.id.clone(),
319            worktree_path: fork.workspace.display().to_string(),
320            timestamp: OffsetDateTime::now_utc(),
321        }))
322        .await;
323        Ok(fork)
324    }
325}
326
327/**
328 * Selects the parent events that seed the child transcript: only
329 * conversation-history records (turn lifecycle and transcript items), never
330 * tool/approval/audit events, so nothing side-effectful is replayed. When
331 * `from_turn_id` is set, events after that turn's records are dropped.
332 */
333fn seed_events_for_child(
334    events: &[EventEnvelope],
335    from_turn_id: Option<&str>,
336) -> anyhow::Result<Vec<EventEnvelope>> {
337    let mut ordered: Vec<&EventEnvelope> = events.iter().collect();
338    ordered.sort_by_key(|envelope| envelope.seq);
339
340    let cutoff = match from_turn_id {
341        Some(turn_id) => {
342            let last = ordered
343                .iter()
344                .rposition(|envelope| envelope.turn_id.as_deref() == Some(turn_id))
345                .ok_or_else(|| {
346                    anyhow::anyhow!("turn {turn_id} was not found in the parent thread")
347                })?;
348            last + 1
349        }
350        None => ordered.len(),
351    };
352
353    Ok(ordered[..cutoff]
354        .iter()
355        .filter(|envelope| match &envelope.event {
356            RoderEvent::TurnStarted(_)
357            | RoderEvent::TurnCompleted(_)
358            | RoderEvent::TurnFailed(_)
359            | RoderEvent::TurnInterrupted(_) => true,
360            RoderEvent::TranscriptItemAppended(event) => event
361                .item
362                .as_ref()
363                .is_some_and(forkable_agent_transcript_item),
364            _ => false,
365        })
366        .map(|envelope| (*envelope).clone())
367        .collect())
368}
369
370fn forkable_agent_transcript_item(item: &roder_api::transcript::TranscriptItem) -> bool {
371    match item {
372        roder_api::transcript::TranscriptItem::UserMessage(_) => true,
373        roder_api::transcript::TranscriptItem::AssistantMessage(message) => message
374            .phase
375            .as_deref()
376            .is_none_or(|phase| phase.is_empty() || phase == "final_answer"),
377        roder_api::transcript::TranscriptItem::ReasoningSummary(_)
378        | roder_api::transcript::TranscriptItem::ToolCall(_)
379        | roder_api::transcript::TranscriptItem::ToolResult(_)
380        | roder_api::transcript::TranscriptItem::FileChange(_)
381        | roder_api::transcript::TranscriptItem::ContextCompaction(_)
382        | roder_api::transcript::TranscriptItem::Error(_)
383        | roder_api::transcript::TranscriptItem::ProviderMetadata(_) => false,
384    }
385}
386
387#[cfg(test)]
388mod tests {
389    use super::*;
390    use roder_api::events::{EventSource, TranscriptItemAppended, TurnCompleted, TurnStarted};
391    use roder_api::transcript::{
392        ContextCompactionRecord, ToolCallRecord, ToolResultRecord, TranscriptItem, UserMessage,
393    };
394
395    fn envelope(seq: u64, turn_id: &str, event: RoderEvent) -> EventEnvelope {
396        EventEnvelope {
397            event_id: format!("event-{seq}"),
398            seq,
399            timestamp: OffsetDateTime::UNIX_EPOCH,
400            source: EventSource::Core,
401            kind: event.kind().to_string(),
402            thread_id: Some("parent".to_string()),
403            turn_id: Some(turn_id.to_string()),
404            event,
405        }
406    }
407
408    fn turn_events(seq: u64, turn_id: &str, text: &str) -> Vec<EventEnvelope> {
409        vec![
410            envelope(
411                seq,
412                turn_id,
413                RoderEvent::TurnStarted(TurnStarted {
414                    thread_id: "parent".to_string(),
415                    turn_id: turn_id.to_string(),
416                    runtime_profile: Default::default(),
417                    timestamp: OffsetDateTime::UNIX_EPOCH,
418                }),
419            ),
420            envelope(
421                seq + 1,
422                turn_id,
423                RoderEvent::TranscriptItemAppended(TranscriptItemAppended {
424                    thread_id: "parent".to_string(),
425                    turn_id: turn_id.to_string(),
426                    item_type: "user_message".to_string(),
427                    item_index: None,
428                    item: Some(TranscriptItem::UserMessage(UserMessage::text(text))),
429                    timestamp: OffsetDateTime::UNIX_EPOCH,
430                }),
431            ),
432            envelope(
433                seq + 2,
434                turn_id,
435                RoderEvent::TurnCompleted(TurnCompleted {
436                    thread_id: "parent".to_string(),
437                    turn_id: turn_id.to_string(),
438                    usage: None,
439                    finish_reason: Some("stop".to_string()),
440                    timestamp: OffsetDateTime::UNIX_EPOCH,
441                }),
442            ),
443        ]
444    }
445
446    #[test]
447    fn seed_events_keep_conversation_records_only() {
448        let mut events = turn_events(1, "turn-1", "hello");
449        events.push(envelope(
450            4,
451            "turn-1",
452            RoderEvent::ToolCallStarted(roder_api::events::ToolCallStarted {
453                thread_id: "parent".to_string(),
454                turn_id: "turn-1".to_string(),
455                tool_id: "call-1".to_string(),
456                tool_name: Some("shell".to_string()),
457                display_payload: None,
458                timestamp: OffsetDateTime::UNIX_EPOCH,
459            }),
460        ));
461        events.push(envelope(
462            7,
463            "turn-1",
464            RoderEvent::TranscriptItemAppended(TranscriptItemAppended {
465                thread_id: "parent".to_string(),
466                turn_id: "turn-1".to_string(),
467                item_type: "context_compaction".to_string(),
468                item_index: None,
469                item: Some(TranscriptItem::ContextCompaction(ContextCompactionRecord {
470                    summary: "private parent compaction".to_string(),
471                })),
472                timestamp: OffsetDateTime::UNIX_EPOCH,
473            }),
474        ));
475        events.push(envelope(
476            5,
477            "turn-1",
478            RoderEvent::TranscriptItemAppended(TranscriptItemAppended {
479                thread_id: "parent".to_string(),
480                turn_id: "turn-1".to_string(),
481                item_type: "tool_call".to_string(),
482                item_index: None,
483                item: Some(TranscriptItem::ToolCall(ToolCallRecord {
484                    id: "spawn-call".to_string(),
485                    name: "spawn_agent".to_string(),
486                    arguments: "{}".to_string(),
487                })),
488                timestamp: OffsetDateTime::UNIX_EPOCH,
489            }),
490        ));
491        events.push(envelope(
492            6,
493            "turn-1",
494            RoderEvent::TranscriptItemAppended(TranscriptItemAppended {
495                thread_id: "parent".to_string(),
496                turn_id: "turn-1".to_string(),
497                item_type: "tool_result".to_string(),
498                item_index: None,
499                item: Some(TranscriptItem::ToolResult(ToolResultRecord {
500                    id: "spawn-call".to_string(),
501                    name: Some("spawn_agent".to_string()),
502                    result: "spawned".to_string(),
503                    display_payload: None,
504                    is_error: false,
505                })),
506                timestamp: OffsetDateTime::UNIX_EPOCH,
507            }),
508        ));
509
510        let seeded = seed_events_for_child(&events, None).unwrap();
511
512        assert_eq!(seeded.len(), 3, "tool records must not be replayed");
513        assert!(
514            seeded
515                .iter()
516                .all(|envelope| !matches!(envelope.event, RoderEvent::ToolCallStarted(_)))
517        );
518        assert!(seeded.iter().all(|envelope| {
519            !matches!(
520                &envelope.event,
521                RoderEvent::TranscriptItemAppended(event)
522                    if matches!(
523                        event.item,
524                        Some(
525                            TranscriptItem::ToolCall(_)
526                                | TranscriptItem::ToolResult(_)
527                                | TranscriptItem::ContextCompaction(_)
528                        )
529                    )
530            )
531        }));
532    }
533
534    #[test]
535    fn seed_events_truncate_at_requested_turn() {
536        let mut events = turn_events(1, "turn-1", "first");
537        events.extend(turn_events(10, "turn-2", "second"));
538
539        let seeded = seed_events_for_child(&events, Some("turn-1")).unwrap();
540        assert_eq!(seeded.len(), 3);
541        assert!(
542            seeded
543                .iter()
544                .all(|envelope| envelope.turn_id.as_deref() == Some("turn-1"))
545        );
546
547        let error = seed_events_for_child(&events, Some("missing-turn")).unwrap_err();
548        assert!(error.to_string().contains("missing-turn"));
549    }
550}
roder_core/conversation_forks.rs

roder_core/
conversation_forks.rs
