Add projected startup cause and secondary failure contracts

Cargo.lock

@@ -4768,6 +4768,7 @@ dependencies = [
  "agen",
  "async-trait",
  "base64 0.22.1",
+ "error-contract",
  "filetime",
  "fs4",
  "futures",
@@ -7272,6 +7273,7 @@ dependencies = [
  "client",
  "config-source",
  "dotenv",
+ "error-contract",
  "flow",
  "fs-operation",
  "fs4",
@@ -7328,6 +7330,7 @@ dependencies = [
  "axum",
  "base64 0.22.1",
  "decodal",
+ "error-contract",
  "flow",
  "fs4",
  "futures",

crates/protocol/src/lib.rs

@@ -2,6 +2,7 @@
 
 pub mod identity;
 pub mod invocation;
+pub mod startup_failure;
 #[cfg(feature = "stream")]
 pub mod stream;
 pub mod subscription;

crates/protocol/src/startup_failure.rs

@@ -0,0 +1,231 @@
+//! Startup diagnostics are observations, never lifecycle, ownership, or retry authority.
+//!
+//! Attachments contain a domain's explicitly projected error-contract envelope. Receivers
+//! decode through the existing codec, and must project again for a different audience;
+//! deserializing stored bytes never creates a disclosure capability. Unknown and corrupt
+//! attachments remain readable alongside metadata instead of breaking its lifecycle fields.
+use error_contract::{
+    Audience, Budget, ContractError, DecodeFailure, ProjectedFailure, WireDetails, decode_failure,
+    encode_projected_failure,
+};
+use serde::{Deserialize, Serialize};
+
+/// A bounded, projected envelope at the Worker/Runtime or metadata boundary.
+/// No constructor accepts Display text, arbitrary JSON fields, or an unprojected error.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(transparent)]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub struct StartupFailureAttachment(String);
+
+impl std::fmt::Debug for StartupFailureAttachment {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StartupFailureAttachment")
+            .finish_non_exhaustive()
+    }
+}
+
+impl StartupFailureAttachment {
+    pub fn from_projected<A: Audience, D: WireDetails>(
+        failure: &ProjectedFailure<A, D>,
+    ) -> Result<Self, ContractError> {
+        let encoded = encode_projected_failure(failure, Self::budget())?;
+        String::from_utf8(encoded.as_bytes().to_vec())
+            .map(Self)
+            .map_err(|_| ContractError::Encoding)
+    }
+
+    /// Missing attachments are represented by the enclosing Option, not a fabricated cause.
+    pub fn decode<D: WireDetails>(&self) -> DecodeFailure<D> {
+        decode_failure(self.0.as_bytes(), Self::budget())
+    }
+
+    fn budget() -> Budget {
+        Budget::new(
+            error_contract::MAX_FAILURE_BYTES,
+            error_contract::MAX_FAILURE_BYTES,
+        )
+    }
+}
+
+/// Owner-selected startup stage; independent of resource cleanup disposition.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupStage {
+    Preflight,
+    Metadata,
+    Reconstruction,
+    Recovery,
+    Binding,
+    Feature,
+    ActivationEffects,
+    Controller,
+}
+
+/// Safe local Runtime cause when no more specific domain attachment was supplied.
+/// The primary cause is selected where the failure occurs, not parsed from its diagnostic.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub struct RuntimeStartupCause {
+    pub stage: StartupStage,
+    pub reason: RuntimeStartupReason,
+}
+
+/// Coarse local causes do not claim a domain authorization verdict.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum RuntimeStartupReason {
+    StartupFailed,
+    SegmentEmpty,
+    SystemPromptMissing,
+    MetadataPending,
+    MetadataMissing,
+    MetadataInactive,
+    ManifestMissing,
+    AllocationFailed,
+    SessionReadFailed,
+    MetadataReadFailed,
+    ProviderFailed,
+    AttributionFailed,
+    FeatureFailed,
+}
+
+impl error_contract::FailureCause for RuntimeStartupCause {
+    fn identity(&self) -> error_contract::FailureIdentity {
+        error_contract::FailureIdentity {
+            domain: "worker_startup",
+            reason: "failed",
+            category: error_contract::Category::Internal,
+            details_version: 1,
+        }
+    }
+    fn validate(&self) -> Result<(), ContractError> {
+        Ok(())
+    }
+}
+impl WireDetails for RuntimeStartupCause {
+    fn supports(domain: &str, reason: &str, version: u16) -> bool {
+        domain == "worker_startup" && reason == "failed" && version == 1
+    }
+}
+
+// This cause contains only an enum. Arbitrary provider messages, IO paths and credentials
+// cannot enter its projection. Domain-specific causes must supply their own projector.
+macro_rules! project_runtime {
+    ($($audience:ty),*) => {$(
+        impl error_contract::ProjectFor<$audience> for RuntimeStartupCause {
+            type Authorization = ();
+            type Details = Self;
+            fn disclose(&self, facts: &error_contract::FailureFacts<Self>, _: &())
+                -> Result<error_contract::Disclosure<Self>, ContractError>
+            {
+                let mut disclosure = error_contract::Disclosure::new(self.clone(), "Worker startup failed.");
+                // Only fixed owner-selected operation/stage labels are published here.
+                for context in &facts.context().operations {
+                    if context.operation == "worker_startup" && context.stage == "runtime" {
+                        disclosure.context.operations.push(context.clone());
+                        if let Some(last) = disclosure.context.operations.last_mut() {
+                            last.operation_ref = None;
+                        }
+                    }
+                }
+                Ok(disclosure)
+            }
+        }
+    )*};
+}
+project_runtime!(
+    error_contract::InternalTransfer,
+    error_contract::Storage,
+    error_contract::Browser
+);
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupFailurePhase {
+    Cleanup,
+    RuntimeRecordDelete,
+    RuntimeSnapshotSave,
+    RuntimeRemovalPublish,
+    MetadataSave,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupCleanupFailure {
+    Ownership,
+    Workdir,
+    Artifacts,
+    Allocation,
+    Feature,
+    Unsupported,
+}
+
+/// A separate secondary fact, not a replacement primary cause or a causal-chain link.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupSecondaryFailure {
+    Cleanup {
+        failure: StartupCleanupFailure,
+    },
+    Persistence {
+        phase: StartupFailurePhase,
+        outcome: StartupPersistenceOutcome,
+    },
+}
+
+/// Persisted is evidence only for this attachment/record write, not execution completion.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupPersistenceOutcome {
+    Persisted,
+    NotSaved,
+    Conflict,
+    OutcomeUnknown,
+}
+
+/// Independent resource observations from the existing execution owner. These diagnostics
+/// never promote a result to RolledBack or authorize replay; the backend disposition does.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupResources {
+    None,
+    WorkerStartup,
+    Unproven,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub enum StartupCleanup {
+    NotNeeded,
+    Required,
+    Confirmed,
+    Failed(StartupCleanupFailure),
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+pub struct StartupResourceFacts {
+    pub resources: StartupResources,
+    pub cleanup: StartupCleanup,
+    pub durable_effects_possible: bool,
+}

crates/runtime-api/src/lib.rs

@@ -10,6 +10,11 @@ use protocol::{CompletionEntry, CompletionKind, Segment, WorkerId, WorkerStateSn
 use serde::{Deserialize, Serialize};
 use workdir::workspace::{MaterializerKind, RuntimeWorkerRef, RuntimeWorkingDirectorySummary};
 
+pub use protocol::startup_failure::{
+    RuntimeStartupCause, StartupCleanupFailure, StartupFailureAttachment, StartupFailurePhase,
+    StartupPersistenceOutcome, StartupSecondaryFailure, StartupStage,
+};
+
 pub const RUNTIME_API_VERSION: &str = "v1";
 pub const RUNTIME_API_BASE_PATH: &str = "/v1";
 /// Runtime transport classification for a materialized Repository source.

crates/session-store/Cargo.toml

@@ -6,6 +6,7 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+error-contract.workspace = true
 base64.workspace = true
 agen = { workspace = true }
 fs4.workspace = true

crates/worker-runtime/Cargo.toml

@@ -27,6 +27,7 @@ http-server = ["dep:axum", "dep:tower", "dep:reqwest"]
 ws-server = ["http-server", "axum/ws", "tokio/sync"]
 
 [dependencies]
+error-contract.workspace = true
 agen.workspace = true
 async-trait.workspace = true
 base64.workspace = true

crates/worker-runtime/src/execution.rs

@@ -94,6 +94,15 @@ pub struct WorkerExecutionResult {
     pub submission: Option<WorkerSubmissionAck>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub notification: Option<WorkerNotificationAck>,
+    /// Projected primary facts. Never stop, cleanup, or retry evidence.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub startup_failure: Option<protocol::startup_failure::StartupFailureAttachment>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub secondary_failures: Vec<protocol::startup_failure::StartupSecondaryFailure>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub startup_resources: Option<protocol::startup_failure::StartupResourceFacts>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub startup_failure_persistence: Option<protocol::startup_failure::StartupPersistenceOutcome>,
 }
 
 /// Backend result class for a Worker execution operation.
@@ -116,6 +125,10 @@ impl WorkerExecutionResult {
             message: None,
             submission: None,
             notification: None,
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
@@ -141,6 +154,10 @@ impl WorkerExecutionResult {
                 disposition,
             }),
             notification: None,
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
@@ -157,6 +174,10 @@ impl WorkerExecutionResult {
             notification: Some(WorkerNotificationAck {
                 notification_request_id: notification_request_id.into(),
             }),
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
@@ -168,6 +189,10 @@ impl WorkerExecutionResult {
             message: Some(message.into()),
             submission: None,
             notification: None,
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
@@ -179,6 +204,10 @@ impl WorkerExecutionResult {
             message: Some(message.into()),
             submission: None,
             notification: None,
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
@@ -190,6 +219,10 @@ impl WorkerExecutionResult {
             message: Some(message.into()),
             submission: None,
             notification: None,
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
@@ -201,9 +234,36 @@ impl WorkerExecutionResult {
             message: Some(message.into()),
             submission: None,
             notification: None,
+            startup_failure: None,
+            secondary_failures: Vec::new(),
+            startup_resources: None,
+            startup_failure_persistence: None,
         }
     }
 
+    pub fn with_startup_failure(
+        mut self,
+        failure: protocol::startup_failure::StartupFailureAttachment,
+    ) -> Self {
+        self.startup_failure = Some(failure);
+        self
+    }
+
+    pub fn startup_failure<D: error_contract::WireDetails>(
+        &self,
+    ) -> Option<error_contract::DecodeFailure<D>> {
+        self.startup_failure
+            .as_ref()
+            .map(|failure| failure.decode())
+    }
+
+    pub fn push_secondary_failure(
+        &mut self,
+        failure: protocol::startup_failure::StartupSecondaryFailure,
+    ) {
+        self.secondary_failures.push(failure);
+    }
+
     pub fn is_accepted(&self) -> bool {
         self.outcome == WorkerExecutionOutcome::Accepted
     }

crates/worker/Cargo.toml

@@ -10,6 +10,7 @@ autobins = false
 default = []
 
 [dependencies]
+error-contract.workspace = true
 async-trait = { workspace = true }
 agen = { workspace = true, features = ["codex"] }
 session-store = { workspace = true }