From 9d8175f52e4537f2937005fc16c81496a1dcb958 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 11:01:56 +0900 Subject: [PATCH] Add projected startup cause and secondary failure contracts --- Cargo.lock | 3 + crates/protocol/src/lib.rs | 1 + crates/protocol/src/startup_failure.rs | 231 +++++++++++++++++++++++++ crates/runtime-api/src/lib.rs | 5 + crates/session-store/Cargo.toml | 1 + crates/worker-runtime/Cargo.toml | 1 + crates/worker-runtime/src/execution.rs | 60 +++++++ crates/worker/Cargo.toml | 1 + 8 files changed, 303 insertions(+) create mode 100644 crates/protocol/src/startup_failure.rs diff --git a/Cargo.lock b/Cargo.lock index 30a5d87ca..52e90cef5 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 7f7f9a096..b8c386bf5 100644 --- a/crates/protocol/src/lib.rs +++ b/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; diff --git a/crates/protocol/src/startup_failure.rs b/crates/protocol/src/startup_failure.rs new file mode 100644 index 000000000..520f05ef2 --- /dev/null +++ b/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( + failure: &ProjectedFailure, + ) -> Result { + 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(&self) -> DecodeFailure { + 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, _: &()) + -> Result, 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, +} diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index a205e7893..c7900fd27 100644 --- a/crates/runtime-api/src/lib.rs +++ b/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. diff --git a/crates/session-store/Cargo.toml b/crates/session-store/Cargo.toml index cc697cef3..d86c6f6a9 100644 --- a/crates/session-store/Cargo.toml +++ b/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 diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index d6fe83605..4ab827331 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/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 diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 72c22e46e..952d4cafe 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -94,6 +94,15 @@ pub struct WorkerExecutionResult { pub submission: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub notification: Option, + /// Projected primary facts. Never stop, cleanup, or retry evidence. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_failure: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub secondary_failures: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_resources: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_failure_persistence: Option, } /// 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( + &self, + ) -> Option> { + 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 } diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 139758d44..837eda114 100644 --- a/crates/worker/Cargo.toml +++ b/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 }