Skip to main content

runifold_workflow/
governance.rs

1//! Capability-scoped control plane for Task tombstone governance.
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    future::Future,
6    pin::Pin,
7    sync::Arc,
8};
9
10use runifold_core::CheckpointId;
11use thiserror::Error;
12
13use crate::{
14    LeaseDuration, WorkerId, WorkflowStoreError, WorkflowTaskCleanupLease, WorkflowTaskLegalHold,
15    WorkflowTaskLegalHoldReason, WorkflowTaskTombstone, WorkflowTaskTombstoneApprovalInboxItem,
16    WorkflowTaskTombstoneApprovalInboxLimit, WorkflowTaskTombstoneApprovalLease,
17    WorkflowTaskTombstoneApprovalWindow, WorkflowTaskTombstoneCursor,
18    WorkflowTaskTombstoneExportReceipt, WorkflowTaskTombstoneGovernanceStore,
19    WorkflowTaskTombstoneLimit, WorkflowTaskTombstonePurgeEvidence, WorkflowTaskTombstonePurgeId,
20    WorkflowTaskTombstonePurgeIntent, WorkflowTaskTombstonePurgeLimit,
21    WorkflowTaskTombstoneRejectionReason, WorkflowTaskTombstoneRetention, WorkflowTenantId,
22};
23
24/// One tenant-scoped tombstone governance authority.
25#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
26#[non_exhaustive]
27pub enum WorkflowTaskGovernancePermission {
28    /// Place a legal hold.
29    PlaceHold,
30    /// Release a legal hold.
31    ReleaseHold,
32    /// Export tombstone details and confirm an archive receipt.
33    Export,
34    /// Prepare a bounded purge intent.
35    PreparePurge,
36    /// Independently approve a purge intent.
37    ApprovePurge,
38    /// Read the tenant's bounded approval inbox.
39    ReadApprovalInbox,
40    /// Claim one approval request for independent review.
41    ClaimPurgeApproval,
42    /// Reject a claimed purge request with a durable reason.
43    RejectPurge,
44    /// Execute an approved purge under a fenced lease.
45    ExecutePurge,
46    /// Read immutable purge evidence.
47    ReadEvidence,
48}
49
50/// Low-cardinality governance outcome for telemetry.
51#[derive(Clone, Copy, Debug, Eq, PartialEq)]
52#[non_exhaustive]
53pub enum WorkflowTaskGovernanceOutcome {
54    /// The operation completed successfully.
55    Succeeded,
56    /// Policy rejected the principal.
57    Denied,
58    /// Authorization infrastructure failed closed.
59    AuthorizationError,
60    /// Durable store execution failed.
61    StoreError,
62    /// External archive execution failed.
63    ArchiveError,
64}
65
66/// Optional observer that must not retain tenant or principal identities.
67pub trait WorkflowTaskGovernanceObserver: Send + Sync {
68    /// Observes one terminal control-plane outcome.
69    fn observe(
70        &self,
71        permission: WorkflowTaskGovernancePermission,
72        outcome: WorkflowTaskGovernanceOutcome,
73    );
74}
75
76#[derive(Debug, Default)]
77struct NoopWorkflowTaskGovernanceObserver;
78
79impl WorkflowTaskGovernanceObserver for NoopWorkflowTaskGovernanceObserver {
80    fn observe(
81        &self,
82        _permission: WorkflowTaskGovernancePermission,
83        _outcome: WorkflowTaskGovernanceOutcome,
84    ) {
85    }
86}
87
88/// Safe authorizer failure. Control-plane operations fail closed.
89#[derive(Clone, Debug, Error, Eq, PartialEq)]
90#[error("Task tombstone governance authorization failed: {message}")]
91pub struct WorkflowTaskGovernanceAuthorizationError {
92    message: String,
93}
94
95impl WorkflowTaskGovernanceAuthorizationError {
96    /// Creates a bounded safe authorization failure.
97    pub fn new(message: impl Into<String>) -> Self {
98        let mut message = message.into();
99        message.truncate(512);
100        Self { message }
101    }
102}
103
104/// Borrowing future returned by a governance authorizer.
105pub type WorkflowTaskGovernanceAuthorizationFuture<'a> = Pin<
106    Box<dyn Future<Output = Result<bool, WorkflowTaskGovernanceAuthorizationError>> + Send + 'a>,
107>;
108
109/// Pluggable policy boundary for principal, tenant, and permission checks.
110pub trait WorkflowTaskGovernanceAuthorizer: Send + Sync {
111    /// Returns `true` only for an explicit grant.
112    fn authorize(
113        &self,
114        principal: &WorkerId,
115        tenant_id: &WorkflowTenantId,
116        permission: WorkflowTaskGovernancePermission,
117    ) -> WorkflowTaskGovernanceAuthorizationFuture<'_>;
118}
119
120/// Immutable in-process authorizer for simple deployments and tests.
121#[derive(Clone, Debug, Default)]
122pub struct StaticWorkflowTaskGovernanceAuthorizer {
123    grants: BTreeMap<(WorkerId, WorkflowTenantId), BTreeSet<WorkflowTaskGovernancePermission>>,
124}
125
126impl StaticWorkflowTaskGovernanceAuthorizer {
127    /// Creates an empty deny-by-default policy.
128    pub fn new() -> Self {
129        Self::default()
130    }
131
132    /// Adds one explicit principal and tenant grant.
133    #[must_use]
134    pub fn with_grant(
135        mut self,
136        principal: WorkerId,
137        tenant_id: WorkflowTenantId,
138        permission: WorkflowTaskGovernancePermission,
139    ) -> Self {
140        self.grants
141            .entry((principal, tenant_id))
142            .or_default()
143            .insert(permission);
144        self
145    }
146}
147
148impl WorkflowTaskGovernanceAuthorizer for StaticWorkflowTaskGovernanceAuthorizer {
149    fn authorize(
150        &self,
151        principal: &WorkerId,
152        tenant_id: &WorkflowTenantId,
153        permission: WorkflowTaskGovernancePermission,
154    ) -> WorkflowTaskGovernanceAuthorizationFuture<'_> {
155        let allowed = self
156            .grants
157            .get(&(principal.clone(), tenant_id.clone()))
158            .is_some_and(|grants| grants.contains(&permission));
159        Box::pin(async move { Ok(allowed) })
160    }
161}
162
163/// Stable failure class for archive retry and observability policy.
164#[derive(Clone, Copy, Debug, Eq, PartialEq)]
165#[non_exhaustive]
166pub enum WorkflowTaskTombstoneArchiveErrorKind {
167    /// Archive configuration or local signing policy is invalid.
168    Configuration,
169    /// Remote credentials or policy rejected the request.
170    Authorization,
171    /// The bounded archive operation elapsed.
172    Timeout,
173    /// The remote archive is temporarily unavailable.
174    Unavailable,
175    /// A committed object did not match the stable batch payload.
176    Integrity,
177    /// Commit state could not be determined after reconciliation.
178    Ambiguous,
179    /// A safe fallback for adapters without a more specific classification.
180    Other,
181}
182
183/// Safe archive adapter failure.
184#[derive(Clone, Debug, Error, Eq, PartialEq)]
185#[error("Task tombstone archive failed ({kind:?}): {message}")]
186pub struct WorkflowTaskTombstoneArchiveError {
187    kind: WorkflowTaskTombstoneArchiveErrorKind,
188    message: String,
189}
190
191impl WorkflowTaskTombstoneArchiveError {
192    /// Creates a bounded safe archive failure with the fallback classification.
193    pub fn new(message: impl Into<String>) -> Self {
194        Self::with_kind(WorkflowTaskTombstoneArchiveErrorKind::Other, message)
195    }
196
197    /// Creates a bounded safe archive failure with a stable classification.
198    pub fn with_kind(
199        kind: WorkflowTaskTombstoneArchiveErrorKind,
200        message: impl Into<String>,
201    ) -> Self {
202        let mut message = message.into();
203        message.truncate(512);
204        Self { kind, message }
205    }
206
207    /// Returns the stable failure class without exposing provider detail.
208    pub const fn kind(&self) -> WorkflowTaskTombstoneArchiveErrorKind {
209        self.kind
210    }
211}
212
213/// Stable idempotency identity for one ordered archive batch.
214#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
215pub struct WorkflowTaskTombstoneArchiveBatchId(String);
216
217impl WorkflowTaskTombstoneArchiveBatchId {
218    /// Validates an externally restored stable batch identity.
219    ///
220    /// # Errors
221    ///
222    /// Rejects blank, oversized, or control-character-bearing values.
223    pub fn parse(value: impl Into<String>) -> Result<Self, WorkflowStoreError> {
224        let value = value.into();
225        if value.trim().is_empty() || value.len() > 512 || value.chars().any(char::is_control) {
226            return Err(WorkflowStoreError::new(
227                crate::WorkflowStoreErrorKind::InvalidInput,
228                "Task tombstone archive batch ID must contain 1..=512 printable bytes",
229            ));
230        }
231        Ok(Self(value))
232    }
233
234    fn from_batch(
235        tenant_id: &WorkflowTenantId,
236        first: WorkflowTaskTombstoneCursor,
237        last: WorkflowTaskTombstoneCursor,
238    ) -> Self {
239        Self(format!(
240            "{}:{}:{}",
241            tenant_id.as_str(),
242            first.get(),
243            last.get()
244        ))
245    }
246
247    /// Returns the stable archive idempotency key.
248    pub fn as_str(&self) -> &str {
249        &self.0
250    }
251}
252
253/// Exact ordered tombstone batch delivered to an external archive.
254#[derive(Clone, Debug, Eq, PartialEq)]
255pub struct WorkflowTaskTombstoneArchiveBatch {
256    /// Stable idempotency key reused after partial failure.
257    pub batch_id: WorkflowTaskTombstoneArchiveBatchId,
258    /// Owning tenant.
259    pub tenant_id: WorkflowTenantId,
260    /// Detailed tombstones in ascending cursor order.
261    pub tombstones: Vec<WorkflowTaskTombstone>,
262}
263
264/// Borrowing future returned by an external tombstone archive.
265pub type WorkflowTaskTombstoneArchiveFuture<'a> = Pin<
266    Box<
267        dyn Future<
268                Output = Result<
269                    WorkflowTaskTombstoneExportReceipt,
270                    WorkflowTaskTombstoneArchiveError,
271                >,
272            > + Send
273            + 'a,
274    >,
275>;
276
277/// Idempotent external archive boundary.
278pub trait WorkflowTaskTombstoneArchive: Send + Sync {
279    /// Stores the exact batch or replays its original receipt.
280    fn archive(
281        &self,
282        batch: WorkflowTaskTombstoneArchiveBatch,
283    ) -> WorkflowTaskTombstoneArchiveFuture<'_>;
284}
285
286/// Result of one bounded archive-and-confirm cycle.
287#[derive(Clone, Debug, Eq, PartialEq)]
288pub struct WorkflowTaskTombstoneArchiveReport {
289    /// Stable batch key, absent when the source page was empty.
290    pub batch_id: Option<WorkflowTaskTombstoneArchiveBatchId>,
291    /// Number of detailed tombstones archived.
292    pub tombstones_exported: u32,
293    /// Confirmed cursor, absent when the source page was empty.
294    pub through: Option<WorkflowTaskTombstoneCursor>,
295    /// Archive receipt persisted with the watermark.
296    pub receipt: Option<WorkflowTaskTombstoneExportReceipt>,
297}
298
299/// Fail-closed tombstone governance failure.
300#[derive(Debug, Error)]
301#[non_exhaustive]
302pub enum WorkflowTaskGovernanceError {
303    /// The principal lacks the exact tenant-scoped permission.
304    #[error("principal is not authorized for {permission:?}")]
305    Denied {
306        /// Rejected operation.
307        permission: WorkflowTaskGovernancePermission,
308    },
309    /// The policy backend failed and the operation was not attempted.
310    #[error(transparent)]
311    Authorization(#[from] WorkflowTaskGovernanceAuthorizationError),
312    /// Durable governance state failed.
313    #[error(transparent)]
314    Store(#[from] WorkflowStoreError),
315    /// External archive failed before durable confirmation.
316    #[error(transparent)]
317    Archive(#[from] WorkflowTaskTombstoneArchiveError),
318    /// A fenced lease was supplied by a different principal.
319    #[error("cleanup lease owner does not match the authenticated principal")]
320    LeasePrincipalMismatch,
321    /// A reviewer lease was supplied by a different principal.
322    #[error("approval lease reviewer does not match the authenticated principal")]
323    ApprovalLeasePrincipalMismatch,
324}
325
326/// Authorized facade over the destructive tombstone governance store.
327pub struct WorkflowTaskGovernanceControlPlane<S, A> {
328    store: Arc<S>,
329    authorizer: Arc<A>,
330    observer: Arc<dyn WorkflowTaskGovernanceObserver>,
331}
332
333impl<S, A> WorkflowTaskGovernanceControlPlane<S, A>
334where
335    S: WorkflowTaskTombstoneGovernanceStore,
336    A: WorkflowTaskGovernanceAuthorizer,
337{
338    /// Creates a fail-closed control plane.
339    pub fn new(store: Arc<S>, authorizer: Arc<A>) -> Self {
340        Self {
341            store,
342            authorizer,
343            observer: Arc::new(NoopWorkflowTaskGovernanceObserver),
344        }
345    }
346
347    /// Attaches a low-cardinality outcome observer.
348    #[must_use]
349    pub fn with_observer(mut self, observer: Arc<dyn WorkflowTaskGovernanceObserver>) -> Self {
350        self.observer = observer;
351        self
352    }
353
354    /// Places a hold using the authenticated principal as the audit actor.
355    ///
356    /// # Errors
357    ///
358    /// Fails closed on denied authority or store failure.
359    pub async fn place_hold(
360        &self,
361        principal: &WorkerId,
362        tenant_id: WorkflowTenantId,
363        checkpoint_id: CheckpointId,
364        reason: WorkflowTaskLegalHoldReason,
365    ) -> Result<WorkflowTaskLegalHold, WorkflowTaskGovernanceError> {
366        let permission = WorkflowTaskGovernancePermission::PlaceHold;
367        self.authorize(principal, &tenant_id, permission).await?;
368        let result = self
369            .store
370            .place_task_tombstone_hold(tenant_id, checkpoint_id, principal.clone(), reason)
371            .await;
372        self.store_result(permission, result)
373    }
374
375    /// Releases a hold using the authenticated principal as the audit actor.
376    ///
377    /// # Errors
378    ///
379    /// Fails closed on denied authority or store failure.
380    pub async fn release_hold(
381        &self,
382        principal: &WorkerId,
383        tenant_id: WorkflowTenantId,
384        checkpoint_id: CheckpointId,
385    ) -> Result<WorkflowTaskLegalHold, WorkflowTaskGovernanceError> {
386        let permission = WorkflowTaskGovernancePermission::ReleaseHold;
387        self.authorize(principal, &tenant_id, permission).await?;
388        let result = self
389            .store
390            .release_task_tombstone_hold(tenant_id, checkpoint_id, principal.clone())
391            .await;
392        self.store_result(permission, result)
393    }
394
395    /// Archives one page and confirms its watermark only after receipt.
396    ///
397    /// # Errors
398    ///
399    /// Fails closed on denied authority, source-store failure, archive
400    /// failure, or durable confirmation failure.
401    pub async fn export_next_page<R>(
402        &self,
403        principal: &WorkerId,
404        tenant_id: WorkflowTenantId,
405        after: Option<WorkflowTaskTombstoneCursor>,
406        limit: WorkflowTaskTombstoneLimit,
407        archive: &R,
408    ) -> Result<WorkflowTaskTombstoneArchiveReport, WorkflowTaskGovernanceError>
409    where
410        R: WorkflowTaskTombstoneArchive,
411    {
412        let permission = WorkflowTaskGovernancePermission::Export;
413        self.authorize(principal, &tenant_id, permission).await?;
414        let tombstones = match self
415            .store
416            .list_task_tombstones(tenant_id.clone(), after, limit)
417            .await
418        {
419            Ok(tombstones) => tombstones,
420            Err(error) => return Err(self.store_error(permission, error)),
421        };
422        let (Some(first), Some(last)) = (tombstones.first(), tombstones.last()) else {
423            self.observer
424                .observe(permission, WorkflowTaskGovernanceOutcome::Succeeded);
425            return Ok(WorkflowTaskTombstoneArchiveReport {
426                batch_id: None,
427                tombstones_exported: 0,
428                through: None,
429                receipt: None,
430            });
431        };
432        let batch_id =
433            WorkflowTaskTombstoneArchiveBatchId::from_batch(&tenant_id, first.cursor, last.cursor);
434        let through = last.cursor;
435        let count = u32::try_from(tombstones.len()).unwrap_or(u32::MAX);
436        let receipt = match archive
437            .archive(WorkflowTaskTombstoneArchiveBatch {
438                batch_id: batch_id.clone(),
439                tenant_id: tenant_id.clone(),
440                tombstones,
441            })
442            .await
443        {
444            Ok(receipt) => receipt,
445            Err(error) => {
446                self.observer
447                    .observe(permission, WorkflowTaskGovernanceOutcome::ArchiveError);
448                return Err(error.into());
449            }
450        };
451        if let Err(error) = self
452            .store
453            .confirm_task_tombstone_export(tenant_id, through, receipt.clone(), principal.clone())
454            .await
455        {
456            return Err(self.store_error(permission, error));
457        }
458        self.observer
459            .observe(permission, WorkflowTaskGovernanceOutcome::Succeeded);
460        Ok(WorkflowTaskTombstoneArchiveReport {
461            batch_id: Some(batch_id),
462            tombstones_exported: count,
463            through: Some(through),
464            receipt: Some(receipt),
465        })
466    }
467
468    /// Prepares a purge only when the lease belongs to the principal.
469    ///
470    /// # Errors
471    ///
472    /// Rejects denied authority, a mismatched lease principal, or store
473    /// failure.
474    pub async fn prepare_purge(
475        &self,
476        principal: &WorkerId,
477        lease: WorkflowTaskCleanupLease,
478        retention: WorkflowTaskTombstoneRetention,
479        limit: WorkflowTaskTombstonePurgeLimit,
480        approval_window: WorkflowTaskTombstoneApprovalWindow,
481    ) -> Result<WorkflowTaskTombstonePurgeIntent, WorkflowTaskGovernanceError> {
482        let permission = WorkflowTaskGovernancePermission::PreparePurge;
483        self.require_lease_principal(principal, &lease, permission)?;
484        self.authorize(principal, &lease.tenant_id, permission)
485            .await?;
486        let result = self
487            .store
488            .prepare_task_tombstone_purge(lease, retention, limit, approval_window)
489            .await;
490        self.store_result(permission, result)
491    }
492
493    /// Approves using the authenticated principal, preserving four-eyes checks.
494    ///
495    /// # Errors
496    ///
497    /// Fails closed on denied authority or store governance failure.
498    pub async fn approve_purge(
499        &self,
500        principal: &WorkerId,
501        tenant_id: WorkflowTenantId,
502        purge_id: WorkflowTaskTombstonePurgeId,
503    ) -> Result<WorkflowTaskTombstonePurgeIntent, WorkflowTaskGovernanceError> {
504        let permission = WorkflowTaskGovernancePermission::ApprovePurge;
505        self.authorize(principal, &tenant_id, permission).await?;
506        let result = self
507            .store
508            .approve_task_tombstone_purge(tenant_id, purge_id, principal.clone())
509            .await;
510        self.store_result(permission, result)
511    }
512
513    /// Lists the tenant's bounded durable approval inbox.
514    ///
515    /// # Errors
516    ///
517    /// Fails closed on denied authority or store failure.
518    pub async fn list_purge_approvals(
519        &self,
520        principal: &WorkerId,
521        tenant_id: WorkflowTenantId,
522        limit: WorkflowTaskTombstoneApprovalInboxLimit,
523    ) -> Result<Vec<WorkflowTaskTombstoneApprovalInboxItem>, WorkflowTaskGovernanceError> {
524        let permission = WorkflowTaskGovernancePermission::ReadApprovalInbox;
525        self.authorize(principal, &tenant_id, permission).await?;
526        let result = self
527            .store
528            .list_task_tombstone_purge_approvals(tenant_id, limit)
529            .await;
530        self.store_result(permission, result)
531    }
532
533    /// Claims the oldest eligible approval using the authenticated reviewer.
534    ///
535    /// # Errors
536    ///
537    /// Fails closed on denied authority or store failure.
538    pub async fn claim_purge_approval(
539        &self,
540        principal: &WorkerId,
541        tenant_id: WorkflowTenantId,
542        lease: LeaseDuration,
543    ) -> Result<Option<WorkflowTaskTombstoneApprovalLease>, WorkflowTaskGovernanceError> {
544        let permission = WorkflowTaskGovernancePermission::ClaimPurgeApproval;
545        self.authorize(principal, &tenant_id, permission).await?;
546        let result = self
547            .store
548            .claim_task_tombstone_purge_approval(tenant_id, principal.clone(), lease)
549            .await;
550        self.store_result(permission, result)
551    }
552
553    /// Approves an exact principal-owned reviewer lease.
554    ///
555    /// # Errors
556    ///
557    /// Rejects a mismatched reviewer, denied authority, or stale lease.
558    pub async fn approve_claimed_purge(
559        &self,
560        principal: &WorkerId,
561        lease: WorkflowTaskTombstoneApprovalLease,
562    ) -> Result<WorkflowTaskTombstonePurgeIntent, WorkflowTaskGovernanceError> {
563        let permission = WorkflowTaskGovernancePermission::ApprovePurge;
564        self.require_approval_principal(principal, &lease, permission)?;
565        self.authorize(principal, &lease.tenant_id, permission)
566            .await?;
567        let result = self.store.approve_claimed_task_tombstone_purge(lease).await;
568        self.store_result(permission, result)
569    }
570
571    /// Rejects an exact principal-owned reviewer lease with durable evidence.
572    ///
573    /// # Errors
574    ///
575    /// Rejects a mismatched reviewer, denied authority, or stale lease.
576    pub async fn reject_claimed_purge(
577        &self,
578        principal: &WorkerId,
579        lease: WorkflowTaskTombstoneApprovalLease,
580        reason: WorkflowTaskTombstoneRejectionReason,
581    ) -> Result<WorkflowTaskTombstoneApprovalInboxItem, WorkflowTaskGovernanceError> {
582        let permission = WorkflowTaskGovernancePermission::RejectPurge;
583        self.require_approval_principal(principal, &lease, permission)?;
584        self.authorize(principal, &lease.tenant_id, permission)
585            .await?;
586        let result = self
587            .store
588            .reject_claimed_task_tombstone_purge(lease, reason)
589            .await;
590        self.store_result(permission, result)
591    }
592
593    /// Executes using an exact principal-owned fenced lease.
594    ///
595    /// # Errors
596    ///
597    /// Rejects denied authority, a mismatched lease principal, or store
598    /// execution failure.
599    pub async fn execute_purge(
600        &self,
601        principal: &WorkerId,
602        lease: WorkflowTaskCleanupLease,
603        purge_id: WorkflowTaskTombstonePurgeId,
604    ) -> Result<WorkflowTaskTombstonePurgeEvidence, WorkflowTaskGovernanceError> {
605        let permission = WorkflowTaskGovernancePermission::ExecutePurge;
606        self.require_lease_principal(principal, &lease, permission)?;
607        self.authorize(principal, &lease.tenant_id, permission)
608            .await?;
609        let result = self
610            .store
611            .execute_task_tombstone_purge(lease, purge_id)
612            .await;
613        self.store_result(permission, result)
614    }
615
616    /// Reads evidence under an explicit tenant-scoped grant.
617    ///
618    /// # Errors
619    ///
620    /// Fails closed on denied authority or store failure.
621    pub async fn get_evidence(
622        &self,
623        principal: &WorkerId,
624        tenant_id: WorkflowTenantId,
625        purge_id: WorkflowTaskTombstonePurgeId,
626    ) -> Result<Option<WorkflowTaskTombstonePurgeEvidence>, WorkflowTaskGovernanceError> {
627        let permission = WorkflowTaskGovernancePermission::ReadEvidence;
628        self.authorize(principal, &tenant_id, permission).await?;
629        let result = self
630            .store
631            .get_task_tombstone_purge_evidence(tenant_id, purge_id)
632            .await;
633        self.store_result(permission, result)
634    }
635
636    async fn authorize(
637        &self,
638        principal: &WorkerId,
639        tenant_id: &WorkflowTenantId,
640        permission: WorkflowTaskGovernancePermission,
641    ) -> Result<(), WorkflowTaskGovernanceError> {
642        match self
643            .authorizer
644            .authorize(principal, tenant_id, permission)
645            .await
646        {
647            Ok(true) => Ok(()),
648            Ok(false) => {
649                self.observer
650                    .observe(permission, WorkflowTaskGovernanceOutcome::Denied);
651                Err(WorkflowTaskGovernanceError::Denied { permission })
652            }
653            Err(error) => {
654                self.observer.observe(
655                    permission,
656                    WorkflowTaskGovernanceOutcome::AuthorizationError,
657                );
658                Err(error.into())
659            }
660        }
661    }
662
663    fn require_lease_principal(
664        &self,
665        principal: &WorkerId,
666        lease: &WorkflowTaskCleanupLease,
667        permission: WorkflowTaskGovernancePermission,
668    ) -> Result<(), WorkflowTaskGovernanceError> {
669        if lease.owner == *principal {
670            Ok(())
671        } else {
672            self.observer
673                .observe(permission, WorkflowTaskGovernanceOutcome::Denied);
674            Err(WorkflowTaskGovernanceError::LeasePrincipalMismatch)
675        }
676    }
677
678    fn require_approval_principal(
679        &self,
680        principal: &WorkerId,
681        lease: &WorkflowTaskTombstoneApprovalLease,
682        permission: WorkflowTaskGovernancePermission,
683    ) -> Result<(), WorkflowTaskGovernanceError> {
684        if lease.reviewer == *principal {
685            Ok(())
686        } else {
687            self.observer
688                .observe(permission, WorkflowTaskGovernanceOutcome::Denied);
689            Err(WorkflowTaskGovernanceError::ApprovalLeasePrincipalMismatch)
690        }
691    }
692
693    fn store_result<T>(
694        &self,
695        permission: WorkflowTaskGovernancePermission,
696        result: Result<T, WorkflowStoreError>,
697    ) -> Result<T, WorkflowTaskGovernanceError> {
698        match result {
699            Ok(value) => {
700                self.observer
701                    .observe(permission, WorkflowTaskGovernanceOutcome::Succeeded);
702                Ok(value)
703            }
704            Err(error) => Err(self.store_error(permission, error)),
705        }
706    }
707
708    fn store_error(
709        &self,
710        permission: WorkflowTaskGovernancePermission,
711        error: WorkflowStoreError,
712    ) -> WorkflowTaskGovernanceError {
713        self.observer
714            .observe(permission, WorkflowTaskGovernanceOutcome::StoreError);
715        error.into()
716    }
717}
718
719impl<S, A> std::fmt::Debug for WorkflowTaskGovernanceControlPlane<S, A> {
720    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
721        formatter
722            .debug_struct("WorkflowTaskGovernanceControlPlane")
723            .field("store", &"<governance-store>")
724            .field("authorizer", &"<governance-authorizer>")
725            .field("observer", &"<governance-observer>")
726            .finish()
727    }
728}
729
730#[cfg(test)]
731mod tests {
732    use super::{WorkflowTaskTombstoneArchiveError, WorkflowTaskTombstoneArchiveErrorKind};
733
734    #[test]
735    fn archive_error_preserves_kind_and_bounds_safe_message() {
736        let error = WorkflowTaskTombstoneArchiveError::with_kind(
737            WorkflowTaskTombstoneArchiveErrorKind::Ambiguous,
738            "x".repeat(1_024),
739        );
740
741        assert_eq!(
742            error.kind(),
743            WorkflowTaskTombstoneArchiveErrorKind::Ambiguous
744        );
745        assert!(error.to_string().len() < 600);
746        assert_eq!(
747            WorkflowTaskTombstoneArchiveError::new("fallback").kind(),
748            WorkflowTaskTombstoneArchiveErrorKind::Other
749        );
750    }
751}