roder_core/
forks.rs

1//! Runtime fork manager (roadmap phase 81, Task 2): resolves configured
2//! `ForkProvider`s from the extension registry and applies path/policy
3//! checks before provider calls. Thread attachment lives in
4//! `conversation_forks`; these are the provider-facing primitives shared by
5//! conversation forks and the `forks/*` app-server surface.
6
7use std::collections::HashMap;
8use std::path::PathBuf;
9use std::sync::Arc;
10
11use roder_api::forks::{
12    ForkCapabilities, ForkId, ForkProvenance, ForkProvider, ForkProviderDescriptor, ForkRequest,
13    ForkStatus, RemoveForkPolicy, RemoveForkResult, WorkspaceFork,
14};
15use roder_api::remote_runner::{RemoteRunnerProvider, RemoteRunnerSession, RunnerDestination};
16use tokio::sync::Mutex;
17
18use crate::Runtime;
19
20/// Default fork provider when a request does not name one.
21pub const DEFAULT_FORK_PROVIDER: &str = "git-worktree";
22
23impl Runtime {
24    /// Lists registered fork providers.
25    pub fn fork_providers(&self) -> Vec<ForkProviderDescriptor> {
26        self.registry
27            .fork_providers
28            .iter()
29            .map(|provider| provider.descriptor())
30            .collect()
31    }
32
33    /// Resolves a fork provider with an actionable error naming the
34    /// available ids.
35    pub fn fork_provider(&self, provider_id: &str) -> anyhow::Result<Arc<dyn ForkProvider>> {
36        self.registry.fork_provider(provider_id).ok_or_else(|| {
37            let available = self
38                .registry
39                .fork_providers
40                .iter()
41                .map(|provider| provider.descriptor().id)
42                .collect::<Vec<_>>()
43                .join(", ");
44            anyhow::anyhow!(
45                "fork provider {provider_id:?} is not installed (available: {})",
46                if available.is_empty() {
47                    "none"
48                } else {
49                    &available
50                }
51            )
52        })
53    }
54
55    /// Creates a workspace fork after host-side validation.
56    pub async fn create_workspace_fork(
57        &self,
58        provider_id: &str,
59        request: ForkRequest,
60    ) -> anyhow::Result<WorkspaceFork> {
61        anyhow::ensure!(
62            request.source_workspace.is_absolute(),
63            "fork source workspace must be an absolute path: {}",
64            request.source_workspace.display()
65        );
66        if let Some(name) = &request.name {
67            anyhow::ensure!(
68                !name.trim().is_empty() && name.len() <= 100,
69                "fork names must be 1-100 characters"
70            );
71        }
72        let provider = self.fork_provider(provider_id)?;
73        anyhow::ensure!(
74            provider.descriptor().capabilities.create,
75            "fork provider {provider_id:?} does not support creation"
76        );
77        provider.create_fork(request).await
78    }
79
80    pub async fn list_workspace_forks(
81        &self,
82        provider_id: &str,
83        source_workspace: &std::path::Path,
84    ) -> anyhow::Result<Vec<WorkspaceFork>> {
85        self.fork_provider(provider_id)?
86            .list_forks(source_workspace)
87            .await
88    }
89
90    pub async fn resume_workspace_fork(
91        &self,
92        provider_id: &str,
93        id: &ForkId,
94    ) -> anyhow::Result<WorkspaceFork> {
95        self.fork_provider(provider_id)?.resume_fork(id).await
96    }
97
98    /// Removes a fork; destructive and always path-confirmed by the
99    /// provider contract.
100    pub async fn remove_workspace_fork(
101        &self,
102        provider_id: &str,
103        id: &ForkId,
104        policy: RemoveForkPolicy,
105    ) -> anyhow::Result<RemoveForkResult> {
106        let provider = self.fork_provider(provider_id)?;
107        if provider_id == DEFAULT_FORK_PROVIDER {
108            let fork = provider.resume_fork(id).await?;
109            if let Some(owner) = fork.ownership.and_then(|owner| owner.owner_thread_id) {
110                anyhow::ensure!(
111                    !self.has_active_turn_for_thread(&owner).await,
112                    "cannot remove a worktree while its owner thread {owner} has an active turn"
113                );
114            }
115        }
116        provider.remove_fork(id, policy).await
117    }
118}
119
120/// Provider id used by the remote-runner fork adapter (roadmap phase 81,
121/// Task 5).
122pub const REMOTE_RUNNER_FORK_PROVIDER_ID: &str = "remote-runner";
123
124/**
125 * Represents a fresh remote-runner session as a `WorkspaceFork` with
126 * `remote_compute = true`. The fork layer owns only lifecycle, provenance,
127 * and attachment — file/process operations stay delegated to the
128 * `RemoteRunnerSession` contract. `remove_fork` closes the runner session
129 * (providers without snapshot deletion simply terminate the session, which
130 * is documented as the deterministic cleanup behavior); `resume_fork`
131 * re-opens it through the runner provider's `resume_session` path, with
132 * snapshot-backed restore when the provider recorded one.
133 */
134pub struct RemoteRunnerForkAdapter {
135    provider: Arc<dyn RemoteRunnerProvider>,
136    destination: RunnerDestination,
137    /// Absolute workspace path on the runner.
138    runner_workspace: PathBuf,
139    sessions: Mutex<HashMap<ForkId, Arc<dyn RemoteRunnerSession>>>,
140}
141
142impl RemoteRunnerForkAdapter {
143    pub fn new(
144        provider: Arc<dyn RemoteRunnerProvider>,
145        destination: RunnerDestination,
146        runner_workspace: PathBuf,
147    ) -> Self {
148        Self {
149            provider,
150            destination,
151            runner_workspace,
152            sessions: Mutex::new(HashMap::new()),
153        }
154    }
155
156    /// The live runner session backing a fork, for tool execution wiring.
157    pub async fn session(&self, id: &ForkId) -> Option<Arc<dyn RemoteRunnerSession>> {
158        self.sessions.lock().await.get(id).cloned()
159    }
160
161    fn fork_for(&self, session: &Arc<dyn RemoteRunnerSession>) -> WorkspaceFork {
162        let state = session.state();
163        WorkspaceFork {
164            ownership: None,
165            id: state.session_id.clone(),
166            provider_id: REMOTE_RUNNER_FORK_PROVIDER_ID.to_string(),
167            source_workspace: self.runner_workspace.clone(),
168            workspace: self.runner_workspace.clone(),
169            status: ForkStatus::Active,
170            provenance: ForkProvenance {
171                branch: None,
172                source_branch: None,
173                source_commit: None,
174                snapshot_id: state
175                    .snapshot
176                    .as_ref()
177                    .map(|snapshot| snapshot.snapshot_id.clone()),
178                session_id: Some(state.session_id),
179                created_at: time::OffsetDateTime::now_utc(),
180            },
181            cleanup: Default::default(),
182            metadata: serde_json::json!({
183                "runnerProviderId": state.provider_id,
184                "destinationId": state.destination_id,
185            }),
186        }
187    }
188}
189
190#[async_trait::async_trait]
191impl ForkProvider for RemoteRunnerForkAdapter {
192    fn descriptor(&self) -> ForkProviderDescriptor {
193        ForkProviderDescriptor {
194            id: REMOTE_RUNNER_FORK_PROVIDER_ID.to_string(),
195            display_name: format!("Remote runner ({})", self.destination.provider_id),
196            capabilities: ForkCapabilities {
197                create: true,
198                list: false,
199                remove: true,
200                resume: true,
201                diff_summary: false,
202                merge_back: false,
203                copy_on_write: false,
204                remote_compute: true,
205            },
206        }
207    }
208
209    async fn create_fork(&self, request: ForkRequest) -> anyhow::Result<WorkspaceFork> {
210        anyhow::ensure!(
211            !request.policy.allow_dirty_source,
212            "remote-runner forks always start from the destination's own state; \
213             allow_dirty_source has no meaning here and must stay false"
214        );
215        let session = self
216            .provider
217            .create_session(self.destination.clone())
218            .await?;
219        let fork = self.fork_for(&session);
220        self.sessions.lock().await.insert(fork.id.clone(), session);
221        Ok(fork)
222    }
223
224    async fn list_forks(&self, _source: &std::path::Path) -> anyhow::Result<Vec<WorkspaceFork>> {
225        // Runner providers own session listing; the adapter only tracks the
226        // sessions it created in this process.
227        let sessions = self.sessions.lock().await;
228        Ok(sessions
229            .values()
230            .map(|session| self.fork_for(session))
231            .collect())
232    }
233
234    async fn resume_fork(&self, id: &ForkId) -> anyhow::Result<WorkspaceFork> {
235        if let Some(session) = self.session(id).await {
236            return Ok(self.fork_for(&session));
237        }
238        // Snapshot-backed resume through the runner provider.
239        let state = roder_api::remote_runner::RunnerSessionState {
240            provider_id: self.destination.provider_id.clone(),
241            session_id: id.clone(),
242            destination_id: self.destination.id.clone(),
243            snapshot: None,
244            metadata: self.destination.config.clone(),
245        };
246        let session = self.provider.resume_session(state).await?;
247        let fork = self.fork_for(&session);
248        self.sessions.lock().await.insert(fork.id.clone(), session);
249        Ok(fork)
250    }
251
252    async fn remove_fork(
253        &self,
254        id: &ForkId,
255        policy: RemoveForkPolicy,
256    ) -> anyhow::Result<RemoveForkResult> {
257        anyhow::ensure!(
258            policy.confirm_workspace == self.runner_workspace,
259            "removal is path-confirmed: confirm the runner workspace {}",
260            self.runner_workspace.display()
261        );
262        let session = self
263            .sessions
264            .lock()
265            .await
266            .remove(id)
267            .ok_or_else(|| anyhow::anyhow!("remote fork {id} is not active in this process"))?;
268        session.close().await?;
269        Ok(RemoveForkResult {
270            id: id.clone(),
271            removed: true,
272            workspace: self.runner_workspace.clone(),
273        })
274    }
275}
roder_core/forks.rs

roder_core/
forks.rs
