buntrail_engine/remote/
mod.rs

1//! Service proxies only. T16 connects these to the engine's document/mutation coordinator.
2mod files;
3pub use files::RemoteOwnedSave;
4mod git;
5mod proof;
6pub use proof::{EntryGroup, ObservationEvent, OwnedObservation, WorktreePreparation};
7mod operations;
8mod search;
9mod terminal;
10mod transport;
11use base64::{Engine as _, engine::general_purpose::STANDARD as B64};
12use buntrail_core::contracts::{
13    async_types::{EventStream, ServiceFuture},
14    *,
15};
16use buntrail_protocol::{services as w, types::Message};
17use buntrail_services::contracts::*;
18use futures_core::Stream;
19use sha2::Digest;
20use std::{
21    pin::Pin,
22    sync::{
23        Arc,
24        atomic::{AtomicBool, Ordering},
25    },
26    task::{Context, Poll},
27};
28fn io_error(e: impl std::fmt::Display) -> CoreError {
29    CoreError::new(ErrorCode::Io, &e.to_string())
30}
31#[derive(Clone)]
32pub struct RemoteServices {
33    transport: transport::Transport,
34    target: TargetIdentity,
35    lease: WorkspaceLease,
36    spools: Arc<buntrail_services::leases::LeasePool>,
37    cancellation: Option<Arc<AtomicBool>>,
38    // A registered view keeps the reusable physical anchor alive, never the reverse.
39    anchor: Option<Arc<RemoteServices>>,
40}
41pub(crate) use transport::Lifetime;
42pub use transport::Shutdown as RemoteShutdown;
43impl RemoteServices {
44    /// The route owner supplies an authenticated byte transport and verified target, never a display hostname.
45    pub fn connect(
46        socket: std::os::unix::net::UnixStream,
47        target: TargetIdentity,
48        lease: WorkspaceLease,
49        spool_directory: std::path::PathBuf,
50    ) -> Result<Self, CoreError> {
51        let spools = buntrail_services::leases::LeasePool::open(spool_directory)?;
52        let transport = transport::Transport::connect(socket, &target, &lease)?;
53        let lease = WorkspaceLease {
54            runtime: transport.runtime.clone(),
55            ..lease
56        };
57        Ok(Self {
58            lease,
59            transport,
60            target,
61            spools,
62            cancellation: None,
63            anchor: None,
64        })
65    }
66    #[cfg(test)]
67    pub(crate) fn shares_connection(&self, other: &Self) -> bool {
68        Arc::ptr_eq(&self.transport.connected, &other.transport.connected)
69    }
70    /// Worker-only registration; all views share the existing wire worker and spool pool.
71    pub fn register_workspace(&self, lease: WorkspaceLease) -> Result<Self, CoreError> {
72        let transport = self.transport.register(lease.clone())?;
73        Ok(Self {
74            transport,
75            lease,
76            ..self.clone()
77        })
78    }
79    /// Worker-only, read-only observation of an exact prior owner.
80    pub fn observe_owner(&self, owner: WorkspaceLease) -> Result<LeaseObservation, CoreError> {
81        if !self
82            .negotiated_capabilities()
83            .iter()
84            .any(|c| c == "leases.observe")
85        {
86            return Err(CoreError::new(
87                ErrorCode::Unavailable,
88                "active helper cannot observe owner retirement; preserve the existing supervisor",
89            ));
90        }
91        if owner.client != self.lease.client {
92            return Err(w::invalid());
93        }
94        let expected = w::LeaseOwner::from_core(&owner, &self.target)?;
95        match self.call(
96            w::Call::LeaseObserve {
97                owner: expected.clone(),
98            },
99            None,
100        )? {
101            w::Reply::LeaseObserved {
102                owner: actual,
103                runtime_generation,
104                presence,
105                workers_retired,
106            } if actual == expected
107                && w::id(&runtime_generation)? == self.runtime().supervisor_boot =>
108            {
109                let presence = match presence {
110                    w::OwnerPresence::Active {} => LeasePresence::Active,
111                    w::OwnerPresence::Detached { remaining_millis }
112                        if remaining_millis <= 1_800_000 =>
113                    {
114                        LeasePresence::Detached { remaining_millis }
115                    }
116                    w::OwnerPresence::Absent {} => LeasePresence::Absent,
117                    w::OwnerPresence::RuntimeLost {}
118                        if owner.runtime != self.runtime() && !workers_retired =>
119                    {
120                        LeasePresence::RuntimeLost
121                    }
122                    _ => return Err(w::invalid()),
123                };
124                if owner.runtime != self.runtime() && presence != LeasePresence::RuntimeLost {
125                    return Err(w::invalid());
126                }
127                Ok(LeaseObservation {
128                    owner,
129                    target: self.target.clone(),
130                    observed_runtime: self.runtime(),
131                    presence,
132                    workers_retired,
133                })
134            }
135            _ => Err(w::invalid()),
136        }
137    }
138    pub fn lease(&self) -> &WorkspaceLease {
139        &self.lease
140    }
141    pub fn target(&self) -> &TargetIdentity {
142        &self.target
143    }
144    pub fn validate_environment_ready(&self) -> Result<(), CoreError> {
145        self.validate_connection_ready()?;
146        if !self.transport.registered() {
147            return Err(CoreError::new(
148                ErrorCode::Closed,
149                "environment registration is released or disconnected; explicit owner resume is required",
150            ));
151        }
152        Ok(())
153    }
154    pub(crate) fn validate_connection_ready(&self) -> Result<(), CoreError> {
155        if self.connection_state() != ConnectionState::Ready
156            || !self.transport.authentication_ready.load(Ordering::SeqCst)
157        {
158            return Err(CoreError::new(
159                ErrorCode::Unavailable,
160                "current connection authentication has not been acknowledged",
161            ));
162        }
163        if self.target.container.is_some()
164            && !self
165                .negotiated_capabilities()
166                .iter()
167                .any(|c| c == "container.binding.v1")
168        {
169            return Err(CoreError::new(
170                ErrorCode::Unavailable,
171                "Active helper cannot verify container identity/runtime; preserve it and reconnect after a compatible supervisor is available",
172            ));
173        }
174        for capability in [
175            "services",
176            "file.resolve",
177            "file.read",
178            "file.save",
179            "file.save_proof",
180            "operations",
181            "leases",
182            "terminal",
183            "leases.retirement",
184            "terminal.ordered_controls",
185            "leases.observe",
186            "git.existing",
187        ] {
188            if !self
189                .negotiated_capabilities()
190                .iter()
191                .any(|c| c == capability)
192            {
193                return Err(CoreError::new(
194                    ErrorCode::Unavailable,
195                    &format!(
196                        "active helper lacks {capability}; preserve its existing sessions and reconnect after a compatible supervisor is available"
197                    ),
198                ));
199            }
200        }
201        Ok(())
202    }
203    pub(crate) fn anchored_workspace(
204        self: &Arc<Self>,
205        lease: WorkspaceLease,
206    ) -> Result<Self, CoreError> {
207        let mut view = self.register_workspace(lease)?;
208        view.anchor = Some(self.clone());
209        Ok(view)
210    }
211    pub(crate) fn bind_connection_permit(
212        &mut self,
213        permit: Arc<crate::environment::ConnectionPermit>,
214    ) {
215        self.transport.connection_permit = Some(permit);
216    }
217    pub(crate) fn physical(&self) -> Option<Arc<crate::environment::PhysicalOwner>> {
218        self.transport.physical.clone()
219    }
220    pub(crate) fn bind_physical(
221        &mut self,
222        channel: buntrail_services::connections::ssh::SshChannel,
223    ) -> Result<(), CoreError> {
224        self.bind_retirement(Box::new(move || Box::pin(channel.close())))
225    }
226    pub(crate) fn bind_retirement(
227        &mut self,
228        retire: Box<dyn FnOnce() -> ServiceFuture<'static, ()> + Send>,
229    ) -> Result<(), CoreError> {
230        self.transport.physical = Some(crate::environment::PhysicalOwner::start(
231            self.transport.lifetime(),
232            retire,
233        )?);
234        Ok(())
235    }
236    pub fn negotiated_capabilities(&self) -> &[String] {
237        &self.transport.capabilities
238    }
239    pub(crate) fn validate_workspace_release(
240        &self,
241        intent: LeaseReleaseIntent,
242    ) -> Result<(), CoreError> {
243        if intent == LeaseReleaseIntent::ResolvedWorkspaceClosure
244            && !self
245                .negotiated_capabilities()
246                .iter()
247                .any(|capability| capability == "leases.retirement")
248        {
249            return Err(CoreError::new(
250                ErrorCode::Unavailable,
251                "helper cannot confirm terminal retirement; keep this workspace until a compatible connection is available",
252            ));
253        }
254        Ok(())
255    }
256    /// Revokes this view's new-work authority; other registered views remain connected.
257    pub fn release_workspace(&self, intent: LeaseReleaseIntent) -> Result<(), CoreError> {
258        self.validate_workspace_release(intent)?;
259        self.transport.release(match intent {
260            LeaseReleaseIntent::ViewDetachment => {
261                buntrail_protocol::types::ReleaseIntent::ViewDetachment
262            }
263            LeaseReleaseIntent::ResolvedWorkspaceClosure => {
264                buntrail_protocol::types::ReleaseIntent::ResolvedWorkspaceClosure
265            }
266        })
267    }
268    pub(crate) fn detach_workspace(&self) -> Result<(), CoreError> {
269        self.transport.detach()
270    }
271    /// Physical-route owner only. Drop the SSH channel separately and await both shutdowns off the owner thread.
272    pub fn close_connection(&self) -> RemoteShutdown {
273        self.transport.close()
274    }
275    pub fn set_authentication(&self, ssh_auth_sock: Option<TargetPath>) -> Result<(), CoreError> {
276        let mut request = self.transport.start(
277            Message::Authentication {
278                ssh_auth_sock: ssh_auth_sock.as_ref().map(w::path).transpose()?,
279            },
280            None,
281            None,
282        )?;
283        if !matches!(request.next()?.header.message, Message::Ack {}) {
284            return Err(w::invalid());
285        }
286        request.finish();
287        self.transport
288            .authentication_ready
289            .store(true, Ordering::SeqCst);
290        Ok(())
291    }
292    pub fn runtime(&self) -> RuntimeGeneration {
293        self.transport.runtime.clone()
294    }
295    pub fn connection_state(&self) -> ConnectionState {
296        if self.transport.connected.load(Ordering::SeqCst) {
297            ConnectionState::Ready
298        } else {
299            ConnectionState::Disconnected
300        }
301    }
302    /// A worker-scoped view; cancellation never changes sibling views or physical ownership.
303    pub(crate) fn with_cancellation(&self, cancellation: Arc<AtomicBool>) -> Self {
304        Self {
305            cancellation: Some(cancellation),
306            ..self.clone()
307        }
308    }
309    fn check_cancellation(&self) -> Result<(), CoreError> {
310        if self
311            .cancellation
312            .as_ref()
313            .is_some_and(|flag| flag.load(Ordering::SeqCst))
314        {
315            Err(CoreError::new(
316                ErrorCode::Cancelled,
317                "remote request cancellation requested",
318            ))
319        } else {
320            Ok(())
321        }
322    }
323    fn next_reply(
324        &self,
325        request: &mut transport::Request,
326    ) -> Result<buntrail_protocol::frame::Frame, CoreError> {
327        if self.cancellation.is_none() {
328            return request.next();
329        }
330        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(120);
331        let mut cx = Context::from_waker(std::task::Waker::noop());
332        loop {
333            // Prefer an already-observed receipt over a concurrent cancellation request.
334            if let Poll::Ready(result) = request.poll(&mut cx) {
335                return result;
336            }
337            self.check_cancellation()?;
338            if std::time::Instant::now() >= deadline {
339                return Err(CoreError::new(
340                    ErrorCode::Unknown,
341                    "Remote request deadline exceeded; inspect original operation without replay",
342                ));
343            }
344            std::thread::park_timeout(std::time::Duration::from_millis(10));
345        }
346    }
347    fn check(&self, t: &TargetIdentity) -> Result<(), CoreError> {
348        if t != &self.target {
349            Err(CoreError::new(
350                ErrorCode::TargetChanged,
351                "Remote target changed",
352            ))
353        } else {
354            Ok(())
355        }
356    }
357    fn stream(&self, call: w::Call, operation: Option<OperationID>) -> Result<Replies, CoreError> {
358        self.stream_payload(call, operation, Vec::new())
359    }
360    fn stream_payload(
361        &self,
362        call: w::Call,
363        operation: Option<OperationID>,
364        payload: Vec<u8>,
365    ) -> Result<Replies, CoreError> {
366        self.check_cancellation()?;
367        Ok(Replies {
368            request: self.transport.start_payload(
369                Message::ServiceCall { call },
370                operation,
371                None,
372                payload,
373            )?,
374            stream_id: None,
375            bytes: Vec::new(),
376            length: None,
377            hash: None,
378            sequence: 0,
379            done: false,
380        })
381    }
382    fn call(&self, call: w::Call, operation: Option<OperationID>) -> Result<w::Reply, CoreError> {
383        self.call_payload(call, operation, Vec::new())
384    }
385    fn call_payload(
386        &self,
387        call: w::Call,
388        operation: Option<OperationID>,
389        payload: Vec<u8>,
390    ) -> Result<w::Reply, CoreError> {
391        let mut stream = self.stream_payload(call, operation, payload)?;
392        loop {
393            let frame = self.next_reply(&mut stream.request)?;
394            if let Some(value) = stream.accept(frame)? {
395                return value;
396            }
397        }
398    }
399    pub fn contains_document(
400        &self,
401        worktree: WorktreeIdentity,
402        file: FileRef,
403    ) -> Result<bool, CoreError> {
404        self.check(&worktree.target)?;
405        self.check(&file.target)?;
406        match self.call(
407            w::Call::Contains {
408                worktree: w::Worktree::from_core(&worktree)?,
409                file: w::File::from_core(&file)?,
410            },
411            None,
412        )? {
413            w::Reply::Contains { value } => Ok(value),
414            _ => Err(w::invalid()),
415        }
416    }
417    pub fn path_version(&self, file: FileRef) -> Result<FileVersion, CoreError> {
418        self.check(&file.target)?;
419        match self.call(
420            w::Call::PathVersion {
421                file: w::File::from_core(&file)?,
422            },
423            None,
424        )? {
425            w::Reply::Version { version } => version.core(&self.target),
426            _ => Err(w::invalid()),
427        }
428    }
429}
430struct Replies {
431    request: transport::Request,
432    stream_id: Option<String>,
433    bytes: Vec<u8>,
434    length: Option<u64>,
435    hash: Option<String>,
436    sequence: u64,
437    done: bool,
438}
439impl Replies {
440    fn accept(
441        &mut self,
442        f: buntrail_protocol::frame::Frame,
443    ) -> Result<Option<Result<w::Reply, CoreError>>, CoreError> {
444        if let Message::Error { code, safe_message } = f.header.message {
445            self.done = true;
446            self.request.finish();
447            return Ok(Some(Err(w::core_error(&code, &safe_message))));
448        }
449        let Message::ServiceEvent {
450            offset,
451            total_length,
452            sha256,
453            terminal,
454        } = f.header.message
455        else {
456            return Err(w::invalid());
457        };
458        let stream_id = f.header.stream_id.clone().ok_or_else(w::invalid)?;
459        if self.stream_id.as_ref().is_some_and(|id| id != &stream_id) {
460            return Err(w::invalid());
461        }
462        self.stream_id = Some(stream_id);
463        if f.header.sequence != Some(self.sequence)
464            || offset != self.bytes.len() as u64
465            || total_length > 32 * 1024 * 1024
466            || self.length.is_some_and(|l| l != total_length)
467            || self.hash.as_ref().is_some_and(|h| h != &sha256)
468        {
469            return Err(w::invalid());
470        }
471        self.length = Some(total_length);
472        self.hash = Some(sha256.clone());
473        self.bytes.extend_from_slice(&f.payload);
474        let complete = self.bytes.len() as u64 == total_length;
475        self.request.credit(
476            f.header.stream_id.as_deref().ok_or_else(w::invalid)?,
477            f.payload.len(),
478        )?;
479        if !complete {
480            return Ok(None);
481        }
482        if w::hex(&sha2::Sha256::digest(&self.bytes)) != sha256 {
483            return Err(w::invalid());
484        }
485        let value: w::Reply = serde_json::from_slice(&self.bytes).map_err(io_error)?;
486        self.bytes.clear();
487        self.length = None;
488        self.hash = None;
489        self.sequence += 1;
490        if terminal {
491            self.done = true;
492            self.request.finish();
493        }
494        Ok(Some(match value {
495            w::Reply::Error { code, message } => Err(w::core_error(&code, &message)),
496            r => Ok(r),
497        }))
498    }
499}
500impl Stream for Replies {
501    type Item = Result<w::Reply, CoreError>;
502    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
503        if self.done {
504            return Poll::Ready(None);
505        }
506        loop {
507            let frame = match self.request.poll(cx) {
508                Poll::Pending => return Poll::Pending,
509                Poll::Ready(Err(e)) => {
510                    self.done = true;
511                    return Poll::Ready(Some(Err(e)));
512                }
513                Poll::Ready(Ok(f)) => f,
514            };
515            match self.accept(frame) {
516                Ok(Some(v)) => return Poll::Ready(Some(v)),
517                Ok(None) => {}
518                Err(e) => {
519                    self.done = true;
520                    return Poll::Ready(Some(Err(e)));
521                }
522            }
523        }
524    }
525}
526
527impl WorkspaceLeaseService for RemoteServices {
528    fn register(&self, lease: WorkspaceLease) -> ServiceFuture<'_, Result<(), CoreError>> {
529        Box::pin(async move {
530            if lease != self.lease {
531                return Err(w::invalid());
532            }
533            self.validate_environment_ready()
534        })
535    }
536    fn release(
537        &self,
538        id: WorkspaceLeaseID,
539        intent: LeaseReleaseIntent,
540    ) -> ServiceFuture<'_, Result<(), CoreError>> {
541        Box::pin(async move {
542            if id != self.lease.lease {
543                return Err(w::invalid());
544            }
545            self.release_workspace(intent)
546        })
547    }
548}
buntrail_engine/remote/mod.rs

buntrail_engine/remote/
mod.rs
