supercode_harness/
runtime_lease.rs

1//! Transport-neutral observer and controller leases for live SDK runtimes.
2//!
3//! HTTP, ACP, local frontends, and compatibility adapters all project this
4//! coordinator. Transports authenticate clients, but they do not get to
5//! invent different ownership or expiry semantics.
6
7use std::collections::{BTreeMap, BTreeSet};
8use std::sync::{Arc, Mutex};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11use async_trait::async_trait;
12use serde::{Deserialize, Serialize};
13
14use crate::{
15    FrontendAttachment, FrontendOperationInvocation, FrontendOperationResult, FrontendResponse,
16    FrontendRuntimeDescriptor, SdkError, SdkOperation, SdkRuntime,
17};
18
19/// Default controller duration. Successful mutations and explicit heartbeats
20/// renew it; disconnect and detach release it immediately.
21pub const DEFAULT_RUNTIME_LEASE_TTL_MS: u64 = 30_000;
22
23/// Capability independently granted to an authenticated runtime client.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum RuntimePermission {
27    /// Discover, describe, attach, and watch canonical runtime events.
28    Observe,
29    /// Own the controller lease and mutate turn state.
30    Interact,
31    /// Answer approval or elicitation requests while controlling the runtime.
32    Approve,
33    /// Explicitly close the SDK-owned runtime.
34    Terminate,
35}
36
37impl RuntimePermission {
38    /// Stable permission spelling used by every binding and error envelope.
39    pub const fn as_str(self) -> &'static str {
40        match self {
41            Self::Observe => "observe",
42            Self::Interact => "interact",
43            Self::Approve => "approve",
44            Self::Terminate => "terminate",
45        }
46    }
47}
48
49/// Permissions carried by one authenticated runtime client.
50#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
51pub struct RuntimeAuthorization {
52    permissions: BTreeSet<RuntimePermission>,
53}
54
55impl RuntimeAuthorization {
56    /// Construct an authorization grant from its exact permission set.
57    pub fn new(permissions: impl IntoIterator<Item = RuntimePermission>) -> Self {
58        Self {
59            permissions: permissions.into_iter().collect(),
60        }
61    }
62
63    /// Full owner grant used by the process that started the runtime.
64    pub fn owner() -> Self {
65        Self::new([
66            RuntimePermission::Observe,
67            RuntimePermission::Interact,
68            RuntimePermission::Approve,
69            RuntimePermission::Terminate,
70        ])
71    }
72
73    /// Interactive frontend grant. Frontends may observe, drive turns, and
74    /// answer requests, but runtime termination remains an operator-only
75    /// control-plane action.
76    pub fn interactive() -> Self {
77        Self::new([
78            RuntimePermission::Observe,
79            RuntimePermission::Interact,
80            RuntimePermission::Approve,
81        ])
82    }
83
84    /// Read-only observer grant.
85    pub fn observer() -> Self {
86        Self::new([RuntimePermission::Observe])
87    }
88
89    /// Whether this grant contains `permission`.
90    pub fn allows(&self, permission: RuntimePermission) -> bool {
91        self.permissions.contains(&permission)
92    }
93
94    /// Stable ordered permission inventory for descriptors and bindings.
95    pub fn permissions(&self) -> impl Iterator<Item = RuntimePermission> + '_ {
96        self.permissions.iter().copied()
97    }
98
99    /// Narrow this grant to permissions also present in `requested`.
100    pub fn restrict_to(&self, requested: &Self) -> Self {
101        Self::new(
102            self.permissions
103                .intersection(&requested.permissions)
104                .copied(),
105        )
106    }
107
108    /// Stable comma-separated representation used only to narrow an already
109    /// authenticated transport credential.
110    pub fn header_value(&self) -> String {
111        self.permissions()
112            .map(RuntimePermission::as_str)
113            .collect::<Vec<_>>()
114            .join(",")
115    }
116
117    /// Decode a narrowing request. Unknown or duplicate-independent names
118    /// fail closed instead of being ignored.
119    pub fn parse_header(value: &str) -> Result<Self, RuntimeLeaseError> {
120        if value.is_empty() {
121            return Ok(Self::default());
122        }
123        let mut permissions = Vec::new();
124        for name in value.split(',') {
125            let permission = match name {
126                "observe" => RuntimePermission::Observe,
127                "interact" => RuntimePermission::Interact,
128                "approve" => RuntimePermission::Approve,
129                "terminate" => RuntimePermission::Terminate,
130                _ => return Err(RuntimeLeaseError::InvalidAuthorization),
131            };
132            permissions.push(permission);
133        }
134        Ok(Self::new(permissions))
135    }
136}
137
138/// Opaque identity scoped to one authenticated frontend connection.
139#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
140#[serde(transparent)]
141pub struct RuntimeClientId(String);
142
143impl RuntimeClientId {
144    /// Validate a transport-provided client identity.
145    pub fn parse(value: impl Into<String>) -> Result<Self, RuntimeLeaseError> {
146        let value = value.into();
147        if value.is_empty()
148            || value.len() > 128
149            || !value
150                .bytes()
151                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
152        {
153            return Err(RuntimeLeaseError::InvalidClientId);
154        }
155        Ok(Self(value))
156    }
157
158    /// Stable opaque spelling used on protocol boundaries.
159    pub fn as_str(&self) -> &str {
160        &self.0
161    }
162}
163
164impl std::fmt::Display for RuntimeClientId {
165    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
166        formatter.write_str(&self.0)
167    }
168}
169
170/// Controller lease visible through runtime describe/list operations.
171#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
172pub struct RuntimeControllerLease {
173    /// Authenticated client that owns mutation rights.
174    pub client_id: RuntimeClientId,
175    /// Absolute epoch-millisecond deadline. At or after this value the lease
176    /// is expired and another authorized client may claim it.
177    pub expires_at_ms: u64,
178}
179
180/// One attached observer and its last authenticated activity.
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
182pub struct RuntimeObserverLease {
183    /// Authenticated observing client.
184    pub client_id: RuntimeClientId,
185    /// Last epoch-millisecond activity used for stale-client reconciliation.
186    pub last_seen_ms: u64,
187}
188
189/// Serializable ownership snapshot included in registry descriptors.
190#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
191pub struct RuntimeLeaseSnapshot {
192    /// Current unexpired controller, if any.
193    pub controller: Option<RuntimeControllerLease>,
194    /// Attached observers in stable client-id order.
195    pub observers: Vec<RuntimeObserverLease>,
196    /// Configured lease duration.
197    pub lease_ttl_ms: u64,
198}
199
200/// Stable coordination failure shared by every transport projection.
201#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
202pub enum RuntimeLeaseError {
203    /// Client identity is absent or not a valid opaque identifier.
204    #[error("invalid runtime client id")]
205    InvalidClientId,
206    /// A transport attempted to request an unknown permission spelling.
207    #[error("invalid runtime authorization grant")]
208    InvalidAuthorization,
209    /// The authenticated client lacks the operation's required grant.
210    #[error("runtime permission `{0:?}` is required")]
211    Unauthorized(RuntimePermission),
212    /// Another client still owns the controller lease.
213    #[error("controller lease is held by `{holder}` until {expires_at_ms}")]
214    ControllerHeld {
215        /// Current controller.
216        holder: RuntimeClientId,
217        /// Current controller deadline.
218        expires_at_ms: u64,
219    },
220    /// A mutation was attempted without owning the controller lease.
221    #[error("controller lease required")]
222    ControllerRequired,
223    /// This client owned the lease, but its deadline has passed.
224    #[error("controller lease expired")]
225    LeaseExpired,
226}
227
228/// One-controller, many-observer coordination state for a live runtime.
229#[derive(Debug)]
230pub struct RuntimeLeaseCoordinator {
231    lease_ttl_ms: u64,
232    controller: Option<RuntimeControllerLease>,
233    observers: BTreeMap<RuntimeClientId, RuntimeObserverLease>,
234    expired_controller: Option<RuntimeClientId>,
235}
236
237impl RuntimeLeaseCoordinator {
238    /// Create a coordinator with a non-zero controller lease duration.
239    pub fn new(lease_ttl_ms: u64) -> Self {
240        assert!(lease_ttl_ms > 0, "runtime lease TTL must be non-zero");
241        Self {
242            lease_ttl_ms,
243            controller: None,
244            observers: BTreeMap::new(),
245            expired_controller: None,
246        }
247    }
248
249    /// Register or refresh one authenticated observer.
250    pub fn attach(
251        &mut self,
252        client_id: RuntimeClientId,
253        authorization: &RuntimeAuthorization,
254        now_ms: u64,
255    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
256        require(authorization, RuntimePermission::Observe)?;
257        self.reconcile(now_ms);
258        self.observers.insert(
259            client_id.clone(),
260            RuntimeObserverLease {
261                client_id,
262                last_seen_ms: now_ms,
263            },
264        );
265        Ok(self.snapshot(now_ms))
266    }
267
268    /// Refresh observation activity without changing controller ownership.
269    pub fn heartbeat(
270        &mut self,
271        client_id: &RuntimeClientId,
272        authorization: &RuntimeAuthorization,
273        now_ms: u64,
274    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
275        require(authorization, RuntimePermission::Observe)?;
276        self.reconcile(now_ms);
277        self.observers
278            .entry(client_id.clone())
279            .and_modify(|observer| observer.last_seen_ms = now_ms)
280            .or_insert_with(|| RuntimeObserverLease {
281                client_id: client_id.clone(),
282                last_seen_ms: now_ms,
283            });
284        Ok(self.snapshot(now_ms))
285    }
286
287    /// Validate a non-controller capability. Observation and lifecycle
288    /// administration are authorized independently from turn ownership;
289    /// interactive and approval mutations must use
290    /// [`Self::authorize_controller`] instead.
291    pub fn authorize(
292        &mut self,
293        authorization: &RuntimeAuthorization,
294        permission: RuntimePermission,
295        now_ms: u64,
296    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
297        require(authorization, permission)?;
298        Ok(self.snapshot(now_ms))
299    }
300
301    /// Claim the controller when unowned, or renew it when already owned by
302    /// this client. Another live controller is never displaced implicitly.
303    pub fn claim_control(
304        &mut self,
305        client_id: RuntimeClientId,
306        authorization: &RuntimeAuthorization,
307        now_ms: u64,
308    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
309        require(authorization, RuntimePermission::Interact)?;
310        self.attach(client_id.clone(), authorization, now_ms)?;
311        if self.controller.is_none() && self.expired_controller.as_ref() == Some(&client_id) {
312            return Err(RuntimeLeaseError::LeaseExpired);
313        }
314        match &self.controller {
315            Some(lease) if lease.client_id != client_id => {
316                return Err(RuntimeLeaseError::ControllerHeld {
317                    holder: lease.client_id.clone(),
318                    expires_at_ms: lease.expires_at_ms,
319                });
320            }
321            _ => {}
322        }
323        self.controller = Some(RuntimeControllerLease {
324            client_id,
325            expires_at_ms: now_ms.saturating_add(self.lease_ttl_ms),
326        });
327        self.expired_controller = None;
328        Ok(self.snapshot(now_ms))
329    }
330
331    /// Explicitly transfer control to another authorized interactive client.
332    pub fn take_control(
333        &mut self,
334        client_id: RuntimeClientId,
335        authorization: &RuntimeAuthorization,
336        now_ms: u64,
337    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
338        require(authorization, RuntimePermission::Interact)?;
339        self.attach(client_id.clone(), authorization, now_ms)?;
340        self.controller = Some(RuntimeControllerLease {
341            client_id,
342            expires_at_ms: now_ms.saturating_add(self.lease_ttl_ms),
343        });
344        self.expired_controller = None;
345        Ok(self.snapshot(now_ms))
346    }
347
348    /// Validate and renew mutation authority for `client_id`.
349    pub fn authorize_controller(
350        &mut self,
351        client_id: &RuntimeClientId,
352        authorization: &RuntimeAuthorization,
353        permission: RuntimePermission,
354        now_ms: u64,
355    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
356        require(authorization, permission)?;
357        let was_expired = self
358            .controller
359            .as_ref()
360            .is_some_and(|lease| lease.client_id == *client_id && lease.expires_at_ms <= now_ms);
361        self.reconcile(now_ms);
362        let Some(controller) = &mut self.controller else {
363            return Err(
364                if was_expired || self.expired_controller.as_ref() == Some(client_id) {
365                    RuntimeLeaseError::LeaseExpired
366                } else {
367                    RuntimeLeaseError::ControllerRequired
368                },
369            );
370        };
371        if controller.client_id != *client_id {
372            return Err(RuntimeLeaseError::ControllerRequired);
373        }
374        controller.expires_at_ms = now_ms.saturating_add(self.lease_ttl_ms);
375        self.expired_controller = None;
376        if let Some(observer) = self.observers.get_mut(client_id) {
377            observer.last_seen_ms = now_ms;
378        }
379        Ok(self.snapshot(now_ms))
380    }
381
382    /// Remove an observer and release its controller lease, if held.
383    pub fn detach(&mut self, client_id: &RuntimeClientId, now_ms: u64) -> RuntimeLeaseSnapshot {
384        self.reconcile(now_ms);
385        self.observers.remove(client_id);
386        if self
387            .controller
388            .as_ref()
389            .is_some_and(|lease| lease.client_id == *client_id)
390        {
391            self.controller = None;
392        }
393        if self.expired_controller.as_ref() == Some(client_id) {
394            self.expired_controller = None;
395        }
396        self.snapshot(now_ms)
397    }
398
399    /// Current state after deterministic expiry reconciliation.
400    pub fn snapshot(&mut self, now_ms: u64) -> RuntimeLeaseSnapshot {
401        self.reconcile(now_ms);
402        RuntimeLeaseSnapshot {
403            controller: self.controller.clone(),
404            observers: self.observers.values().cloned().collect(),
405            lease_ttl_ms: self.lease_ttl_ms,
406        }
407    }
408
409    fn reconcile(&mut self, now_ms: u64) {
410        if self
411            .controller
412            .as_ref()
413            .is_some_and(|lease| lease.expires_at_ms <= now_ms)
414        {
415            self.expired_controller = self
416                .controller
417                .take()
418                .map(|controller| controller.client_id);
419        }
420    }
421}
422
423/// Shared transport-neutral owner of a runtime and its client leases.
424///
425/// A transport authenticates a request, validates its opaque client id, and
426/// asks this object for a [`CoordinatedRuntimeClient`]. HTTP, ACP, local
427/// embedding, and compatibility adapters therefore execute identical
428/// authorization, takeover, and expiry behavior.
429pub struct CoordinatedRuntime {
430    runtime: Arc<dyn SdkRuntime>,
431    leases: Mutex<RuntimeLeaseCoordinator>,
432}
433
434impl CoordinatedRuntime {
435    /// Wrap one SDK runtime with the default controller lease duration.
436    pub fn new(runtime: Arc<dyn SdkRuntime>) -> Arc<Self> {
437        Self::with_lease_ttl(runtime, DEFAULT_RUNTIME_LEASE_TTL_MS)
438    }
439
440    /// Wrap one SDK runtime with an explicit lease duration. Tests and
441    /// embedders use this to exercise deterministic expiry without sleeping.
442    pub fn with_lease_ttl(runtime: Arc<dyn SdkRuntime>, lease_ttl_ms: u64) -> Arc<Self> {
443        Arc::new(Self {
444            runtime,
445            leases: Mutex::new(RuntimeLeaseCoordinator::new(lease_ttl_ms)),
446        })
447    }
448
449    /// Bind an authenticated identity and exact permission set to this
450    /// runtime. The returned client never owns the underlying runtime.
451    pub fn client(
452        self: &Arc<Self>,
453        client_id: RuntimeClientId,
454        authorization: RuntimeAuthorization,
455    ) -> Arc<CoordinatedRuntimeClient> {
456        Arc::new(CoordinatedRuntimeClient {
457            coordinator: self.clone(),
458            client_id,
459            authorization,
460        })
461    }
462
463    fn leases(&self) -> std::sync::MutexGuard<'_, RuntimeLeaseCoordinator> {
464        self.leases
465            .lock()
466            .unwrap_or_else(std::sync::PoisonError::into_inner)
467    }
468}
469
470/// Authenticated client projection of one coordinated SDK runtime.
471pub struct CoordinatedRuntimeClient {
472    coordinator: Arc<CoordinatedRuntime>,
473    client_id: RuntimeClientId,
474    authorization: RuntimeAuthorization,
475}
476
477impl CoordinatedRuntimeClient {
478    /// Opaque authenticated identity used by lease snapshots and transports.
479    pub fn client_id(&self) -> &RuntimeClientId {
480        &self.client_id
481    }
482
483    /// Attach or refresh this observer and return current ownership state.
484    pub fn observe(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
485        self.coordinator
486            .leases()
487            .attach(self.client_id.clone(), &self.authorization, epoch_ms())
488            .map_err(|error| lease_sdk_error(error, SdkOperation::Events))
489    }
490
491    /// Explicitly take the controller lease from another interactive client.
492    pub fn take_control(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
493        self.coordinator
494            .leases()
495            .take_control(self.client_id.clone(), &self.authorization, epoch_ms())
496            .map_err(|error| lease_sdk_error(error, SdkOperation::Input))
497    }
498
499    /// Acquire control only when it is unowned (or already ours). A live
500    /// controller is never displaced by this operation.
501    pub fn acquire_control(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
502        self.coordinator
503            .leases()
504            .claim_control(self.client_id.clone(), &self.authorization, epoch_ms())
505            .map_err(|error| lease_sdk_error(error, SdkOperation::Input))
506    }
507
508    /// Refresh this client's observer activity and controller lease when it
509    /// currently owns control.
510    pub fn heartbeat(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
511        let now_ms = epoch_ms();
512        let mut leases = self.coordinator.leases();
513        let mut snapshot = leases
514            .heartbeat(&self.client_id, &self.authorization, now_ms)
515            .map_err(|error| lease_sdk_error(error, SdkOperation::Events))?;
516        if snapshot
517            .controller
518            .as_ref()
519            .is_some_and(|lease| lease.client_id == self.client_id)
520        {
521            snapshot = leases
522                .authorize_controller(
523                    &self.client_id,
524                    &self.authorization,
525                    RuntimePermission::Interact,
526                    now_ms,
527                )
528                .map_err(|error| lease_sdk_error(error, SdkOperation::Input))?;
529        }
530        Ok(snapshot)
531    }
532
533    /// Release observation and controller ownership without closing the
534    /// SDK-owned runtime.
535    pub fn detach(&self) -> RuntimeLeaseSnapshot {
536        self.coordinator
537            .leases()
538            .detach(&self.client_id, epoch_ms())
539    }
540
541    /// Current ownership state, visible only to observers.
542    pub fn lease_snapshot(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
543        self.coordinator
544            .leases()
545            .authorize(&self.authorization, RuntimePermission::Observe, epoch_ms())
546            .map_err(|error| lease_sdk_error(error, SdkOperation::Events))
547    }
548
549    fn authorize_controller(
550        &self,
551        permission: RuntimePermission,
552        operation: SdkOperation,
553    ) -> Result<(), SdkError> {
554        let now_ms = epoch_ms();
555        let mut leases = self.coordinator.leases();
556        leases
557            .claim_control(self.client_id.clone(), &self.authorization, now_ms)
558            .map_err(|error| lease_sdk_error(error, operation))?;
559        if permission != RuntimePermission::Interact {
560            leases
561                .authorize_controller(&self.client_id, &self.authorization, permission, now_ms)
562                .map_err(|error| lease_sdk_error(error, operation))?;
563        }
564        Ok(())
565    }
566
567    fn authorize_lifecycle(
568        &self,
569        permission: RuntimePermission,
570        operation: SdkOperation,
571    ) -> Result<(), SdkError> {
572        self.coordinator
573            .leases()
574            .authorize(&self.authorization, permission, epoch_ms())
575            .map(|_| ())
576            .map_err(|error| lease_sdk_error(error, operation))
577    }
578
579    async fn descriptor(&self) -> Result<FrontendRuntimeDescriptor, SdkError> {
580        let mut descriptor = self.coordinator.runtime.describe().await?;
581        descriptor.actions.submit &= self.authorization.allows(RuntimePermission::Interact);
582        descriptor.actions.interrupt &= self.authorization.allows(RuntimePermission::Interact);
583        descriptor.actions.steer &= self.authorization.allows(RuntimePermission::Interact);
584        descriptor.actions.respond &= self.authorization.allows(RuntimePermission::Approve)
585            && self.authorization.allows(RuntimePermission::Interact);
586        descriptor.actions.close &= self.authorization.allows(RuntimePermission::Terminate);
587        descriptor.actions.detach &= self.authorization.allows(RuntimePermission::Observe);
588        Ok(descriptor)
589    }
590}
591
592#[async_trait]
593impl SdkRuntime for CoordinatedRuntimeClient {
594    async fn describe(&self) -> Result<FrontendRuntimeDescriptor, SdkError> {
595        self.authorize_lifecycle(RuntimePermission::Observe, SdkOperation::Events)?;
596        self.descriptor().await
597    }
598
599    async fn attach(&self, history_limit: usize) -> Result<FrontendAttachment, SdkError> {
600        self.observe()?;
601        let mut attachment = self.coordinator.runtime.attach(history_limit).await?;
602        attachment.descriptor = self.descriptor().await?;
603        Ok(attachment)
604    }
605
606    async fn send_input(self: Arc<Self>, prompt: String) -> Result<(), SdkError> {
607        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
608        self.coordinator.runtime.clone().send_input(prompt).await
609    }
610
611    async fn send_input_with_images(
612        self: Arc<Self>,
613        prompt: String,
614        image_urls: Vec<String>,
615    ) -> Result<(), SdkError> {
616        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
617        self.coordinator
618            .runtime
619            .clone()
620            .send_input_with_images(prompt, image_urls)
621            .await
622    }
623
624    async fn submit(&self, prompt: String) -> Result<String, SdkError> {
625        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
626        self.coordinator.runtime.submit(prompt).await
627    }
628
629    async fn submit_with_images(
630        &self,
631        prompt: String,
632        image_urls: Vec<String>,
633    ) -> Result<String, SdkError> {
634        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
635        self.coordinator
636            .runtime
637            .submit_with_images(prompt, image_urls)
638            .await
639    }
640
641    async fn interrupt(&self) -> Result<bool, SdkError> {
642        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Interrupt)?;
643        self.coordinator.runtime.interrupt().await
644    }
645
646    async fn steer(&self, prompt: String) -> Result<(), SdkError> {
647        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Steer)?;
648        self.coordinator.runtime.steer(prompt).await
649    }
650
651    async fn respond(&self, response: FrontendResponse) -> Result<(), SdkError> {
652        self.authorize_controller(RuntimePermission::Approve, SdkOperation::Respond)?;
653        self.coordinator.runtime.respond(response).await
654    }
655
656    async fn invoke(
657        &self,
658        operation: FrontendOperationInvocation,
659    ) -> Result<FrontendOperationResult, SdkError> {
660        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
661        self.coordinator.runtime.invoke(operation).await
662    }
663
664    async fn lease_snapshot(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
665        CoordinatedRuntimeClient::lease_snapshot(self)
666    }
667
668    async fn take_control(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
669        CoordinatedRuntimeClient::take_control(self)
670    }
671
672    async fn acquire_control(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
673        CoordinatedRuntimeClient::acquire_control(self)
674    }
675
676    async fn heartbeat(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
677        CoordinatedRuntimeClient::heartbeat(self)
678    }
679
680    async fn detach(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
681        Ok(CoordinatedRuntimeClient::detach(self))
682    }
683
684    async fn close(&self) -> Result<(), SdkError> {
685        self.authorize_lifecycle(RuntimePermission::Terminate, SdkOperation::Close)?;
686        self.coordinator.runtime.close().await
687    }
688}
689
690fn lease_sdk_error(error: RuntimeLeaseError, operation: SdkOperation) -> SdkError {
691    match error {
692        error @ (RuntimeLeaseError::InvalidClientId | RuntimeLeaseError::InvalidAuthorization) => {
693            SdkError::InvalidArgument {
694                operation,
695                message: error.to_string(),
696            }
697        }
698        RuntimeLeaseError::Unauthorized(permission) => SdkError::Unauthorized {
699            permission: permission.as_str().into(),
700        },
701        RuntimeLeaseError::ControllerHeld {
702            holder,
703            expires_at_ms,
704        } => SdkError::ControllerRequired {
705            holder: Some(holder.to_string()),
706            expires_at_ms: Some(expires_at_ms),
707        },
708        RuntimeLeaseError::ControllerRequired => SdkError::ControllerRequired {
709            holder: None,
710            expires_at_ms: None,
711        },
712        RuntimeLeaseError::LeaseExpired => SdkError::LeaseExpired,
713    }
714}
715
716fn epoch_ms() -> u64 {
717    SystemTime::now()
718        .duration_since(UNIX_EPOCH)
719        .unwrap_or_default()
720        .as_millis()
721        .min(u64::MAX as u128) as u64
722}
723
724fn require(
725    authorization: &RuntimeAuthorization,
726    permission: RuntimePermission,
727) -> Result<(), RuntimeLeaseError> {
728    if authorization.allows(permission) {
729        Ok(())
730    } else {
731        Err(RuntimeLeaseError::Unauthorized(permission))
732    }
733}
734
735#[cfg(test)]
736mod tests {
737    use super::*;
738
739    fn client(value: &str) -> RuntimeClientId {
740        RuntimeClientId::parse(value).unwrap()
741    }
742
743    #[test]
744    fn many_observers_share_one_explicit_controller() {
745        let mut leases = RuntimeLeaseCoordinator::new(100);
746        let owner = RuntimeAuthorization::owner();
747        let observer = RuntimeAuthorization::observer();
748        leases.attach(client("viewer-a"), &observer, 10).unwrap();
749        leases.attach(client("viewer-b"), &observer, 11).unwrap();
750        let snapshot = leases.claim_control(client("owner"), &owner, 12).unwrap();
751        assert_eq!(snapshot.observers.len(), 3);
752        assert_eq!(snapshot.controller.unwrap().client_id, client("owner"));
753
754        let error = leases
755            .claim_control(client("viewer-b"), &owner, 13)
756            .unwrap_err();
757        assert!(matches!(error, RuntimeLeaseError::ControllerHeld { .. }));
758    }
759
760    #[test]
761    fn observer_cannot_claim_control_approve_or_terminate() {
762        let mut leases = RuntimeLeaseCoordinator::new(100);
763        let observer = RuntimeAuthorization::observer();
764        leases.attach(client("viewer"), &observer, 1).unwrap();
765        assert_eq!(
766            leases
767                .claim_control(client("viewer"), &observer, 2)
768                .unwrap_err(),
769            RuntimeLeaseError::Unauthorized(RuntimePermission::Interact)
770        );
771        assert_eq!(
772            leases
773                .authorize_controller(&client("viewer"), &observer, RuntimePermission::Approve, 2)
774                .unwrap_err(),
775            RuntimeLeaseError::Unauthorized(RuntimePermission::Approve)
776        );
777        assert_eq!(
778            leases
779                .authorize(&observer, RuntimePermission::Terminate, 2)
780                .unwrap_err(),
781            RuntimeLeaseError::Unauthorized(RuntimePermission::Terminate)
782        );
783    }
784
785    #[test]
786    fn expiry_is_deterministic_and_requires_a_new_claim() {
787        let mut leases = RuntimeLeaseCoordinator::new(10);
788        let owner = RuntimeAuthorization::owner();
789        leases.claim_control(client("a"), &owner, 5).unwrap();
790        assert_eq!(
791            leases.claim_control(client("a"), &owner, 15).unwrap_err(),
792            RuntimeLeaseError::LeaseExpired
793        );
794        assert_eq!(
795            leases
796                .authorize_controller(&client("a"), &owner, RuntimePermission::Interact, 15)
797                .unwrap_err(),
798            RuntimeLeaseError::LeaseExpired
799        );
800        let snapshot = leases.claim_control(client("b"), &owner, 15).unwrap();
801        assert_eq!(snapshot.controller.unwrap().client_id, client("b"));
802    }
803
804    #[test]
805    fn successful_mutation_renews_controller_and_observer_activity() {
806        let mut leases = RuntimeLeaseCoordinator::new(10);
807        let owner = RuntimeAuthorization::owner();
808        leases.claim_control(client("a"), &owner, 5).unwrap();
809        let snapshot = leases
810            .authorize_controller(&client("a"), &owner, RuntimePermission::Approve, 9)
811            .unwrap();
812        assert_eq!(snapshot.controller.unwrap().expires_at_ms, 19);
813        assert_eq!(snapshot.observers[0].last_seen_ms, 9);
814    }
815
816    #[test]
817    fn takeover_and_detach_are_explicit_and_release_control() {
818        let mut leases = RuntimeLeaseCoordinator::new(10);
819        let owner = RuntimeAuthorization::owner();
820        leases.claim_control(client("a"), &owner, 1).unwrap();
821        let snapshot = leases.take_control(client("b"), &owner, 2).unwrap();
822        assert_eq!(snapshot.controller.unwrap().client_id, client("b"));
823        let snapshot = leases.detach(&client("b"), 3);
824        assert!(snapshot.controller.is_none());
825        assert_eq!(
826            snapshot
827                .observers
828                .into_iter()
829                .map(|observer| observer.client_id)
830                .collect::<Vec<_>>(),
831            vec![client("a")]
832        );
833    }
834
835    #[test]
836    fn client_ids_are_opaque_bounded_and_header_safe() {
837        for invalid in ["", "space here", "slash/here", "💥"] {
838            assert_eq!(
839                RuntimeClientId::parse(invalid).unwrap_err(),
840                RuntimeLeaseError::InvalidClientId
841            );
842        }
843        assert_eq!(
844            RuntimeClientId::parse("a".repeat(129)).unwrap_err(),
845            RuntimeLeaseError::InvalidClientId
846        );
847        assert_eq!(
848            RuntimeClientId::parse("client-1.v2_ok").unwrap().as_str(),
849            "client-1.v2_ok"
850        );
851        let owner = RuntimeAuthorization::owner();
852        let requested = RuntimeAuthorization::parse_header("observe,interact").unwrap();
853        assert_eq!(
854            owner.restrict_to(&requested).header_value(),
855            "observe,interact"
856        );
857        assert_eq!(
858            RuntimeAuthorization::parse_header("observe,admin").unwrap_err(),
859            RuntimeLeaseError::InvalidAuthorization
860        );
861    }
862}
supercode_harness/runtime_lease.rs

supercode_harness/
runtime_lease.rs
