rustev-core 0.2.0

//! Seam traits (spec 002, 3.1.2, amended by spec 003, 3.2.6; design section
//! 8), declared as signatures over `std::future` only. The runtime (spec 003)
//! composes them. They are transport-neutral; network implementations live
//! under `integrations/` (spec 001, 3.4.3).
//!
//! Nothing here reads a clock, performs I/O or spawns: time reaches a seam as
//! a remaining duration, and cancellation as a [`CancelSignal`].

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::task::{Context, Poll, Waker};

use rustev_contract::acquisition::AcquisitionRequest;
use rustev_contract::descriptor::BackendDescriptor;
use rustev_contract::dynamic::{DynamicCapability, DynamicOutcome, DynamicRawOutput, TemplateRef};
use rustev_contract::eval_report::EvalReport;
use rustev_contract::output::RawOutput;
use rustev_contract::remote::ServedIdentity;
use rustev_contract::run::{Charge, CostBound, CostModel, RunRecord};
use rustev_contract::snapshot::Snapshot;
use rustev_contract::source::SourceDescriptor;
use rustev_contract::time::DurationMs;

/// A boxed future, so the traits stay object-safe without an executor.
pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Passed to every seam call. Time is a value, not a clock read.
#[derive(Debug, Clone)]
pub struct CallContext {
    /// Time left before the decision's deadline when the call was made, by
    /// the runtime's monotonic clock.
    pub remaining_ms: DurationMs,
    pub trace_id: String,
    /// Opaque to Rustev: passed through for the host's authorization and
    /// cache isolation, never interpreted as authority (spec 001, 3.2).
    pub principal_handle: Vec<u8>,
}

/// Supplies an authorized context snapshot. Returns only what the caller may
/// read; a missing field is reported missing, never filled.
pub trait ContextSource: Send + Sync {
    fn fetch<'a>(
        &'a self,
        fields: &'a [String],
        cx: &'a CallContext,
    ) -> BoxFuture<'a, Result<Snapshot, String>>;
}

/// One dispatched attempt (spec 003, 3.2.6).
pub struct AttemptCall<'a> {
    /// Canonical projection, exactly as the core listed it.
    pub projection: &'a [u8],
    /// Stable across timing; usable as an idempotency key.
    pub attempt_id: &'a str,
    /// Raised on caller cancellation, deadline or attempt timeout.
    pub cancel: CancelSignal,
    pub cx: &'a CallContext,
}

/// How an adapter says an attempt failed. Invalid outputs are judged by the
/// core, not reported here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AdapterFailure {
    /// May not recur.
    Transient,
    /// Refused for load.
    Overloaded,
    /// Will recur for this input.
    Permanent,
    /// Stopped because the attempt's cancellation signal was raised.
    Cancelled,
}

/// The adapter's answer to a raised cancellation signal.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CancelAck {
    /// No cancellation was observed.
    NotRequested,
    /// Work and charges stopped; the adapter can establish this.
    Stopped,
    /// The adapter stopped waiting but cannot establish a remote stop.
    Unconfirmed,
}

/// Whether remote work may outlive a report the adapter returned of its own
/// accord (spec 013, 3.1).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RemoteEnd {
    /// The remote work, if any, is over as far as the adapter can establish.
    /// Every in-process backend reports this.
    Finished,
    /// Request bytes left the process and nothing establishes that the remote
    /// side stopped.
    PossiblyContinuing,
}

/// What one attempt produced, what it cost and how cancellation went.
#[derive(Debug, Clone, PartialEq)]
pub struct AttemptReport {
    pub result: Result<RawOutput, (AdapterFailure, String)>,
    pub charge: Charge,
    pub cancel: CancelAck,
    /// Read only for a report returned of the adapter's own accord; the
    /// cancellation path decides from `cancel` (spec 013, 3.2).
    pub remote: RemoteEnd,
}

/// A backend's explicit declaration that one projection may participate in a
/// general runtime batch. Equal compatibility identifiers mean the backend
/// promises compatible semantics. Work units are backend-defined and are
/// used only against the plan's hard `max_backend_work` bound.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BatchCapability {
    pub compatibility: String,
    pub work_units: u64,
}

/// One owned member of a general runtime batch.
#[derive(Debug, Clone)]
pub struct BatchMemberCall {
    pub projection: Vec<u8>,
    pub attempt_id: String,
    pub cancel: CancelSignal,
    pub cx: CallContext,
}

/// One general batch dispatch. Member order is stable and evidence-visible.
pub struct BatchAttemptCall {
    pub batch_id: String,
    pub cancel: CancelSignal,
    pub members: Vec<BatchMemberCall>,
}

/// A member result is keyed by attempt id. Duplicate, unknown, or missing ids
/// are malformed for the affected member and never fill another member.
#[derive(Debug, Clone, PartialEq)]
pub struct BatchMemberReport {
    pub attempt_id: String,
    pub result: Result<RawOutput, (AdapterFailure, String)>,
}

/// One backend-wide batch report. A top-level failure is attributed to every
/// dispatched member. The batch charge is allocated by the plan policy.
#[derive(Debug, Clone, PartialEq)]
pub struct BatchAttemptReport {
    pub result: Result<Vec<BatchMemberReport>, (AdapterFailure, String)>,
    pub charge: Charge,
    pub cancel: CancelAck,
    pub remote: RemoteEnd,
}

/// Receives complete candidate fields during a generation attempt. The runtime
/// validates each field before exposing it. Core never invokes this seam.
pub trait GenerationFields: Send + Sync {
    fn field(&self, name: &str, value: rustev_contract::output::TextValue);
}

/// Answers semantic requests. States its capabilities and its cost; never
/// manufactures a distribution or a calibration claim.
pub trait DecisionBackend: Send + Sync {
    fn descriptor(&self) -> &BackendDescriptor;
    /// Static disclosure: whether every call has an enforceable upper bound.
    fn cost_model(&self) -> CostModel;
    /// Per-call disclosure before dispatch. A `bounded` model returns
    /// `bounded`; fractional costs are rounded up.
    fn cost_bound(&self, projection: &[u8]) -> CostBound;
    fn infer<'a>(&'a self, call: AttemptCall<'a>) -> BoxFuture<'a, AttemptReport>;

    /// Optional generation streaming. Existing backends return their ordinary
    /// complete report without emitting provisional fields.
    fn infer_generation<'a>(
        &'a self,
        call: AttemptCall<'a>,
        _fields: &'a dyn GenerationFields,
    ) -> BoxFuture<'a, AttemptReport> {
        self.infer(call)
    }

    /// Opt-in general batch declaration. `None` preserves unbatched behavior.
    fn batch_capability(&self, _projection: &[u8]) -> Option<BatchCapability> {
        None
    }

    /// The remote evidence of an attempt this backend ran, for a
    /// `rustev.run/2` record (spec 020, 3.2), taken once: the adapter's
    /// part of the entry (exchange reference, identities, binding, code,
    /// charge provenance); the runtime sets the step, instance and final
    /// remote state. `None`, the default, for an in-process backend.
    fn take_remote_evidence(
        &self,
        _attempt_id: &str,
    ) -> Option<rustev_contract::run::RemoteEvidence> {
        None
    }

    /// Dispatch a declared-compatible batch. The default is unreachable for
    /// a backend that keeps the default `batch_capability` implementation.
    fn infer_batch<'a>(&'a self, _call: BatchAttemptCall) -> BoxFuture<'a, BatchAttemptReport> {
        Box::pin(async {
            BatchAttemptReport {
                result: Err((
                    AdapterFailure::Permanent,
                    "backend did not implement declared batch semantics".into(),
                )),
                charge: Charge::Unknown,
                cancel: CancelAck::NotRequested,
                remote: RemoteEnd::PossiblyContinuing,
            }
        })
    }
}

/// One dynamic question's single attempt (spec 019, 3.5).
pub struct DynamicCall<'a> {
    /// The canonical `DynamicProjection` bytes: data only.
    pub projection: &'a [u8],
    /// The template the capability declared; the adapter maps the
    /// projection through it and nothing else.
    pub template: &'a TemplateRef,
    pub attempt_id: &'a str,
    /// Raised on caller cancellation or the deadline.
    pub cancel: CancelSignal,
    pub cx: &'a CallContext,
}

/// What a dynamic attempt produced, who served it, what it cost and how
/// cancellation went. Invalid answers are judged by the core.
#[derive(Debug, Clone, PartialEq)]
pub struct DynamicReport {
    pub result: Result<DynamicRawOutput, (AdapterFailure, String)>,
    /// The served identity as the adapter can state it; `unknown` is a
    /// value, never replaced by the requested artifact.
    pub served: ServedIdentity,
    pub charge: Charge,
    pub cancel: CancelAck,
    pub remote: RemoteEnd,
}

/// Answers dynamic questions (spec 019, 3.4). Separate from
/// [`DecisionBackend`]: implementing that trait never makes a backend
/// dynamic, and a backend answers only when it returns a dynamic
/// capability. It is offered no tool, function or retrieval surface.
pub trait DynamicBackend: Send + Sync {
    fn backend_id(&self) -> &str;
    /// The opt-in declaration, or `None`: then every request is refused
    /// before dispatch.
    fn dynamic_capability(&self) -> Option<&DynamicCapability>;
    /// Per-call disclosure before dispatch; only `bounded` is admitted.
    fn cost_bound(&self, projection: &[u8]) -> CostBound;
    fn ask<'a>(&'a self, call: DynamicCall<'a>) -> BoxFuture<'a, DynamicReport>;
}

/// Receives dynamic outcomes (spec 019, 3.6). A receipt establishes that
/// the sink accepted the outcome under its own durability terms, nothing
/// more.
pub trait DynamicEvidenceSink: Send + Sync {
    fn deliver<'a>(&'a self, outcome: &'a DynamicOutcome) -> BoxFuture<'a, Result<String, String>>;
}

/// A task adapter producing an evaluation report envelope (R-06).
pub trait Evaluator: Send + Sync {
    fn evaluate<'a>(&'a self, cx: &'a CallContext) -> BoxFuture<'a, EvalReport>;
}

/// Receives run records (spec 003, 3.9). A receipt establishes that the sink
/// accepted the record for its decision id under the sink's own durability
/// terms, nothing more.
pub trait EvidenceSink: Send + Sync {
    fn deliver<'a>(&'a self, record: &'a RunRecord) -> BoxFuture<'a, Result<String, String>>;

    /// Whether the sink accepts run records of `schema` (spec 032, 3.11.2).
    /// A sink accepts only `rustev.run/1` unless it says otherwise.
    fn accepts(&self, schema: &str) -> bool {
        schema == rustev_contract::schema::RUN
    }

    /// The one run-record schema the sink selected at construction for a
    /// plan 2 decision (spec 020, 2 and 3.2): `rustev.run/1`, the default,
    /// or `rustev.run/2`. A sink that selects version 2 is never given
    /// version 1 bytes in its place, and a rejected version 2 record is
    /// never retried as version 1.
    fn run_schema(&self) -> &'static str {
        rustev_contract::schema::RUN
    }
}

/// A one-way, clonable cancellation signal with children, using `std` only.
/// Raising a signal raises every child created from it; a child can be
/// raised alone.
#[derive(Clone, Default)]
pub struct CancelSignal(Arc<Inner>);

#[derive(Default)]
struct Inner {
    raised: AtomicBool,
    state: Mutex<State>,
}

#[derive(Default)]
struct State {
    wakers: Vec<Waker>,
    children: Vec<Weak<Inner>>,
}

impl std::fmt::Debug for CancelSignal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_tuple("CancelSignal")
            .field(&self.is_raised())
            .finish()
    }
}

fn lock(m: &Mutex<State>) -> std::sync::MutexGuard<'_, State> {
    m.lock().unwrap_or_else(|e| e.into_inner())
}

impl Inner {
    fn raise(&self) {
        if self.raised.swap(true, Ordering::SeqCst) {
            return;
        }
        let (wakers, children) = {
            let mut s = lock(&self.state);
            (
                std::mem::take(&mut s.wakers),
                std::mem::take(&mut s.children),
            )
        };
        for w in wakers {
            w.wake();
        }
        for c in children.iter().filter_map(Weak::upgrade) {
            c.raise();
        }
    }
}

impl CancelSignal {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn raise(&self) {
        self.0.raise();
    }

    pub fn is_raised(&self) -> bool {
        self.0.raised.load(Ordering::SeqCst)
    }

    /// A signal raised when this one is, or on its own.
    pub fn child(&self) -> CancelSignal {
        let child = CancelSignal::new();
        {
            let mut s = lock(&self.0.state);
            if !self.is_raised() {
                s.children.retain(|w| w.strong_count() > 0);
                s.children.push(Arc::downgrade(&child.0));
                return child;
            }
        }
        child.raise();
        child
    }

    /// Ready once raised; registers the task's waker otherwise.
    pub fn poll_raised(&self, cx: &mut Context<'_>) -> Poll<()> {
        if self.is_raised() {
            return Poll::Ready(());
        }
        let mut s = lock(&self.0.state);
        if self.is_raised() {
            return Poll::Ready(());
        }
        if !s.wakers.iter().any(|w| w.will_wake(cx.waker())) {
            s.wakers.push(cx.waker().clone());
        }
        Poll::Pending
    }

    /// Wakers and live children registered on this signal, for diagnostics:
    /// a signal shared across many waiters must not accumulate them.
    pub fn registered(&self) -> usize {
        let s = lock(&self.0.state);
        s.wakers.len() + s.children.iter().filter(|w| w.strong_count() > 0).count()
    }

    /// Resolves once raised.
    pub fn raised(&self) -> impl Future<Output = ()> + Send + '_ {
        std::future::poll_fn(move |cx| self.poll_raised(cx))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn raising_a_parent_raises_its_children_and_not_the_reverse() {
        let parent = CancelSignal::new();
        let a = parent.child();
        let b = parent.child();
        b.raise();
        assert!(b.is_raised() && !a.is_raised() && !parent.is_raised());
        parent.raise();
        assert!(a.is_raised());
        let late = parent.child();
        assert!(late.is_raised(), "a child of a raised signal starts raised");
    }
}

/// An acquisition adapter. Dispatch belongs to the caller, never to the core.
/// Its future uses only std and has no executor or runtime dependency.
pub trait AcquisitionSource: Send + Sync {
    fn descriptor(&self) -> &SourceDescriptor;
    fn cost_bound(&self, request: &AcquisitionRequest) -> CostBound;
    fn acquire<'a>(&'a self, call: AcquireCall<'a>) -> BoxFuture<'a, AcquireReport>;
}

/// One acquisition attempt, with the core's canonical request and identity.
pub struct AcquireCall<'a> {
    pub operation: &'a str,
    pub request: &'a rustev_contract::acquisition::AcquisitionRequest,
    pub request_id: &'a rustev_contract::ids::AcquisitionRequestId,
    pub attempt_id: &'a str,
    pub cancel: CancelSignal,
    pub cx: &'a CallContext,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AcquisitionFailure {
    Transient,
    Overloaded,
    Permanent,
    Denied,
}

/// Raw page bytes are always validated by the core before becoming values.
#[derive(Debug, Clone, PartialEq)]
pub struct AcquireReport {
    pub result: Result<Vec<u8>, (AcquisitionFailure, String)>,
    pub charge: Charge,
    pub cancel: CancelAck,
}