feat(workflow): mount the remaining Work/workflow product surface

crates/tracedecay-api/src/workflow.rs

@@ -122,6 +122,21 @@ impl WorkflowOperation {
             .find(|operation| operation.operation_key() == key)
     }
 
+    /// Whether the operation reads without producing a durable effect.
+    /// Parity with the catalog's effect class is pinned by
+    /// `read_only_operations_mirror_the_catalog_effect_class`.
+    pub const fn is_read_only(self) -> bool {
+        matches!(
+            self,
+            Self::ValidateDefinition
+                | Self::GetDefinition
+                | Self::ListDefinitions
+                | Self::DefinitionHistory
+                | Self::DiffDefinition
+                | Self::GetRun
+        )
+    }
+
     pub fn from_cli_name(name: &str) -> Option<Self> {
         Self::ALL.iter().copied().find(|operation| {
             operation.operation_key() == name || operation.route_segment() == name
@@ -356,6 +371,27 @@ mod tests {
         }
     }
 
+    #[test]
+    fn read_only_operations_mirror_the_catalog_effect_class() {
+        let registry = tracedecay_application::workflow_executable_binding_registry()
+            .expect("canonical Workflow executable registry");
+        for operation in WorkflowOperation::ALL {
+            let operation_id =
+                tracedecay_tool_catalog::OperationId::new(operation.operation_id_str().to_owned())
+                    .expect("catalog operation id");
+            let binding = registry
+                .get(&operation_id)
+                .and_then(|availability| availability.binding())
+                .expect("every mounted Workflow operation has an executable binding");
+            assert_eq!(
+                operation.is_read_only(),
+                binding.effect() == tracedecay_tool_catalog::EffectClass::Read,
+                "{} read-only declaration must mirror the catalog effect class",
+                operation.operation_key()
+            );
+        }
+    }
+
     #[tokio::test]
     async fn router_dispatches_every_advertised_definition_and_runtime_operation() {
         let seen = Arc::new(Mutex::new(Vec::new()));

crates/tracedecay-application/src/lib.rs

@@ -82,6 +82,7 @@ pub mod work_retry;
 pub mod work_run_control;
 pub mod work_synthesis;
 pub mod work_topology_view;
+pub mod workflow_admission;
 pub mod workflow_catalog;
 pub mod workflow_coordination;
 pub mod workflow_effect;
@@ -371,6 +372,7 @@ pub use work_retry::*;
 pub use work_run_control::*;
 pub use work_synthesis::*;
 pub use work_topology_view::*;
+pub use workflow_admission::*;
 pub use workflow_catalog::*;
 pub use workflow_coordination::*;
 pub use workflow_effect::*;

crates/tracedecay-application/src/workflow_admission.rs

@@ -0,0 +1,109 @@
+//! Tool-catalog semantic admission for workflow definitions.
+//!
+//! Plan 32: unknown operations and incompatible schemas reject before
+//! activation. Structural validation proves only the DAG shape, so activation
+//! additionally admits every step operation against the canonical Work
+//! executable catalog — the registry fan-out lowers steps into. The schema
+//! and capability halves of the check are the catalog digest pin:
+//! [`crate::work_executable_catalog_digest`] hashes every capability manifest
+//! and schema authority, so a stale `pinned_catalog_digest` is a typed
+//! denial, never a silent re-pin.
+
+use std::fmt::{self, Display};
+
+use tracedecay_domain::{ManifestDigest, WorkflowDefinition, WorkflowOperationRef, WorkflowStepId};
+use tracedecay_tool_catalog::{CatalogValidationError, OperationId};
+
+use crate::work_catalog::{work_executable_binding_registry, work_executable_catalog_digest};
+
+/// Typed denial produced by workflow catalog admission.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum WorkflowCatalogAdmissionError {
+    /// The step names an operation the executable catalog does not know.
+    UnknownOperation {
+        step_id: WorkflowStepId,
+        operation: WorkflowOperationRef,
+    },
+    /// The operation is cataloged but currently has no executable binding.
+    OperationUnavailable {
+        step_id: WorkflowStepId,
+        operation: WorkflowOperationRef,
+    },
+    /// The definition pins a catalog other than the live executable catalog,
+    /// so its operations were authored against different schemas or
+    /// capability contracts.
+    CatalogPinMismatch {
+        pinned: ManifestDigest,
+        current: ManifestDigest,
+    },
+    /// The canonical catalog itself could not be composed.
+    CatalogUnavailable(CatalogValidationError),
+}
+
+impl Display for WorkflowCatalogAdmissionError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::UnknownOperation { step_id, operation } => write!(
+                formatter,
+                "workflow step {step_id} names unknown catalog operation {operation}"
+            ),
+            Self::OperationUnavailable { step_id, operation } => write!(
+                formatter,
+                "workflow step {step_id} names catalog operation {operation} without an executable binding"
+            ),
+            Self::CatalogPinMismatch { pinned, current } => write!(
+                formatter,
+                "workflow definition pins catalog {pinned} but the live executable catalog is {current}"
+            ),
+            Self::CatalogUnavailable(error) => {
+                write!(
+                    formatter,
+                    "workflow executable catalog unavailable: {error}"
+                )
+            }
+        }
+    }
+}
+
+impl std::error::Error for WorkflowCatalogAdmissionError {}
+
+/// Admit every step operation of one workflow definition against the
+/// canonical Work executable catalog: the definition must pin the live
+/// catalog digest and every step operation must resolve to an available
+/// executable binding. The first violation is a typed denial naming the
+/// offending step and operation.
+pub(crate) fn admit_workflow_definition_operations(
+    definition: &WorkflowDefinition,
+) -> Result<(), WorkflowCatalogAdmissionError> {
+    let registry = work_executable_binding_registry()
+        .map_err(WorkflowCatalogAdmissionError::CatalogUnavailable)?;
+    let current = work_executable_catalog_digest()
+        .map_err(WorkflowCatalogAdmissionError::CatalogUnavailable)?;
+    if definition.pinned_catalog_digest() != &current {
+        return Err(WorkflowCatalogAdmissionError::CatalogPinMismatch {
+            pinned: definition.pinned_catalog_digest().clone(),
+            current,
+        });
+    }
+    for step in definition.steps() {
+        let Ok(operation_id) = OperationId::new(step.operation.as_str().to_owned()) else {
+            return Err(WorkflowCatalogAdmissionError::UnknownOperation {
+                step_id: step.step_id.clone(),
+                operation: step.operation.clone(),
+            });
+        };
+        let Some(availability) = registry.get(&operation_id) else {
+            return Err(WorkflowCatalogAdmissionError::UnknownOperation {
+                step_id: step.step_id.clone(),
+                operation: step.operation.clone(),
+            });
+        };
+        if availability.binding().is_none() {
+            return Err(WorkflowCatalogAdmissionError::OperationUnavailable {
+                step_id: step.step_id.clone(),
+                operation: step.operation.clone(),
+            });
+        }
+    }
+    Ok(())
+}

crates/tracedecay-application/src/workflow_coordination.rs

@@ -9,6 +9,9 @@ use std::fmt::{self, Display};
 
 use crate::RequestContext;
 use crate::work_handoff_frontier::WorkHandoffFrontierV1;
+use crate::workflow_admission::{
+    WorkflowCatalogAdmissionError, admit_workflow_definition_operations,
+};
 use schemars::JsonSchema;
 use serde::{Deserialize, Deserializer, Serialize};
 use tracedecay_domain::{
@@ -341,6 +344,7 @@ pub struct WorkflowDefinitionRejectRequest {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum WorkflowCoordinationError {
     InvalidDefinition,
+    CatalogAdmissionDenied(WorkflowCatalogAdmissionError),
     ScopeMismatch,
     ImmutableDefinitionConflict,
     DefinitionNotFound,
@@ -353,6 +357,9 @@ impl Display for WorkflowCoordinationError {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::InvalidDefinition => formatter.write_str("workflow definition is invalid"),
+            Self::CatalogAdmissionDenied(denial) => {
+                write!(formatter, "workflow catalog admission denied: {denial}")
+            }
             Self::ScopeMismatch => {
                 formatter.write_str("workflow definition is outside the admitted project")
             }
@@ -413,13 +420,17 @@ where
         }
     }
 
+    /// The preflight for activation: structural shape plus tool-catalog
+    /// semantic admission of every step operation.
     pub fn validate(
         &self,
         definition: WorkflowDefinition,
     ) -> Result<WorkflowDefinitionValidation, WorkflowCoordinationError> {
         definition
             .validate()
             .map_err(|_| WorkflowCoordinationError::InvalidDefinition)?;
+        admit_workflow_definition_operations(&definition)
+            .map_err(WorkflowCoordinationError::CatalogAdmissionDenied)?;
         Ok(WorkflowDefinitionValidation { definition })
     }
 
@@ -452,24 +463,39 @@ where
             .map_err(coordination_authority_error)
     }
 
+    /// Admission every activation must clear before its lifecycle transition
+    /// is journaled: structural revalidation plus tool-catalog admission of
+    /// every step operation. The one authority both activation paths — this
+    /// service and the daemon's journaled effect — run.
+    pub fn admit_activation(
+        &self,
+        definition_id: &WorkflowDefinitionId,
+        definition_version: u64,
+    ) -> Result<(), WorkflowCoordinationError> {
+        let definition = self.get(definition_id, definition_version)?;
+        definition
+            .validate()
+            .map_err(|_| WorkflowCoordinationError::InvalidDefinition)?;
+        admit_workflow_definition_operations(&definition)
+            .map_err(WorkflowCoordinationError::CatalogAdmissionDenied)
+    }
+
     /// Advances a registered definition version to `active`.
     ///
     /// Plan 32: "Unknown operations, cycles, dangling references, incompatible
     /// schemas, unbounded fan-out, privilege expansion, unsupported effects,
     /// or recursive generic execution reject before activation." The stored
-    /// payload is revalidated here, and the `candidate -> validated -> active`
-    /// path is recorded as immutable history entries by the authority.
+    /// payload is revalidated and catalog-admitted here, and the
+    /// `candidate -> validated -> active` path is recorded as immutable
+    /// history entries by the authority.
     pub fn activate(
         &self,
         definition_id: &WorkflowDefinitionId,
         definition_version: u64,
         expected_revision: u64,
         transitioned_at: UtcMicros,
     ) -> Result<WorkflowDefinitionDisposition, WorkflowCoordinationError> {
-        let definition = self.get(definition_id, definition_version)?;
-        definition
-            .validate()
-            .map_err(|_| WorkflowCoordinationError::InvalidDefinition)?;
+        self.admit_activation(definition_id, definition_version)?;
         self.apply_lifecycle(WorkflowDefinitionLifecycleCommand {
             definition_id: definition_id.clone(),
             definition_version,