From c4a34cf0ccf03fdb2b061e9068bca4ca1b72d8a9 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 15:25:24 +0900 Subject: [PATCH 1/2] feat: retain typed session registration causes across host startup --- crates/protocol/src/lib.rs | 1 + crates/protocol/src/session_failure.rs | 620 ++++++++++++++++++ crates/server-api/src/lib.rs | 13 + crates/standalone/src/subjektiv.rs | 76 ++- crates/standalone/src/subjektiv/tests.rs | 50 ++ crates/worker/src/feature/builtin/journal.rs | 4 +- crates/worker/src/startup_failure.rs | 72 ++ crates/worker/src/subjektiv.rs | 118 +++- crates/worker/src/worker.rs | 191 ++---- crates/workspace-server/src/server.rs | 187 +++++- .../src/server/session_registration.rs | 212 ++++++ .../src/server/subjektiv_journal.rs | 34 +- openapi/server-api.json | 17 +- .../src/lib/generated/companion-api.ts | 6 +- .../src/lib/generated/repository-api.ts | 6 +- .../src/lib/generated/runtime-api.ts | 6 +- web/workspace/src/lib/generated/ticket-api.ts | 6 +- .../src/lib/generated/workdir-api.ts | 6 +- .../src/lib/generated/worker-launch-api.ts | 6 +- 19 files changed, 1414 insertions(+), 217 deletions(-) create mode 100644 crates/protocol/src/session_failure.rs create mode 100644 crates/workspace-server/src/server/session_registration.rs diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index b8c386bf5..1cc668265 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 session_failure; pub mod startup_failure; #[cfg(feature = "stream")] pub mod stream; diff --git a/crates/protocol/src/session_failure.rs b/crates/protocol/src/session_failure.rs new file mode 100644 index 000000000..26104d755 --- /dev/null +++ b/crates/protocol/src/session_failure.rs @@ -0,0 +1,620 @@ +//! Typed session-registration failures, not lifecycle or retry authority. +//! +//! Causes contain only fixed enum values. Stored or transferred attachments are +//! observations, never permission to disclose: Browser/Tool projections require +//! the caller's current visibility decision and fail closed without it. + +use error_contract::{ + Browser, Category, ContractError, Disclosure, FailureCause, FailureFacts, FailureIdentity, + InternalTransfer, OperationContext, ProjectFor, Storage, Tool, WireDetails, project_failure, +}; +use serde::{Deserialize, Serialize}; + +use crate::startup_failure::StartupFailureAttachment; + +#[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 SessionRegistrationReason { + PermissionDenied, + AttributionMissing, + BindingMismatch, + Transport, + OutcomeUnknown, + InvalidResponse, +} + +#[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 SessionPrincipal { + Operator, + Subject, + Unknown, +} + +#[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 enum SessionOperation { + #[serde(rename = "session.record")] + Record, +} + +/// No account/session identifiers, paths, or free-form diagnostics are retained. +#[derive(Clone, Copy, 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 SessionRegistrationCause { + pub reason: SessionRegistrationReason, + pub principal: SessionPrincipal, + pub required_operation: SessionOperation, +} + +impl SessionRegistrationCause { + pub const fn denied(principal: SessionPrincipal) -> Self { + Self::new(SessionRegistrationReason::PermissionDenied, principal) + } + + pub const fn attribution_missing() -> Self { + Self::new( + SessionRegistrationReason::AttributionMissing, + SessionPrincipal::Subject, + ) + } + + pub const fn binding_mismatch() -> Self { + Self::new( + SessionRegistrationReason::BindingMismatch, + SessionPrincipal::Subject, + ) + } + + pub const fn transport() -> Self { + Self::new( + SessionRegistrationReason::Transport, + SessionPrincipal::Unknown, + ) + } + + pub const fn outcome_unknown() -> Self { + Self::new( + SessionRegistrationReason::OutcomeUnknown, + SessionPrincipal::Unknown, + ) + } + + pub const fn invalid_response() -> Self { + Self::new( + SessionRegistrationReason::InvalidResponse, + SessionPrincipal::Unknown, + ) + } + + const fn new(reason: SessionRegistrationReason, principal: SessionPrincipal) -> Self { + Self { + reason, + principal, + required_operation: SessionOperation::Record, + } + } + + fn explanation(&self) -> &'static str { + match self.reason { + SessionRegistrationReason::PermissionDenied => { + "Session registration permission denied." + } + SessionRegistrationReason::AttributionMissing => { + "Session registration requires Subject attribution." + } + SessionRegistrationReason::BindingMismatch => { + "Session registration Subject binding does not match." + } + SessionRegistrationReason::Transport => "Session registration transport failed.", + SessionRegistrationReason::OutcomeUnknown => "Session registration outcome is unknown.", + SessionRegistrationReason::InvalidResponse => { + "Session registration returned an invalid response." + } + } + } + + fn disclosure(&self, facts: &FailureFacts) -> Disclosure { + let mut disclosure = Disclosure::new(*self, self.explanation()); + for context in &facts.context().operations { + if matches!( + (context.operation.as_str(), context.stage.as_str()), + ("session.record", "authorization" | "registration") + | ("worker_startup", "binding" | "runtime") + ) { + let mut context = context.clone(); + context.operation_ref = None; + disclosure.context.operations.push(context); + } + } + disclosure + } +} + +impl FailureCause for SessionRegistrationCause { + fn identity(&self) -> FailureIdentity { + let (reason, category) = match self.reason { + SessionRegistrationReason::PermissionDenied => { + ("permission_denied", Category::Authorization) + } + SessionRegistrationReason::AttributionMissing => { + ("attribution_missing", Category::Authorization) + } + SessionRegistrationReason::BindingMismatch => { + ("binding_mismatch", Category::Authorization) + } + SessionRegistrationReason::Transport => ("transport", Category::Dependency), + SessionRegistrationReason::OutcomeUnknown => ("outcome_unknown", Category::Unknown), + SessionRegistrationReason::InvalidResponse => ("invalid_response", Category::Internal), + }; + FailureIdentity { + domain: "session_registration", + reason, + category, + details_version: 1, + } + } + + fn validate(&self) -> Result<(), ContractError> { + // All fields are closed enums; arbitrary diagnostic data cannot enter. + Ok(()) + } +} + +impl WireDetails for SessionRegistrationCause { + fn supports(domain: &str, reason: &str, version: u16) -> bool { + domain == "session_registration" + && version == 1 + && matches!( + reason, + "permission_denied" + | "attribution_missing" + | "binding_mismatch" + | "transport" + | "outcome_unknown" + | "invalid_response" + ) + } +} + +impl std::fmt::Display for SessionRegistrationCause { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.explanation()) + } +} + +impl std::error::Error for SessionRegistrationCause {} + +/// A current caller-supplied visibility decision, not a serialized capability. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct SessionFailureVisibility { + pub can_view_cause: bool, +} + +impl ProjectFor for SessionRegistrationCause { + type Authorization = (); + type Details = Self; + + fn disclose( + &self, + facts: &FailureFacts, + _: &(), + ) -> Result, ContractError> { + Ok(self.disclosure(facts)) + } +} + +impl ProjectFor for SessionRegistrationCause { + type Authorization = (); + type Details = Self; + + fn disclose( + &self, + facts: &FailureFacts, + _: &(), + ) -> Result, ContractError> { + Ok(self.disclosure(facts)) + } +} + +impl ProjectFor for SessionRegistrationCause { + type Authorization = SessionFailureVisibility; + type Details = Self; + + fn disclose( + &self, + facts: &FailureFacts, + authorization: &SessionFailureVisibility, + ) -> Result, ContractError> { + if !authorization.can_view_cause { + // Redaction must not substitute a different reason/category identity. + return Err(ContractError::InvalidDetails); + } + Ok(self.disclosure(facts)) + } +} + +impl ProjectFor for SessionRegistrationCause { + type Authorization = SessionFailureVisibility; + type Details = Self; + + fn disclose( + &self, + facts: &FailureFacts, + authorization: &SessionFailureVisibility, + ) -> Result, ContractError> { + if !authorization.can_view_cause { + return Err(ContractError::InvalidDetails); + } + Ok(self.disclosure(facts)) + } +} + +/// Attach fixed registration/binding observations using internal-transfer policy. +/// This does not assert registration rollback, retry safety, or startup ownership. +pub fn session_registration_attachment( + cause: SessionRegistrationCause, +) -> Result { + let mut facts = FailureFacts::new(cause)?; + facts.with_context(OperationContext::new("session.record", "registration")?)?; + facts.with_context(OperationContext::new("worker_startup", "binding")?)?; + let projected = project_failure::(&facts, &())?; + StartupFailureAttachment::from_projected(&projected) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::startup_failure::StoredStartupFailureAttachment; + use error_contract::{ + Audience, Budget, CausalLink, CorrelationRef, DecodeFailure, ProjectedFailure, + decode_failure, encode_projected_failure, + }; + use serde_json::json; + + fn budget() -> Budget { + Budget::new( + error_contract::MAX_FAILURE_BYTES, + error_contract::MAX_FAILURE_BYTES, + ) + } + + fn cases() -> [( + SessionRegistrationCause, + &'static str, + Category, + &'static str, + ); 6] { + [ + ( + SessionRegistrationCause::denied(SessionPrincipal::Operator), + "permission_denied", + Category::Authorization, + "Session registration permission denied.", + ), + ( + SessionRegistrationCause::attribution_missing(), + "attribution_missing", + Category::Authorization, + "Session registration requires Subject attribution.", + ), + ( + SessionRegistrationCause::binding_mismatch(), + "binding_mismatch", + Category::Authorization, + "Session registration Subject binding does not match.", + ), + ( + SessionRegistrationCause::transport(), + "transport", + Category::Dependency, + "Session registration transport failed.", + ), + ( + SessionRegistrationCause::outcome_unknown(), + "outcome_unknown", + Category::Unknown, + "Session registration outcome is unknown.", + ), + ( + SessionRegistrationCause::invalid_response(), + "invalid_response", + Category::Internal, + "Session registration returned an invalid response.", + ), + ] + } + + #[test] + fn constructors_serde_identity_and_fixed_explanations() { + for (cause, reason, category, explanation) in cases() { + let principal = match cause.reason { + SessionRegistrationReason::PermissionDenied => "operator", + SessionRegistrationReason::AttributionMissing + | SessionRegistrationReason::BindingMismatch => "subject", + _ => "unknown", + }; + let value = serde_json::to_value(cause).unwrap(); + assert_eq!( + value, + json!({ + "reason": reason, + "principal": principal, + "required_operation": "session.record" + }) + ); + assert_eq!( + serde_json::from_value::(value).unwrap(), + cause + ); + assert_eq!(cause.required_operation, SessionOperation::Record); + assert_eq!( + cause.identity(), + FailureIdentity { + domain: "session_registration", + reason, + category, + details_version: 1, + } + ); + assert!(SessionRegistrationCause::supports( + "session_registration", + reason, + 1 + )); + assert_eq!(cause.to_string(), explanation); + assert!(std::error::Error::source(&cause).is_none()); + let facts = FailureFacts::new(cause).unwrap(); + let projected = project_failure::(&facts, &()).unwrap(); + assert_eq!(projected.explanation(), explanation); + let encoded = encode_projected_failure(&projected, budget()).unwrap(); + match decode_failure::(encoded.as_bytes(), budget()) { + DecodeFailure::Known(received) => assert_eq!(received.facts().cause(), &cause), + _ => panic!("expected known session-registration failure"), + } + } + for (principal, wire) in [ + (SessionPrincipal::Operator, "operator"), + (SessionPrincipal::Subject, "subject"), + (SessionPrincipal::Unknown, "unknown"), + ] { + let cause = SessionRegistrationCause::denied(principal); + assert_eq!(cause.principal, principal); + assert_eq!( + serde_json::to_value(cause).unwrap()["principal"], + json!(wire) + ); + } + assert!(!SessionRegistrationCause::supports( + "worker_startup", + "transport", + 1 + )); + assert!(!SessionRegistrationCause::supports( + "session_registration", + "future", + 1 + )); + assert!(!SessionRegistrationCause::supports( + "session_registration", + "transport", + 2 + )); + } + + #[test] + fn serde_rejects_untyped_diagnostic_fields_and_unknown_enum_values() { + let valid = serde_json::to_value(SessionRegistrationCause::attribution_missing()).unwrap(); + for (field, value) in [ + ("reason", json!("raw_error")), + ("principal", json!("account-secret")), + ("required_operation", json!("session.delete")), + ("diagnostic", json!("/private/path TEST_ONLY_SECRET")), + ] { + let mut changed = valid.clone(); + changed[field] = value; + assert!(serde_json::from_value::(changed).is_err()); + } + for field in ["reason", "principal", "required_operation"] { + let mut changed = valid.clone(); + changed.as_object_mut().unwrap().remove(field); + assert!(serde_json::from_value::(changed).is_err()); + } + } + + #[test] + fn codec_distinguishes_unsupported_and_corrupt_and_replaces_received_prose() { + let facts = FailureFacts::new(SessionRegistrationCause::transport()).unwrap(); + let projected = project_failure::(&facts, &()).unwrap(); + let encoded = encode_projected_failure(&projected, budget()).unwrap(); + let original: serde_json::Value = serde_json::from_slice(encoded.as_bytes()).unwrap(); + for (field, value) in [ + ("domain", json!("future_domain")), + ("reason", json!("future_reason")), + ("details_version", json!(2)), + ] { + let mut changed = original.clone(); + changed[field] = value; + assert!(matches!( + decode_failure::( + &serde_json::to_vec(&changed).unwrap(), + budget() + ), + DecodeFailure::Unsupported(_) + )); + } + for (field, value) in [ + ("reason", json!("permission_denied")), + ("category", json!("authorization")), + ] { + let mut changed = original.clone(); + changed[field] = value; + assert!(matches!( + decode_failure::( + &serde_json::to_vec(&changed).unwrap(), + budget() + ), + DecodeFailure::Corrupt(ContractError::InvalidDetails) + )); + } + let mut changed = original; + changed["explanation"] = json!("TEST_ONLY_SECRET received prose"); + changed["context"]["diagnostics"] = json!([ + {"code": "TEST_ONLY_SECRET", "explanation": "TEST_ONLY_SECRET diagnostic"} + ]); + let bytes = serde_json::to_vec(&changed).unwrap(); + let DecodeFailure::Known(received) = + decode_failure::(&bytes, budget()) + else { + panic!("expected known failure with untrusted supplementary fields"); + }; + let browser = project_failure::( + received.facts(), + &SessionFailureVisibility { + can_view_cause: true, + }, + ) + .unwrap(); + assert_eq!( + browser.explanation(), + "Session registration transport failed." + ); + assert!(browser.context().diagnostics.is_empty()); + let encoded = encode_projected_failure(&browser, budget()).unwrap(); + assert!( + !std::str::from_utf8(encoded.as_bytes()) + .unwrap() + .contains("TEST_ONLY_SECRET") + ); + } + + #[test] + fn attachment_storage_and_current_visibility_preserve_identity_without_granting_access() { + let allowed = SessionFailureVisibility { + can_view_cause: true, + }; + let denied = SessionFailureVisibility::default(); + for (cause, _, _, _) in cases() { + let attachment = session_registration_attachment(cause).unwrap(); + match attachment.decode::() { + DecodeFailure::Known(received) => { + assert_eq!(received.facts().cause(), &cause); + assert_eq!( + received.facts().context().operations, + vec![ + OperationContext::new("session.record", "registration").unwrap(), + OperationContext::new("worker_startup", "binding").unwrap(), + ] + ); + } + _ => panic!("expected typed startup attachment"), + } + let storage = attachment + .project_for::(&()) + .unwrap(); + let stored = StoredStartupFailureAttachment::from_projected(&storage).unwrap(); + let loaded: StoredStartupFailureAttachment = + serde_json::from_str(&serde_json::to_string(&stored).unwrap()).unwrap(); + let browser = loaded + .project_for::(&allowed) + .unwrap(); + let tool = loaded + .project_for::(&allowed) + .unwrap(); + assert_eq!(browser.details(), &cause); + assert_eq!(tool.details(), &cause); + assert_eq!(browser.identity(), cause.identity()); + assert_eq!(tool.identity(), cause.identity()); + assert!(matches!( + loaded.project_for::(&denied), + Err(ContractError::InvalidDetails) + )); + assert!(matches!( + loaded.project_for::(&denied), + Err(ContractError::InvalidDetails) + )); + // Even a previous public projection cannot serve as current permission. + let previously_public = StartupFailureAttachment::from_projected(&browser).unwrap(); + assert!(matches!( + previously_public.project_for::(&denied), + Err(ContractError::InvalidDetails) + )); + assert!(matches!( + previously_public.project_for::(&denied), + Err(ContractError::InvalidDetails) + )); + } + } + + fn assert_redacted(projected: &ProjectedFailure) { + assert_eq!( + projected.context().operations, + vec![ + OperationContext::new("session.record", "authorization").unwrap(), + OperationContext::new("session.record", "registration").unwrap(), + OperationContext::new("worker_startup", "binding").unwrap(), + OperationContext::new("worker_startup", "runtime").unwrap(), + ] + ); + assert!(projected.context().correlation.is_none()); + assert!(projected.context().causes.is_empty()); + assert!(projected.context().diagnostics.is_empty()); + let encoded = encode_projected_failure(projected, budget()).unwrap(); + let wire = std::str::from_utf8(encoded.as_bytes()).unwrap(); + for secret in ["TEST_ONLY_SECRET", "/private/path", "raw_stage"] { + assert!(!wire.contains(secret)); + } + } + + #[test] + fn every_audience_allowlists_operation_stage_pairs_and_drops_references() { + let mut facts = FailureFacts::new(SessionRegistrationCause::binding_mismatch()).unwrap(); + for (operation, stage) in [ + ("session.record", "authorization"), + ("session.record", "registration"), + ("worker_startup", "binding"), + ("worker_startup", "runtime"), + ("session.record", "binding"), + ("worker_startup", "authorization"), + ("/private/path", "registration"), + ("session.record", "raw_stage"), + ] { + let mut context = OperationContext::new(operation, stage).unwrap(); + context.operation_ref = Some("TEST_ONLY_SECRET operation reference".into()); + facts.with_context(context).unwrap(); + } + facts + .set_correlation(CorrelationRef::new("TEST_ONLY_SECRET correlation").unwrap()) + .unwrap(); + facts + .with_causal_link(CausalLink { + domain: "TEST_ONLY_SECRET domain".into(), + reason: "TEST_ONLY_SECRET reason".into(), + correlation: CorrelationRef::new("TEST_ONLY_SECRET causal reference").unwrap(), + }) + .unwrap(); + let allowed = SessionFailureVisibility { + can_view_cause: true, + }; + assert_redacted(&project_failure::(&facts, &()).unwrap()); + assert_redacted(&project_failure::(&facts, &()).unwrap()); + assert_redacted(&project_failure::(&facts, &allowed).unwrap()); + assert_redacted(&project_failure::(&facts, &allowed).unwrap()); + // Projection selects output without mutating the owner's facts. + assert_eq!(facts.context().operations.len(), 8); + assert!( + facts + .context() + .operations + .iter() + .all(|context| context.operation_ref.is_some()) + ); + } +} diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 01a9f9988..f7e83d0d9 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -88,6 +88,10 @@ pub struct RepositoryApiError { pub message: String, #[serde(default)] pub diagnostics: Vec, + /// An origin-projected error-contract attachment, not raw HTTP error text. + /// Receiving it does not authorize redisclosure to a different audience. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure: Option, #[serde(skip, default = "default_repository_error_status")] status: u16, } @@ -107,9 +111,18 @@ impl RepositoryApiError { error: error.into(), message: message.into(), diagnostics, + failure: None, status, } } + + pub fn with_failure( + mut self, + failure: protocol::startup_failure::StartupFailureAttachment, + ) -> Self { + self.failure = Some(failure); + self + } } impl api_macros::HttpError for RepositoryApiError { diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index 371ae8495..82a47b836 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -30,6 +30,7 @@ use std::sync::{ use feature_storage::FeatureStorage; use fs4::fs_std::FileExt; +use protocol::session_failure::{SessionPrincipal, SessionRegistrationCause}; use serde::{Deserialize, Serialize}; use server_api::*; use subjektiv::{SubjectRecord, SubjectRole, SubjectSessionAttribution, SubjektivStore}; @@ -418,6 +419,49 @@ impl LocalSubjectHost { } Ok(()) } + fn registration_authority( + &self, + context: &SubjektivHostContext, + ) -> Result<(), SubjektivHostError> { + if !self.live.load(Ordering::Acquire) { + return Err(SubjektivHostError::SessionRegistration { + cause: SessionRegistrationCause::denied(SessionPrincipal::Subject), + source: None, + }); + } + if !self + .catalog + .store + .journal_writer_active(&self.writer) + .map_err(|error| { + SubjektivHostError::session_registration( + SessionRegistrationCause::outcome_unknown(), + error, + ) + })? + { + return Err(SubjektivHostError::SessionRegistration { + cause: SessionRegistrationCause::binding_mismatch(), + source: None, + }); + } + if context.worker_id != self.worker_key && context.worker_id != self.worker_id.to_string() { + return Err(SubjektivHostError::SessionRegistration { + cause: SessionRegistrationCause::binding_mismatch(), + source: None, + }); + } + context + .session_id + .parse::() + .map_err(|error| { + SubjektivHostError::session_registration( + SessionRegistrationCause::invalid_response(), + error, + ) + })?; + Ok(()) + } fn attribution( &self, context: &SubjektivHostContext, @@ -427,15 +471,16 @@ impl LocalSubjectHost { let store = &self.catalog.store; if let Some(record) = store .session_attribution(&context.session_id) - .map_err(domain_error)? + .map_err(registration_domain_error)? { if record.subject_id != self.binding.subject_id || record.runtime_id != self.binding.scope_id || record.worker_id != self.worker_id.to_string() { - return Err(SubjektivHostError::InvalidArgument( - "Session does not belong to this local Subject/Worker scope".into(), - )); + return Err(SubjektivHostError::SessionRegistration { + cause: SessionRegistrationCause::binding_mismatch(), + source: None, + }); } store .bind_journal_session(&self.writer, record.clone()) @@ -443,9 +488,10 @@ impl LocalSubjectHost { return Ok(record); } if !create { - return Err(SubjektivHostError::InvalidArgument( - "Session attribution is unavailable".into(), - )); + return Err(SubjektivHostError::SessionRegistration { + cause: SessionRegistrationCause::attribution_missing(), + source: None, + }); } // The capability is bound before controller startup. The Worker supplies // the Session it owns; models cannot call this Host API or select one. @@ -455,10 +501,10 @@ impl LocalSubjectHost { self.worker_id.to_string(), &context.session_id, ) - .map_err(domain_error)?; + .map_err(registration_domain_error)?; let record = store .record_session_attribution(record) - .map_err(domain_error)?; + .map_err(registration_domain_error)?; store .bind_journal_session(&self.writer, record.clone()) .map_err(journal::journal_error)?; @@ -466,6 +512,17 @@ impl LocalSubjectHost { } } +fn registration_domain_error(error: subjektiv::SubjektivError) -> SubjektivHostError { + let cause = match &error { + subjektiv::SubjektivError::SessionSubjectConflict { .. } + | subjektiv::SubjektivError::SessionAttributionConflict { .. } => { + SessionRegistrationCause::binding_mismatch() + } + _ => SessionRegistrationCause::outcome_unknown(), + }; + SubjektivHostError::session_registration(cause, error) +} + fn domain_error(error: subjektiv::SubjektivError) -> SubjektivHostError { operation_error(subjektiv::api::subjektiv_store_error(error)) } @@ -1036,6 +1093,7 @@ impl SubjektivHost for LocalSubjectHost { context: &SubjektivHostContext, create_if_missing: bool, ) -> Result { + self.registration_authority(context)?; let attribution = self.attribution(context, create_if_missing)?; self.reserve_journal_session(context)?; Ok(SubjektivRecordSessionResponse { diff --git a/crates/standalone/src/subjektiv/tests.rs b/crates/standalone/src/subjektiv/tests.rs index 9104f0795..7966751ef 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -267,3 +267,53 @@ fn consolidation_operation_policy_is_attempt_bound_and_does_not_inherit_body_rig .is_err() ); } + +#[test] +fn session_registration_uses_real_capability_and_attribution_decisions() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("state"); + let catalog = StandaloneSubjects::open(&root).unwrap(); + let subject = catalog.create("registration-owner", None).unwrap(); + let id = protocol::WorkerId::now_v7(); + let mut connection = catalog.connect(&subject.id, id).unwrap(); + let port = connection.host.clone(); + let context = SubjektivHostContext { + worker_id: format!("standalone-{id}"), + session_id: session_store::SessionId::now_v7().to_string(), + }; + assert_eq!( + port.record_session(&context, false) + .unwrap_err() + .session_registration_cause(), + SessionRegistrationCause::attribution_missing(), + ); + let foreign = SubjektivHostContext { + worker_id: "unbound-private-worker".into(), + ..context.clone() + }; + assert_eq!( + port.record_session(&foreign, true) + .unwrap_err() + .session_registration_cause(), + SessionRegistrationCause::binding_mismatch(), + ); + let success = port.record_session(&context, true).unwrap(); + assert_eq!(success.subject_id, subject.id); + assert_eq!(success.session_id, context.session_id); + assert!(port.record_session(&context, false).is_ok()); + let reopened = StandaloneSubjects::open_existing(&root).unwrap(); + let actual = reopened + .store + .session_attribution(&context.session_id) + .unwrap() + .unwrap(); + assert_eq!(actual.subject_id, subject.id); + assert_eq!(actual.worker_id, id.to_string()); + connection.close().unwrap(); + assert_eq!( + port.record_session(&context, true) + .unwrap_err() + .session_registration_cause(), + SessionRegistrationCause::denied(SessionPrincipal::Subject), + ); +} diff --git a/crates/worker/src/feature/builtin/journal.rs b/crates/worker/src/feature/builtin/journal.rs index 9a4ab8499..3f805af1a 100644 --- a/crates/worker/src/feature/builtin/journal.rs +++ b/crates/worker/src/feature/builtin/journal.rs @@ -83,7 +83,9 @@ impl JournalCompactionFailure { SubjektivHostError::PermissionDenied(_) => Self::SnapshotPermissionDenied, SubjektivHostError::Conflict { .. } => Self::SnapshotConflict, SubjektivHostError::InvalidArgument(_) => Self::SnapshotInvalidArgument, - SubjektivHostError::Unavailable(_) => Self::SnapshotUnavailable, + SubjektivHostError::Unavailable(_) | SubjektivHostError::SessionRegistration { .. } => { + Self::SnapshotUnavailable + } } } diff --git a/crates/worker/src/startup_failure.rs b/crates/worker/src/startup_failure.rs index dc856231a..2c48c7138 100644 --- a/crates/worker/src/startup_failure.rs +++ b/crates/worker/src/startup_failure.rs @@ -9,6 +9,18 @@ use protocol::startup_failure::{ }; impl WorkerError { + pub fn session_registration_cause( + &self, + ) -> Option { + match self { + Self::SessionRegistration { source } => Some(source.session_registration_cause()), + Self::StartupFailure { source, .. } + | Self::ConstructionCleanup { source, .. } + | Self::RestoreActivationEffects { source } => source.session_registration_cause(), + _ => None, + } + } + pub fn with_startup_failure(self, failure: StartupFailureAttachment) -> Self { Self::StartupFailure { source: Box::new(self), @@ -55,6 +67,9 @@ impl WorkerError { if let Some(failure) = self.startup_failure() { return Ok(failure.clone()); } + if let Some(cause) = self.session_registration_cause() { + return protocol::session_failure::session_registration_attachment(cause); + } let reason = match self { Self::ConstructionCleanup { source, .. } | Self::RestoreActivationEffects { source } => return source.startup_failure_at(stage), @@ -91,6 +106,63 @@ pub fn runtime_startup_attachment( StartupFailureAttachment::from_projected(&projected) } +/// Dispatch by the domain attested in process, not by parsing attachment fields. +pub(crate) fn persist_binding_startup_failure( + store: &S, + expected: &session_store::worker_metadata::StartupFailureExpectation, + source: WorkerError, +) -> WorkerError { + if source.session_registration_cause().is_some() { + persist_startup_failure::<_, protocol::session_failure::SessionRegistrationCause>( + store, + expected, + source, + StartupStage::Binding, + &(), + ) + } else { + persist_startup_failure::<_, RuntimeStartupCause>( + store, + expected, + source, + StartupStage::Binding, + &(), + ) + } +} + +/// An attribution-state write failure supplements the registration decision. It +/// neither establishes the write's absence nor replaces the primary Error source. +pub(crate) fn retain_attribution_save_failure( + source: WorkerError, + saved: Result<(), WorkerError>, +) -> WorkerError { + use protocol::startup_failure::{StartupFailurePhase, StartupPersistenceOutcome}; + let Err(error) = saved else { + return source; + }; + let outcome = match error { + WorkerError::SubjektivSessionAttribution { + state: "attribution_conflict", + .. + } => StartupPersistenceOutcome::Conflict, + WorkerError::WorkerStore(_) => StartupPersistenceOutcome::OutcomeUnknown, + _ => StartupPersistenceOutcome::NotSaved, + }; + let failure = source + .startup_failure_at(StartupStage::Binding) + .expect("fixed Session cause fits the shared codec budget"); + WorkerError::StartupFailure { + source: Box::new(source), + failure, + secondary: vec![StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::MetadataSave, + outcome, + }], + metadata_save: None, + } +} + /// Bind a caller-owned cause to the exact pre-observed Session/attribution. Saving /// never replaces the original error, changes ownership, or issues retry authority. pub fn persist_startup_failure( diff --git a/crates/worker/src/subjektiv.rs b/crates/worker/src/subjektiv.rs index d8ace3664..3e7c9b62d 100644 --- a/crates/worker/src/subjektiv.rs +++ b/crates/worker/src/subjektiv.rs @@ -64,6 +64,12 @@ pub struct SubjektivHostSettings { #[derive(Debug, thiserror::Error)] pub enum SubjektivHostError { + #[error("{cause}")] + SessionRegistration { + cause: protocol::session_failure::SessionRegistrationCause, + #[source] + source: Option>, + }, #[error("{message}")] Conflict { code: String, message: String }, #[error("{0}")] @@ -74,6 +80,30 @@ pub enum SubjektivHostError { Unavailable(String), } +impl SubjektivHostError { + pub fn session_registration( + cause: protocol::session_failure::SessionRegistrationCause, + source: impl std::error::Error + Send + Sync + 'static, + ) -> Self { + Self::SessionRegistration { + cause, + source: Some(Box::new(source)), + } + } + + pub fn session_registration_cause( + &self, + ) -> protocol::session_failure::SessionRegistrationCause { + use protocol::session_failure::SessionRegistrationCause; + match self { + Self::SessionRegistration { cause, .. } => cause.clone(), + // Old Host implementations have not attested a domain verdict. Their prose + // must not become a permission or binding decision at the caller. + _ => SessionRegistrationCause::outcome_unknown(), + } + } +} + impl From for ToolError { fn from(error: SubjektivHostError) -> Self { match error { @@ -85,6 +115,9 @@ impl From for ToolError { Self::ExecutionFailed(format!("Subjektiv operation permission denied: {message}")) } SubjektivHostError::Unavailable(message) => Self::ExecutionFailed(message), + SubjektivHostError::SessionRegistration { cause, .. } => { + Self::ExecutionFailed(cause.to_string()) + } } } } @@ -249,6 +282,84 @@ impl BackendSubjektivHost { } } +/// Session registration is an internal operation boundary, not a Browser HTTP +/// classifier. Only the domain's codec attestation can establish a denial. +/// Neither legacy bodies nor transport error prose are retained in metadata. +pub(crate) fn record_backend_session( + client: &dyn WorkspaceClient, + session_id: &str, + create_if_missing: bool, +) -> Result { + use protocol::session_failure::SessionRegistrationCause; + let response = client + .execute_server_operation(WorkspaceServerOperation::SubjektivRecordSession( + SubjektivRecordSessionRequest { + session_id: session_id.into(), + create_if_missing, + }, + )) + .map_err(|error| { + let cause = match &error { + crate::WorkspaceClientError::Unavailable(_) => { + SessionRegistrationCause::transport() + } + crate::WorkspaceClientError::Request(_) => { + SessionRegistrationCause::outcome_unknown() + } + crate::WorkspaceClientError::InvalidPath(_) => { + SessionRegistrationCause::invalid_response() + } + }; + SubjektivHostError::session_registration(cause, error) + })?; + if !response.is_success() { + let cause = match serde_json::from_str::(&response.body) { + Ok(error) => match error.failure { + Some(failure) => match failure.decode::() { + error_contract::DecodeFailure::Known(received) => { + received.facts().cause().clone() + } + error_contract::DecodeFailure::Unsupported(_) => { + SessionRegistrationCause::outcome_unknown() + } + error_contract::DecodeFailure::Corrupt(_) => { + SessionRegistrationCause::invalid_response() + } + }, + None => SessionRegistrationCause::outcome_unknown(), + }, + Err(_) => SessionRegistrationCause::invalid_response(), + }; + return Err(SubjektivHostError::SessionRegistration { + cause, + source: None, + }); + } + let output = serde_json::from_str(&response.body).map_err(|error| { + SubjektivHostError::session_registration( + SessionRegistrationCause::invalid_response(), + error, + ) + })?; + validate_session_registration(output, session_id) +} + +pub(crate) fn validate_session_registration( + output: SubjektivRecordSessionResponse, + session_id: &str, +) -> Result { + if output.session_id != session_id + || output.subject_id.trim().is_empty() + || output.subject_id.chars().any(char::is_control) + { + return Err(SubjektivHostError::SessionRegistration { + cause: protocol::session_failure::SessionRegistrationCause::invalid_response(), + source: None, + }); + } + Ok(output) +} + fn decode( response: crate::WorkspaceResponse, ) -> Result { @@ -490,12 +601,7 @@ impl SubjektivHost for BackendSubjektivHost { context: &SubjektivHostContext, create_if_missing: bool, ) -> Result { - self.execute(WorkspaceServerOperation::SubjektivRecordSession( - SubjektivRecordSessionRequest { - session_id: context.session_id.clone(), - create_if_missing, - }, - )) + record_backend_session(self.client.as_ref(), &context.session_id, create_if_missing) } fn request_consolidation( &self, diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 707752550..a19c433dc 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -9406,14 +9406,6 @@ fn incomplete_subjektiv_attribution_state( } } -fn incomplete_subjektiv_attribution_label(create_if_missing: bool) -> &'static str { - if create_if_missing { - "outcome_unknown" - } else { - "legacy_unknown" - } -} - fn bounded_attribution_diagnostic(mut diagnostic: String) -> String { const MAX_CHARS: usize = 1024; if diagnostic.chars().count() > MAX_CHARS { @@ -9721,15 +9713,10 @@ where &mut expected, ) .map_err(|error| match expected { - Some(expected) => crate::startup_failure::persist_startup_failure::< - _, - protocol::startup_failure::RuntimeStartupCause, - >( + Some(expected) => crate::startup_failure::persist_binding_startup_failure( &self.store, &expected, error, - protocol::startup_failure::StartupStage::Binding, - &(), ), None => error, }) @@ -9855,145 +9842,55 @@ where pending_attribution.clone(), )?; let output = if self.subjektiv_host.is_some() { - let connection = self - .subjektiv_host_connection() - .map_err(|error| WorkerError::SubjektivSessionAttribution { - state: "failed", - message: error.to_string(), - })? - .ok_or_else(|| WorkerError::SubjektivSessionAttribution { - state: "failed", - message: "Session attribution requires enabled explicit Host capability".into(), - })?; - match connection - .host - .record_session(&connection.context, create_if_missing) - { - Ok(output) => output, - Err(error) => { - let message = bounded_attribution_diagnostic(error.to_string()); - let rejected = - !matches!(error, crate::subjektiv::SubjektivHostError::Unavailable(_)); - let state = if create_if_missing && rejected { - SubjektivSessionAttributionState::Failed { - session_id, - diagnostic: message.clone(), - } - } else { - incomplete_subjektiv_attribution_state( - session_id, - create_if_missing, - message.clone(), - ) - }; - persist_attribution(Some(&pending_attribution), state)?; - return Err(WorkerError::SubjektivSessionAttribution { - state: if create_if_missing && rejected { - "failed" - } else { - incomplete_subjektiv_attribution_label(create_if_missing) - }, - message, - }); - } - } + self.subjektiv_host_connection() + .map_err(|error| crate::subjektiv::SubjektivHostError::session_registration( + protocol::session_failure::SessionRegistrationCause::outcome_unknown(), error, + )) + .and_then(|connection| connection.ok_or_else(|| { + crate::subjektiv::SubjektivHostError::SessionRegistration { + cause: protocol::session_failure::SessionRegistrationCause::binding_mismatch(), + source: None, + } + })) + .and_then(|connection| connection.host.record_session(&connection.context, create_if_missing)) } else { - let response = match self.workspace_client().execute_server_operation( - WorkspaceServerOperation::SubjektivRecordSession( - server_api::SubjektivRecordSessionRequest { - session_id: session_id.to_string(), - create_if_missing, - }, - ), - ) { - Ok(response) => response, - Err(error) => { - let message = bounded_attribution_diagnostic(format!( - "Session attribution transport outcome is unknown: {error}" - )); - persist_attribution( - Some(&pending_attribution), - incomplete_subjektiv_attribution_state( - session_id, - create_if_missing, - message.clone(), - ), - )?; - return Err(WorkerError::SubjektivSessionAttribution { - state: incomplete_subjektiv_attribution_label(create_if_missing), - message, - }); - } - }; - if !response.is_success() { - let message = bounded_attribution_diagnostic(format!( - "Session attribution was rejected with HTTP {}: {}", - response.status, response.body - )); - let state = if create_if_missing { + crate::subjektiv::record_backend_session( + self.workspace_client(), &session_id.to_string(), create_if_missing, + ) + }.and_then(|output| crate::subjektiv::validate_session_registration(output, &session_id.to_string())); + let output = match output { + Ok(output) => output, + Err(error) => { + use protocol::session_failure::SessionRegistrationReason; + let cause = error.session_registration_cause(); + let rejected = matches!( + cause.reason, + SessionRegistrationReason::PermissionDenied + | SessionRegistrationReason::AttributionMissing + | SessionRegistrationReason::BindingMismatch + ); + // The legacy lifecycle diagnostic is a fixed domain label, never a + // transport body, credential, path or upstream Display string. + let diagnostic = cause.to_string(); + let next = if create_if_missing && rejected { SubjektivSessionAttributionState::Failed { session_id, - diagnostic: message.clone(), + diagnostic, } } else { - SubjektivSessionAttributionState::LegacyUnknown { + incomplete_subjektiv_attribution_state( session_id, - diagnostic: message.clone(), - } + create_if_missing, + diagnostic, + ) }; - persist_attribution(Some(&pending_attribution), state)?; - return Err(WorkerError::SubjektivSessionAttribution { - state: if create_if_missing { - "failed" - } else { - "legacy_unknown" - }, - message, - }); + let source = WorkerError::SessionRegistration { source: error }; + let saved = persist_attribution(Some(&pending_attribution), next); + return Err(crate::startup_failure::retain_attribution_save_failure( + source, saved, + )); } - let output: server_api::SubjektivRecordSessionResponse = - match serde_json::from_str(&response.body) { - Ok(output) => output, - Err(error) => { - let message = bounded_attribution_diagnostic(format!( - "Session attribution response outcome is unknown: {error}" - )); - persist_attribution( - Some(&pending_attribution), - incomplete_subjektiv_attribution_state( - session_id, - create_if_missing, - message.clone(), - ), - )?; - return Err(WorkerError::SubjektivSessionAttribution { - state: incomplete_subjektiv_attribution_label(create_if_missing), - message, - }); - } - }; - output }; - if output.session_id != session_id.to_string() - || output.subject_id.trim().is_empty() - || output.subject_id.chars().any(char::is_control) - { - let message = - "Session attribution response did not confirm the requested immutable identity" - .to_string(); - persist_attribution( - Some(&pending_attribution), - incomplete_subjektiv_attribution_state( - session_id, - create_if_missing, - message.clone(), - ), - )?; - return Err(WorkerError::SubjektivSessionAttribution { - state: incomplete_subjektiv_attribution_label(create_if_missing), - message, - }); - } persist_attribution( Some(&pending_attribution), SubjektivSessionAttributionState::Confirmed { @@ -11555,6 +11452,12 @@ pub enum WorkerError { #[error("feature lifecycle rejected operation: {0}")] FeatureLifecycle(String), + #[error("Session registration failed: {source}")] + SessionRegistration { + #[source] + source: crate::subjektiv::SubjektivHostError, + }, + #[error("Subjektiv Session attribution lifecycle is {state}: {message}")] SubjektivSessionAttribution { state: &'static str, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index c2d247e92..ccfa2c8b1 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -28,6 +28,7 @@ use subjektiv::api::*; mod backend_job_profile; mod chat; mod external_operation_audit; +mod session_registration; mod subject_authorization; mod subject_body; mod subjektiv_deletion; @@ -19529,7 +19530,12 @@ fn source_ref_readiness_blocker(code: &str) -> &'static str { } fn remap_source_ref_error(error: ApiError) -> ApiError { - let ApiError { error, diagnostics } = error; + let ApiError { + error, + diagnostics, + session_registration_cause, + session_registration_source, + } = error; match error { Error::RuntimeOperationFailed { runtime_id, @@ -19541,7 +19547,12 @@ fn remap_source_ref_error(error: ApiError) -> ApiError { message, } .into(), - error => ApiError { error, diagnostics }, + error => ApiError { + error, + diagnostics, + session_registration_cause, + session_registration_source, + }, } } @@ -20145,7 +20156,9 @@ async fn scoped_list_merge_requests( } fn merge_ref_diagnostic(error: ApiError) -> MergeRequestRefDiagnostic { - let ApiError { error, diagnostics } = error; + let ApiError { + error, diagnostics, .. + } = error; diagnostics .into_iter() .next() @@ -24275,19 +24288,51 @@ fn subjektiv_worker_scope_from_store( api: &WorkspaceApi, workspace_id: &str, context: &server_api::ServerRequestContext, +) -> ApiResult { + subjektiv_worker_scope_with_judgment( + api, + workspace_id, + context, + session_registration::Judgment::Other, + ) +} + +fn subjektiv_worker_scope_with_judgment( + api: &WorkspaceApi, + workspace_id: &str, + context: &server_api::ServerRequestContext, + judgment: session_registration::Judgment, ) -> ApiResult { let source = context.runtime_source.as_ref().ok_or_else(|| { - Error::WorkspacePermissionDenied( - "subjektiv operations require authenticated Runtime-owned source authority".to_string(), + judgment.error( + Error::WorkspacePermissionDenied( + "subjektiv operations require authenticated Runtime-owned source authority" + .to_string(), + ), + protocol::session_failure::SessionRegistrationCause::denied( + if context.actor.is_some() { + protocol::session_failure::SessionPrincipal::Operator + } else { + protocol::session_failure::SessionPrincipal::Unknown + }, + ), ) })?; let worker_id = source.worker_id.as_deref().ok_or_else(|| { - Error::WorkspacePermissionDenied( - "subjektiv operations require Runtime-bound Worker identity".to_string(), + judgment.error( + Error::WorkspacePermissionDenied( + "subjektiv operations require Runtime-bound Worker identity".to_string(), + ), + protocol::session_failure::SessionRegistrationCause::binding_mismatch(), ) })?; let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id); - if let Some(grant) = subjektiv_job::active_grant(api, workspace_id, &worker)? { + if let Some(grant) = + subjektiv_job::active_grant(api, workspace_id, &worker).map_err(|error| { + judgment + .permission_rejection(error, protocol::session_failure::SessionPrincipal::Subject) + })? + { return Ok(SubjektivStoredWorkerScope { subject_id: grant.subject_id, worker, @@ -24296,13 +24341,18 @@ fn subjektiv_worker_scope_from_store( } let lease = api .store - .require_current_worker_singleton_owner(workspace_id, &worker)? + .require_current_worker_singleton_owner(workspace_id, &worker) + .map_err(|error| judgment.binding_rejection(error.into()))? .ok_or_else(|| { - Error::WorkspacePermissionDenied( - "current Worker has no keyed singleton ownership".to_string(), + judgment.error( + Error::WorkspacePermissionDenied( + "current Worker has no keyed singleton ownership".to_string(), + ), + protocol::session_failure::SessionRegistrationCause::binding_mismatch(), ) })?; - let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key)?; + let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key) + .map_err(|error| judgment.binding_rejection(error))?; let authority = match authority { SubjektivWorkerAuthority::Subject => SubjektivStoredScopeAuthority::Subject, SubjektivWorkerAuthority::Consolidation => { @@ -24367,6 +24417,14 @@ fn authorize_subjektiv_operation( resource: resource.clone(), }, ) + .map_err(|error| { + if operation == "session.record" { + session_registration::Judgment::SessionRecord + .permission_rejection(error, protocol::session_failure::SessionPrincipal::Subject) + } else { + error + } + }) // The binding query returns only the authorized concrete target, never a // wildcard ceiling or a broader permission that happened to cover it. .map(|scope| scope.map(|_| resource)) @@ -24387,18 +24445,23 @@ fn require_subjektiv_worker_authority( } } -fn subjektiv_session_attribution( +fn subjektiv_session_attribution_with_judgment( api: &WorkspaceApi, workspace_id: &str, context: &server_api::ServerRequestContext, session_id: &str, + judgment: session_registration::Judgment, ) -> ApiResult { - let scope = subjektiv_worker_scope_from_store(api, workspace_id, context)?; + let scope = subjektiv_worker_scope_with_judgment(api, workspace_id, context, judgment)?; if !matches!(scope.authority, SubjektivStoredScopeAuthority::Subject) { - return Err(Error::WorkspacePermissionDenied( - "subject Session attribution requires the current subject Worker".into(), - ) - .into()); + return Err(judgment.error( + Error::WorkspacePermissionDenied( + "subject Session attribution requires the current subject Worker".into(), + ), + protocol::session_failure::SessionRegistrationCause::denied( + protocol::session_failure::SessionPrincipal::Subject, + ), + )); } let session_id = session_id.trim(); if session_id.is_empty() || session_id.chars().any(char::is_control) { @@ -24419,24 +24482,52 @@ fn require_existing_subjektiv_session_attribution( context: &server_api::ServerRequestContext, session_id: &str, ) -> ApiResult { - let expected = subjektiv_session_attribution(api, workspace_id, context, session_id)?; + require_existing_subjektiv_session_attribution_with_judgment( + api, + workspace_id, + context, + session_id, + session_registration::Judgment::Other, + ) +} + +fn require_existing_subjektiv_session_attribution_with_judgment( + api: &WorkspaceApi, + workspace_id: &str, + context: &server_api::ServerRequestContext, + session_id: &str, + judgment: session_registration::Judgment, +) -> ApiResult { + let expected = subjektiv_session_attribution_with_judgment( + api, + workspace_id, + context, + session_id, + judgment, + )?; let existing = open_subjektiv_store(api)? .session_attribution(&expected.session_id) .map_err(subjektiv_store_error)? .ok_or_else(|| { - Error::RepositoryConflict(format!( - "subjektiv_session_attribution_incomplete: Session {} has not completed its Host attribution lifecycle", - expected.session_id - )) + judgment.error( + Error::RepositoryConflict(format!( + "subjektiv_session_attribution_incomplete: Session {} has not completed its Host attribution lifecycle", + expected.session_id + )), + protocol::session_failure::SessionRegistrationCause::attribution_missing(), + ) })?; if existing.subject_id != expected.subject_id || existing.runtime_id != expected.runtime_id || existing.worker_id != expected.worker_id { - return Err(Error::WorkspacePermissionDenied( - "existing Session attribution does not match the authorized subject Worker".to_string(), - ) - .into()); + return Err(judgment.error( + Error::WorkspacePermissionDenied( + "existing Session attribution does not match the authorized subject Worker" + .to_string(), + ), + protocol::session_failure::SessionRegistrationCause::binding_mismatch(), + )); } Ok(existing) } @@ -24855,18 +24946,20 @@ async fn scoped_record_subjektiv_session( .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let attribution = if request.create_if_missing { - subjektiv_session_attribution( + subjektiv_session_attribution_with_judgment( &api, &path.workspace_id, &context, &request.session_id, + session_registration::Judgment::SessionRecord, )? } else { - require_existing_subjektiv_session_attribution( + require_existing_subjektiv_session_attribution_with_judgment( &api, &path.workspace_id, &context, &request.session_id, + session_registration::Judgment::SessionRecord, )? }; authorize_subjektiv_operation( @@ -24876,13 +24969,14 @@ async fn scoped_record_subjektiv_session( &attribution.subject_id, "session.record", )?; - subjektiv_journal::bind_host_session(&api, attribution.clone())?; + session_registration::bind_host_session(&api, attribution.clone())?; Ok(Json(server_api::SubjektivRecordSessionResponse { subject_id: attribution.subject_id, session_id: attribution.session_id, })) }) .await + .map_err(session_registration::execution_error) } const SUBJEKTIV_SESSION_DEFAULT_LIMIT: usize = 20; @@ -39369,6 +39463,8 @@ fn workspace_id_mismatch_error() -> ApiError { struct ApiError { error: Error, diagnostics: Vec, + session_registration_cause: Option, + session_registration_source: Option>, } #[derive(Debug, Clone)] @@ -39582,7 +39678,12 @@ impl From for ApiError { } _ => Vec::new(), }; - Self { error, diagnostics } + Self { + error, + diagnostics, + session_registration_cause: None, + session_registration_source: None, + } } } @@ -39599,13 +39700,21 @@ fn subjektiv_diagnostic_code(message: &str) -> Option<&'static str> { impl ApiError { fn with_diagnostics(error: Error, diagnostics: Vec) -> Self { - Self { error, diagnostics } + Self { + error, + diagnostics, + session_registration_cause: None, + session_registration_source: None, + } } fn into_repository_api_error(self) -> server_api::RepositoryApiError { let status = api_error_status(&self.error); - let message = api_error_response_message(&self.error); - server_api::RepositoryApiError::new( + let message = self + .session_registration_cause + .map(|cause| cause.to_string()) + .unwrap_or_else(|| api_error_response_message(&self.error)); + let response = server_api::RepositoryApiError::new( status.as_u16(), match self.error { Error::RestoreObservationConflict => "restore_observation_conflict", @@ -39616,7 +39725,15 @@ impl ApiError { }, message, working_directory_diagnostics(self.diagnostics), - ) + ); + match self.session_registration_cause { + Some(cause) => response.with_failure( + protocol::session_failure::session_registration_attachment(cause).expect( + "enum-only Session registration cause must satisfy the transfer contract", + ), + ), + None => response, + } } } diff --git a/crates/workspace-server/src/server/session_registration.rs b/crates/workspace-server/src/server/session_registration.rs new file mode 100644 index 000000000..b59cf3b6a --- /dev/null +++ b/crates/workspace-server/src/server/session_registration.rs @@ -0,0 +1,212 @@ +//! Session-record judgment projection. Only explicit domain decisions issue causes; +//! status codes, diagnostic prose and arbitrary backend errors are not classifiers. +use super::*; +use protocol::session_failure::{SessionPrincipal, SessionRegistrationCause}; + +#[derive(Clone, Copy)] +pub(super) enum Judgment { + Other, + SessionRecord, +} + +impl Judgment { + pub(super) fn error(self, error: Error, cause: SessionRegistrationCause) -> ApiError { + self.attach(error.into(), cause) + } + + fn attach(self, mut error: ApiError, cause: SessionRegistrationCause) -> ApiError { + if matches!(self, Self::SessionRecord) && error.session_registration_cause.is_none() { + error.session_registration_cause = Some(cause); + } + error + } + + pub(super) fn permission_rejection( + self, + error: ApiError, + principal: SessionPrincipal, + ) -> ApiError { + if matches!(&error.error, Error::WorkspacePermissionDenied(_)) { + self.attach(error, SessionRegistrationCause::denied(principal)) + } else { + error + } + } + + pub(super) fn binding_rejection(self, error: ApiError) -> ApiError { + if matches!( + &error.error, + Error::WorkspacePermissionDenied(_) | Error::SubjectBodySuperseded + ) { + self.attach(error, SessionRegistrationCause::binding_mismatch()) + } else { + error + } + } +} + +pub(super) fn bind_host_session( + api: &WorkspaceApi, + attribution: crate::subjektiv::SubjectSessionAttribution, +) -> ApiResult<()> { + subjektiv_journal::bind_host_session_for_registration(api, attribution) + .map_err(|error| Judgment::SessionRecord.binding_rejection(error)) +} + +pub(super) fn journal_error(error: subjektiv::journal::JournalError) -> ApiError { + let cause = match &error { + subjektiv::journal::JournalError::PermissionDenied(_) => { + SessionRegistrationCause::binding_mismatch() + } + _ => SessionRegistrationCause::outcome_unknown(), + }; + let mut failure = Judgment::SessionRecord.error(Error::Store(cause.to_string()), cause); + failure.session_registration_source = Some(Box::new(error)); + failure +} + +pub(super) fn domain_error(error: crate::subjektiv::SubjektivError) -> ApiError { + let cause = match &error { + crate::subjektiv::SubjektivError::SessionSubjectConflict { .. } + | crate::subjektiv::SubjektivError::SessionAttributionConflict { .. } => { + SessionRegistrationCause::binding_mismatch() + } + _ => SessionRegistrationCause::outcome_unknown(), + }; + let mut failure = Judgment::SessionRecord.error(Error::InvalidInput(cause.to_string()), cause); + failure.session_registration_source = Some(Box::new(error)); + failure +} + +pub(super) fn execution_error(error: ApiError) -> ApiError { + // This typed executor result means the accepted callback's outcome cannot + // be established. It is not an authorization denial or proof of rollback. + if matches!(&error.error, Error::DbOperationOutcomeUnknown) { + Judgment::SessionRecord.attach(error, SessionRegistrationCause::outcome_unknown()) + } else { + error + } +} + +impl std::fmt::Display for ApiError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + std::fmt::Display::fmt(&self.error, formatter) + } +} + +impl std::error::Error for ApiError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self.session_registration_source.as_deref() { + Some(source) => Some(source), + None => Some(&self.error), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use api_macros::HttpError; + use protocol::session_failure::SessionRegistrationReason; + + #[test] + fn permission_projection_keeps_principal_and_original_error() { + for principal in [SessionPrincipal::Operator, SessionPrincipal::Subject] { + let error = Judgment::SessionRecord.permission_rejection( + Error::WorkspacePermissionDenied("wording is not classification input".into()) + .into(), + principal, + ); + let cause = error.session_registration_cause.as_ref().unwrap(); + assert_eq!(cause.reason, SessionRegistrationReason::PermissionDenied); + assert_eq!(cause.principal, principal); + assert_eq!(api_error_status(&error.error), StatusCode::FORBIDDEN); + assert!( + std::error::Error::source(&error) + .unwrap() + .downcast_ref::() + .is_some() + ); + } + } + + #[test] + fn unrelated_operations_and_non_denial_errors_stay_unclassified() { + let unrelated = Judgment::Other.permission_rejection( + Error::WorkspacePermissionDenied("session.record permission denied".into()).into(), + SessionPrincipal::Subject, + ); + assert!(unrelated.session_registration_cause.is_none()); + let backend = Judgment::SessionRecord.permission_rejection( + Error::Store("session.record permission denied".into()).into(), + SessionPrincipal::Subject, + ); + assert!(backend.session_registration_cause.is_none()); + assert!(backend.into_repository_api_error().failure.is_none()); + } + + #[test] + fn known_causes_are_not_replaced_by_outer_projection() { + for cause in [ + SessionRegistrationCause::attribution_missing(), + SessionRegistrationCause::binding_mismatch(), + ] { + let error = Judgment::SessionRecord.error( + Error::WorkspacePermissionDenied("private attribution diagnostic".into()), + cause, + ); + let before = serde_json::to_value(error.session_registration_cause.as_ref()).unwrap(); + let error = + Judgment::SessionRecord.permission_rejection(error, SessionPrincipal::Subject); + assert_eq!( + before, + serde_json::to_value(error.session_registration_cause.as_ref()).unwrap() + ); + } + } + + #[test] + fn unknown_executor_outcome_is_not_permission_denied() { + let error = execution_error(Error::DbOperationOutcomeUnknown.into()); + assert_eq!( + error.session_registration_cause.unwrap().reason, + SessionRegistrationReason::OutcomeUnknown + ); + assert!( + execution_error(Error::DbExecutorUnavailable.into()) + .session_registration_cause + .is_none() + ); + } + + #[test] + fn all_enum_only_causes_satisfy_the_transfer_contract() { + for cause in [ + SessionRegistrationCause::denied(SessionPrincipal::Operator), + SessionRegistrationCause::denied(SessionPrincipal::Subject), + SessionRegistrationCause::denied(SessionPrincipal::Unknown), + SessionRegistrationCause::attribution_missing(), + SessionRegistrationCause::binding_mismatch(), + SessionRegistrationCause::transport(), + SessionRegistrationCause::outcome_unknown(), + SessionRegistrationCause::invalid_response(), + ] { + protocol::session_failure::session_registration_attachment(cause).unwrap(); + } + } + + #[test] + fn internal_transfer_contains_no_original_error_text() { + let error = Judgment::SessionRecord + .error( + Error::WorkspacePermissionDenied("secret-subject-id /private/path raw-body".into()), + SessionRegistrationCause::denied(SessionPrincipal::Subject), + ) + .into_repository_api_error(); + assert_eq!(error.status_code(), StatusCode::FORBIDDEN.as_u16()); + let attachment = serde_json::to_string(error.failure.as_ref().unwrap()).unwrap(); + for private in ["secret-subject-id", "/private/path", "raw-body"] { + assert!(!attachment.contains(private)); + } + } +} diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index a8c90419d..c5da3b83b 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -449,6 +449,13 @@ pub(super) fn bind_host_session( api: &WorkspaceApi, attr: crate::subjektiv::SubjectSessionAttribution, ) -> Result<()> { + bind_host_session_for_registration(api, attr).map_err(|failure| failure.error) +} + +pub(super) fn bind_host_session_for_registration( + api: &WorkspaceApi, + attr: crate::subjektiv::SubjectSessionAttribution, +) -> ApiResult<()> { let gate = lifecycle_lock(api, &attr.subject_id)?; let _guard = gate .lock() @@ -465,16 +472,18 @@ pub(super) fn bind_host_session( { return Err(Error::RepositoryConflict( "Subject release intent fences Controller publication".into(), - )); + ) + .into()); } let store = store(api)?; if store .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id) - .map_err(domain)? + .map_err(session_registration::journal_error)? { return Err(Error::WorkspacePermissionDenied( "released body cannot reacquire Journal authority".into(), - )); + ) + .into()); } // A released Backend body is never rebound by a delayed create/restore callback. New // Backend bodies have new Worker IDs; their first callback has no historical generation. @@ -484,7 +493,8 @@ pub(super) fn bind_host_session( .into_iter() .filter(|a| a.runtime_id == attr.runtime_id && a.worker_id == attr.worker_id) { - let host = JournalHostContext::validated_session(&store, prior).map_err(domain)?; + let host = JournalHostContext::validated_session(&store, prior) + .map_err(session_registration::journal_error)?; if let Some(id) = host.writer_binding_id() { let binding = JournalWriterBinding { subject_id: attr.subject_id.clone(), @@ -492,10 +502,14 @@ pub(super) fn bind_host_session( worker_id: attr.worker_id.clone(), binding_id: id.into(), }; - if !store.journal_writer_active(&binding).map_err(domain)? { + if !store + .journal_writer_active(&binding) + .map_err(session_registration::journal_error)? + { return Err(Error::WorkspacePermissionDenied( "released body Session cannot reacquire Journal authority".into(), - )); + ) + .into()); } } } @@ -503,11 +517,13 @@ pub(super) fn bind_host_session( // authenticated before a recovery intent must not leave a late attribution. let attr = store .record_session_attribution(attr) - .map_err(|error| Error::InvalidInput(error.to_string()))?; + .map_err(session_registration::domain_error)?; let writer = store .claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id) - .map_err(domain)?; - store.bind_journal_session(&writer, attr).map_err(domain)?; + .map_err(session_registration::journal_error)?; + store + .bind_journal_session(&writer, attr) + .map_err(session_registration::journal_error)?; Ok(()) } diff --git a/openapi/server-api.json b/openapi/server-api.json index 63e3f6bbe..864bc1037 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -11817,6 +11817,17 @@ "error": { "type": "string" }, + "failure": { + "anyOf": [ + { + "$ref": "#/components/schemas/StartupFailureAttachment" + }, + { + "type": "null" + } + ], + "description": "An origin-projected error-contract attachment, not raw HTTP error text.\nReceiving it does not authorize redisclosure to a different audience." + }, "message": { "type": "string" } @@ -14799,6 +14810,10 @@ ], "type": "object" }, + "StartupFailureAttachment": { + "description": "A bounded, projected envelope at the Worker/Runtime or metadata boundary.\nNo constructor accepts Display text, arbitrary JSON fields, or an unprojected error.", + "type": "string" + }, "SubjectBodyApiError": { "additionalProperties": false, "description": "Only action handlers with a causally bound request may provide resolution evidence.", @@ -23348,7 +23363,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321" + "x-yoi-source-digest": "sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e" }, "openapi": "3.1.0", "paths": { diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index 382d570bf..ed272d568 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) +// Input contract: 0.1.0 (sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check @@ -29,7 +29,9 @@ export type Diagnostic = { code: string; message: string; severity: DiagnosticSe export type DiagnosticSeverity = "info" | "warning" | "error"; -export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; }; +export type RepositoryApiError = { diagnostics?: Array; error: string; failure?: StartupFailureAttachment | null; message: string; }; + +export type StartupFailureAttachment = string; export type WorkspaceWorkerDiscoveryItem = { display_name: string; profile?: string | null; resource_key: string; status?: string | null; subject: WorkspaceWorkerSubject; }; diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index cfffdfe22..1984f7db8 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) +// Input contract: 0.1.0 (sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check @@ -29,7 +29,7 @@ export type HostSummary = { arch: string; diagnostics: Array; host_i export type HostWorkerListResponse = { diagnostics: Array; items: Array; limit: number; source: string; workspace_id: string; }; -export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; }; +export type RepositoryApiError = { diagnostics?: Array; error: string; failure?: StartupFailureAttachment | null; message: string; }; export type RepositoryDetailResponse = { item: RepositorySummary; source: string; workspace_id: string; }; @@ -57,6 +57,8 @@ export type RepositorySshHostTrust = { created_at: string; fingerprint: string; export type RepositorySummary = { default_selector?: string | null; diagnostics?: Array | null; git?: GitRepositorySummary | null; kind: string; observed_at?: string | null; observed_status: RepositoryObservedStatus; provider: string; record_authority: string; repository_key: string; source: RepositorySource; source_fingerprint: string; }; +export type StartupFailureAttachment = string; + export type SubscriptionWorkerAvailability = "observed" | "unavailable"; export type WorkerBusyState = { kind: "run"; state: WorkerRunState; } | { kind: "maintenance"; state: WorkerMaintenanceState; }; diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index ad88cbaf8..b2570df8c 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) +// Input contract: 0.1.0 (sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check @@ -63,7 +63,7 @@ export type PasteArtifactRef = { artifact_id: string; availability: PasteArtifac export type RemoveRuntimeRequest = { expected_binding_id: string; operation_id: string; }; -export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; }; +export type RepositoryApiError = { diagnostics?: Array; error: string; failure?: StartupFailureAttachment | null; message: string; }; export type RevokeRuntimeTrustKeyRequest = { expected_binding_id: string; }; @@ -155,6 +155,8 @@ export type RuntimeWorkerTicketAssignmentRequest = { operation_id: string; orche export type Segment = { content: string; kind: "text"; } | { chars: number; content: string; id: number; kind: "paste"; lines: number; } | { artifact: PasteArtifactRef; kind: "paste_artifact"; } | { file: UploadedFileRef; kind: "uploaded_file"; } | { kind: "file_ref"; path: string; } | { invocation: FeatureInvocation; kind: "feature_invoke"; } | { kind: "flow"; selector: string; } | { kind: "unknown"; }; +export type StartupFailureAttachment = string; + export type SubscriptionWorkerAvailability = "observed" | "unavailable"; export type UpdateRemoteRuntimeRequest = { display_name?: string | null; endpoint: string; }; diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index 428cdc13a..b23760724 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) +// Input contract: 0.1.0 (sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check @@ -87,7 +87,7 @@ export type ObjectiveSummary = { created_at?: string | null; id: string; linked_ export type QueryPage = { has_more: boolean; limit: number; next_cursor?: string | null; returned: number; sort: string; source_limit?: number | null; source_truncated: boolean; }; -export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; }; +export type RepositoryApiError = { diagnostics?: Array; error: string; failure?: StartupFailureAttachment | null; message: string; }; export type ReviewDecision = "approve" | "request_changes"; @@ -95,6 +95,8 @@ export type ReviewFinding = { body: string; code?: string | null; line?: number export type SetTicketRoleAssignmentRequest = { expected_assignment_id?: string | null; operation_id: string; orchestrator_assignment_id?: string | null; principal: TicketAssignmentPrincipal; workdir_bindings?: Array; }; +export type StartupFailureAttachment = string; + export type TicketActionEligibility = { blockers: Array; can_assign_orchestrator: boolean; can_queue: boolean; can_set_parent: boolean; can_start_manual_worker: boolean; can_unassign_orchestrator: boolean; queue_tickets: Array; }; export type TicketAssignmentAnswer = { assignment_id: string; body: string; expected_last_event_id: string; operation_key: string; }; diff --git a/web/workspace/src/lib/generated/workdir-api.ts b/web/workspace/src/lib/generated/workdir-api.ts index 154a884c8..986c6fe70 100644 --- a/web/workspace/src/lib/generated/workdir-api.ts +++ b/web/workspace/src/lib/generated/workdir-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-workdir-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:6a3d9c468ff204b44fc0b11d47e45f153b8131b9ec99c54ba2ebfdf2c0d14580) +// Input contract: 0.1.0 (sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e) // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/workdir-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check @@ -15,7 +15,9 @@ export type ExternalWorkdirGrantResponse = { display_name: string; expires_at?: export type ExternalWorkdirPermissions = { command: boolean; read: boolean; write: boolean; }; -export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; }; +export type RepositoryApiError = { diagnostics?: Array; error: string; failure?: StartupFailureAttachment | null; message: string; }; + +export type StartupFailureAttachment = string; export type WorkingDirectoryCleanupTarget = { kind: string; repository_key: string; working_directory_id: string; }; diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index 722c0733b..9571bddb5 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321) +// Input contract: 0.1.0 (sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check @@ -37,7 +37,7 @@ export type PasteArtifactRef = { artifact_id: string; availability: PasteArtifac export type PermissionResource = { installation_id: string; kind: "plugin_installation"; workspace_id: string; } | { job_id: string; kind: "job"; workspace_id: string; } | { kind: "ticket"; ticket_id: string; workspace_id: string; } | { kind: "objective"; objective_id: string; workspace_id: string; } | { kind: "merge_request"; merge_request_id: string; workspace_id: string; } | { kind: "subject"; subject_id: string; workspace_id: string; } | { kind: "drive_entry"; node_id: string; workspace_id: string; } | { kind: "drive_request"; request_id: string; workspace_id: string; } | { kind: "repository"; repository_key: string; workspace_id: string; } | { connection_id: string; kind: "workdir_attachment"; workspace_id: string; } | { kind: "workspace_config"; path: string; recursive: boolean; workspace_id: string; } | { kind: "network"; url: string; } | { kind: "sub_worker"; name: string; runtime_id: string; worker_id: string; } | { kind: "feature"; } | { kind: "workspace"; workspace_id: string; } | { kind: "workdir"; path: string; recursive: boolean; working_directory_id: string; } | { kind: "worker_operation"; workspace_id: string; } | { kind: "worker"; runtime_id: string; worker_id: string; }; -export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; }; +export type RepositoryApiError = { diagnostics?: Array; error: string; failure?: StartupFailureAttachment | null; message: string; }; export type RuntimeWorkerWorkdirAttachmentSummary = { alias: string; working_directory: RuntimeWorkingDirectorySummary; }; @@ -49,6 +49,8 @@ export type ScopedPermission = { operation: string; resource: PermissionResource export type Segment = { content: string; kind: "text"; } | { chars: number; content: string; id: number; kind: "paste"; lines: number; } | { artifact: PasteArtifactRef; kind: "paste_artifact"; } | { file: UploadedFileRef; kind: "uploaded_file"; } | { kind: "file_ref"; path: string; } | { invocation: FeatureInvocation; kind: "feature_invoke"; } | { kind: "flow"; selector: string; } | { kind: "unknown"; }; +export type StartupFailureAttachment = string; + export type SubjektivWorkerConnectionRequest = { subject_id: string; }; export type SubscriptionWorkerAvailability = "observed" | "unavailable"; From 7873f4359ab62c83107591afd1a8d9f1311bd26b Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 15:50:30 +0900 Subject: [PATCH 2/2] feat: transfer typed startup causes over Runtime HTTP with real registration fixture --- crates/runtime-api/src/lib.rs | 16 + crates/worker-runtime/src/http_server.rs | 39 +- .../src/http_server/runtime_management_api.rs | 5 +- crates/worker/src/startup_failure.rs | 14 + crates/workspace-server/src/hosts.rs | 43 +- crates/workspace-server/src/server.rs | 50 +++ .../session_registration_connected_tests.rs | 369 ++++++++++++++++++ .../tests/subject_authorization_tests.rs | 14 + 8 files changed, 543 insertions(+), 7 deletions(-) create mode 100644 crates/workspace-server/src/server/tests/session_registration_connected_tests.rs diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 533b657ae..cdb819085 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -105,10 +105,25 @@ pub struct RuntimeApiError { status: u16, } +/// Transient, allowlisted startup observations copied from the execution owner. +/// This is not an operation result, lifecycle record, or permission to retry. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct WorkerStartupFailure { + pub primary: StartupFailureAttachment, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub secondary: Vec, + pub resources: Option, + pub persistence: Option, + pub io_kind: Option, +} + #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct RuntimeApiErrorDetail { pub code: String, pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup: Option, } impl RuntimeApiError { @@ -117,6 +132,7 @@ impl RuntimeApiError { error: RuntimeApiErrorDetail { code: code.into(), message: message.into(), + startup: None, }, status, } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index d591008a4..5d04e4c6d 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -542,6 +542,8 @@ pub struct RuntimeHttpErrorResponse { pub struct RuntimeHttpErrorDetail { pub code: String, pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup: Option, } #[cfg(feature = "ws-server")] @@ -2325,6 +2327,7 @@ struct RuntimeHttpRestError { status: StatusCode, code: String, message: String, + startup: Option, } impl RuntimeHttpRestError { @@ -2333,13 +2336,28 @@ impl RuntimeHttpRestError { status, code: code.into(), message: message.into(), + startup: None, } } fn runtime(error: RuntimeError) -> Self { let status = status_for_runtime_error(&error); let code = code_for_runtime_error(&error); - Self::new(status, code, error.to_string()) + let startup = match &error { + RuntimeError::WorkerExecutionRejected { result, .. } => startup_transfer(result), + _ => None, + }; + let message = if startup.is_some() { + "Worker startup failed".to_owned() + } else { + error.to_string() + }; + Self { + status, + code, + message, + startup, + } } fn working_directory_join(_error: tokio::task::JoinError) -> Self { @@ -2375,12 +2393,31 @@ impl IntoResponse for RuntimeHttpRestError { error: RuntimeHttpErrorDetail { code: self.code, message: self.message, + startup: self.startup, }, }; (self.status, Json(body)).into_response() } } +/// Explicit transfer mapping, not another execution/lifecycle result. HTTP request +/// authentication has already established this internal Runtime caller's scope. +fn startup_transfer( + result: &crate::execution::WorkerExecutionResult, +) -> Option { + let primary = worker::startup_failure::reproject_internal_startup_failure( + result.startup_failure.as_ref()?, + ) + .ok()?; + Some(runtime_api::WorkerStartupFailure { + primary, + secondary: result.secondary_failures.clone(), + resources: result.startup_resources, + persistence: result.startup_failure_persistence, + io_kind: result.startup_io_kind, + }) +} + fn status_for_runtime_error(error: &RuntimeError) -> StatusCode { match error { RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => { diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index 860a8f266..cac820371 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -224,7 +224,10 @@ where } fn api_error(error: RuntimeHttpRestError) -> runtime_api::RuntimeApiError { - runtime_api::RuntimeApiError::new(error.status.as_u16(), error.code, error.message) + let mut response = + runtime_api::RuntimeApiError::new(error.status.as_u16(), error.code, error.message); + response.error.startup = error.startup; + response } impl runtime_api::RuntimeApi for RuntimeManagementApi { diff --git a/crates/worker/src/startup_failure.rs b/crates/worker/src/startup_failure.rs index 2c48c7138..b0c4c59fa 100644 --- a/crates/worker/src/startup_failure.rs +++ b/crates/worker/src/startup_failure.rs @@ -106,6 +106,20 @@ pub fn runtime_startup_attachment( StartupFailureAttachment::from_projected(&projected) } +/// The internal HTTP/Host transfer admits only implemented startup domains. +/// Unknown/corrupt envelopes cannot be blindly forwarded or classified from text. +pub fn reproject_internal_startup_failure( + failure: &StartupFailureAttachment, +) -> Result { + if let Ok(projected) = failure + .project_for::(&()) + { + return StartupFailureAttachment::from_projected(&projected); + } + let projected = failure.project_for::(&())?; + StartupFailureAttachment::from_projected(&projected) +} + /// Dispatch by the domain attested in process, not by parsing attachment fields. pub(crate) fn persist_binding_startup_failure( store: &S, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 7f4d9a847..9975b5f45 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -117,6 +117,9 @@ where #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct RuntimeDiagnostic { + /// Internal startup observations; public consumers must separately authorize projection. + #[serde(skip)] + pub startup: Option, pub code: String, pub severity: HostDiagnosticSeverity, pub message: String, @@ -6649,7 +6652,7 @@ fn sanitize_embedded_execution_detail(summary: &str, message: &str) -> String { } fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnostic { - match error { + let mut projection = match error { EmbeddedRuntimeError::RestoreAdmissionBusy { .. } => diagnostic( "restore_admission_busy", HostDiagnosticSeverity::Warning, @@ -6767,7 +6770,19 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti HostDiagnosticSeverity::Error, "Embedded Runtime state is unavailable".to_string(), ), + }; + if let EmbeddedRuntimeError::WorkerExecutionRejected { result, .. } = error { + projection.startup = result.startup_failure.as_ref().and_then(|primary| { + internal_startup_transfer(runtime_api::WorkerStartupFailure { + primary: primary.clone(), + secondary: result.secondary_failures.clone(), + resources: result.startup_resources, + persistence: result.startup_failure_persistence, + io_kind: result.startup_io_kind, + }) + }); } + projection } fn host_id_for_embedded_workspace(workspace_id: &str) -> String { @@ -6846,6 +6861,7 @@ fn runtime_api_diagnostic( ) -> RuntimeDiagnostic { match error { runtime_api::client_support::ClientError::Public { status, error } => { + let startup = error.error.startup.and_then(internal_startup_transfer); let remote_code = error.error.code; let message = sanitize_remote_runtime_message(&remote_code, &error.error.message); let (code, severity) = match status { @@ -6862,13 +6878,15 @@ fn runtime_api_diagnostic( }, ), }; - diagnostic( + let mut projection = diagnostic( code, severity, format!( "Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}" ), - ) + ); + projection.startup = startup; + projection } runtime_api::client_support::ClientError::Failure(failure) => { let code = match failure { @@ -6945,11 +6963,25 @@ fn remote_http_status_diagnostic( } _ => (remote_code, HostDiagnosticSeverity::Warning), }; - diagnostic( + let mut projection = diagnostic( code, severity, format!("Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {remote_message}"), - ) + ); + projection.startup = error + .and_then(|error| error.error.startup) + .and_then(internal_startup_transfer); + projection +} + +/// Treat transferred details as untrusted observations and reproject through the +/// same domain allowlist; HTTP status and text never supply a missing cause. +fn internal_startup_transfer( + mut startup: runtime_api::WorkerStartupFailure, +) -> Option { + startup.primary = + worker::startup_failure::reproject_internal_startup_failure(&startup.primary).ok()?; + Some(startup) } fn diagnostic( @@ -6958,6 +6990,7 @@ fn diagnostic( message: impl Into, ) -> RuntimeDiagnostic { RuntimeDiagnostic { + startup: None, code: code.into(), severity, message: message.into(), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ccfa2c8b1..d79a839ff 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -8612,6 +8612,7 @@ fn worker_input_result_to_api( // distinguishable without collapsing it into a definite refusal or replaying input. if result.disposition == WorkerInputDisposition::Unknown { result.diagnostics.push(RuntimeDiagnostic { + startup: None, code: "worker_input_outcome_unknown".into(), severity: HostDiagnosticSeverity::Warning, message: "Worker input delivery outcome is unknown; inspect the current session before retrying".into(), @@ -27020,6 +27021,7 @@ fn skill_api_error(error: skills::SkillError) -> ApiError { message: format!("unknown Skill `{name}`"), }, vec![RuntimeDiagnostic { + startup: None, code: "skill_not_found".to_string(), severity: HostDiagnosticSeverity::Error, message: format!("unknown Skill `{name}`"), @@ -30426,6 +30428,7 @@ fn prepare_workspace_workdir_create( .to_string(), ), vec![RuntimeDiagnostic { + startup: None, code: "default_runtime_not_configured".to_string(), severity: HostDiagnosticSeverity::Error, message: "Workspace default Runtime is not configured".to_string(), @@ -30580,6 +30583,7 @@ async fn create_workspace_working_directory( return Err(ApiError::with_diagnostics( Error::UnknownRuntime(reserved.resolved_runtime_id), vec![RuntimeDiagnostic { + startup: None, code: "runtime_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "Selected Runtime is not available".to_string(), @@ -30603,6 +30607,7 @@ async fn create_workspace_working_directory( message: "Selected Runtime is not connected".to_string(), }, vec![RuntimeDiagnostic { + startup: None, code: "runtime_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "Selected Runtime is not available".to_string(), @@ -30659,6 +30664,7 @@ async fn create_workspace_working_directory( return Err(ApiError::with_diagnostics( error.into_error(), vec![RuntimeDiagnostic { + startup: None, code: failure_code, severity: HostDiagnosticSeverity::Error, message: "Selected Runtime could not inspect the reserved Workdir".to_string(), @@ -30697,6 +30703,7 @@ async fn create_workspace_working_directory( return Err(ApiError::with_diagnostics( error.into_error(), vec![RuntimeDiagnostic { + startup: None, code: failure_code, severity: HostDiagnosticSeverity::Error, message: "Selected Runtime rejected Workdir creation".to_string(), @@ -30736,6 +30743,7 @@ async fn create_workspace_working_directory( diagnostics.insert( 0, RuntimeDiagnostic { + startup: None, code: failure_code.clone(), severity: HostDiagnosticSeverity::Error, message: "Runtime did not create the reserved Workdir".to_string(), @@ -31490,6 +31498,7 @@ fn build_runtime_cleanup_plan( let mut diagnostics = error.diagnostics; if diagnostics.is_empty() { diagnostics.push(RuntimeDiagnostic { + startup: None, code: "workspace_cleanup_runtime_observation_unavailable".to_string(), severity: HostDiagnosticSeverity::Warning, message: sanitize_backend_error(&error.error.to_string()), @@ -35599,6 +35608,7 @@ fn reject_workdir_for_embedded_runtime(runtime_id: &str, has_workdir: bool) -> A message: "The embedded Runtime does not accept working directories".to_string(), }, vec![RuntimeDiagnostic { + startup: None, code: "embedded_worker_workdir_unsupported".to_string(), severity: HostDiagnosticSeverity::Error, message: "Choose a non-embedded Runtime for workspace-file Workers; embedded Workers are no-workdir Workspace-API workers.".to_string(), @@ -35689,6 +35699,7 @@ fn finalize_worker_spawn_stage( .map(|assignment| assignment.operation_id.as_str()) .unwrap_or("none"); let mut diagnostics = vec![RuntimeDiagnostic { + startup: None, code: format!("worker_spawn_finalize_{}_failed", stage.as_str()), severity: HostDiagnosticSeverity::Error, message: format!( @@ -35704,6 +35715,7 @@ fn finalize_worker_spawn_stage( diagnostics.extend(compensation_errors); if !compensation_failed { diagnostics.push(RuntimeDiagnostic { + startup: None, code: "worker_spawn_compensated".to_string(), severity: HostDiagnosticSeverity::Info, message: format!( @@ -36039,6 +36051,7 @@ fn lifecycle_failure_detail( fn spawn_compensation_diagnostic(code: &str, message: String) -> RuntimeDiagnostic { RuntimeDiagnostic { + startup: None, code: code.to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -37044,6 +37057,7 @@ fn project_workspace_worker_with_attachments( summary.state = "unavailable".to_string(); summary.worker_state = None; summary.diagnostics.push(RuntimeDiagnostic { + startup: None, code: "worker_singleton_fenced".to_string(), severity: HostDiagnosticSeverity::Warning, message: "This historical singleton generation is fenced by Backend ownership." @@ -37749,6 +37763,7 @@ fn working_directory_summaries( } if let Some(grant_id) = legacy_external_workdir_grant_id(api, &record)? { diagnostics.push(RuntimeDiagnostic { + startup: None, code: "external_workdir_legacy_permissions".to_string(), severity: HostDiagnosticSeverity::Warning, message: format!( @@ -37955,6 +37970,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> InternalWorker }, workdir_attachments: Vec::new(), diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "backend_worker_missing".to_string(), severity: HostDiagnosticSeverity::Info, message: @@ -37999,6 +38015,7 @@ fn worker_summary_from_projection( Vec::new() } else { vec![RuntimeDiagnostic { + startup: None, code: "worker_observation_unavailable".to_string(), severity: HostDiagnosticSeverity::Info, message: "The Worker remains in the Backend catalog, but its Runtime observation is currently unavailable.".to_string(), @@ -38645,6 +38662,7 @@ fn validate_working_directory_claims_for_browser( message: "working directory relative_cwd must stay inside the Runtime-owned working directory".to_string(), }, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_relative_cwd_invalid".to_string(), severity: HostDiagnosticSeverity::Error, message: "relative_cwd must be a relative path without parent traversal".to_string(), @@ -39239,6 +39257,7 @@ fn settings_diagnostic( message: impl Into, ) -> RuntimeDiagnostic { RuntimeDiagnostic { + startup: None, code: code.into(), severity, message: message.into(), @@ -39260,6 +39279,7 @@ fn repository_lookup(result: std::result::Result) - message: message.clone(), }, vec![RuntimeDiagnostic { + startup: None, code: "repository_not_configured".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39277,6 +39297,7 @@ fn repository_lookup(result: std::result::Result) - message: message.clone(), }, vec![RuntimeDiagnostic { + startup: None, code: "repository_provider_unsupported".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39292,6 +39313,7 @@ fn repository_lookup(result: std::result::Result) - message: message.clone(), }, vec![RuntimeDiagnostic { + startup: None, code: "repository_evidence_invalid".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39452,6 +39474,7 @@ fn workspace_id_mismatch_error() -> ApiError { ApiError::with_diagnostics( Error::WorkspaceIdMismatch, vec![RuntimeDiagnostic { + startup: None, code: "workspace_id_mismatch".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39561,6 +39584,7 @@ impl ApiErrorLog { .diagnostics .iter() .map(|diagnostic| RuntimeDiagnostic { + startup: None, code: diagnostic.code.clone(), severity: match diagnostic.severity { server_api::DiagnosticSeverity::Info => HostDiagnosticSeverity::Info, @@ -39598,6 +39622,7 @@ impl From for ApiError { Error::DbCapacityExceeded | Error::DbExecutorUnavailable | Error::DbOperationOutcomeUnknown => vec![RuntimeDiagnostic { + startup: None, code: match &error { Error::DbCapacityExceeded => "db_capacity_exceeded", Error::DbExecutorUnavailable => "db_executor_unavailable", @@ -39608,21 +39633,25 @@ impl From for ApiError { message: error.to_string(), }], Error::WorkdirSessionOpenFailed { transport, .. } => vec![RuntimeDiagnostic { + startup: None, code: "workdir_session_open_failed".to_string(), severity: HostDiagnosticSeverity::Error, message: transport.message.clone(), }], Error::RuntimeOperationFailed { code, message, .. } => vec![RuntimeDiagnostic { + startup: None, code: code.clone(), severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(message), }], Error::WorkspaceSigningIdentity { code, message } => vec![RuntimeDiagnostic { + startup: None, code: code.clone(), severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(message), }], Error::Ticket(ticket_error) => vec![RuntimeDiagnostic { + startup: None, code: match ticket_error { ticket::TicketError::NotFound(_) => "ticket_not_found", ticket::TicketError::Ambiguous { .. } => "ticket_ambiguous", @@ -39669,6 +39698,7 @@ impl From for ApiError { subjektiv_diagnostic_code(message) .map(|code| { vec![RuntimeDiagnostic { + startup: None, code: code.to_string(), severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(message), @@ -43074,6 +43104,7 @@ mod tests { worker: None, acceptance_evidence: Vec::new(), diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "fixture_runtime_rejected".to_string(), severity: HostDiagnosticSeverity::Error, message: "fixture rejected Worker create".to_string(), @@ -44981,6 +45012,7 @@ mod tests { kind: "ticket_backend_error".to_string(), message: "sqlite error: FOREIGN KEY constraint failed".to_string(), diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "ticket_backend_error".to_string(), severity: HostDiagnosticSeverity::Error, message: "FOREIGN KEY constraint failed".to_string(), @@ -51633,6 +51665,7 @@ mod tests { #[test] fn unrecoverable_pending_workspace_restore_is_typed_for_replacement() { let diagnostics = [RuntimeDiagnostic { + startup: None, code: "embedded_worker_execution_rejected".to_string(), severity: HostDiagnosticSeverity::Error, message: "Restore Errored: pending Workspace Worker restore requires operation-owned launch material; generic restore must not reconstruct it from current Workspace config".to_string(), @@ -51644,6 +51677,7 @@ mod tests { #[test] fn workdir_runtime_miss_uses_exact_typed_code() { let typed_not_found = [RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Warning, message: "missing".to_string(), @@ -51655,6 +51689,7 @@ mod tests { assert!(workdir_runtime_miss_is_not_found(&typed_not_found)); let unrelated = [RuntimeDiagnostic { + startup: None, code: "some_other_not_found".to_string(), severity: HostDiagnosticSeverity::Warning, message: "not a typed workdir miss".to_string(), @@ -63358,6 +63393,7 @@ mod tests { &api, deleted, &[RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Warning, message: "missing".to_string(), @@ -63384,6 +63420,7 @@ mod tests { &api, unknown, &[RuntimeDiagnostic { + startup: None, code: "runtime_unavailable".to_string(), severity: HostDiagnosticSeverity::Warning, message: "temporarily unavailable".to_string(), @@ -65080,6 +65117,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".into(), severity: HostDiagnosticSeverity::Warning, message: "authoritatively absent".into(), @@ -65131,6 +65169,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_cleanup_resource_busy".into(), severity: HostDiagnosticSeverity::Warning, message: "busy".into(), @@ -65360,6 +65399,7 @@ mod tests { let (missing_operation, _) = reserve_removal_fixture(&api, "provider-not-found"); let not_found = RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Error, message: "missing".to_string(), @@ -65395,6 +65435,7 @@ mod tests { InternalWorkerOperationState::Accepted, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_provider_timeout".to_string(), severity: HostDiagnosticSeverity::Error, message: "timeout".to_string(), @@ -65511,6 +65552,7 @@ mod tests { InternalWorkerOperationState::Unsupported, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_unsupported".to_string(), severity: HostDiagnosticSeverity::Error, message: "unsupported".to_string(), @@ -65571,6 +65613,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: code.into(), severity: HostDiagnosticSeverity::Error, message: "secret /private/path".into(), @@ -65602,6 +65645,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_cleanup_mount_present".into(), severity: HostDiagnosticSeverity::Error, message: "/private/mount credentials must not escape".into(), @@ -65752,6 +65796,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_cleanup_permission_denied".into(), severity: HostDiagnosticSeverity::Error, message: "permission denied /private/path".into(), @@ -65938,6 +65983,7 @@ mod tests { state: InternalWorkerOperationState::Rejected, working_directory: None, diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Error, message: "missing".to_string(), @@ -65949,6 +65995,7 @@ mod tests { state: InternalWorkerOperationState::Rejected, working_directory: None, diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "working_directory_provider_timeout".to_string(), severity: HostDiagnosticSeverity::Error, message: "timeout".to_string(), @@ -67641,11 +67688,13 @@ mod tests { fn provider_rejection_classification_prefers_stable_error_diagnostic() { let diagnostics = vec![ RuntimeDiagnostic { + startup: None, code: "runtime_capacity_warning".to_string(), severity: HostDiagnosticSeverity::Warning, message: "capacity is low".to_string(), }, RuntimeDiagnostic { + startup: None, code: "runtime_capacity_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "capacity is exhausted".to_string(), @@ -67705,6 +67754,7 @@ mod tests { }) .unwrap(); let diagnostics = vec![RuntimeDiagnostic { + startup: None, code: "runtime_capacity_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "capacity is exhausted".to_string(), diff --git a/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs b/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs new file mode 100644 index 000000000..8ad590fb5 --- /dev/null +++ b/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs @@ -0,0 +1,369 @@ +//! Focused T840 connection: real Subject policy -> signed Session POST -> real +//! ProfileRuntimeWorkerFactory startup -> RuntimeHTTP/Host diagnostic -> saved +//! Worker metadata. Only the unused external model is replaced. No restore, +//! operation replay, singleton-release, or Browser rendering contract is asserted. +use super::*; +use protocol::session_failure::{ + SessionPrincipal, SessionRegistrationCause, session_registration_attachment, +}; +use session_store::WorkerMetadataStore as _; +use std::sync::atomic::{AtomicUsize, Ordering}; +use worker_runtime::worker_backend::RuntimeWorkerFactory as _; + +#[derive(Clone, Default)] +struct UnusedModel(Arc); + +#[async_trait::async_trait] +impl agen::llm_client::LlmClient for UnusedModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + _: agen::llm_client::Request, + ) -> std::result::Result + { + self.0.fetch_add(1, Ordering::SeqCst); + panic!("Session registration does not require a model turn") + } +} + +struct RegistrationFixture { + connected: ConnectedWorkerHarness, + subject: String, + callbacks: Arc>>, + model: UnusedModel, +} + +impl RegistrationFixture { + async fn new(revoke_on_registration: bool) -> Self { + // Armed only after real Subject creation with its normal default policy. + // The callback is a deterministic synchronization point, not a fake Host + // rejection: the original signed request and real handler remain intact. + let revoke = Arc::new(Mutex::new(None::<(WorkspaceApi, String, PathBuf)>)); + let hook = revoke.clone(); + let callbacks = Arc::new(Mutex::new(Vec::new())); + let observed = callbacks.clone(); + let model = UnusedModel::default(); + let connected = ConnectedWorkerHarness::start_with_layer(model.clone(), move |router| { + router.layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| { + let hook = hook.clone(); + let observed = observed.clone(); + async move { + if request.method() != axum::http::Method::POST + || request.uri().path() + != format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions") + { + return next.run(request).await; + } + assert!(request.headers().contains_key( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER + )); + let armed = hook.lock().unwrap().take(); + let pin_root = armed.as_ref().map(|(_, _, root)| root.clone()); + if let Some((api, subject, _)) = armed { + let store = open_subjektiv_store(&api).unwrap(); + let policy = store.subject_permissions(&subject).unwrap().unwrap(); + assert!( + policy + .execution + .iter() + .any(|p| p.operation == "session.record") + ); + store + .set_subject_permissions( + &subject, + &SubjectPermissionsUpdate { + operation_id: "session-registration-revoke-at-callback" + .into(), + expected_change_id: Some(policy.change_id), + execution: policy + .execution + .into_iter() + .filter(|p| p.operation != "session.record") + .collect(), + delegation: policy.delegation, + }, + ) + .unwrap(); + } + let (parts, body) = request.into_parts(); + let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); + let input: Value = serde_json::from_slice(&bytes).unwrap(); + if let Some(root) = pin_root { + let request: server_api::SubjektivRecordSessionRequest = serde_json::from_slice(&bytes).unwrap(); + let session = request.session_id.parse().unwrap(); + let retention = worker_runtime::retention::runtime_session_retention(&root); + retention.pin(session, session_store::retention::SessionRetentionPin(uuid::Uuid::now_v7()), || { + let entries = std::fs::read_dir(root.join("workers"))?; + for entry in entries { + let entry = entry?; + let name = format!("worker-runtime-{}", entry.file_name().to_string_lossy()); + let store = session_store::WorkerAggregateStore::open_read_only(&entry.path(), &name) + .map_err(|error| session_store::StoreError::Io(std::io::Error::other(error)))?; + if store.read_by_name(&name).unwrap().and_then(|m| m.active).is_some_and(|a| a.session_id == session) { return Ok(true); } + } + Ok(false) + }).unwrap(); + } + let response = next + .run(Request::from_parts(parts, Body::from(bytes))) + .await; + observed.lock().unwrap().push((response.status(), input)); + response + } + }, + )) + }) + .await; + let subject = open_subjektiv_store(&connected.api) + .unwrap() + .create_subject( + crate::subjektiv::SubjectRole::new("session-registration-connected").unwrap(), + ) + .unwrap() + .id; + let initial = open_subjektiv_store(&connected.api) + .unwrap() + .subject_permissions(&subject) + .unwrap() + .unwrap(); + assert!( + initial + .execution + .iter() + .any(|p| p.operation == "session.record") + ); + if revoke_on_registration { + *revoke.lock().unwrap() = Some(( + connected.api.clone(), + subject.clone(), + connected.runtime_root.clone(), + )); + } + Self { + connected, + subject, + callbacks, + model, + } + } + + fn request(&self) -> CreateWorkspaceWorkerRequest { + CreateWorkspaceWorkerRequest { + runtime_id: CONNECTED_RUNTIME.into(), + display_name: "Connected Session registration".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: self.subject.clone(), + }), + }, + control_operation_id: None, + execution_grant: None, + } + } + + fn assert_callback(&self, expected: StatusCode) -> String { + let callbacks = self.callbacks.lock().unwrap(); + assert!(!callbacks.is_empty(), "real signed startup callback"); + assert!(callbacks.iter().all(|(status, body)| *status == expected + && body["session_id"] == callbacks[0].1["session_id"])); + let (status, input) = &callbacks[0]; + assert_eq!(*status, expected); + assert_eq!(input["create_if_missing"], true); + assert_eq!(self.model.0.load(Ordering::SeqCst), 0); + input["session_id"].as_str().unwrap().to_owned() + } +} + +// This deliberately inspects only persisted serde shape, not rendered error text +// or an error-contract dependency. The expected value is a typed live attachment. +fn contains_saved_value(saved: &Value, expected: &Value) -> bool { + saved == expected + || match saved { + Value::Object(fields) => fields + .values() + .any(|value| contains_saved_value(value, expected)), + Value::Array(values) => values + .iter() + .any(|value| contains_saved_value(value, expected)), + _ => false, + } +} + +fn read_metadata(root: &std::path::Path, worker: &str) -> Value { + // WorkerAggregateStore owns metadata.json; worker.json is the Runtime + // catalog. Open a new file handle on each read. + let path = root.join("workers").join(worker).join("metadata.json"); + serde_json::from_reader(std::fs::File::open(path).unwrap()).unwrap() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn real_session_record_denial_survives_runtime_host_and_metadata_reopen() { + let fixture = RegistrationFixture::new(true).await; + // The real create preflight sees normal defaults. Revocation happens only + // after the actual factory has issued its signed Session-registration POST. + let error = scoped_create_workspace_worker( + State(fixture.connected.api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(fixture.request()), + ) + .await + .unwrap_err(); + fixture.assert_callback(StatusCode::FORBIDDEN); + let store = open_subjektiv_store(&fixture.connected.api).unwrap(); + assert!( + store + .subject_session_attributions(&fixture.subject) + .unwrap() + .is_empty() + ); + assert!( + !store + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap() + .execution + .iter() + .any(|p| p.operation == "session.record") + ); + + let failures = fixture.connected.runtime_startup_failures.lock().unwrap(); + assert_eq!( + failures.len(), + 1, + "actual RuntimeHTTP startup error envelope" + ); + let wire = &failures[0]; + let expected_primary = session_registration_attachment(SessionRegistrationCause::denied( + SessionPrincipal::Subject, + )) + .unwrap(); + assert_eq!(wire.primary, expected_primary); + assert_eq!( + wire.persistence, + Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted) + ); + let host = error + .diagnostics + .iter() + .filter_map(|d| d.startup.as_ref()) + .find(|startup| startup.primary == wire.primary) + .expect("RemoteWorkerRuntime must preserve the typed startup primary in Host diagnostics"); + assert_eq!(host.primary, wire.primary); + assert!( + host.secondary.iter().any(|failure| matches!( + failure, + protocol::startup_failure::StartupSecondaryFailure::Persistence { + phase: protocol::startup_failure::StartupFailurePhase::RuntimeRecordDelete, + .. + } + )), + "a real Session retention pin rejects outer Runtime deletion without hiding the primary" + ); + let expected = serde_json::to_value(&wire.primary).unwrap(); + assert!( + expected.is_string(), + "the shared attachment contains a codec envelope" + ); + + // A fresh fixture has exactly one attempted Worker. Discover its canonical + // saved aggregate, without making any assumptions about retained ownership + // or public catalog visibility after a failed create. + let workers = std::fs::read_dir(fixture.connected.runtime_root.join("workers")) + .unwrap() + .map(|entry| entry.unwrap().file_name().into_string().unwrap()) + .collect::>(); + assert_eq!( + workers.len(), + 1, + "factory attempt must have real saved metadata" + ); + let saved = read_metadata(&fixture.connected.runtime_root, &workers[0]); + assert!( + contains_saved_value(&saved, &expected), + "saved metadata must retain the original typed Session registration denial" + ); + let reopened = read_metadata(&fixture.connected.runtime_root, &workers[0]); + assert_eq!( + reopened, saved, + "read-only reopen, not a startup operation retry" + ); + assert!(contains_saved_value(&reopened, &expected)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn permitted_session_registration_reaches_real_factory_and_saved_confirmation() { + let fixture = RegistrationFixture::new(false).await; + let initial = open_subjektiv_store(&fixture.connected.api) + .unwrap() + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap(); + let Json(created) = scoped_create_workspace_worker( + State(fixture.connected.api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(fixture.request()), + ) + .await + .unwrap(); + let session_id = fixture.assert_callback(StatusCode::OK); + let store = open_subjektiv_store(&fixture.connected.api).unwrap(); + assert_eq!( + store + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap(), + initial + ); + let attributions = store + .subject_session_attributions(&fixture.subject) + .unwrap(); + assert_eq!(attributions.len(), 1); + assert_eq!(attributions[0].session_id, session_id); + assert_eq!(attributions[0].worker_id, created.worker_id); + assert_eq!(attributions[0].runtime_id, CONNECTED_RUNTIME); + assert!( + fixture + .connected + .runtime_startup_failures + .lock() + .unwrap() + .is_empty() + ); + + let name = format!("worker-runtime-{}", created.worker_id); + let metadata = session_store::WorkerAggregateStore::open_read_only( + &fixture + .connected + .runtime_root + .join("workers") + .join(&created.worker_id), + &name, + ) + .unwrap(); + let saved = metadata.read_by_name(&name).unwrap().unwrap(); + assert!(matches!( + saved.subjektiv_session_attribution, + Some(session_store::SubjektivSessionAttributionState::Confirmed { .. }) + )); + let source = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + fixture + .connected + .factory + .retained_session_snapshot(&worker_runtime::identity::WorkerRef::try_from(&source).unwrap()) + .expect("real factory retained the permitted Worker Session"); +} diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index ab95afe7a..514fed0de 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs @@ -9,6 +9,8 @@ use super::*; mod chat_delegation_connected; #[path = "chat_tests.rs"] mod chat_tests; +#[path = "session_registration_connected_tests.rs"] +mod session_registration_connected; #[path = "subject_body_cleanup_connected_tests.rs"] mod subject_body_cleanup_connected; #[path = "subject_body_connected_tests.rs"] @@ -2737,6 +2739,7 @@ pub(super) struct ConnectedWorkerHarness { materializer: worker_runtime::working_directory::RuntimeGitMaterializer, authorizations: Arc>>, runtime_errors: Arc>>, + runtime_startup_failures: Arc>>, _backend_server: AuthorizationCallbackServer, _runtime_server: ConnectedRuntimeServer, _dir: tempfile::TempDir, @@ -2942,6 +2945,8 @@ impl ConnectedWorkerHarness { .unwrap(); let runtime_errors = Arc::new(Mutex::new(Vec::::new())); let captured_errors = runtime_errors.clone(); + let runtime_startup_failures = Arc::new(Mutex::new(Vec::new())); + let captured_startup_failures = runtime_startup_failures.clone(); let runtime_server = ConnectedRuntimeServer::start( runtime_listener, worker_runtime::http_server::runtime_http_router_with_workspace_auth( @@ -2955,6 +2960,7 @@ impl ConnectedWorkerHarness { .layer(axum::middleware::from_fn( move |request: axum::extract::Request, next: axum::middleware::Next| { let errors = captured_errors.clone(); + let startup_failures = captured_startup_failures.clone(); async move { let response = next.run(request).await; if !response.status().is_client_error() @@ -2965,6 +2971,13 @@ impl ConnectedWorkerHarness { let (parts, body) = response.into_parts(); let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); let body: Value = serde_json::from_slice(&bytes).unwrap(); + if let Some(startup) = body["error"].get("startup").filter(|v| !v.is_null()) + { + startup_failures.lock().unwrap().push( + serde_json::from_value(startup.clone()) + .expect("RuntimeHTTP structured startup failure"), + ); + } errors.lock().unwrap().push( body["error"]["message"] .as_str() @@ -3045,6 +3058,7 @@ impl ConnectedWorkerHarness { materializer, authorizations, runtime_errors, + runtime_startup_failures, _backend_server: backend_server, _runtime_server: runtime_server, _dir: dir,