roder_core/

conversation_forks.rs

//! Conversation forks (roadmap phases 90 + 81).
//!
//! Forks an existing thread into a child thread backed by a workspace fork
//! from any registered `ForkProvider` (default: `git-worktree`): the child
//! starts from the parent transcript (no side-effectful tool replay — only
//! conversation history records are copied) and all subsequent tool
//! execution resolves against the fork workspace because the child's
//! `ThreadMetadata.workspace` points at it. Cleanup is explicit and
//! path-confirmed; the parent workspace is never modified.

use std::path::PathBuf;
use std::sync::Arc;

use roder_api::events::{
    EventEnvelope, RoderEvent, ThreadCreated, ThreadForkFailed, ThreadForkRemoved,
    ThreadForkRequested, ThreadForked, ThreadId, TurnId,
};
use roder_api::forks::{
    ForkPolicy, ForkReason, ForkRequest, ForkStatus, RemoveForkPolicy, WorkspaceFork,
};
use roder_api::thread::{ThreadMetadata, ThreadStore};
use time::OffsetDateTime;

use crate::Runtime;
use crate::forks::DEFAULT_FORK_PROVIDER;

mod history;
use history::seed_events_for_child;

#[derive(Debug, Clone)]
pub struct ForkThreadRequest {
    pub parent_thread_id: ThreadId,
    /// User-facing fork name; the provider sanitizes it into its naming
    /// scheme (directories, branches, snapshot names).
    pub name: String,
    /// Fork at a specific parent turn; `None` forks at the latest turn.
    pub from_turn_id: Option<TurnId>,
    /// Fork provider id; `None` uses [`DEFAULT_FORK_PROVIDER`].
    pub provider_id: Option<String>,
    /// Provider-specific options (never secrets).
    pub provider_config: serde_json::Value,
}

impl ForkThreadRequest {
    pub fn new(parent_thread_id: ThreadId, name: impl Into<String>) -> Self {
        Self {
            parent_thread_id,
            name: name.into(),
            from_turn_id: None,
            provider_id: None,
            provider_config: serde_json::json!({}),
        }
    }
}

#[derive(Debug, Clone)]
pub struct ForkThreadOutcome {
    pub child: ThreadMetadata,
    pub warnings: Vec<String>,
}

impl Runtime {
    /// Seeds a long-lived collaboration agent with a safe subset of the parent
    /// conversation. Unlike `fork_thread`, this keeps the same workspace: only
    /// transcript/lifecycle records are copied, never executable tool or approval
    /// events.
    pub(crate) async fn seed_agent_thread_history(
        &self,
        parent_thread_id: &ThreadId,
        child_thread_id: &ThreadId,
        fork_turns: &str,
    ) -> anyhow::Result<()> {
        if fork_turns == "none" {
            return Ok(());
        }
        let Some(store) = self.thread_store.clone() else {
            return Ok(());
        };
        let Some(parent) = store.load_thread(parent_thread_id).await? else {
            return Ok(());
        };
        let mut events = seed_events_for_child(&parent.events, None)?;
        if fork_turns != "all" {
            let turn_count = fork_turns.parse::<usize>().map_err(|_| {
                anyhow::anyhow!("fork_turns must be one of none, all, or a positive integer")
            })?;
            anyhow::ensure!(turn_count > 0, "fork_turns integer must be positive");
            let mut ordered_turns = Vec::<TurnId>::new();
            for event in &events {
                if let Some(turn_id) = event.turn_id.as_ref()
                    && ordered_turns.last() != Some(turn_id)
                {
                    ordered_turns.push(turn_id.clone());
                }
            }
            let keep_from = ordered_turns.len().saturating_sub(turn_count);
            let kept = &ordered_turns[keep_from..];
            events.retain(|event| {
                event
                    .turn_id
                    .as_ref()
                    .is_some_and(|turn_id| kept.contains(turn_id))
            });
        }
        for event in &events {
            store.append_event(child_thread_id, event).await?;
        }
        Ok(())
    }

    /// Forks `parent_thread_id` into a new child thread backed by a fresh
    /// workspace fork of the parent workspace.
    pub async fn fork_thread(
        &self,
        request: ForkThreadRequest,
    ) -> anyhow::Result<ForkThreadOutcome> {
        self.emit(RoderEvent::ThreadForkRequested(ThreadForkRequested {
            parent_thread_id: request.parent_thread_id.clone(),
            name: request.name.clone(),
            timestamp: OffsetDateTime::now_utc(),
        }))
        .await;
        match self.fork_thread_inner(&request).await {
            Ok(outcome) => Ok(outcome),
            Err(error) => {
                self.emit(RoderEvent::ThreadForkFailed(ThreadForkFailed {
                    parent_thread_id: request.parent_thread_id.clone(),
                    name: request.name.clone(),
                    message: error.to_string(),
                    timestamp: OffsetDateTime::now_utc(),
                }))
                .await;
                Err(error)
            }
        }
    }

    async fn fork_thread_inner(
        &self,
        request: &ForkThreadRequest,
    ) -> anyhow::Result<ForkThreadOutcome> {
        let store = self
            .thread_store
            .clone()
            .ok_or_else(|| anyhow::anyhow!("conversation forks require a thread store"))?;
        let parent = store
            .load_thread(&request.parent_thread_id)
            .await?
            .ok_or_else(|| {
                anyhow::anyhow!("parent thread {} was not found", request.parent_thread_id)
            })?;
        let parent_metadata = parent.metadata.clone().ok_or_else(|| {
            anyhow::anyhow!(
                "parent thread {} has no metadata to fork from",
                request.parent_thread_id
            )
        })?;
        // Validate the history boundary before allocating a checkout. The
        // child identity is recorded atomically with ownership at creation.
        let seed_events = seed_events_for_child(&parent.events, request.from_turn_id.as_deref())?;
        let child_id = uuid::Uuid::new_v4().to_string();

        // Materialize the workspace fork first; thread creation only
        // proceeds once an isolated workspace exists.
        let provider_id = request
            .provider_id
            .clone()
            .unwrap_or_else(|| DEFAULT_FORK_PROVIDER.to_string());
        let fork = self
            .create_workspace_fork(
                &provider_id,
                ForkRequest {
                    ownership: Some(roder_api::forks::ForkOwnership {
                        creator_thread_id: Some(request.parent_thread_id.clone()),
                        owner_thread_id: Some(child_id.clone()),
                    }),
                    source_workspace: PathBuf::from(&parent_metadata.workspace),
                    name: Some(request.name.clone()),
                    reason: ForkReason::ConversationFork,
                    policy: ForkPolicy::default(),
                    provider_config: request.provider_config.clone(),
                },
            )
            .await?;

        let now = OffsetDateTime::now_utc();
        let mut warnings = Vec::new();
        if request.from_turn_id.is_none() && seed_events.is_empty() && !parent.events.is_empty() {
            warnings.push(
                "parent thread has events but none were conversation records; the fork starts \
                 with an empty transcript"
                    .to_string(),
            );
        }

        let child_metadata = ThreadMetadata {
            thread_id: child_id.clone(),
            title: Some(match &parent_metadata.title {
                Some(title) => format!("{title} (fork: {})", request.name),
                None => format!("fork: {}", request.name),
            }),
            workspace: fork.workspace.display().to_string(),
            // The fork workspace lives outside registered workspace roots.
            workspace_id: None,
            root_id: None,
            provider: parent_metadata.provider.clone(),
            model: parent_metadata.model.clone(),
            selection_mode: parent_metadata.selection_mode.clone(),
            tool_allowlist: parent_metadata.tool_allowlist.clone(),
            service_tier: parent_metadata.service_tier.clone(),
            developer_instructions: parent_metadata.developer_instructions.clone(),
            external_tools: parent_metadata.external_tools.clone(),
            // Local workspace forks never inherit runner bindings.
            runner_destination: None,
            runner_state: None,
            runner_binding: None,
            created_at: now,
            updated_at: now,
            message_count: 0,
            usage: None,
            parent_thread_id: Some(request.parent_thread_id.clone()),
            forked_from_turn_id: request.from_turn_id.clone(),
            workspace_fork: Some(fork.clone()),
        };

        let seed = async {
            self.seed_child_thread(&store, child_metadata.clone(), &child_id, seed_events)
                .await?;
            self.goals
                .inherit_thread_goal_snapshot(&request.parent_thread_id, &child_id)
                .await
        }
        .await;
        let inherited_goal = match seed {
            Ok(goal) => goal,
            Err(error) => {
                // Best-effort cleanup so a failed fork does not leak a workspace.
                let _ = self
                    .remove_workspace_fork(
                        &provider_id,
                        &fork.id,
                        RemoveForkPolicy {
                            confirm_workspace: fork.workspace.clone(),
                        },
                    )
                    .await;
                return Err(error);
            }
        };

        self.emit(RoderEvent::ThreadCreated(ThreadCreated {
            thread_id: child_id.clone(),
            timestamp: OffsetDateTime::now_utc(),
        }))
        .await;
        if let Some(goal) = inherited_goal {
            self.goals.emit_goal_updated(goal).await;
        }
        self.emit(RoderEvent::ThreadForked(ThreadForked {
            parent_thread_id: request.parent_thread_id.clone(),
            child_thread_id: child_id.clone(),
            fork,
            timestamp: OffsetDateTime::now_utc(),
        }))
        .await;

        let child = store
            .load_thread_metadata(&child_id)
            .await?
            .unwrap_or(child_metadata);
        Ok(ForkThreadOutcome { child, warnings })
    }

    async fn seed_child_thread(
        &self,
        store: &Arc<dyn ThreadStore>,
        child_metadata: ThreadMetadata,
        child_id: &ThreadId,
        seed_events: Vec<EventEnvelope>,
    ) -> anyhow::Result<()> {
        store.create_thread(child_metadata).await?;
        for envelope in &seed_events {
            store.append_event(child_id, envelope).await?;
        }
        Ok(())
    }

    /**
     * Removes the workspace fork behind a forked thread. Destructive and
     * explicit: `confirm_path` must match the fork workspace exactly. The
     * thread itself is kept (status flips to `Removed`) so the conversation
     * stays readable.
     */
    pub async fn remove_thread_workspace_fork(
        &self,
        thread_id: &ThreadId,
        confirm_path: &str,
    ) -> anyhow::Result<WorkspaceFork> {
        let store = self
            .thread_store
            .clone()
            .ok_or_else(|| anyhow::anyhow!("conversation forks require a thread store"))?;
        let mut metadata = store
            .load_thread_metadata(thread_id)
            .await?
            .ok_or_else(|| anyhow::anyhow!("thread {thread_id} was not found"))?;
        let mut fork = metadata
            .workspace_fork
            .clone()
            .ok_or_else(|| anyhow::anyhow!("thread {thread_id} is not a workspace fork"))?;
        anyhow::ensure!(
            fork.status == ForkStatus::Active,
            "fork {} was already removed",
            fork.id
        );
        anyhow::ensure!(
            std::path::Path::new(confirm_path) == fork.workspace,
            "confirmation path does not match the fork workspace {}; removal is \
             path-confirmed to prevent accidental deletion",
            fork.workspace.display()
        );

        self.remove_workspace_fork(
            &fork.provider_id.clone(),
            &fork.id.clone(),
            RemoveForkPolicy {
                confirm_workspace: fork.workspace.clone(),
            },
        )
        .await?;

        fork.status = ForkStatus::Removed;
        metadata.workspace_fork = Some(fork.clone());
        metadata.updated_at = OffsetDateTime::now_utc();
        store.update_thread_metadata(metadata).await?;

        self.emit(RoderEvent::ThreadForkRemoved(ThreadForkRemoved {
            thread_id: thread_id.clone(),
            fork_id: fork.id.clone(),
            worktree_path: fork.workspace.display().to_string(),
            timestamp: OffsetDateTime::now_utc(),
        }))
        .await;
        Ok(fork)
    }
}