Merge commit '7873f4359ab62c83107591afd1a8d9f1311bd26b' into work/t838-source-policy-adoption

This commit is contained in:
2026-11-04 15:54:35 +09:00
25 changed files with 1957 additions and 224 deletions
+1
View File
@@ -2,6 +2,7 @@
pub mod identity; pub mod identity;
pub mod invocation; pub mod invocation;
pub mod session_failure;
pub mod startup_failure; pub mod startup_failure;
#[cfg(feature = "stream")] #[cfg(feature = "stream")]
pub mod stream; pub mod stream;
+620
View File
@@ -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<Self>) -> Disclosure<Self> {
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<InternalTransfer> for SessionRegistrationCause {
type Authorization = ();
type Details = Self;
fn disclose(
&self,
facts: &FailureFacts<Self>,
_: &(),
) -> Result<Disclosure<Self>, ContractError> {
Ok(self.disclosure(facts))
}
}
impl ProjectFor<Storage> for SessionRegistrationCause {
type Authorization = ();
type Details = Self;
fn disclose(
&self,
facts: &FailureFacts<Self>,
_: &(),
) -> Result<Disclosure<Self>, ContractError> {
Ok(self.disclosure(facts))
}
}
impl ProjectFor<Browser> for SessionRegistrationCause {
type Authorization = SessionFailureVisibility;
type Details = Self;
fn disclose(
&self,
facts: &FailureFacts<Self>,
authorization: &SessionFailureVisibility,
) -> Result<Disclosure<Self>, 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<Tool> for SessionRegistrationCause {
type Authorization = SessionFailureVisibility;
type Details = Self;
fn disclose(
&self,
facts: &FailureFacts<Self>,
authorization: &SessionFailureVisibility,
) -> Result<Disclosure<Self>, 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<StartupFailureAttachment, ContractError> {
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::<InternalTransfer, _>(&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::<SessionRegistrationCause>(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::<InternalTransfer, _>(&facts, &()).unwrap();
assert_eq!(projected.explanation(), explanation);
let encoded = encode_projected_failure(&projected, budget()).unwrap();
match decode_failure::<SessionRegistrationCause>(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::<SessionRegistrationCause>(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::<SessionRegistrationCause>(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::<InternalTransfer, _>(&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::<SessionRegistrationCause>(
&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::<SessionRegistrationCause>(
&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::<SessionRegistrationCause>(&bytes, budget())
else {
panic!("expected known failure with untrusted supplementary fields");
};
let browser = project_failure::<Browser, _>(
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::<SessionRegistrationCause>() {
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::<Storage, SessionRegistrationCause>(&())
.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::<Browser, SessionRegistrationCause>(&allowed)
.unwrap();
let tool = loaded
.project_for::<Tool, SessionRegistrationCause>(&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::<Browser, SessionRegistrationCause>(&denied),
Err(ContractError::InvalidDetails)
));
assert!(matches!(
loaded.project_for::<Tool, SessionRegistrationCause>(&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::<Browser, SessionRegistrationCause>(&denied),
Err(ContractError::InvalidDetails)
));
assert!(matches!(
previously_public.project_for::<Tool, SessionRegistrationCause>(&denied),
Err(ContractError::InvalidDetails)
));
}
}
fn assert_redacted<A: Audience>(projected: &ProjectedFailure<A, SessionRegistrationCause>) {
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::<InternalTransfer, _>(&facts, &()).unwrap());
assert_redacted(&project_failure::<Storage, _>(&facts, &()).unwrap());
assert_redacted(&project_failure::<Browser, _>(&facts, &allowed).unwrap());
assert_redacted(&project_failure::<Tool, _>(&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())
);
}
}
+16
View File
@@ -105,10 +105,25 @@ pub struct RuntimeApiError {
status: u16, 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<StartupSecondaryFailure>,
pub resources: Option<StartupResourceFacts>,
pub persistence: Option<StartupPersistenceOutcome>,
pub io_kind: Option<StartupIoKind>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct RuntimeApiErrorDetail { pub struct RuntimeApiErrorDetail {
pub code: String, pub code: String,
pub message: String, pub message: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub startup: Option<WorkerStartupFailure>,
} }
impl RuntimeApiError { impl RuntimeApiError {
@@ -117,6 +132,7 @@ impl RuntimeApiError {
error: RuntimeApiErrorDetail { error: RuntimeApiErrorDetail {
code: code.into(), code: code.into(),
message: message.into(), message: message.into(),
startup: None,
}, },
status, status,
} }
+13
View File
@@ -89,6 +89,10 @@ pub struct RepositoryApiError {
pub message: String, pub message: String,
#[serde(default)] #[serde(default)]
pub diagnostics: Vec<Diagnostic>, pub diagnostics: Vec<Diagnostic>,
/// 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<protocol::startup_failure::StartupFailureAttachment>,
#[serde(skip, default = "default_repository_error_status")] #[serde(skip, default = "default_repository_error_status")]
status: u16, status: u16,
} }
@@ -108,9 +112,18 @@ impl RepositoryApiError {
error: error.into(), error: error.into(),
message: message.into(), message: message.into(),
diagnostics, diagnostics,
failure: None,
status, status,
} }
} }
pub fn with_failure(
mut self,
failure: protocol::startup_failure::StartupFailureAttachment,
) -> Self {
self.failure = Some(failure);
self
}
} }
impl api_macros::HttpError for RepositoryApiError { impl api_macros::HttpError for RepositoryApiError {
+67 -9
View File
@@ -30,6 +30,7 @@ use std::sync::{
use feature_storage::FeatureStorage; use feature_storage::FeatureStorage;
use fs4::fs_std::FileExt; use fs4::fs_std::FileExt;
use protocol::session_failure::{SessionPrincipal, SessionRegistrationCause};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use server_api::*; use server_api::*;
use subjektiv::{SubjectRecord, SubjectRole, SubjectSessionAttribution, SubjektivStore}; use subjektiv::{SubjectRecord, SubjectRole, SubjectSessionAttribution, SubjektivStore};
@@ -418,6 +419,49 @@ impl LocalSubjectHost {
} }
Ok(()) 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::<session_store::SessionId>()
.map_err(|error| {
SubjektivHostError::session_registration(
SessionRegistrationCause::invalid_response(),
error,
)
})?;
Ok(())
}
fn attribution( fn attribution(
&self, &self,
context: &SubjektivHostContext, context: &SubjektivHostContext,
@@ -427,15 +471,16 @@ impl LocalSubjectHost {
let store = &self.catalog.store; let store = &self.catalog.store;
if let Some(record) = store if let Some(record) = store
.session_attribution(&context.session_id) .session_attribution(&context.session_id)
.map_err(domain_error)? .map_err(registration_domain_error)?
{ {
if record.subject_id != self.binding.subject_id if record.subject_id != self.binding.subject_id
|| record.runtime_id != self.binding.scope_id || record.runtime_id != self.binding.scope_id
|| record.worker_id != self.worker_id.to_string() || record.worker_id != self.worker_id.to_string()
{ {
return Err(SubjektivHostError::InvalidArgument( return Err(SubjektivHostError::SessionRegistration {
"Session does not belong to this local Subject/Worker scope".into(), cause: SessionRegistrationCause::binding_mismatch(),
)); source: None,
});
} }
store store
.bind_journal_session(&self.writer, record.clone()) .bind_journal_session(&self.writer, record.clone())
@@ -443,9 +488,10 @@ impl LocalSubjectHost {
return Ok(record); return Ok(record);
} }
if !create { if !create {
return Err(SubjektivHostError::InvalidArgument( return Err(SubjektivHostError::SessionRegistration {
"Session attribution is unavailable".into(), cause: SessionRegistrationCause::attribution_missing(),
)); source: None,
});
} }
// The capability is bound before controller startup. The Worker supplies // The capability is bound before controller startup. The Worker supplies
// the Session it owns; models cannot call this Host API or select one. // the Session it owns; models cannot call this Host API or select one.
@@ -455,10 +501,10 @@ impl LocalSubjectHost {
self.worker_id.to_string(), self.worker_id.to_string(),
&context.session_id, &context.session_id,
) )
.map_err(domain_error)?; .map_err(registration_domain_error)?;
let record = store let record = store
.record_session_attribution(record) .record_session_attribution(record)
.map_err(domain_error)?; .map_err(registration_domain_error)?;
store store
.bind_journal_session(&self.writer, record.clone()) .bind_journal_session(&self.writer, record.clone())
.map_err(journal::journal_error)?; .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 { fn domain_error(error: subjektiv::SubjektivError) -> SubjektivHostError {
operation_error(subjektiv::api::subjektiv_store_error(error)) operation_error(subjektiv::api::subjektiv_store_error(error))
} }
@@ -1036,6 +1093,7 @@ impl SubjektivHost for LocalSubjectHost {
context: &SubjektivHostContext, context: &SubjektivHostContext,
create_if_missing: bool, create_if_missing: bool,
) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> { ) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
self.registration_authority(context)?;
let attribution = self.attribution(context, create_if_missing)?; let attribution = self.attribution(context, create_if_missing)?;
self.reserve_journal_session(context)?; self.reserve_journal_session(context)?;
Ok(SubjektivRecordSessionResponse { Ok(SubjektivRecordSessionResponse {
+50
View File
@@ -267,3 +267,53 @@ fn consolidation_operation_policy_is_attempt_bound_and_does_not_inherit_body_rig
.is_err() .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),
);
}
+38 -1
View File
@@ -542,6 +542,8 @@ pub struct RuntimeHttpErrorResponse {
pub struct RuntimeHttpErrorDetail { pub struct RuntimeHttpErrorDetail {
pub code: String, pub code: String,
pub message: String, pub message: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub startup: Option<runtime_api::WorkerStartupFailure>,
} }
#[cfg(feature = "ws-server")] #[cfg(feature = "ws-server")]
@@ -2325,6 +2327,7 @@ struct RuntimeHttpRestError {
status: StatusCode, status: StatusCode,
code: String, code: String,
message: String, message: String,
startup: Option<runtime_api::WorkerStartupFailure>,
} }
impl RuntimeHttpRestError { impl RuntimeHttpRestError {
@@ -2333,13 +2336,28 @@ impl RuntimeHttpRestError {
status, status,
code: code.into(), code: code.into(),
message: message.into(), message: message.into(),
startup: None,
} }
} }
fn runtime(error: RuntimeError) -> Self { fn runtime(error: RuntimeError) -> Self {
let status = status_for_runtime_error(&error); let status = status_for_runtime_error(&error);
let code = code_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 { fn working_directory_join(_error: tokio::task::JoinError) -> Self {
@@ -2375,12 +2393,31 @@ impl IntoResponse for RuntimeHttpRestError {
error: RuntimeHttpErrorDetail { error: RuntimeHttpErrorDetail {
code: self.code, code: self.code,
message: self.message, message: self.message,
startup: self.startup,
}, },
}; };
(self.status, Json(body)).into_response() (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<runtime_api::WorkerStartupFailure> {
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 { fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
match error { match error {
RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => { RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => {
@@ -224,7 +224,10 @@ where
} }
fn api_error(error: RuntimeHttpRestError) -> runtime_api::RuntimeApiError { 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 { impl runtime_api::RuntimeApi for RuntimeManagementApi {
+3 -1
View File
@@ -83,7 +83,9 @@ impl JournalCompactionFailure {
SubjektivHostError::PermissionDenied(_) => Self::SnapshotPermissionDenied, SubjektivHostError::PermissionDenied(_) => Self::SnapshotPermissionDenied,
SubjektivHostError::Conflict { .. } => Self::SnapshotConflict, SubjektivHostError::Conflict { .. } => Self::SnapshotConflict,
SubjektivHostError::InvalidArgument(_) => Self::SnapshotInvalidArgument, SubjektivHostError::InvalidArgument(_) => Self::SnapshotInvalidArgument,
SubjektivHostError::Unavailable(_) => Self::SnapshotUnavailable, SubjektivHostError::Unavailable(_) | SubjektivHostError::SessionRegistration { .. } => {
Self::SnapshotUnavailable
}
} }
} }
+86
View File
@@ -9,6 +9,18 @@ use protocol::startup_failure::{
}; };
impl WorkerError { impl WorkerError {
pub fn session_registration_cause(
&self,
) -> Option<protocol::session_failure::SessionRegistrationCause> {
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 { pub fn with_startup_failure(self, failure: StartupFailureAttachment) -> Self {
Self::StartupFailure { Self::StartupFailure {
source: Box::new(self), source: Box::new(self),
@@ -55,6 +67,9 @@ impl WorkerError {
if let Some(failure) = self.startup_failure() { if let Some(failure) = self.startup_failure() {
return Ok(failure.clone()); return Ok(failure.clone());
} }
if let Some(cause) = self.session_registration_cause() {
return protocol::session_failure::session_registration_attachment(cause);
}
let reason = match self { let reason = match self {
Self::ConstructionCleanup { source, .. } Self::ConstructionCleanup { source, .. }
| Self::RestoreActivationEffects { source } => return source.startup_failure_at(stage), | Self::RestoreActivationEffects { source } => return source.startup_failure_at(stage),
@@ -91,6 +106,77 @@ pub fn runtime_startup_attachment(
StartupFailureAttachment::from_projected(&projected) 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<StartupFailureAttachment, error_contract::ContractError> {
if let Ok(projected) = failure
.project_for::<InternalTransfer, protocol::session_failure::SessionRegistrationCause>(&())
{
return StartupFailureAttachment::from_projected(&projected);
}
let projected = failure.project_for::<InternalTransfer, RuntimeStartupCause>(&())?;
StartupFailureAttachment::from_projected(&projected)
}
/// Dispatch by the domain attested in process, not by parsing attachment fields.
pub(crate) fn persist_binding_startup_failure<S: session_store::WorkerMetadataStore>(
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 /// Bind a caller-owned cause to the exact pre-observed Session/attribution. Saving
/// never replaces the original error, changes ownership, or issues retry authority. /// never replaces the original error, changes ownership, or issues retry authority.
pub fn persist_startup_failure<S, D>( pub fn persist_startup_failure<S, D>(
+112 -6
View File
@@ -64,6 +64,12 @@ pub struct SubjektivHostSettings {
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum SubjektivHostError { pub enum SubjektivHostError {
#[error("{cause}")]
SessionRegistration {
cause: protocol::session_failure::SessionRegistrationCause,
#[source]
source: Option<Box<dyn std::error::Error + Send + Sync>>,
},
#[error("{message}")] #[error("{message}")]
Conflict { code: String, message: String }, Conflict { code: String, message: String },
#[error("{0}")] #[error("{0}")]
@@ -74,6 +80,30 @@ pub enum SubjektivHostError {
Unavailable(String), 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<SubjektivHostError> for ToolError { impl From<SubjektivHostError> for ToolError {
fn from(error: SubjektivHostError) -> Self { fn from(error: SubjektivHostError) -> Self {
match error { match error {
@@ -85,6 +115,9 @@ impl From<SubjektivHostError> for ToolError {
Self::ExecutionFailed(format!("Subjektiv operation permission denied: {message}")) Self::ExecutionFailed(format!("Subjektiv operation permission denied: {message}"))
} }
SubjektivHostError::Unavailable(message) => Self::ExecutionFailed(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<SubjektivRecordSessionResponse, SubjektivHostError> {
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::<RepositoryApiError>(&response.body) {
Ok(error) => match error.failure {
Some(failure) => match failure.decode::<SessionRegistrationCause>() {
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<SubjektivRecordSessionResponse, SubjektivHostError> {
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<T: serde::de::DeserializeOwned>( fn decode<T: serde::de::DeserializeOwned>(
response: crate::WorkspaceResponse, response: crate::WorkspaceResponse,
) -> Result<T, SubjektivHostError> { ) -> Result<T, SubjektivHostError> {
@@ -490,12 +601,7 @@ impl SubjektivHost for BackendSubjektivHost {
context: &SubjektivHostContext, context: &SubjektivHostContext,
create_if_missing: bool, create_if_missing: bool,
) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> { ) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
self.execute(WorkspaceServerOperation::SubjektivRecordSession( record_backend_session(self.client.as_ref(), &context.session_id, create_if_missing)
SubjektivRecordSessionRequest {
session_id: context.session_id.clone(),
create_if_missing,
},
))
} }
fn request_consolidation( fn request_consolidation(
&self, &self,
+42 -139
View File
@@ -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 { fn bounded_attribution_diagnostic(mut diagnostic: String) -> String {
const MAX_CHARS: usize = 1024; const MAX_CHARS: usize = 1024;
if diagnostic.chars().count() > MAX_CHARS { if diagnostic.chars().count() > MAX_CHARS {
@@ -9721,15 +9713,10 @@ where
&mut expected, &mut expected,
) )
.map_err(|error| match expected { .map_err(|error| match expected {
Some(expected) => crate::startup_failure::persist_startup_failure::< Some(expected) => crate::startup_failure::persist_binding_startup_failure(
_,
protocol::startup_failure::RuntimeStartupCause,
>(
&self.store, &self.store,
&expected, &expected,
error, error,
protocol::startup_failure::StartupStage::Binding,
&(),
), ),
None => error, None => error,
}) })
@@ -9855,145 +9842,55 @@ where
pending_attribution.clone(), pending_attribution.clone(),
)?; )?;
let output = if self.subjektiv_host.is_some() { let output = if self.subjektiv_host.is_some() {
let connection = self self.subjektiv_host_connection()
.subjektiv_host_connection() .map_err(|error| crate::subjektiv::SubjektivHostError::session_registration(
.map_err(|error| WorkerError::SubjektivSessionAttribution { protocol::session_failure::SessionRegistrationCause::outcome_unknown(), error,
state: "failed", ))
message: error.to_string(), .and_then(|connection| connection.ok_or_else(|| {
})? crate::subjektiv::SubjektivHostError::SessionRegistration {
.ok_or_else(|| WorkerError::SubjektivSessionAttribution { cause: protocol::session_failure::SessionRegistrationCause::binding_mismatch(),
state: "failed", source: None,
message: "Session attribution requires enabled explicit Host capability".into(), }
})?; }))
match connection .and_then(|connection| connection.host.record_session(&connection.context, create_if_missing))
.host } else {
.record_session(&connection.context, 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, Ok(output) => output,
Err(error) => { Err(error) => {
let message = bounded_attribution_diagnostic(error.to_string()); use protocol::session_failure::SessionRegistrationReason;
let rejected = let cause = error.session_registration_cause();
!matches!(error, crate::subjektiv::SubjektivHostError::Unavailable(_)); let rejected = matches!(
let state = if create_if_missing && rejected { 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 { SubjektivSessionAttributionState::Failed {
session_id, session_id,
diagnostic: message.clone(), diagnostic,
} }
} else { } else {
incomplete_subjektiv_attribution_state( incomplete_subjektiv_attribution_state(
session_id, session_id,
create_if_missing, create_if_missing,
message.clone(), diagnostic,
) )
}; };
persist_attribution(Some(&pending_attribution), state)?; let source = WorkerError::SessionRegistration { source: error };
return Err(WorkerError::SubjektivSessionAttribution { let saved = persist_attribution(Some(&pending_attribution), next);
state: if create_if_missing && rejected { return Err(crate::startup_failure::retain_attribution_save_failure(
"failed" source, saved,
} else {
incomplete_subjektiv_attribution_label(create_if_missing)
},
message,
});
}
}
} 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 {
SubjektivSessionAttributionState::Failed {
session_id,
diagnostic: message.clone(),
}
} else {
SubjektivSessionAttributionState::LegacyUnknown {
session_id,
diagnostic: message.clone(),
}
};
persist_attribution(Some(&pending_attribution), state)?;
return Err(WorkerError::SubjektivSessionAttribution {
state: if create_if_missing {
"failed"
} else {
"legacy_unknown"
},
message,
});
}
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( persist_attribution(
Some(&pending_attribution), Some(&pending_attribution),
SubjektivSessionAttributionState::Confirmed { SubjektivSessionAttributionState::Confirmed {
@@ -11555,6 +11452,12 @@ pub enum WorkerError {
#[error("feature lifecycle rejected operation: {0}")] #[error("feature lifecycle rejected operation: {0}")]
FeatureLifecycle(String), FeatureLifecycle(String),
#[error("Session registration failed: {source}")]
SessionRegistration {
#[source]
source: crate::subjektiv::SubjektivHostError,
},
#[error("Subjektiv Session attribution lifecycle is {state}: {message}")] #[error("Subjektiv Session attribution lifecycle is {state}: {message}")]
SubjektivSessionAttribution { SubjektivSessionAttribution {
state: &'static str, state: &'static str,
+38 -5
View File
@@ -117,6 +117,9 @@ where
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeDiagnostic { pub struct RuntimeDiagnostic {
/// Internal startup observations; public consumers must separately authorize projection.
#[serde(skip)]
pub startup: Option<runtime_api::WorkerStartupFailure>,
pub code: String, pub code: String,
pub severity: HostDiagnosticSeverity, pub severity: HostDiagnosticSeverity,
pub message: String, pub message: String,
@@ -6649,7 +6652,7 @@ fn sanitize_embedded_execution_detail(summary: &str, message: &str) -> String {
} }
fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnostic { fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnostic {
match error { let mut projection = match error {
EmbeddedRuntimeError::RestoreAdmissionBusy { .. } => diagnostic( EmbeddedRuntimeError::RestoreAdmissionBusy { .. } => diagnostic(
"restore_admission_busy", "restore_admission_busy",
HostDiagnosticSeverity::Warning, HostDiagnosticSeverity::Warning,
@@ -6767,7 +6770,19 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti
HostDiagnosticSeverity::Error, HostDiagnosticSeverity::Error,
"Embedded Runtime state is unavailable".to_string(), "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 { fn host_id_for_embedded_workspace(workspace_id: &str) -> String {
@@ -6846,6 +6861,7 @@ fn runtime_api_diagnostic(
) -> RuntimeDiagnostic { ) -> RuntimeDiagnostic {
match error { match error {
runtime_api::client_support::ClientError::Public { status, 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 remote_code = error.error.code;
let message = sanitize_remote_runtime_message(&remote_code, &error.error.message); let message = sanitize_remote_runtime_message(&remote_code, &error.error.message);
let (code, severity) = match status { let (code, severity) = match status {
@@ -6862,13 +6878,15 @@ fn runtime_api_diagnostic(
}, },
), ),
}; };
diagnostic( let mut projection = diagnostic(
code, code,
severity, severity,
format!( format!(
"Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}" "Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}"
), ),
) );
projection.startup = startup;
projection
} }
runtime_api::client_support::ClientError::Failure(failure) => { runtime_api::client_support::ClientError::Failure(failure) => {
let code = match failure { let code = match failure {
@@ -6945,11 +6963,25 @@ fn remote_http_status_diagnostic(
} }
_ => (remote_code, HostDiagnosticSeverity::Warning), _ => (remote_code, HostDiagnosticSeverity::Warning),
}; };
diagnostic( let mut projection = diagnostic(
code, code,
severity, severity,
format!("Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {remote_message}"), 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<runtime_api::WorkerStartupFailure> {
startup.primary =
worker::startup_failure::reproject_internal_startup_failure(&startup.primary).ok()?;
Some(startup)
} }
fn diagnostic( fn diagnostic(
@@ -6958,6 +6990,7 @@ fn diagnostic(
message: impl Into<String>, message: impl Into<String>,
) -> RuntimeDiagnostic { ) -> RuntimeDiagnostic {
RuntimeDiagnostic { RuntimeDiagnostic {
startup: None,
code: code.into(), code: code.into(),
severity, severity,
message: message.into(), message: message.into(),
+193 -26
View File
@@ -28,6 +28,7 @@ use subjektiv::api::*;
mod backend_job_profile; mod backend_job_profile;
mod chat; mod chat;
mod external_operation_audit; mod external_operation_audit;
mod session_registration;
mod subject_authorization; mod subject_authorization;
mod subject_body; mod subject_body;
mod subjektiv_deletion; mod subjektiv_deletion;
@@ -8611,6 +8612,7 @@ fn worker_input_result_to_api(
// distinguishable without collapsing it into a definite refusal or replaying input. // distinguishable without collapsing it into a definite refusal or replaying input.
if result.disposition == WorkerInputDisposition::Unknown { if result.disposition == WorkerInputDisposition::Unknown {
result.diagnostics.push(RuntimeDiagnostic { result.diagnostics.push(RuntimeDiagnostic {
startup: None,
code: "worker_input_outcome_unknown".into(), code: "worker_input_outcome_unknown".into(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "Worker input delivery outcome is unknown; inspect the current session before retrying".into(), message: "Worker input delivery outcome is unknown; inspect the current session before retrying".into(),
@@ -19529,7 +19531,12 @@ fn source_ref_readiness_blocker(code: &str) -> &'static str {
} }
fn remap_source_ref_error(error: ApiError) -> ApiError { 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 { match error {
Error::RuntimeOperationFailed { Error::RuntimeOperationFailed {
runtime_id, runtime_id,
@@ -19541,7 +19548,12 @@ fn remap_source_ref_error(error: ApiError) -> ApiError {
message, message,
} }
.into(), .into(),
error => ApiError { error, diagnostics }, error => ApiError {
error,
diagnostics,
session_registration_cause,
session_registration_source,
},
} }
} }
@@ -20145,7 +20157,9 @@ async fn scoped_list_merge_requests(
} }
fn merge_ref_diagnostic(error: ApiError) -> MergeRequestRefDiagnostic { fn merge_ref_diagnostic(error: ApiError) -> MergeRequestRefDiagnostic {
let ApiError { error, diagnostics } = error; let ApiError {
error, diagnostics, ..
} = error;
diagnostics diagnostics
.into_iter() .into_iter()
.next() .next()
@@ -24275,19 +24289,51 @@ fn subjektiv_worker_scope_from_store(
api: &WorkspaceApi, api: &WorkspaceApi,
workspace_id: &str, workspace_id: &str,
context: &server_api::ServerRequestContext, context: &server_api::ServerRequestContext,
) -> ApiResult<SubjektivStoredWorkerScope> {
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<SubjektivStoredWorkerScope> { ) -> ApiResult<SubjektivStoredWorkerScope> {
let source = context.runtime_source.as_ref().ok_or_else(|| { let source = context.runtime_source.as_ref().ok_or_else(|| {
judgment.error(
Error::WorkspacePermissionDenied( Error::WorkspacePermissionDenied(
"subjektiv operations require authenticated Runtime-owned source authority".to_string(), "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(|| { let worker_id = source.worker_id.as_deref().ok_or_else(|| {
judgment.error(
Error::WorkspacePermissionDenied( Error::WorkspacePermissionDenied(
"subjektiv operations require Runtime-bound Worker identity".to_string(), "subjektiv operations require Runtime-bound Worker identity".to_string(),
),
protocol::session_failure::SessionRegistrationCause::binding_mismatch(),
) )
})?; })?;
let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id); 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 { return Ok(SubjektivStoredWorkerScope {
subject_id: grant.subject_id, subject_id: grant.subject_id,
worker, worker,
@@ -24296,13 +24342,18 @@ fn subjektiv_worker_scope_from_store(
} }
let lease = api let lease = api
.store .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(|| { .ok_or_else(|| {
judgment.error(
Error::WorkspacePermissionDenied( Error::WorkspacePermissionDenied(
"current Worker has no keyed singleton ownership".to_string(), "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 { let authority = match authority {
SubjektivWorkerAuthority::Subject => SubjektivStoredScopeAuthority::Subject, SubjektivWorkerAuthority::Subject => SubjektivStoredScopeAuthority::Subject,
SubjektivWorkerAuthority::Consolidation => { SubjektivWorkerAuthority::Consolidation => {
@@ -24367,6 +24418,14 @@ fn authorize_subjektiv_operation(
resource: resource.clone(), 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 // The binding query returns only the authorized concrete target, never a
// wildcard ceiling or a broader permission that happened to cover it. // wildcard ceiling or a broader permission that happened to cover it.
.map(|scope| scope.map(|_| resource)) .map(|scope| scope.map(|_| resource))
@@ -24387,18 +24446,23 @@ fn require_subjektiv_worker_authority(
} }
} }
fn subjektiv_session_attribution( fn subjektiv_session_attribution_with_judgment(
api: &WorkspaceApi, api: &WorkspaceApi,
workspace_id: &str, workspace_id: &str,
context: &server_api::ServerRequestContext, context: &server_api::ServerRequestContext,
session_id: &str, session_id: &str,
judgment: session_registration::Judgment,
) -> ApiResult<crate::subjektiv::SubjectSessionAttribution> { ) -> ApiResult<crate::subjektiv::SubjectSessionAttribution> {
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) { if !matches!(scope.authority, SubjektivStoredScopeAuthority::Subject) {
return Err(Error::WorkspacePermissionDenied( return Err(judgment.error(
Error::WorkspacePermissionDenied(
"subject Session attribution requires the current subject Worker".into(), "subject Session attribution requires the current subject Worker".into(),
) ),
.into()); protocol::session_failure::SessionRegistrationCause::denied(
protocol::session_failure::SessionPrincipal::Subject,
),
));
} }
let session_id = session_id.trim(); let session_id = session_id.trim();
if session_id.is_empty() || session_id.chars().any(char::is_control) { if session_id.is_empty() || session_id.chars().any(char::is_control) {
@@ -24419,24 +24483,52 @@ fn require_existing_subjektiv_session_attribution(
context: &server_api::ServerRequestContext, context: &server_api::ServerRequestContext,
session_id: &str, session_id: &str,
) -> ApiResult<crate::subjektiv::SubjectSessionAttribution> { ) -> ApiResult<crate::subjektiv::SubjectSessionAttribution> {
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<crate::subjektiv::SubjectSessionAttribution> {
let expected = subjektiv_session_attribution_with_judgment(
api,
workspace_id,
context,
session_id,
judgment,
)?;
let existing = open_subjektiv_store(api)? let existing = open_subjektiv_store(api)?
.session_attribution(&expected.session_id) .session_attribution(&expected.session_id)
.map_err(subjektiv_store_error)? .map_err(subjektiv_store_error)?
.ok_or_else(|| { .ok_or_else(|| {
judgment.error(
Error::RepositoryConflict(format!( Error::RepositoryConflict(format!(
"subjektiv_session_attribution_incomplete: Session {} has not completed its Host attribution lifecycle", "subjektiv_session_attribution_incomplete: Session {} has not completed its Host attribution lifecycle",
expected.session_id expected.session_id
)) )),
protocol::session_failure::SessionRegistrationCause::attribution_missing(),
)
})?; })?;
if existing.subject_id != expected.subject_id if existing.subject_id != expected.subject_id
|| existing.runtime_id != expected.runtime_id || existing.runtime_id != expected.runtime_id
|| existing.worker_id != expected.worker_id || existing.worker_id != expected.worker_id
{ {
return Err(Error::WorkspacePermissionDenied( return Err(judgment.error(
"existing Session attribution does not match the authorized subject Worker".to_string(), Error::WorkspacePermissionDenied(
) "existing Session attribution does not match the authorized subject Worker"
.into()); .to_string(),
),
protocol::session_failure::SessionRegistrationCause::binding_mismatch(),
));
} }
Ok(existing) Ok(existing)
} }
@@ -24855,18 +24947,20 @@ async fn scoped_record_subjektiv_session(
.run(move || { .run(move || {
validate_workspace_scope(&api, &path.workspace_id)?; validate_workspace_scope(&api, &path.workspace_id)?;
let attribution = if request.create_if_missing { let attribution = if request.create_if_missing {
subjektiv_session_attribution( subjektiv_session_attribution_with_judgment(
&api, &api,
&path.workspace_id, &path.workspace_id,
&context, &context,
&request.session_id, &request.session_id,
session_registration::Judgment::SessionRecord,
)? )?
} else { } else {
require_existing_subjektiv_session_attribution( require_existing_subjektiv_session_attribution_with_judgment(
&api, &api,
&path.workspace_id, &path.workspace_id,
&context, &context,
&request.session_id, &request.session_id,
session_registration::Judgment::SessionRecord,
)? )?
}; };
authorize_subjektiv_operation( authorize_subjektiv_operation(
@@ -24876,13 +24970,14 @@ async fn scoped_record_subjektiv_session(
&attribution.subject_id, &attribution.subject_id,
"session.record", "session.record",
)?; )?;
subjektiv_journal::bind_host_session(&api, attribution.clone())?; session_registration::bind_host_session(&api, attribution.clone())?;
Ok(Json(server_api::SubjektivRecordSessionResponse { Ok(Json(server_api::SubjektivRecordSessionResponse {
subject_id: attribution.subject_id, subject_id: attribution.subject_id,
session_id: attribution.session_id, session_id: attribution.session_id,
})) }))
}) })
.await .await
.map_err(session_registration::execution_error)
} }
const SUBJEKTIV_SESSION_DEFAULT_LIMIT: usize = 20; const SUBJEKTIV_SESSION_DEFAULT_LIMIT: usize = 20;
@@ -26926,6 +27021,7 @@ fn skill_api_error(error: skills::SkillError) -> ApiError {
message: format!("unknown Skill `{name}`"), message: format!("unknown Skill `{name}`"),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "skill_not_found".to_string(), code: "skill_not_found".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: format!("unknown Skill `{name}`"), message: format!("unknown Skill `{name}`"),
@@ -30332,6 +30428,7 @@ fn prepare_workspace_workdir_create(
.to_string(), .to_string(),
), ),
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "default_runtime_not_configured".to_string(), code: "default_runtime_not_configured".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Workspace default Runtime is not configured".to_string(), message: "Workspace default Runtime is not configured".to_string(),
@@ -30486,6 +30583,7 @@ async fn create_workspace_working_directory(
return Err(ApiError::with_diagnostics( return Err(ApiError::with_diagnostics(
Error::UnknownRuntime(reserved.resolved_runtime_id), Error::UnknownRuntime(reserved.resolved_runtime_id),
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "runtime_unavailable".to_string(), code: "runtime_unavailable".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Selected Runtime is not available".to_string(), message: "Selected Runtime is not available".to_string(),
@@ -30509,6 +30607,7 @@ async fn create_workspace_working_directory(
message: "Selected Runtime is not connected".to_string(), message: "Selected Runtime is not connected".to_string(),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "runtime_unavailable".to_string(), code: "runtime_unavailable".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Selected Runtime is not available".to_string(), message: "Selected Runtime is not available".to_string(),
@@ -30565,6 +30664,7 @@ async fn create_workspace_working_directory(
return Err(ApiError::with_diagnostics( return Err(ApiError::with_diagnostics(
error.into_error(), error.into_error(),
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: failure_code, code: failure_code,
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Selected Runtime could not inspect the reserved Workdir".to_string(), message: "Selected Runtime could not inspect the reserved Workdir".to_string(),
@@ -30603,6 +30703,7 @@ async fn create_workspace_working_directory(
return Err(ApiError::with_diagnostics( return Err(ApiError::with_diagnostics(
error.into_error(), error.into_error(),
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: failure_code, code: failure_code,
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Selected Runtime rejected Workdir creation".to_string(), message: "Selected Runtime rejected Workdir creation".to_string(),
@@ -30642,6 +30743,7 @@ async fn create_workspace_working_directory(
diagnostics.insert( diagnostics.insert(
0, 0,
RuntimeDiagnostic { RuntimeDiagnostic {
startup: None,
code: failure_code.clone(), code: failure_code.clone(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Runtime did not create the reserved Workdir".to_string(), message: "Runtime did not create the reserved Workdir".to_string(),
@@ -31396,6 +31498,7 @@ fn build_runtime_cleanup_plan(
let mut diagnostics = error.diagnostics; let mut diagnostics = error.diagnostics;
if diagnostics.is_empty() { if diagnostics.is_empty() {
diagnostics.push(RuntimeDiagnostic { diagnostics.push(RuntimeDiagnostic {
startup: None,
code: "workspace_cleanup_runtime_observation_unavailable".to_string(), code: "workspace_cleanup_runtime_observation_unavailable".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: sanitize_backend_error(&error.error.to_string()), message: sanitize_backend_error(&error.error.to_string()),
@@ -35505,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(), message: "The embedded Runtime does not accept working directories".to_string(),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "embedded_worker_workdir_unsupported".to_string(), code: "embedded_worker_workdir_unsupported".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "Choose a non-embedded Runtime for workspace-file Workers; embedded Workers are no-workdir Workspace-API workers.".to_string(), message: "Choose a non-embedded Runtime for workspace-file Workers; embedded Workers are no-workdir Workspace-API workers.".to_string(),
@@ -35595,6 +35699,7 @@ fn finalize_worker_spawn_stage<T>(
.map(|assignment| assignment.operation_id.as_str()) .map(|assignment| assignment.operation_id.as_str())
.unwrap_or("none"); .unwrap_or("none");
let mut diagnostics = vec![RuntimeDiagnostic { let mut diagnostics = vec![RuntimeDiagnostic {
startup: None,
code: format!("worker_spawn_finalize_{}_failed", stage.as_str()), code: format!("worker_spawn_finalize_{}_failed", stage.as_str()),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: format!( message: format!(
@@ -35610,6 +35715,7 @@ fn finalize_worker_spawn_stage<T>(
diagnostics.extend(compensation_errors); diagnostics.extend(compensation_errors);
if !compensation_failed { if !compensation_failed {
diagnostics.push(RuntimeDiagnostic { diagnostics.push(RuntimeDiagnostic {
startup: None,
code: "worker_spawn_compensated".to_string(), code: "worker_spawn_compensated".to_string(),
severity: HostDiagnosticSeverity::Info, severity: HostDiagnosticSeverity::Info,
message: format!( message: format!(
@@ -35945,6 +36051,7 @@ fn lifecycle_failure_detail(
fn spawn_compensation_diagnostic(code: &str, message: String) -> RuntimeDiagnostic { fn spawn_compensation_diagnostic(code: &str, message: String) -> RuntimeDiagnostic {
RuntimeDiagnostic { RuntimeDiagnostic {
startup: None,
code: code.to_string(), code: code.to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message, message,
@@ -36950,6 +37057,7 @@ fn project_workspace_worker_with_attachments(
summary.state = "unavailable".to_string(); summary.state = "unavailable".to_string();
summary.worker_state = None; summary.worker_state = None;
summary.diagnostics.push(RuntimeDiagnostic { summary.diagnostics.push(RuntimeDiagnostic {
startup: None,
code: "worker_singleton_fenced".to_string(), code: "worker_singleton_fenced".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "This historical singleton generation is fenced by Backend ownership." message: "This historical singleton generation is fenced by Backend ownership."
@@ -37655,6 +37763,7 @@ fn working_directory_summaries(
} }
if let Some(grant_id) = legacy_external_workdir_grant_id(api, &record)? { if let Some(grant_id) = legacy_external_workdir_grant_id(api, &record)? {
diagnostics.push(RuntimeDiagnostic { diagnostics.push(RuntimeDiagnostic {
startup: None,
code: "external_workdir_legacy_permissions".to_string(), code: "external_workdir_legacy_permissions".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: format!( message: format!(
@@ -37861,6 +37970,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> InternalWorker
}, },
workdir_attachments: Vec::new(), workdir_attachments: Vec::new(),
diagnostics: vec![RuntimeDiagnostic { diagnostics: vec![RuntimeDiagnostic {
startup: None,
code: "backend_worker_missing".to_string(), code: "backend_worker_missing".to_string(),
severity: HostDiagnosticSeverity::Info, severity: HostDiagnosticSeverity::Info,
message: message:
@@ -37905,6 +38015,7 @@ fn worker_summary_from_projection(
Vec::new() Vec::new()
} else { } else {
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "worker_observation_unavailable".to_string(), code: "worker_observation_unavailable".to_string(),
severity: HostDiagnosticSeverity::Info, severity: HostDiagnosticSeverity::Info,
message: "The Worker remains in the Backend catalog, but its Runtime observation is currently unavailable.".to_string(), message: "The Worker remains in the Backend catalog, but its Runtime observation is currently unavailable.".to_string(),
@@ -38551,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(), message: "working directory relative_cwd must stay inside the Runtime-owned working directory".to_string(),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_relative_cwd_invalid".to_string(), code: "working_directory_relative_cwd_invalid".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "relative_cwd must be a relative path without parent traversal".to_string(), message: "relative_cwd must be a relative path without parent traversal".to_string(),
@@ -39145,6 +39257,7 @@ fn settings_diagnostic(
message: impl Into<String>, message: impl Into<String>,
) -> RuntimeDiagnostic { ) -> RuntimeDiagnostic {
RuntimeDiagnostic { RuntimeDiagnostic {
startup: None,
code: code.into(), code: code.into(),
severity, severity,
message: message.into(), message: message.into(),
@@ -39166,6 +39279,7 @@ fn repository_lookup<T>(result: std::result::Result<T, RepositoryLookupError>) -
message: message.clone(), message: message.clone(),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "repository_not_configured".to_string(), code: "repository_not_configured".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message, message,
@@ -39183,6 +39297,7 @@ fn repository_lookup<T>(result: std::result::Result<T, RepositoryLookupError>) -
message: message.clone(), message: message.clone(),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "repository_provider_unsupported".to_string(), code: "repository_provider_unsupported".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message, message,
@@ -39198,6 +39313,7 @@ fn repository_lookup<T>(result: std::result::Result<T, RepositoryLookupError>) -
message: message.clone(), message: message.clone(),
}, },
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "repository_evidence_invalid".to_string(), code: "repository_evidence_invalid".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message, message,
@@ -39358,6 +39474,7 @@ fn workspace_id_mismatch_error() -> ApiError {
ApiError::with_diagnostics( ApiError::with_diagnostics(
Error::WorkspaceIdMismatch, Error::WorkspaceIdMismatch,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "workspace_id_mismatch".to_string(), code: "workspace_id_mismatch".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message, message,
@@ -39369,6 +39486,8 @@ fn workspace_id_mismatch_error() -> ApiError {
struct ApiError { struct ApiError {
error: Error, error: Error,
diagnostics: Vec<RuntimeDiagnostic>, diagnostics: Vec<RuntimeDiagnostic>,
session_registration_cause: Option<protocol::session_failure::SessionRegistrationCause>,
session_registration_source: Option<Box<dyn std::error::Error + Send + Sync>>,
} }
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
@@ -39465,6 +39584,7 @@ impl ApiErrorLog {
.diagnostics .diagnostics
.iter() .iter()
.map(|diagnostic| RuntimeDiagnostic { .map(|diagnostic| RuntimeDiagnostic {
startup: None,
code: diagnostic.code.clone(), code: diagnostic.code.clone(),
severity: match diagnostic.severity { severity: match diagnostic.severity {
server_api::DiagnosticSeverity::Info => HostDiagnosticSeverity::Info, server_api::DiagnosticSeverity::Info => HostDiagnosticSeverity::Info,
@@ -39502,6 +39622,7 @@ impl From<Error> for ApiError {
Error::DbCapacityExceeded Error::DbCapacityExceeded
| Error::DbExecutorUnavailable | Error::DbExecutorUnavailable
| Error::DbOperationOutcomeUnknown => vec![RuntimeDiagnostic { | Error::DbOperationOutcomeUnknown => vec![RuntimeDiagnostic {
startup: None,
code: match &error { code: match &error {
Error::DbCapacityExceeded => "db_capacity_exceeded", Error::DbCapacityExceeded => "db_capacity_exceeded",
Error::DbExecutorUnavailable => "db_executor_unavailable", Error::DbExecutorUnavailable => "db_executor_unavailable",
@@ -39512,21 +39633,25 @@ impl From<Error> for ApiError {
message: error.to_string(), message: error.to_string(),
}], }],
Error::WorkdirSessionOpenFailed { transport, .. } => vec![RuntimeDiagnostic { Error::WorkdirSessionOpenFailed { transport, .. } => vec![RuntimeDiagnostic {
startup: None,
code: "workdir_session_open_failed".to_string(), code: "workdir_session_open_failed".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: transport.message.clone(), message: transport.message.clone(),
}], }],
Error::RuntimeOperationFailed { code, message, .. } => vec![RuntimeDiagnostic { Error::RuntimeOperationFailed { code, message, .. } => vec![RuntimeDiagnostic {
startup: None,
code: code.clone(), code: code.clone(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: sanitize_backend_error(message), message: sanitize_backend_error(message),
}], }],
Error::WorkspaceSigningIdentity { code, message } => vec![RuntimeDiagnostic { Error::WorkspaceSigningIdentity { code, message } => vec![RuntimeDiagnostic {
startup: None,
code: code.clone(), code: code.clone(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: sanitize_backend_error(message), message: sanitize_backend_error(message),
}], }],
Error::Ticket(ticket_error) => vec![RuntimeDiagnostic { Error::Ticket(ticket_error) => vec![RuntimeDiagnostic {
startup: None,
code: match ticket_error { code: match ticket_error {
ticket::TicketError::NotFound(_) => "ticket_not_found", ticket::TicketError::NotFound(_) => "ticket_not_found",
ticket::TicketError::Ambiguous { .. } => "ticket_ambiguous", ticket::TicketError::Ambiguous { .. } => "ticket_ambiguous",
@@ -39573,6 +39698,7 @@ impl From<Error> for ApiError {
subjektiv_diagnostic_code(message) subjektiv_diagnostic_code(message)
.map(|code| { .map(|code| {
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: code.to_string(), code: code.to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: sanitize_backend_error(message), message: sanitize_backend_error(message),
@@ -39582,7 +39708,12 @@ impl From<Error> for ApiError {
} }
_ => Vec::new(), _ => Vec::new(),
}; };
Self { error, diagnostics } Self {
error,
diagnostics,
session_registration_cause: None,
session_registration_source: None,
}
} }
} }
@@ -39599,13 +39730,21 @@ fn subjektiv_diagnostic_code(message: &str) -> Option<&'static str> {
impl ApiError { impl ApiError {
fn with_diagnostics(error: Error, diagnostics: Vec<RuntimeDiagnostic>) -> Self { fn with_diagnostics(error: Error, diagnostics: Vec<RuntimeDiagnostic>) -> Self {
Self { error, diagnostics } Self {
error,
diagnostics,
session_registration_cause: None,
session_registration_source: None,
}
} }
fn into_repository_api_error(self) -> server_api::RepositoryApiError { fn into_repository_api_error(self) -> server_api::RepositoryApiError {
let status = api_error_status(&self.error); let status = api_error_status(&self.error);
let message = api_error_response_message(&self.error); let message = self
server_api::RepositoryApiError::new( .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(), status.as_u16(),
match self.error { match self.error {
Error::RestoreObservationConflict => "restore_observation_conflict", Error::RestoreObservationConflict => "restore_observation_conflict",
@@ -39616,7 +39755,15 @@ impl ApiError {
}, },
message, message,
working_directory_diagnostics(self.diagnostics), 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,
}
} }
} }
@@ -42957,6 +43104,7 @@ mod tests {
worker: None, worker: None,
acceptance_evidence: Vec::new(), acceptance_evidence: Vec::new(),
diagnostics: vec![RuntimeDiagnostic { diagnostics: vec![RuntimeDiagnostic {
startup: None,
code: "fixture_runtime_rejected".to_string(), code: "fixture_runtime_rejected".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "fixture rejected Worker create".to_string(), message: "fixture rejected Worker create".to_string(),
@@ -44864,6 +45012,7 @@ mod tests {
kind: "ticket_backend_error".to_string(), kind: "ticket_backend_error".to_string(),
message: "sqlite error: FOREIGN KEY constraint failed".to_string(), message: "sqlite error: FOREIGN KEY constraint failed".to_string(),
diagnostics: vec![RuntimeDiagnostic { diagnostics: vec![RuntimeDiagnostic {
startup: None,
code: "ticket_backend_error".to_string(), code: "ticket_backend_error".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "FOREIGN KEY constraint failed".to_string(), message: "FOREIGN KEY constraint failed".to_string(),
@@ -51516,6 +51665,7 @@ mod tests {
#[test] #[test]
fn unrecoverable_pending_workspace_restore_is_typed_for_replacement() { fn unrecoverable_pending_workspace_restore_is_typed_for_replacement() {
let diagnostics = [RuntimeDiagnostic { let diagnostics = [RuntimeDiagnostic {
startup: None,
code: "embedded_worker_execution_rejected".to_string(), code: "embedded_worker_execution_rejected".to_string(),
severity: HostDiagnosticSeverity::Error, 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(), message: "Restore Errored: pending Workspace Worker restore requires operation-owned launch material; generic restore must not reconstruct it from current Workspace config".to_string(),
@@ -51527,6 +51677,7 @@ mod tests {
#[test] #[test]
fn workdir_runtime_miss_uses_exact_typed_code() { fn workdir_runtime_miss_uses_exact_typed_code() {
let typed_not_found = [RuntimeDiagnostic { let typed_not_found = [RuntimeDiagnostic {
startup: None,
code: "working_directory_not_found".to_string(), code: "working_directory_not_found".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "missing".to_string(), message: "missing".to_string(),
@@ -51538,6 +51689,7 @@ mod tests {
assert!(workdir_runtime_miss_is_not_found(&typed_not_found)); assert!(workdir_runtime_miss_is_not_found(&typed_not_found));
let unrelated = [RuntimeDiagnostic { let unrelated = [RuntimeDiagnostic {
startup: None,
code: "some_other_not_found".to_string(), code: "some_other_not_found".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "not a typed workdir miss".to_string(), message: "not a typed workdir miss".to_string(),
@@ -63241,6 +63393,7 @@ mod tests {
&api, &api,
deleted, deleted,
&[RuntimeDiagnostic { &[RuntimeDiagnostic {
startup: None,
code: "working_directory_not_found".to_string(), code: "working_directory_not_found".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "missing".to_string(), message: "missing".to_string(),
@@ -63267,6 +63420,7 @@ mod tests {
&api, &api,
unknown, unknown,
&[RuntimeDiagnostic { &[RuntimeDiagnostic {
startup: None,
code: "runtime_unavailable".to_string(), code: "runtime_unavailable".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "temporarily unavailable".to_string(), message: "temporarily unavailable".to_string(),
@@ -64963,6 +65117,7 @@ mod tests {
InternalWorkerOperationState::Rejected, InternalWorkerOperationState::Rejected,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_not_found".into(), code: "working_directory_not_found".into(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "authoritatively absent".into(), message: "authoritatively absent".into(),
@@ -65014,6 +65169,7 @@ mod tests {
InternalWorkerOperationState::Rejected, InternalWorkerOperationState::Rejected,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_cleanup_resource_busy".into(), code: "working_directory_cleanup_resource_busy".into(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "busy".into(), message: "busy".into(),
@@ -65243,6 +65399,7 @@ mod tests {
let (missing_operation, _) = reserve_removal_fixture(&api, "provider-not-found"); let (missing_operation, _) = reserve_removal_fixture(&api, "provider-not-found");
let not_found = RuntimeDiagnostic { let not_found = RuntimeDiagnostic {
startup: None,
code: "working_directory_not_found".to_string(), code: "working_directory_not_found".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "missing".to_string(), message: "missing".to_string(),
@@ -65278,6 +65435,7 @@ mod tests {
InternalWorkerOperationState::Accepted, InternalWorkerOperationState::Accepted,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_provider_timeout".to_string(), code: "working_directory_provider_timeout".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "timeout".to_string(), message: "timeout".to_string(),
@@ -65394,6 +65552,7 @@ mod tests {
InternalWorkerOperationState::Unsupported, InternalWorkerOperationState::Unsupported,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_unsupported".to_string(), code: "working_directory_unsupported".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "unsupported".to_string(), message: "unsupported".to_string(),
@@ -65454,6 +65613,7 @@ mod tests {
InternalWorkerOperationState::Rejected, InternalWorkerOperationState::Rejected,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: code.into(), code: code.into(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "secret /private/path".into(), message: "secret /private/path".into(),
@@ -65485,6 +65645,7 @@ mod tests {
InternalWorkerOperationState::Rejected, InternalWorkerOperationState::Rejected,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_cleanup_mount_present".into(), code: "working_directory_cleanup_mount_present".into(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "/private/mount credentials must not escape".into(), message: "/private/mount credentials must not escape".into(),
@@ -65635,6 +65796,7 @@ mod tests {
InternalWorkerOperationState::Rejected, InternalWorkerOperationState::Rejected,
None, None,
vec![RuntimeDiagnostic { vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_cleanup_permission_denied".into(), code: "working_directory_cleanup_permission_denied".into(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "permission denied /private/path".into(), message: "permission denied /private/path".into(),
@@ -65821,6 +65983,7 @@ mod tests {
state: InternalWorkerOperationState::Rejected, state: InternalWorkerOperationState::Rejected,
working_directory: None, working_directory: None,
diagnostics: vec![RuntimeDiagnostic { diagnostics: vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_not_found".to_string(), code: "working_directory_not_found".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "missing".to_string(), message: "missing".to_string(),
@@ -65832,6 +65995,7 @@ mod tests {
state: InternalWorkerOperationState::Rejected, state: InternalWorkerOperationState::Rejected,
working_directory: None, working_directory: None,
diagnostics: vec![RuntimeDiagnostic { diagnostics: vec![RuntimeDiagnostic {
startup: None,
code: "working_directory_provider_timeout".to_string(), code: "working_directory_provider_timeout".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "timeout".to_string(), message: "timeout".to_string(),
@@ -67524,11 +67688,13 @@ mod tests {
fn provider_rejection_classification_prefers_stable_error_diagnostic() { fn provider_rejection_classification_prefers_stable_error_diagnostic() {
let diagnostics = vec![ let diagnostics = vec![
RuntimeDiagnostic { RuntimeDiagnostic {
startup: None,
code: "runtime_capacity_warning".to_string(), code: "runtime_capacity_warning".to_string(),
severity: HostDiagnosticSeverity::Warning, severity: HostDiagnosticSeverity::Warning,
message: "capacity is low".to_string(), message: "capacity is low".to_string(),
}, },
RuntimeDiagnostic { RuntimeDiagnostic {
startup: None,
code: "runtime_capacity_unavailable".to_string(), code: "runtime_capacity_unavailable".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "capacity is exhausted".to_string(), message: "capacity is exhausted".to_string(),
@@ -67588,6 +67754,7 @@ mod tests {
}) })
.unwrap(); .unwrap();
let diagnostics = vec![RuntimeDiagnostic { let diagnostics = vec![RuntimeDiagnostic {
startup: None,
code: "runtime_capacity_unavailable".to_string(), code: "runtime_capacity_unavailable".to_string(),
severity: HostDiagnosticSeverity::Error, severity: HostDiagnosticSeverity::Error,
message: "capacity is exhausted".to_string(), message: "capacity is exhausted".to_string(),
@@ -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::<Error>()
.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));
}
}
}
@@ -449,6 +449,13 @@ pub(super) fn bind_host_session(
api: &WorkspaceApi, api: &WorkspaceApi,
attr: crate::subjektiv::SubjectSessionAttribution, attr: crate::subjektiv::SubjectSessionAttribution,
) -> Result<()> { ) -> 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 gate = lifecycle_lock(api, &attr.subject_id)?;
let _guard = gate let _guard = gate
.lock() .lock()
@@ -465,16 +472,18 @@ pub(super) fn bind_host_session(
{ {
return Err(Error::RepositoryConflict( return Err(Error::RepositoryConflict(
"Subject release intent fences Controller publication".into(), "Subject release intent fences Controller publication".into(),
)); )
.into());
} }
let store = store(api)?; let store = store(api)?;
if store if store
.journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id) .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( return Err(Error::WorkspacePermissionDenied(
"released body cannot reacquire Journal authority".into(), "released body cannot reacquire Journal authority".into(),
)); )
.into());
} }
// A released Backend body is never rebound by a delayed create/restore callback. New // 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. // 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() .into_iter()
.filter(|a| a.runtime_id == attr.runtime_id && a.worker_id == attr.worker_id) .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() { if let Some(id) = host.writer_binding_id() {
let binding = JournalWriterBinding { let binding = JournalWriterBinding {
subject_id: attr.subject_id.clone(), subject_id: attr.subject_id.clone(),
@@ -492,10 +502,14 @@ pub(super) fn bind_host_session(
worker_id: attr.worker_id.clone(), worker_id: attr.worker_id.clone(),
binding_id: id.into(), 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( return Err(Error::WorkspacePermissionDenied(
"released body Session cannot reacquire Journal authority".into(), "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. // authenticated before a recovery intent must not leave a late attribution.
let attr = store let attr = store
.record_session_attribution(attr) .record_session_attribution(attr)
.map_err(|error| Error::InvalidInput(error.to_string()))?; .map_err(session_registration::domain_error)?;
let writer = store let writer = store
.claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id) .claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
.map_err(domain)?; .map_err(session_registration::journal_error)?;
store.bind_journal_session(&writer, attr).map_err(domain)?; store
.bind_journal_session(&writer, attr)
.map_err(session_registration::journal_error)?;
Ok(()) Ok(())
} }
@@ -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<AtomicUsize>);
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for UnusedModel {
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
async fn stream(
&self,
_: agen::llm_client::Request,
) -> std::result::Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError>
{
self.0.fetch_add(1, Ordering::SeqCst);
panic!("Session registration does not require a model turn")
}
}
struct RegistrationFixture {
connected: ConnectedWorkerHarness,
subject: String,
callbacks: Arc<Mutex<Vec<(StatusCode, Value)>>>,
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::<Vec<_>>();
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");
}
@@ -9,6 +9,8 @@ use super::*;
mod chat_delegation_connected; mod chat_delegation_connected;
#[path = "chat_tests.rs"] #[path = "chat_tests.rs"]
mod chat_tests; mod chat_tests;
#[path = "session_registration_connected_tests.rs"]
mod session_registration_connected;
#[path = "subject_body_cleanup_connected_tests.rs"] #[path = "subject_body_cleanup_connected_tests.rs"]
mod subject_body_cleanup_connected; mod subject_body_cleanup_connected;
#[path = "subject_body_connected_tests.rs"] #[path = "subject_body_connected_tests.rs"]
@@ -2737,6 +2739,7 @@ pub(super) struct ConnectedWorkerHarness {
materializer: worker_runtime::working_directory::RuntimeGitMaterializer, materializer: worker_runtime::working_directory::RuntimeGitMaterializer,
authorizations: Arc<Mutex<Vec<(Value, StatusCode)>>>, authorizations: Arc<Mutex<Vec<(Value, StatusCode)>>>,
runtime_errors: Arc<Mutex<Vec<String>>>, runtime_errors: Arc<Mutex<Vec<String>>>,
runtime_startup_failures: Arc<Mutex<Vec<runtime_api::WorkerStartupFailure>>>,
_backend_server: AuthorizationCallbackServer, _backend_server: AuthorizationCallbackServer,
_runtime_server: ConnectedRuntimeServer, _runtime_server: ConnectedRuntimeServer,
_dir: tempfile::TempDir, _dir: tempfile::TempDir,
@@ -2942,6 +2945,8 @@ impl ConnectedWorkerHarness {
.unwrap(); .unwrap();
let runtime_errors = Arc::new(Mutex::new(Vec::<String>::new())); let runtime_errors = Arc::new(Mutex::new(Vec::<String>::new()));
let captured_errors = runtime_errors.clone(); 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( let runtime_server = ConnectedRuntimeServer::start(
runtime_listener, runtime_listener,
worker_runtime::http_server::runtime_http_router_with_workspace_auth( worker_runtime::http_server::runtime_http_router_with_workspace_auth(
@@ -2955,6 +2960,7 @@ impl ConnectedWorkerHarness {
.layer(axum::middleware::from_fn( .layer(axum::middleware::from_fn(
move |request: axum::extract::Request, next: axum::middleware::Next| { move |request: axum::extract::Request, next: axum::middleware::Next| {
let errors = captured_errors.clone(); let errors = captured_errors.clone();
let startup_failures = captured_startup_failures.clone();
async move { async move {
let response = next.run(request).await; let response = next.run(request).await;
if !response.status().is_client_error() if !response.status().is_client_error()
@@ -2965,6 +2971,13 @@ impl ConnectedWorkerHarness {
let (parts, body) = response.into_parts(); let (parts, body) = response.into_parts();
let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap();
let body: Value = serde_json::from_slice(&bytes).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( errors.lock().unwrap().push(
body["error"]["message"] body["error"]["message"]
.as_str() .as_str()
@@ -3045,6 +3058,7 @@ impl ConnectedWorkerHarness {
materializer, materializer,
authorizations, authorizations,
runtime_errors, runtime_errors,
runtime_startup_failures,
_backend_server: backend_server, _backend_server: backend_server,
_runtime_server: runtime_server, _runtime_server: runtime_server,
_dir: dir, _dir: dir,
+16 -1
View File
@@ -11817,6 +11817,17 @@
"error": { "error": {
"type": "string" "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": { "message": {
"type": "string" "type": "string"
} }
@@ -14799,6 +14810,10 @@
], ],
"type": "object" "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": { "SubjectBodyApiError": {
"additionalProperties": false, "additionalProperties": false,
"description": "Only action handlers with a causally bound request may provide resolution evidence.", "description": "Only action handlers with a causally bound request may provide resolution evidence.",
@@ -23348,7 +23363,7 @@
"info": { "info": {
"title": "Yoi Server API", "title": "Yoi Server API",
"version": "0.1.0", "version": "0.1.0",
"x-yoi-source-digest": "sha256:28ee9c6c6e2a4eb71829386849b508fcee758fbd7ecdbc0cdcc72cd3b86be321" "x-yoi-source-digest": "sha256:894ccff36a463bec91b635956924114aa0db87d8b3f815230d6e8395511d2f6e"
}, },
"openapi": "3.1.0", "openapi": "3.1.0",
"paths": { "paths": {
@@ -1,6 +1,6 @@
// @generated by yoi-companion-openapi-typescript@1. Do not edit. // @generated by yoi-companion-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json // 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 // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/companion-api.ts // Output: web/workspace/src/lib/generated/companion-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check // 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 DiagnosticSeverity = "info" | "warning" | "error";
export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; message: string; }; export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; 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; }; export type WorkspaceWorkerDiscoveryItem = { display_name: string; profile?: string | null; resource_key: string; status?: string | null; subject: WorkspaceWorkerSubject; };
@@ -1,6 +1,6 @@
// @generated by yoi-repository-openapi-typescript@1. Do not edit. // @generated by yoi-repository-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json // 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 // 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 // Output: web/workspace/src/lib/generated/repository-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check // 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<Diagnostic>; host_i
export type HostWorkerListResponse = { diagnostics: Array<Diagnostic>; items: Array<WorkerSummary>; limit: number; source: string; workspace_id: string; }; export type HostWorkerListResponse = { diagnostics: Array<Diagnostic>; items: Array<WorkerSummary>; limit: number; source: string; workspace_id: string; };
export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; message: string; }; export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; failure?: StartupFailureAttachment | null; message: string; };
export type RepositoryDetailResponse = { item: RepositorySummary; source: string; workspace_id: 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<RepositoryDiagnostic> | 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 RepositorySummary = { default_selector?: string | null; diagnostics?: Array<RepositoryDiagnostic> | 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 SubscriptionWorkerAvailability = "observed" | "unavailable";
export type WorkerBusyState = { kind: "run"; state: WorkerRunState; } | { kind: "maintenance"; state: WorkerMaintenanceState; }; export type WorkerBusyState = { kind: "run"; state: WorkerRunState; } | { kind: "maintenance"; state: WorkerMaintenanceState; };
@@ -1,6 +1,6 @@
// @generated by yoi-runtime-openapi-typescript@1. Do not edit. // @generated by yoi-runtime-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json // 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 // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/runtime-api.ts // Output: web/workspace/src/lib/generated/runtime-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check // 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 RemoveRuntimeRequest = { expected_binding_id: string; operation_id: string; };
export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; message: string; }; export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; failure?: StartupFailureAttachment | null; message: string; };
export type RevokeRuntimeTrustKeyRequest = { expected_binding_id: 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 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 SubscriptionWorkerAvailability = "observed" | "unavailable";
export type UpdateRemoteRuntimeRequest = { display_name?: string | null; endpoint: string; }; export type UpdateRemoteRuntimeRequest = { display_name?: string | null; endpoint: string; };
@@ -1,6 +1,6 @@
// @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json // 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 // 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 // Output: web/workspace/src/lib/generated/ticket-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check // 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 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<Diagnostic>; error: string; message: string; }; export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; failure?: StartupFailureAttachment | null; message: string; };
export type ReviewDecision = "approve" | "request_changes"; 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<TicketWorkerAttachmentBinding>; }; export type SetTicketRoleAssignmentRequest = { expected_assignment_id?: string | null; operation_id: string; orchestrator_assignment_id?: string | null; principal: TicketAssignmentPrincipal; workdir_bindings?: Array<TicketWorkerAttachmentBinding>; };
export type StartupFailureAttachment = string;
export type TicketActionEligibility = { blockers: Array<string>; can_assign_orchestrator: boolean; can_queue: boolean; can_set_parent: boolean; can_start_manual_worker: boolean; can_unassign_orchestrator: boolean; queue_tickets: Array<string>; }; export type TicketActionEligibility = { blockers: Array<string>; can_assign_orchestrator: boolean; can_queue: boolean; can_set_parent: boolean; can_start_manual_worker: boolean; can_unassign_orchestrator: boolean; queue_tickets: Array<string>; };
export type TicketAssignmentAnswer = { assignment_id: string; body: string; expected_last_event_id: string; operation_key: string; }; export type TicketAssignmentAnswer = { assignment_id: string; body: string; expected_last_event_id: string; operation_key: string; };
@@ -1,6 +1,6 @@
// @generated by yoi-workdir-openapi-typescript@1. Do not edit. // @generated by yoi-workdir-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json // 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 // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/workdir-api.ts // Output: web/workspace/src/lib/generated/workdir-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check // 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 ExternalWorkdirPermissions = { command: boolean; read: boolean; write: boolean; };
export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; message: string; }; export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; failure?: StartupFailureAttachment | null; message: string; };
export type StartupFailureAttachment = string;
export type WorkingDirectoryCleanupTarget = { kind: string; repository_key: string; working_directory_id: string; }; export type WorkingDirectoryCleanupTarget = { kind: string; repository_key: string; working_directory_id: string; };
@@ -1,6 +1,6 @@
// @generated by yoi-worker-openapi-typescript@1. Do not edit. // @generated by yoi-worker-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json // 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 // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/worker-launch-api.ts // 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 // 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 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<Diagnostic>; error: string; message: string; }; export type RepositoryApiError = { diagnostics?: Array<Diagnostic>; error: string; failure?: StartupFailureAttachment | null; message: string; };
export type RuntimeWorkerWorkdirAttachmentSummary = { alias: string; working_directory: RuntimeWorkingDirectorySummary; }; 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 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 SubjektivWorkerConnectionRequest = { subject_id: string; };
export type SubscriptionWorkerAvailability = "observed" | "unavailable"; export type SubscriptionWorkerAvailability = "observed" | "unavailable";