liminal_server/auth_lys/
drafts.rs

1//! Atomic target receipts accompany explicit, authenticated connector pulls.
2//!
3//! Connector pages omit approving people; their authority is recorded as `lys`
4//! with the draft id, never as a person inferred from the requesting holder.
5
6use std::collections::BTreeSet;
7use std::num::NonZeroUsize;
8
9use lys_pass::drafts::{ApprovedDraft, Execution, Executor};
10use serde::{Deserialize, Serialize};
11
12#[path = "../http/admin_drafts.rs"]
13pub mod admin;
14#[path = "grant_stream.rs"]
15mod pulls;
16#[path = "../agents/waiting.rs"]
17pub mod waiting;
18pub use admin::DraftPuller;
19pub use projection::AgentProjection;
20pub use pulls::AgentDraftStore;
21pub use target::TargetDraftStore;
22#[path = "drafts_projection.rs"]
23mod projection;
24#[path = "drafts_target.rs"]
25mod target;
26
27/// Explicit resource limits for a single requested pull.
28#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
29#[serde(deny_unknown_fields)]
30pub struct DraftLimits {
31    /// Bounds pages traversed by one caller so a moving population cannot
32    /// hold the administration lane indefinitely. No timer ends a pull.
33    pub pages: NonZeroUsize,
34    /// Bounds distinct draft ids handled per pull, including replays; nothing is evicted
35    /// from the durable receipt ledger when the limit is reached.
36    pub drafts: NonZeroUsize,
37    /// Bounds authenticated approver identities before the transition is kept;
38    /// this must accommodate the grant's required number of distinct people.
39    pub approvers: NonZeroUsize,
40    /// Bounds exact immutable words before any execution or retained-state write.
41    pub words_bytes: NonZeroUsize,
42    /// Bounds ids, grant ids, connector credentials and opaque page cursors.
43    pub metadata_bytes: NonZeroUsize,
44}
45
46/// Approval evidence already authenticated by Lys, bound to the exact words.
47#[derive(Clone, Serialize, Deserialize)]
48#[serde(deny_unknown_fields)]
49pub struct ApprovalEvidence {
50    /// The immutable draft whose approvals were read from Lys.
51    pub draft: String,
52    /// The immutable request approved by the named people.
53    pub digest: String,
54    /// Actual approvers when served, otherwise the explicit issuer marker `lys`.
55    pub approved_by: Vec<String>,
56    /// Prepared held mode; an issuer-only refusal uses one authority marker.
57    pub mode: super::super::GrantMode,
58}
59
60/// Stored in the same durable operation as the target mutation and receipt.
61#[derive(Clone, Serialize, Deserialize)]
62#[serde(deny_unknown_fields)]
63pub struct ApprovalTransition {
64    /// The Lys draft that authorized this transition.
65    pub draft: String,
66    /// SHA-256 of the exact approved words.
67    pub digest: String,
68    /// Served approving people or `lys`; the accompanying draft identifies the issuer evidence.
69    pub approved_by: Vec<String>,
70    /// Milliseconds since the epoch, supplied by the server's operation clock.
71    pub executed_at_ms: u64,
72}
73
74/// The existing registry's atomic target interface; it owns all prepared bytes.
75///
76/// Implementations must bind draft id, grant, kind, id, action and digest to the
77/// stored prepared act. They revalidate the current target and exact words under
78/// the commit hold, and store a named refusal for missing/retired targets.
79/// Each owning registry executes its typed prepared command through this seam.
80/// The human words never select an unapproved dispatcher or command.
81pub trait AtomicDraftStore: Send {
82    /// Reads a committed receipt by immutable draft id.
83    ///
84    /// # Errors
85    /// Durable ledger read failure.
86    fn receipt(&self, draft: &str) -> Result<Option<Execution>, DraftError>;
87
88    /// Reads authenticated approver evidence for this exact draft and digest.
89    ///
90    /// # Errors
91    /// Missing evidence or durable read failure; the act remains unapplied.
92    fn approval(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError>;
93
94    /// One durable operation applies the prepared target mutation, transition,
95    /// receipt and outcome outbox, or commits a named refusal without mutation.
96    /// Concurrent/replayed ids answer the first receipt. A stored digest,
97    /// target, grant or words mismatch never applies a new mutation.
98    ///
99    /// # Errors
100    /// Storage or execution failure. Business refusals are durable `Execution`
101    /// values so another pull never attempts the rejected act again.
102    fn commit(
103        &mut self,
104        draft: &ApprovedDraft,
105        transition: &ApprovalTransition,
106    ) -> Result<Execution, DraftError>;
107
108    /// Whether Lys acknowledged this receipt; the marker survives restart.
109    ///
110    /// # Errors
111    /// Durable ledger read failure.
112    fn acknowledged(&self, draft: &str) -> Result<bool, DraftError>;
113
114    /// Keeps the successful acknowledgment without changing the target again.
115    /// A marker can be set only when `receipt` matches the committed outcome.
116    ///
117    /// # Errors
118    /// Storage failure or a mismatching receipt.
119    fn acknowledge(&mut self, draft: &str, receipt: &Execution) -> Result<(), DraftError>;
120
121    /// Reads at most `limit` locally committed but unacknowledged drafts.
122    /// A close whose response was lost may already be absent from Lys's page.
123    ///
124    /// # Errors
125    /// Durable outbox read failure; no uncommitted act is returned here.
126    fn pending_acknowledgments(
127        &self,
128        limit: NonZeroUsize,
129    ) -> Result<Vec<ApprovedDraft>, DraftError>;
130
131    /// Delivers the committed outbox through the retained state writer and
132    /// inbox with server provenance. The same operation is idempotent after
133    /// restart. An unrelated holder is never rebound from a request's name.
134    ///
135    /// # Errors
136    /// State or inbox delivery failure; the receipt remains committed.
137    fn deliver_outcome(
138        &mut self,
139        draft: &ApprovedDraft,
140        receipt: &Execution,
141    ) -> Result<(), DraftError>;
142}
143
144/// Named failures contain no token, prepared words or HTTP response body.
145#[derive(Debug, thiserror::Error)]
146pub enum DraftError {
147    /// Lys's shared consumer refused the HTTP contract by its own name.
148    #[error("draft_lys_failed: {0}")]
149    Lys(String),
150    /// A configured positive byte, page or count bound was exceeded.
151    #[error("draft_bound_exceeded")]
152    Bound,
153    /// A continuation revisited an earlier cursor.
154    #[error("draft_cursor_repeated")]
155    CursorRepeated,
156    /// The same id was presented again in one pull.
157    #[error("draft_id_repeated")]
158    IdRepeated,
159    /// No authenticated approver evidence can be read for these exact words.
160    #[error("draft_approval_evidence_missing")]
161    ApprovalUnavailable,
162    /// The durable receipt describes another immutable request.
163    #[error("draft_receipt_digest_mismatch")]
164    ReceiptMismatch,
165    /// A target or receipt could not be read, committed or published.
166    #[error("draft_store_failed")]
167    Store,
168    /// Startup has already signalled its one pull.
169    #[error("draft_startup_pull_repeated")]
170    StartupRepeated,
171    /// Lys could not acknowledge a receipt that is already committed.
172    #[error("draft_acknowledgment_failed: {draft}: {name}")]
173    Acknowledgment {
174        /// The draft whose local receipt remains committed.
175        draft: String,
176        /// The issuer consumer's safe machine failure name.
177        name: String,
178    },
179    /// The shared consumer refused the immutable execution contract.
180    #[error("draft_contract_refused")]
181    Contract,
182}
183
184/// The only events which request an approved-draft pull.
185#[derive(Clone, Copy, Debug, PartialEq, Eq)]
186pub enum PullSignal {
187    /// Once, after the target registries are loaded.
188    Startup,
189    /// A person's console approval has returned from Lys.
190    Console,
191    /// The stream owner delivered an explicit draft decision row.
192    Decision,
193}
194
195/// A future stream row is recognized by its declared name; unrelated grant
196/// changes never imply a draft approval. Lys currently publishes no such row.
197#[must_use]
198pub fn stream_signal(kind: &str) -> Option<PullSignal> {
199    (kind == "draft_decision").then_some(PullSignal::Decision)
200}
201
202/// Join for the one grant-stream owner; no thread, timer or second reader.
203#[async_trait::async_trait]
204pub trait DraftDecisionHook: Send + Sync {
205    /// Requests one pull after the owner accepts a named draft decision row.
206    ///
207    /// # Errors
208    /// The explicit pull's named failure; the owner retains it in diagnostics.
209    async fn draft_decided(&self) -> Result<PullReport, DraftError>;
210}
211
212/// Counts from one explicit pull, without carrying draft words or credentials.
213#[derive(Default, Debug, Serialize)]
214pub struct PullReport {
215    /// Pages answered by Lys.
216    pub pages: usize,
217    /// New executed receipts.
218    pub executed: usize,
219    /// New refused receipts.
220    pub refused: usize,
221    /// Previously committed receipts returned without another mutation.
222    pub replayed: usize,
223    /// Outcomes newly acknowledged by Lys.
224    pub acknowledged: usize,
225}
226
227/// Lys's executor over the existing target owner's atomic operation.
228pub struct DraftExecutor<S> {
229    pub(crate) store: S,
230    pub(crate) limits: DraftLimits,
231    pub(crate) at_ms: u64,
232}
233
234impl<S: AtomicDraftStore> Executor for DraftExecutor<S> {
235    type Error = DraftError;
236
237    fn receipt(&self, draft: &str) -> Result<Option<Execution>, DraftError> {
238        self.store.receipt(draft)
239    }
240
241    fn execute_and_record(&mut self, draft: &ApprovedDraft) -> Result<Execution, DraftError> {
242        let evidence = self.store.approval(draft)?;
243        let mut people = BTreeSet::new();
244        let needed = match evidence.mode {
245            super::super::GrantMode::ByDraft => 1,
246            super::super::GrantMode::ByTwo => 2,
247            super::super::GrantMode::Outright => return Err(DraftError::ApprovalUnavailable),
248        };
249        if evidence.draft != draft.id
250            || evidence.digest != draft.request_digest
251            || evidence.approved_by.len() != needed
252            || evidence.approved_by.len() > self.limits.approvers.get()
253            || evidence.approved_by.iter().any(|person| {
254                !valid_metadata(person, self.limits.metadata_bytes) || !people.insert(person)
255            })
256        {
257            return Err(DraftError::ApprovalUnavailable);
258        }
259        self.store.commit(
260            draft,
261            &ApprovalTransition {
262                draft: draft.id.clone(),
263                digest: draft.request_digest.clone(),
264                approved_by: evidence.approved_by,
265                executed_at_ms: self.at_ms,
266            },
267        )
268    }
269}
270
271pub(crate) fn valid_metadata(value: &str, bound: NonZeroUsize) -> bool {
272    !value.is_empty() && value.len() <= bound.get() && !value.chars().any(char::is_control)
273}
274
275pub(crate) fn validate_bounds(
276    draft: &ApprovedDraft,
277    limits: DraftLimits,
278) -> Result<(), DraftError> {
279    bounded(
280        &draft.id,
281        &draft.app,
282        &draft.grant,
283        &draft.target,
284        &draft.words,
285        limits,
286    )
287}
288
289/// The same bounds for a request Lys holds, before any approval exists.
290pub(crate) fn validate_request_bounds(
291    request: &lys_pass::drafts::DraftRequest,
292    limits: DraftLimits,
293) -> Result<(), DraftError> {
294    bounded(
295        &request.operation,
296        "liminal",
297        &request.grant,
298        &request.target,
299        &request.words,
300        limits,
301    )
302}
303
304fn bounded(
305    id: &str,
306    app: &str,
307    grant: &str,
308    target: &lys_pass::Target,
309    words: &str,
310    limits: DraftLimits,
311) -> Result<(), DraftError> {
312    if words.is_empty()
313        || words.len() > limits.words_bytes.get()
314        || [id, app, grant, &target.kind, &target.id, &target.action]
315            .iter()
316            .any(|value| !valid_metadata(value, limits.metadata_bytes))
317    {
318        return Err(DraftError::Bound);
319    }
320    Ok(())
321}
322
323/// The immutable digest shared by either terminal outcome.
324#[must_use]
325pub fn execution_digest(receipt: &Execution) -> &str {
326    match receipt {
327        Execution::Executed { request_digest, .. }
328        | Execution::RefusedOnExecution { request_digest, .. } => request_digest,
329    }
330}
331
332/// Compares every immutable member before a prepared target is applied.
333#[must_use]
334pub fn same_request(request: &lys_pass::drafts::DraftRequest, draft: &ApprovedDraft) -> bool {
335    draft.app == "liminal"
336        && request.grant == draft.grant
337        && request.target == draft.target
338        && request.request_digest == draft.request_digest
339        && request.words == draft.words
340}
341
342/// Approval evidence from the authenticated connector page. The issuer marker
343/// records the limitation of that page; it does not name a person as approver.
344///
345/// # Errors
346/// An unavailable prepared ledger or invalid held mode.
347pub fn issuer_approval(
348    journal: &waiting::durable::DraftJournal,
349    draft: &ApprovedDraft,
350) -> Result<ApprovalEvidence, DraftError> {
351    // Lys names every approver and the receipt's mode on the approved draft
352    // itself (lys-pass 0.3.4), bound to its id and digest. A preparation this
353    // exchange recorded must agree with that mode; an unknown preparation is
354    // judged by the issuer's evidence alone.
355    let mode = match draft.mode {
356        lys_pass::Mode::Outright => super::super::GrantMode::Outright,
357        lys_pass::Mode::ByDraft => super::super::GrantMode::ByDraft,
358        lys_pass::Mode::ByTwo => super::super::GrantMode::ByTwo,
359    };
360    let prepared = journal.read(|snapshot| {
361        snapshot
362            .prepared
363            .values()
364            .find(|prepared| prepared.draft.as_deref() == Some(&draft.id))
365            .map(|prepared| prepared.mode)
366    })?;
367    if prepared.is_some_and(|prepared| prepared != mode) {
368        return Err(DraftError::ApprovalUnavailable);
369    }
370    Ok(ApprovalEvidence {
371        draft: draft.id.clone(),
372        digest: draft.request_digest.clone(),
373        approved_by: draft.approved_by.clone(),
374        mode,
375    })
376}
377
378impl crate::services::lifecycle::site::executor::ApprovalReader for waiting::durable::DraftJournal {
379    fn read(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError> {
380        issuer_approval(self, draft)
381    }
382}
383
384/// Completes one synchronous permission call without nesting a runtime.
385pub(super) fn block<T, F>(handle: &tokio::runtime::Handle, future: F) -> Result<T, lys_pass::Error>
386where
387    T: Send,
388    F: std::future::Future<Output = Result<T, lys_pass::Error>> + Send,
389{
390    std::thread::scope(|scope| scope.spawn(move || handle.block_on(future)).join())
391        .unwrap_or_else(|_| Err(lys_pass::Error::CannotAsk("the live ask did not complete")))
392}
393
394pub(super) fn prepared_failure(_: DraftError) -> lys_pass::Error {
395    lys_pass::Error::CannotAsk("draft target journal failed")
396}
397
398/// Explicit capacity for durable evidence, which is never evicted to admit work.
399#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
400#[serde(deny_unknown_fields)]
401pub struct JournalLimits {
402    /// Bounds native target history and migration allocation before admission.
403    pub rows: NonZeroUsize,
404    /// Bounds prepared operations and receipts retained for immutable replays.
405    pub operations: NonZeroUsize,
406    /// Bounds the whole atomic request before it reaches the durable engine.
407    pub snapshot_bytes: NonZeroUsize,
408    /// Bounds each read of an old installation during its one-time migration.
409    pub replay_page_rows: NonZeroUsize,
410}
411
412impl AgentDraftStore {
413    /// Shares the server's existing native inbox and retirement projection.
414    #[must_use]
415    pub fn with_projection(mut self, projection: std::sync::Arc<dyn AgentProjection>) -> Self {
416        self.projection = projection;
417        self
418    }
419}
420
421/// Operator-selected admission, journal, retained-history and HTTP capacities.
422#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
423#[serde(deny_unknown_fields)]
424pub struct DraftConfig {
425    pub pull: DraftLimits,
426    pub journal: JournalLimits,
427    pub waiting: waiting::WaitingLimits,
428    pub http: admin::AdminLimits,
429}
430
431#[cfg(test)]
432#[path = "drafts_tests.rs"]
433pub(crate) mod tests;
liminal_server/auth_lys/drafts.rs

liminal_server/auth_lys/
drafts.rs
