Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

merge: reconcile integrated Session cause fixes into Job recovery

Browse Source
This commit is contained in:
Hare
2026-11-05 01:09:59 +09:00
parent 0d1840249b ddfea7660e
commit 7181287dfc
16 changed files with 896 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -5040,6 +5040,7 @@ dependencies = [
"agen",
"async-trait",
"client",
"error-contract",
"feature-storage",
"fs4",
"futures",
@@ -7579,6 +7580,7 @@ dependencies = [
"content-history",
"content-history-adapters",
"content-history-store",
"error-contract",
"feature-permissions",
"feature-storage",
"flate2",
crates/standalone/Cargo.toml
+1
View File
@@ -28,6 +28,7 @@ uuid = { workspace = true, features = ["v7"] }
worker.workspace = true
[dev-dependencies]
error-contract.workspace = true
async-trait.workspace = true
futures.workspace = true
tempfile.workspace = true
crates/standalone/src/subjektiv.rs
+31 -5
View File
@@ -276,6 +276,8 @@ impl StandaloneSubjects {
writer,
self_weak: weak.clone(),
journal_dispatch: Mutex::new(false),
#[cfg(test)]
registration_before_attribution: Mutex::new(None),
language: self.memory_language.clone(),
catalog: self,
binding: binding.clone(),
@@ -374,6 +376,8 @@ pub(crate) struct LocalSubjectHost {
writer: subjektiv::journal::JournalWriterBinding,
self_weak: std::sync::Weak<Self>,
journal_dispatch: Mutex<bool>,
#[cfg(test)]
registration_before_attribution: Mutex<Option<Box<dyn FnOnce() + Send>>>,
catalog: StandaloneSubjects,
binding: StandaloneSubjectBinding,
worker_id: protocol::WorkerId,
@@ -468,7 +472,11 @@ impl LocalSubjectHost {
context: &SubjektivHostContext,
create: bool,
) -> Result<SubjectSessionAttribution, SubjektivHostError> {
self.body(context)?;
self.registration_authority(context)?;
#[cfg(test)]
if let Some(checkpoint) = self.registration_before_attribution.lock().unwrap().take() {
checkpoint();
}
let store = &self.catalog.store;
if let Some(record) = store
.session_attribution(&context.session_id)
@@ -485,7 +493,7 @@ impl LocalSubjectHost {
}
store
.bind_journal_session(&self.writer, record.clone())
.map_err(journal::journal_error)?;
.map_err(registration_journal_error)?;
return Ok(record);
}
if !create {
@@ -508,11 +516,25 @@ impl LocalSubjectHost {
.map_err(registration_domain_error)?;
store
.bind_journal_session(&self.writer, record.clone())
.map_err(journal::journal_error)?;
.map_err(registration_journal_error)?;
Ok(record)
}
}
fn registration_journal_error(error: subjektiv::journal::JournalError) -> SubjektivHostError {
let cause = match &error {
subjektiv::journal::JournalError::PermissionDenied(_) => {
SessionRegistrationCause::binding_mismatch()
}
subjektiv::journal::JournalError::Subject(
subjektiv::SubjektivError::SessionSubjectConflict { .. }
| subjektiv::SubjektivError::SessionAttributionConflict { .. },
) => SessionRegistrationCause::binding_mismatch(),
_ => SessionRegistrationCause::outcome_unknown(),
};
SubjektivHostError::session_registration(cause, error)
}
fn registration_domain_error(error: subjektiv::SubjektivError) -> SubjektivHostError {
let cause = match &error {
subjektiv::SubjektivError::SessionSubjectConflict { .. }
@@ -984,7 +1006,11 @@ impl SubjektivHost for LocalSubjectHost {
// An explicit operator connection issues this local body capability.
// Revalidate its live writer on every access/supply, not Feature enablement.
// Job decisions and Backend conversations are not local body capabilities.
self.body(context)?;
if operation == "session.record" {
self.registration_authority(context)?;
} else {
self.body(context)?;
}
match operation {
"session.record"
| "memory.resident.read"
@@ -1211,7 +1237,7 @@ impl SubjektivHost for LocalSubjectHost {
) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
self.registration_authority(context)?;
let attribution = self.attribution(context, create_if_missing)?;
self.reserve_journal_session(context)?;
self.reserve_journal_session(context, &attribution)?;
Ok(SubjektivRecordSessionResponse {
subject_id: attribution.subject_id,
session_id: attribution.session_id,
crates/standalone/src/subjektiv/journal.rs
+34 -17
View File
@@ -220,6 +220,12 @@ pub(super) fn journal_error(error: JournalError) -> SubjektivHostError {
other => SubjektivHostError::Unavailable(other.to_string()),
}
}
fn registration_reservation_error(
error: impl std::error::Error + Send + Sync + 'static,
) -> SubjektivHostError {
SubjektivHostError::session_registration(SessionRegistrationCause::outcome_unknown(), error)
}
fn source_error(error: session_store::StoreError) -> SubjektivHostError {
SubjektivHostError::Unavailable(error.to_string())
}
@@ -368,45 +374,56 @@ impl LocalSubjectHost {
pub(super) fn reserve_journal_session(
&self,
context: &SubjektivHostContext,
attribution: &SubjectSessionAttribution,
) -> Result<(), SubjektivHostError> {
self.body(context)?;
if self.journal_session_erased(&context.session_id)? {
self.registration_authority(context)?;
let journal_context =
JournalHostContext::validated_session(&self.catalog.store, attribution.clone())
.map_err(registration_journal_error)?;
if self
.catalog
.store
.journal_session_erased(&journal_context)
.map_err(registration_journal_error)?
{
return Ok(());
}
let session: session_store::SessionId = context
.session_id
.parse()
.map_err(|_| SubjektivHostError::InvalidArgument("invalid Session".into()))?;
let session: session_store::SessionId = context.session_id.parse().map_err(|error| {
SubjektivHostError::session_registration(
SessionRegistrationCause::invalid_response(),
error,
)
})?;
let root = self
.catalog
.state_root
.join(self.worker_id.to_string())
.join("sessions");
reject_symlink_components(&root)
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
reject_symlink_components(&root).map_err(registration_reservation_error)?;
let source_store = session_store::FsStore::new_with_retention(
&root,
crate::store::session_retention(&self.catalog.state_root),
)
.map_err(source_error)?;
.map_err(registration_reservation_error)?;
{
let _guard = source_store
.retention()
.lock_live(session)
.map_err(source_error)?;
.map_err(registration_reservation_error)?;
let path = root.join(session.to_string());
match fs::create_dir(&path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
let metadata = fs::symlink_metadata(&path)
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
let metadata =
fs::symlink_metadata(&path).map_err(registration_reservation_error)?;
if !metadata.is_dir() || metadata.file_type().is_symlink() {
return Err(SubjektivHostError::InvalidArgument(
"Session retention source is not an owned directory".into(),
));
return Err(registration_reservation_error(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"Session retention source is not an owned directory",
)));
}
}
Err(error) => return Err(SubjektivHostError::Unavailable(error.to_string())),
Err(error) => return Err(registration_reservation_error(error)),
}
}
// Establish retention at attribution, before the first Run can publish bytes. This
@@ -416,7 +433,7 @@ impl LocalSubjectHost {
session,
session_store::retention::SessionRetentionPin(session),
)
.map_err(source_error)
.map_err(registration_reservation_error)
}
pub(super) fn admit_journal(
crates/standalone/src/subjektiv/tests.rs
+139
View File
@@ -767,3 +767,142 @@ fn session_registration_uses_real_capability_and_attribution_decisions() {
SessionRegistrationCause::denied(SessionPrincipal::Subject),
);
}
#[derive(Clone)]
struct RegistrationUnusedModel;
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for RegistrationUnusedModel {
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
async fn stream(
&self,
_: agen::llm_client::Request,
) -> Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError> {
panic!("Session registration must not invoke a model")
}
}
#[tokio::test]
async fn local_writer_revocation_during_registration_retains_known_cause_and_real_journal_source() {
use session_store::{CombinedStore, FsStore, FsWorkerStore, WorkerMetadataStore};
use worker::{
WorkerBootstrap, WorkerBootstrapLayout, WorkerFilesystemAuthority, WorkerWorkspaceContext,
};
// The direct non-Browser caller uses the real bootstrap, local capability and
// transactional Journal store. Only the unused model and revocation timing are substituted.
for existing in [false, true] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("state");
let catalog = StandaloneSubjects::open(&root).unwrap();
let subject = catalog.create("registration-race", None).unwrap();
let id = protocol::WorkerId::now_v7();
let mut connection = catalog.connect(&subject.id, id).unwrap();
let port = connection.host.clone();
let name = format!("standalone-{id}");
let sessions = FsStore::new_with_retention(
root.join(id.to_string()).join("sessions"),
crate::store::session_retention(&root),
)
.unwrap();
let metadata_root = temp.path().join("metadata");
let metadata = FsWorkerStore::new(&metadata_root).unwrap();
let mut manifest = manifest::WorkerManifest::from_toml(&format!("[worker]\nname='{name}'\n[model]\nscheme='anthropic'\nmodel_id='unused'\n[engine]\n[scope]\nallow=[]")).unwrap();
manifest.feature.subjektiv.profile.enabled = true;
manifest.feature.subjektiv.profile.extraction.enabled = false;
let mut prepared = WorkerBootstrap::new(
manifest,
CombinedStore::new(sessions, metadata.clone()),
worker::PromptCatalogSource::builtins_only(),
WorkerWorkspaceContext::no_workspace(),
WorkerFilesystemAuthority::None,
WorkerBootstrapLayout::Direct {
runtime_base: temp.path().join("runtime"),
},
)
.with_model_client(RegistrationUnusedModel)
.prepare()
.await
.unwrap();
let worker = prepared.worker_mut();
worker.enable_worker_metadata_write_through().unwrap();
worker.bind_subjektiv_host(Some(port.clone()));
let session = worker.session_id();
let context = SubjektivHostContext {
worker_id: name.clone(),
session_id: session.to_string(),
};
if existing {
port.record_session(&context, true).unwrap();
}
let revoke_store = port.catalog.store.clone();
let writer = port.writer.clone();
*port.registration_before_attribution.lock().unwrap() = Some(Box::new(move || {
// This actual release follows the last preflight, before transactional bind.
revoke_store.release_journal_writer(&writer).unwrap();
}));
let error = worker
.finalize_subjektiv_session_attribution_retaining_startup_failure(
true,
worker::SubjektivSessionAttributionLifecycle::NewSession,
)
.unwrap_err();
assert_eq!(
error.session_registration_cause(),
Some(SessionRegistrationCause::binding_mismatch())
);
let mut source = Some(&error as &(dyn std::error::Error + 'static));
let mut journal_source = false;
while let Some(current) = source {
if let Some(journal) = current.downcast_ref::<subjektiv::journal::JournalError>() {
assert!(matches!(
journal,
subjektiv::journal::JournalError::PermissionDenied(_)
));
journal_source = true;
}
source = current.source();
}
assert!(
journal_source,
"original store refusal must survive the Worker wrapper"
);
assert_eq!(
error.startup_metadata_save(),
Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted)
);
drop(prepared);
drop(metadata);
let reopened = FsWorkerStore::new(metadata_root)
.unwrap()
.read_by_name(&name)
.unwrap()
.unwrap();
let session_store::worker_metadata::StartupFailureRead::Present(
error_contract::DecodeFailure::Known(cause),
) = reopened.decode_startup_failure::<SessionRegistrationCause>()
else {
panic!("typed cause must really reopen")
};
assert_eq!(
*cause.facts().cause(),
SessionRegistrationCause::binding_mismatch()
);
assert!(
port.catalog
.store
.session_attribution(&session.to_string())
.unwrap()
.is_some(),
"committed attribution is independent from denied writer binding"
);
assert!(
!port
.catalog
.store
.journal_writer_active(&port.writer)
.unwrap()
);
connection.close().unwrap();
}
}
crates/worker-runtime/src/http_server.rs
+2 -2
View File
@@ -2347,8 +2347,8 @@ impl RuntimeHttpRestError {
RuntimeError::WorkerExecutionRejected { result, .. } => startup_transfer(result),
_ => None,
};
let message = if startup.is_some() {
"Worker startup failed".to_owned()
let message = if let Some(startup) = &startup {
worker::startup_failure::internal_startup_failure_summary(&startup.primary).to_owned()
} else {
error.to_string()
};
crates/worker-runtime/src/worker_backend.rs
+47 -10
View File
@@ -271,7 +271,9 @@ enum RuntimeRestoreCleanup {
/// Diagnostics are fixed owner-selected causes, not arbitrary provider messages,
/// paths, configuration, or credentials. Cleanup cannot overwrite the first cause.
#[derive(Debug, Clone, PartialEq, Eq)]
/// The original in-process source is privately shared across clones; it is neither
/// formatted into diagnostics nor included in the wire/storage projection.
#[derive(Clone)]
pub struct RuntimeRestoreFailure {
pub stage: RuntimeRestoreStage,
pub resources: RuntimeRestoreResources,
@@ -284,6 +286,7 @@ pub struct RuntimeRestoreFailure {
secondary: Vec<protocol::startup_failure::StartupSecondaryFailure>,
legacy_diagnostic: Option<String>,
metadata_save: Option<protocol::startup_failure::StartupPersistenceOutcome>,
source: Option<Arc<dyn std::error::Error + Send + Sync>>,
}
impl RuntimeRestoreFailure {
@@ -315,7 +318,13 @@ impl RuntimeRestoreFailure {
self.io_kind
}
fn from_worker(stage: RuntimeRestoreStage, error: &WorkerError) -> Self {
fn from_worker(stage: RuntimeRestoreStage, error: WorkerError) -> Self {
let mut failure = Self::from_worker_ref(stage, &error);
failure.source = Some(Arc::new(error));
failure
}
fn from_worker_ref(stage: RuntimeRestoreStage, error: &WorkerError) -> Self {
let mut failure = Self::started(stage, restore_worker_cause(error));
failure.primary = error
.startup_failure_at(stage.into())
@@ -368,6 +377,7 @@ impl RuntimeRestoreFailure {
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
source: None,
}
}
@@ -384,6 +394,7 @@ impl RuntimeRestoreFailure {
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
source: None,
}
}
@@ -400,6 +411,7 @@ impl RuntimeRestoreFailure {
secondary: Vec::new(),
legacy_diagnostic: None,
metadata_save: None,
source: None,
}
}
@@ -411,6 +423,25 @@ impl RuntimeRestoreFailure {
}
}
impl std::fmt::Debug for RuntimeRestoreFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RuntimeRestoreFailure")
.field("stage", &self.stage)
.field("resources", &self.resources)
.field("cause", &self.cause)
.field("cleanup", &self.cleanup)
.field("io_kind", &self.io_kind)
.field("cleanup_failure", &self.cleanup_failure)
.finish_non_exhaustive()
}
}
impl std::error::Error for RuntimeRestoreFailure {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
self.source.as_deref().map(|source| source as _)
}
}
impl std::fmt::Display for RuntimeRestoreFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let cause: String = self
@@ -2352,7 +2383,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
if let WorkerBootstrapError::Worker(source) = &error {
self.retain_spawn_construction_failure(&request.worker_ref, source);
}
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, &error)
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, error)
})?;
let worker = prepared.worker_mut();
{
@@ -2449,7 +2480,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
if let WorkerBootstrapError::Worker(source) = &error {
self.retain_spawn_construction_failure(&request.worker_ref, source);
}
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, &error)
RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, error)
})?;
self.failed_spawn_resources
.lock()
@@ -2696,7 +2727,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
.await
.map_err(|error| {
self.retain_construction_failure(&request, &error);
let failure = RuntimeRestoreFailure::from_worker(stage, &error);
let failure = RuntimeRestoreFailure::from_worker(stage, error);
let cause = failure.cause;
preparation_failure = Some(failure);
cause
@@ -2972,7 +3003,7 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate {
Some(source.kind()),
),
};
let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, &error);
let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, error);
failure.cause = cause;
failure.durable_effects_possible = true;
failure.io_kind = io_kind;
@@ -3244,11 +3275,17 @@ impl ProfileRuntimeWorkerFactory {
WorkerBootstrapError::Worker(error) => {
stage = RuntimeRestoreStage::Reconstruction;
self.retain_construction_failure(request, &error);
primary_failure = Some(RuntimeRestoreFailure::from_worker(stage, &error));
restore_worker_cause(&error)
let cause = restore_worker_cause(&error);
primary_failure = Some(RuntimeRestoreFailure::from_worker(stage, error));
cause
}
WorkerBootstrapError::Controller { source, .. } => {
io_kind = Some(source.kind());
error @ WorkerBootstrapError::Controller { .. } => {
let failure = RuntimeRestoreFailure::from_bootstrap(
RuntimeRestoreStage::Controller,
error,
);
io_kind = failure.io_kind;
primary_failure = Some(failure);
"restored Worker preparation failed"
}
})?;
crates/worker-runtime/src/worker_backend/restore_recovery.rs
+28 -9
View File
@@ -116,7 +116,7 @@ impl ProfileRuntimeWorkerFactory {
.await;
let mut worker = worker_result.map_err(|error| {
self.retain_construction_failure(request, &error);
cause(restore_worker_cause(&error))
RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Recovery, error)
})?;
let tasks = worker.retain_startup_feature_cleanup();
let sessions = worker.workdir_sessions();
@@ -141,11 +141,13 @@ impl ProfileRuntimeWorkerFactory {
.map_err(|_| cause("recovery Backend Job binding failed"))?;
}
worker
.finalize_subjektiv_session_attribution(
.finalize_subjektiv_session_attribution_retaining_startup_failure(
request.request.subjektiv_attached,
SubjektivSessionAttributionLifecycle::RestoredSession,
)
.map_err(|error| cause(restore_worker_cause(&error)))?;
.map_err(|error| {
RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Recovery, error)
})?;
if pending {
worker::WorkerController::recover_pending_head(&mut worker, &root)
.await
@@ -161,21 +163,38 @@ impl ProfileRuntimeWorkerFactory {
tasks
.shutdown()
.await
.map_err(|_| cause("recovery Feature settlement failed"))?;
.map_err(|_| RuntimeRestoreCleanupFailure::Feature)?;
sessions
.close_all()
.await
.map_err(|_| cause("recovery Workdir close failed"))?;
.map_err(|_| RuntimeRestoreCleanupFailure::Workdir)?;
if let Some(allocation) = allocation {
allocation
.release()
.map_err(|_| cause("recovery allocation release failed"))?;
.map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?;
}
Ok::<_, RuntimeRestoreFailure>(())
Ok::<_, RuntimeRestoreCleanupFailure>(())
}
.await;
result?;
cleanup?;
match (result, cleanup) {
(Err(mut primary), cleanup) => {
primary.cleanup = match cleanup {
Ok(()) => RuntimeRestoreCleanup::Confirmed,
Err(failure) => {
primary.cleanup_failure = Some(failure);
RuntimeRestoreCleanup::Failed(failure)
}
};
return Err(primary);
}
(Ok(()), Err(cleanup)) => {
let mut failure = cause("restore recovery cleanup failed");
failure.cleanup_failure = Some(cleanup);
failure.cleanup = RuntimeRestoreCleanup::Failed(cleanup);
return Err(failure);
}
(Ok(()), Ok(())) => {}
}
self.failed_restore_resources
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
crates/worker-runtime/src/worker_backend/startup_binding_tests.rs
+130 -8
View File
@@ -1,8 +1,12 @@
mod startup_binding_tests {
//! Exercise the factory's production binding step with a constructed Worker,
//! real aggregate/Session stores and reopen. Only the Workspace response is
//! synthetic; this suite does not test domain authorization or network transport.
//! synthetic in the binding cases; the direct internal Profile factory case uses
//! a refused real loopback transport. Domain authorization is tested by Server.
use super::*;
use protocol::session_failure::{
SessionPrincipal, SessionRegistrationCause, session_registration_attachment,
};
use protocol::startup_failure::{RuntimeStartupCause, StartupPersistenceOutcome};
use session_store::worker_metadata::StartupFailureRead;
use worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
@@ -67,13 +71,27 @@ mod startup_binding_tests {
_ => self.store.read_by_name(&self.name).unwrap().unwrap(),
};
if matches!(self.response, Response::Lost) {
return Err(WorkspaceClientError::Unavailable(
return Err(WorkspaceClientError::Request(
"response deliberately lost".into(),
));
}
Ok(WorkspaceResponse {
status: 403,
body: "synthetic rejection with private token /internal/path".into(),
body: serde_json::to_string(
&server_api::RepositoryApiError::new(
403,
"PermissionDenied",
"synthetic rejection with private token /internal/path",
vec![],
)
.with_failure(
session_registration_attachment(SessionRegistrationCause::denied(
SessionPrincipal::Subject,
))
.unwrap(),
),
)
.unwrap(),
})
}
}
@@ -137,9 +155,41 @@ mod startup_binding_tests {
let failure = startup_failure::finalize_startup_binding(&worker, true)
.err()
.expect("failure must remain a failure");
if matches!(response, Response::Lost | Response::Rejected) {
let mut source = std::error::Error::source(&failure);
let mut worker = false;
let mut host = false;
let mut client = false;
while let Some(current) = source {
worker |= current.is::<WorkerError>();
host |= current.is::<worker::subjektiv::SubjektivHostError>();
client |= current.is::<WorkspaceClientError>();
source = current.source();
}
assert!(worker && host, "common factory binding step retains in-process domain sources");
assert_eq!(client, matches!(response, Response::Lost));
assert!(!format!("{failure:?} {failure}").contains("response deliberately lost"));
}
let result = failure.execution_result(WorkerExecutionOperation::Spawn);
let primary = result.startup_failure::<RuntimeStartupCause>().unwrap();
assert!(matches!(primary, error_contract::DecodeFailure::Known(_)));
if matches!(response, Response::BeforeSession | Response::BeforeOwner) {
let primary = result.startup_failure::<RuntimeStartupCause>().unwrap();
assert!(matches!(primary, error_contract::DecodeFailure::Known(_)));
} else {
let error_contract::DecodeFailure::Known(primary) = result
.startup_failure::<SessionRegistrationCause>()
.unwrap()
else {
panic!("a Session decision must not be relabeled as a generic Runtime failure");
};
assert_eq!(
*primary.facts().cause(),
if matches!(response, Response::Lost) {
SessionRegistrationCause::outcome_unknown()
} else {
SessionRegistrationCause::denied(SessionPrincipal::Subject)
}
);
}
drop(worker);
drop(store);
let reopened = WorkerAggregateStore::new(root.path(), name).unwrap();
@@ -151,13 +201,17 @@ mod startup_binding_tests {
Some(StartupPersistenceOutcome::Persisted)
);
let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) =
metadata.decode_startup_failure::<RuntimeStartupCause>()
metadata.decode_startup_failure::<SessionRegistrationCause>()
else {
panic!("same-Session failure must actually save a typed cause")
};
assert_eq!(
cause.facts().cause().reason,
protocol::startup_failure::RuntimeStartupReason::AttributionFailed
*cause.facts().cause(),
if matches!(response, Response::Lost) {
SessionRegistrationCause::outcome_unknown()
} else {
SessionRegistrationCause::denied(SessionPrincipal::Subject)
}
);
assert_eq!(metadata.active.unwrap().session_id, origin);
let saved = serde_json::to_string(&metadata.startup_failure).unwrap();
@@ -222,6 +276,74 @@ mod startup_binding_tests {
}
}
}
#[cfg(feature = "http-server")]
#[tokio::test]
#[serial_test::serial(worker_allocation)]
async fn internal_profile_factory_transport_failure_retains_source_chain_and_reopened_cause() {
require_restore_test_environment();
let root = tempfile::tempdir().unwrap();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let endpoint = format!("http://{}", listener.local_addr().unwrap());
drop(listener);
let worker_ref = WorkerRef::new(WorkerId::now_v7());
let mut request = create_request("internal factory source");
request.worker_id = worker_ref.worker_id;
request.workspace_api = Some(WorkspaceApiRef { workspace_id: "adapter-test".into(), base_url: endpoint.clone() });
request.subjektiv_attached = true;
request.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { workspace_id: "adapter-test".into(), language: "English".into() });
let archive = crate::profile_archive::ProfileSourceArchive::build_evaluated_profile(
"profile-source-archive:internal-source".into(), "builtin:companion".into(),
serde_json::json!({"slug":"internal-source", "description":"internal source test", "scope":"workspace_read",
"model":{"scheme":"anthropic", "model_id":"unused", "auth":{"kind":"none"}},
"engine":{"max_tokens":100}, "feature":{"subjektiv":{"enabled":true,"extraction":{"enabled":false}}}})
).unwrap();
let mut bundle = test_bundle();
bundle.profile_source_archive = Some(archive.clone());
let bundle = bundle.with_computed_digest();
request.profile_source = crate::catalog::ProfileSourceArchiveSource::Embedded { archive };
request.config_bundle = Some(ConfigBundleRef { id: bundle.metadata.id.clone(), digest: bundle.metadata.digest.clone() });
let request = WorkerExecutionSpawnRequest {
worker_ref: worker_ref.clone(), request,
workspace_scope: None,
context: test_execution_context(worker_ref.clone()),
workdir_attachments: BTreeMap::new(), config_bundle: Some(bundle),
};
// Direct internal caller, not a Browser handler or Runtime result decoder.
// The real transport is refused at an unused loopback socket; only the model is replaced.
let factory = ProfileRuntimeWorkerFactory::new(root.path())
.with_runtime_id("internal-source-runtime")
.with_runtime_store_dir(root.path().join("runtime"))
.with_spawn_model_client_factory(|| Box::new(MockClient::new(vec![])));
let failure = match factory.spawn_controller(request).await {
Err(failure) => failure,
Ok(_) => panic!("disconnected Backend must not create a Controller"),
};
assert_eq!(failure.stage, RuntimeRestoreStage::Binding);
let mut source = std::error::Error::source(&failure);
let mut worker = false; let mut host = false; let mut transport = false;
while let Some(current) = source {
worker |= current.is::<WorkerError>();
host |= current.is::<worker::subjektiv::SubjektivHostError>();
if let Some(error) = current.downcast_ref::<WorkspaceClientError>() {
assert!(matches!(error, WorkspaceClientError::Request(_)));
transport = true;
}
source = current.source();
}
assert!(worker && host && transport, "factory must retain original typed in-process source");
let result = failure.execution_result(WorkerExecutionOperation::Spawn);
let Some(error_contract::DecodeFailure::Known(cause)) = result.startup_failure::<SessionRegistrationCause>() else { panic!("Session cause must remain typed") };
assert_eq!(*cause.facts().cause(), SessionRegistrationCause::outcome_unknown());
assert_eq!(result.startup_failure_persistence, Some(StartupPersistenceOutcome::Persisted));
let aggregate = root.path().join("runtime/workers").join(worker_ref.worker_id.to_string());
let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref);
let metadata = WorkerAggregateStore::new(aggregate, &name).unwrap().read_by_name(&name).unwrap().unwrap();
let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) = metadata.decode_startup_failure::<SessionRegistrationCause>() else { panic!("saved factory cause must reopen") };
assert_eq!(*cause.facts().cause(), SessionRegistrationCause::outcome_unknown());
assert!(!format!("{failure:?} {failure}").contains(&endpoint));
assert!(!serde_json::to_string(&result).unwrap().contains(&endpoint));
}
#[tokio::test]
async fn factory_rejection_saves_same_session_cause_after_own_attribution_transition() {
check_binding(Response::Rejected).await;
crates/worker-runtime/src/worker_backend/startup_failure.rs
+8 -6
View File
@@ -32,9 +32,9 @@ impl From<RuntimeRestoreCleanupFailure> for protocol::startup_failure::StartupCl
}
impl RuntimeRestoreFailure {
pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: &WorkerBootstrapError) -> Self {
match error {
WorkerBootstrapError::Worker(source) => Self::from_worker(stage, source),
pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: WorkerBootstrapError) -> Self {
let mut failure = match &error {
WorkerBootstrapError::Worker(source) => Self::from_worker_ref(stage, source),
WorkerBootstrapError::Controller { source, .. } => {
let mut failure = Self::started(
RuntimeRestoreStage::Controller,
@@ -43,7 +43,9 @@ impl RuntimeRestoreFailure {
failure.io_kind = Some(source.kind());
failure
}
}
};
failure.source = Some(Arc::new(error));
failure
}
}
@@ -51,7 +53,7 @@ impl From<session_store::WorkerStoreError> for RuntimeRestoreFailure {
fn from(error: session_store::WorkerStoreError) -> Self {
Self::from_worker(
RuntimeRestoreStage::Metadata,
&WorkerError::WorkerStore(error),
WorkerError::WorkerStore(error),
)
}
}
@@ -114,7 +116,7 @@ where
attached,
SubjektivSessionAttributionLifecycle::NewSession,
)
.map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error))
.map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, error))
}
#[cfg(test)]
crates/worker/src/startup_failure.rs
+44
View File
@@ -120,6 +120,20 @@ pub fn reproject_internal_startup_failure(
StartupFailureAttachment::from_projected(&projected)
}
/// Fixed, identifier-free summary for an already authenticated internal caller.
/// Keep the required operation actionable without exposing the source's Display.
/// This neither projects a Browser cause nor chooses an HTTP status.
pub fn internal_startup_failure_summary(failure: &StartupFailureAttachment) -> &'static str {
if failure
.project_for::<InternalTransfer, protocol::session_failure::SessionRegistrationCause>(&())
.is_ok()
{
"Worker startup failed while registering Session (session.record)"
} else {
"Worker startup failed"
}
}
/// 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,
@@ -249,6 +263,36 @@ mod tests {
WorkerStoreError,
};
#[test]
fn internal_summary_retains_only_attested_required_operation() {
use protocol::session_failure::{SessionPrincipal, SessionRegistrationCause};
let session = protocol::session_failure::session_registration_attachment(
SessionRegistrationCause::denied(SessionPrincipal::Subject),
)
.unwrap();
assert_eq!(
internal_startup_failure_summary(&session),
"Worker startup failed while registering Session (session.record)"
);
let runtime = runtime_startup_attachment(RuntimeStartupCause {
stage: StartupStage::Feature,
reason: RuntimeStartupReason::FeatureFailed,
})
.unwrap();
assert_eq!(
internal_startup_failure_summary(&runtime),
"Worker startup failed"
);
let corrupt = serde_json::from_value(serde_json::json!(
"session.record denied TEST_ONLY_CREDENTIAL /private/path raw_http_body"
))
.unwrap();
assert_eq!(
internal_startup_failure_summary(&corrupt),
"Worker startup failed"
);
}
struct LostSaveResponse(FsWorkerStore);
impl WorkerMetadataStore for LostSaveResponse {
fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> {
crates/workspace-server/Cargo.toml
+1
View File
@@ -70,6 +70,7 @@ webauthn-rs = { workspace = true, features = ["conditional-ui"] }
webauthn-rs-proto = "0.5.5"
[dev-dependencies]
error-contract.workspace = true
rusqlite = { workspace = true, features = ["hooks"] }
feature-storage = { workspace = true, features = ["test-support"] }
native-tls = "0.2"
crates/workspace-server/src/hosts.rs
+121 -5
View File
@@ -6950,8 +6950,11 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti
),
};
if let EmbeddedRuntimeError::WorkerExecutionRejected { result, .. } = error {
if result.startup_failure.is_some() {
projection.message = "Embedded Runtime Worker startup failed".into();
if let Some(primary) = &result.startup_failure {
projection.message = format!(
"Embedded Runtime: {}",
worker::startup_failure::internal_startup_failure_summary(primary)
);
}
projection.startup = result.startup_failure.as_ref().and_then(|primary| {
internal_startup_transfer(runtime_api::WorkerStartupFailure {
@@ -7042,7 +7045,10 @@ fn runtime_api_diagnostic(
) -> RuntimeDiagnostic {
match error {
runtime_api::client_support::ClientError::Public { status, error } => {
let startup = error.error.startup.and_then(internal_startup_transfer);
if let Some(startup) = error.error.startup {
return remote_startup_diagnostic(runtime_id, status, startup);
}
let startup = None;
let remote_code = error.error.code;
let message = sanitize_remote_runtime_message(&remote_code, &error.error.message);
let (code, severity) = match status {
@@ -7121,6 +7127,9 @@ fn remote_http_status_diagnostic(
.ok()
.filter(|_| body.len() <= MAX_RUNTIME_PING_RESPONSE_BYTES)
.and_then(|_| serde_json::from_slice::<RuntimeHttpErrorResponse>(&body).ok());
if let Some(startup) = error.as_ref().and_then(|error| error.error.startup.clone()) {
return remote_startup_diagnostic(runtime_id, status, startup);
}
let remote_code = error
.as_ref()
.map(|error| error.error.code.as_str())
@@ -7155,6 +7164,29 @@ fn remote_http_status_diagnostic(
projection
}
fn remote_startup_diagnostic(
runtime_id: &str,
status: StatusCode,
startup: runtime_api::WorkerStartupFailure,
) -> RuntimeDiagnostic {
let startup = internal_startup_transfer(startup);
let message = startup.as_ref().map_or("Worker startup failed", |startup| {
worker::startup_failure::internal_startup_failure_summary(&startup.primary)
});
let code = if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
"remote_runtime_auth_failed"
} else {
"remote_runtime_startup_failed"
};
let mut projection = diagnostic(
code,
HostDiagnosticSeverity::Error,
format!("Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}"),
);
projection.startup = startup;
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(
@@ -7394,7 +7426,7 @@ mod tests {
let mut wire = runtime_api::RuntimeApiError::new(
500,
"worker_execution_rejected",
"unused remote prose",
"TEST_ONLY_CREDENTIAL private-subject-id raw_http_body",
);
wire.error.startup = embedded.startup.clone();
let received = serde_json::from_str(&serde_json::to_string(&wire).unwrap()).unwrap();
@@ -7412,8 +7444,14 @@ mod tests {
public.get("startup").is_none(),
"no blind public/storage attachment forwarding"
);
for secret in ["TEST_ONLY_CREDENTIAL", "/private/path", "raw_http_body"] {
for secret in [
"TEST_ONLY_CREDENTIAL",
"private-subject-id",
"/private/path",
"raw_http_body",
] {
assert!(!serde_json::to_string(&embedded).unwrap().contains(secret));
assert!(!serde_json::to_string(&remote).unwrap().contains(secret));
}
let legacy = runtime_api_diagnostic(
@@ -7432,6 +7470,84 @@ mod tests {
assert!(internal_startup_transfer(corrupt).is_none());
}
#[test]
fn remote_startup_receivers_discard_untrusted_prose_and_code_even_without_paths() {
use protocol::session_failure::{
SessionRegistrationCause, session_registration_attachment,
};
let valid = runtime_api::WorkerStartupFailure {
primary: session_registration_attachment(SessionRegistrationCause::binding_mismatch())
.unwrap(),
secondary: vec![],
persistence: None,
resources: None,
io_kind: None,
};
for corrupt in [false, true] {
for status in [StatusCode::FORBIDDEN, StatusCode::INTERNAL_SERVER_ERROR] {
let mut startup = valid.clone();
if corrupt {
startup.primary = serde_json::from_value(json!("corrupt attachment")).unwrap();
}
let mut wire = runtime_api::RuntimeApiError::new(
status.as_u16(),
"PRIVATE_CODE_SENTINEL",
"TEST_ONLY_CREDENTIAL private-subject-id raw_http_body",
);
wire.error.startup = Some(startup);
let body = serde_json::to_string(&wire).unwrap();
let typed = runtime_api_diagnostic(
"remote",
runtime_api::client_support::ClientError::Public {
status,
error: serde_json::from_str(&body).unwrap(),
},
);
let (url, server) = serve_mock_http(vec![mock_response(
"GET",
"/",
false,
status.as_u16(),
body,
)]);
let response = reqwest::blocking::Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()
.unwrap()
.get(url)
.send()
.unwrap();
let http = remote_http_status_diagnostic("remote", status, response);
server.join().unwrap();
for diagnostic in [typed, http] {
assert_eq!(diagnostic.startup.is_some(), !corrupt);
assert_eq!(diagnostic.message.contains("session.record"), !corrupt);
assert_eq!(
diagnostic.code,
if status == StatusCode::FORBIDDEN {
"remote_runtime_auth_failed"
} else {
"remote_runtime_startup_failed"
}
);
let public = serde_json::to_string(&diagnostic).unwrap();
for secret in [
"PRIVATE_CODE_SENTINEL",
"TEST_ONLY_CREDENTIAL",
"private-subject-id",
"raw_http_body",
"corrupt attachment",
] {
assert!(
!public.contains(secret),
"received prose/code must never bypass startup projection: {public}"
);
}
}
}
}
}
#[tokio::test]
async fn async_workdir_authorization_uses_store_capacity_without_bearer_fallback() {
use crate::db_executor::{DbExecutor, DbExecutorLimits};
crates/workspace-server/src/server/subjektiv_journal.rs
+8 -4
View File
@@ -423,10 +423,14 @@ pub(super) fn bind_host_session_for_registration(
.config_store
.subject_release_pending(api.workspace_id(), &worker)?
{
return Err(Error::RepositoryConflict(
"Subject release intent fences Controller publication".into(),
)
.into());
// The current body is explicitly fenced from publication, not an
// unobserved callback outcome. Keep the conflict separate from its cause.
return Err(session_registration::Judgment::SessionRecord.error(
Error::RepositoryConflict(
"Subject release intent fences Controller publication".into(),
),
protocol::session_failure::SessionRegistrationCause::binding_mismatch(),
));
}
let store = store(api)?;
if store
crates/workspace-server/src/server/tests/session_registration_connected_tests.rs
+285 -20
View File
@@ -3,6 +3,7 @@
//! Worker metadata. Only the unused external model is replaced. No restore,
//! operation replay, singleton-release, or Browser rendering contract is asserted.
use super::*;
use api_macros::HttpError as _;
use protocol::session_failure::{
SessionPrincipal, SessionRegistrationCause, session_registration_attachment,
};
@@ -29,15 +30,29 @@ impl agen::llm_client::LlmClient for UnusedModel {
}
}
type ReleaseFence = (
crate::store::WorkerSingletonLease,
subjektiv::journal::JournalWriterBinding,
);
struct RegistrationFixture {
connected: ConnectedWorkerHarness,
subject: String,
callbacks: Arc<Mutex<Vec<(StatusCode, Value)>>>,
model: UnusedModel,
release_fence: Arc<Mutex<Option<ReleaseFence>>>,
}
#[derive(Clone, Copy)]
enum RegistrationFault {
Allow,
Deny,
DenyWithMetadataReadFailure,
ReleasePending,
}
impl RegistrationFixture {
async fn new(revoke_on_registration: bool) -> Self {
async fn new(fault: RegistrationFault) -> 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.
@@ -46,11 +61,14 @@ impl RegistrationFixture {
let callbacks = Arc::new(Mutex::new(Vec::new()));
let observed = callbacks.clone();
let model = UnusedModel::default();
let release_fence = Arc::new(Mutex::new(None));
let reserved = release_fence.clone();
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();
let reserved = reserved.clone();
async move {
if request.method() != axum::http::Method::POST
|| request.uri().path()
@@ -63,6 +81,9 @@ impl RegistrationFixture {
));
let armed = hook.lock().unwrap().take();
let pin_root = armed.as_ref().map(|(_, _, root)| root.clone());
let metadata_fault_root = pin_root.clone().filter(|_| {
matches!(fault, RegistrationFault::DenyWithMetadataReadFailure)
});
if let Some((api, subject, _)) = armed {
let store = open_subjektiv_store(&api).unwrap();
let policy = store.subject_permissions(&subject).unwrap().unwrap();
@@ -72,22 +93,39 @@ impl RegistrationFixture {
.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();
if matches!(fault, RegistrationFault::ReleasePending) {
let key = crate::subjektiv::subject_worker_singleton_key(&subject).unwrap();
let lease = api.store
.current_worker_singleton_owner(TEST_WORKSPACE_ID, &key)
.unwrap().unwrap();
let writer = store.claim_journal_writer(
&subject, &lease.worker.runtime_id, &lease.worker.worker_id,
).unwrap();
api.config_store.reserve_subject_release(
TEST_WORKSPACE_ID, &lease, &writer.binding_id,
"registration-release-fence",
).unwrap();
assert!(store.journal_writer_active(&writer).unwrap());
assert_eq!(api.store.current_worker_singleton_owner(TEST_WORKSPACE_ID, &key).unwrap(), Some(lease.clone()));
*reserved.lock().unwrap() = Some((lease, writer));
} else {
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();
@@ -111,6 +149,19 @@ impl RegistrationFixture {
let response = next
.run(Request::from_parts(parts, Body::from(bytes)))
.await;
if let Some(root) = metadata_fault_root {
assert_eq!(response.status(), StatusCode::FORBIDDEN);
// The real denial has already been decided. Replace only this
// fixture's metadata file with a directory before Worker save:
// a real read error, not a fake authorization/store response.
let workers = std::fs::read_dir(root.join("workers"))
.unwrap().map(|entry| entry.unwrap().path()).collect::<Vec<_>>();
assert_eq!(workers.len(), 1);
let metadata = workers[0].join("metadata.json");
assert!(metadata.is_file());
std::fs::remove_file(&metadata).unwrap();
std::fs::create_dir(&metadata).unwrap();
}
observed.lock().unwrap().push((response.status(), input));
response
}
@@ -136,7 +187,7 @@ impl RegistrationFixture {
.iter()
.any(|p| p.operation == "session.record")
);
if revoke_on_registration {
if !matches!(fault, RegistrationFault::Allow) {
*revoke.lock().unwrap() = Some((
connected.api.clone(),
subject.clone(),
@@ -148,6 +199,7 @@ impl RegistrationFixture {
subject,
callbacks,
model,
release_fence,
}
}
@@ -296,7 +348,7 @@ fn assert_registration_cause(body: Value, cause: SessionRegistrationCause) {
#[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;
let fixture = RegistrationFixture::new(RegistrationFault::Deny).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(
@@ -391,9 +443,222 @@ async fn real_session_record_denial_survives_runtime_host_and_metadata_reopen()
assert!(contains_saved_value(&reopened, &expected));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
async fn real_session_record_denial_and_metadata_save_failure_remain_separate() {
use protocol::startup_failure::{
StartupFailurePhase, StartupPersistenceOutcome, StartupSecondaryFailure,
};
let fixture = RegistrationFixture::new(RegistrationFault::DenyWithMetadataReadFailure).await;
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 failures = fixture.connected.runtime_startup_failures.lock().unwrap();
assert_eq!(
failures.len(),
1,
"one real startup attempt, not an operation replay"
);
let wire = &failures[0];
assert_eq!(
wire.primary,
session_registration_attachment(SessionRegistrationCause::denied(
SessionPrincipal::Subject
),)
.unwrap()
);
assert_eq!(wire.persistence, Some(StartupPersistenceOutcome::NotSaved));
assert!(wire.secondary.iter().any(|failure| matches!(
failure,
StartupSecondaryFailure::Persistence {
phase: StartupFailurePhase::MetadataSave,
outcome: StartupPersistenceOutcome::NotSaved
}
)));
let host = error
.diagnostics
.iter()
.filter_map(|d| d.startup.as_ref())
.find(|startup| startup.primary == wire.primary)
.unwrap();
assert_eq!(
host, wire,
"Remote Host preserves the refusal and independent save failure"
);
assert!(
open_subjektiv_store(&fixture.connected.api)
.unwrap()
.subject_session_attributions(&fixture.subject)
.unwrap()
.is_empty()
);
let workers = std::fs::read_dir(fixture.connected.runtime_root.join("workers"))
.unwrap()
.map(|entry| entry.unwrap().path())
.collect::<Vec<_>>();
assert_eq!(workers.len(), 1);
let name = format!(
"worker-runtime-{}",
workers[0].file_name().unwrap().to_str().unwrap()
);
let reopened = session_store::WorkerAggregateStore::open_read_only(&workers[0], &name).unwrap();
assert!(
reopened.read_by_name(&name).is_err(),
"the real read failure remains, not a false saved cause"
);
}
#[tokio::test]
async fn pending_release_binding_refusal_retains_original_conflict_source() {
let f = Fixture::new().await;
let store = open_subjektiv_store(&f.api).unwrap();
let lease = f
.api
.store
.require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &f.source)
.unwrap()
.unwrap();
let writer = store
.claim_journal_writer(&f.subject, &f.source.runtime_id, &f.source.worker_id)
.unwrap();
f.api
.config_store
.reserve_subject_release(
TEST_WORKSPACE_ID,
&lease,
&writer.binding_id,
"registration-source-release",
)
.unwrap();
let session = session_store::new_session_id().to_string();
let attr = crate::subjektiv::SubjectSessionAttribution::new(
&f.subject,
&f.source.runtime_id,
&f.source.worker_id,
&session,
)
.unwrap();
let error = session_registration::bind_host_session(&f.api, attr).unwrap_err();
assert_eq!(
error.session_registration_cause,
Some(SessionRegistrationCause::binding_mismatch())
);
assert!(matches!(
std::error::Error::source(&error)
.unwrap()
.downcast_ref::<Error>(),
Some(Error::RepositoryConflict(_))
));
assert_eq!(
error.into_repository_api_error().status_code(),
StatusCode::CONFLICT.as_u16()
);
assert!(store.session_attribution(&session).unwrap().is_none());
// A conflict without this explicit decision must not acquire a denial.
let unrelated = session_registration::Judgment::SessionRecord
.binding_rejection(Error::RepositoryConflict("an unrelated conflict".into()).into());
assert!(unrelated.session_registration_cause.is_none());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
async fn release_fence_refusal_survives_signed_registration_and_metadata_reopen() {
let fixture = RegistrationFixture::new(RegistrationFault::ReleasePending).await;
let api = &fixture.connected.api;
let store = open_subjektiv_store(api).unwrap();
let policy = store
.subject_permissions(&fixture.subject)
.unwrap()
.unwrap();
let error = scoped_create_workspace_worker(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: TEST_WORKSPACE_ID.into(),
}),
HeaderMap::new(),
Json(fixture.request()),
)
.await
.unwrap_err();
let session = fixture.assert_callback(StatusCode::CONFLICT);
let (lease, writer) = fixture.release_fence.lock().unwrap().clone().unwrap();
assert_eq!(
store.subject_permissions(&fixture.subject).unwrap(),
Some(policy)
);
assert!(store.session_attribution(&session).unwrap().is_none());
assert!(store.journal_writer_active(&writer).unwrap());
api.config_store
.require_subject_release_intent(
TEST_WORKSPACE_ID,
&lease,
&writer.binding_id,
"registration-release-fence",
)
.unwrap();
assert!(
api.config_store
.subject_release_pending(TEST_WORKSPACE_ID, &lease.worker)
.unwrap()
);
assert!(
!api.config_store
.subject_release_proven(
TEST_WORKSPACE_ID,
&lease,
&writer.binding_id,
"registration-release-fence"
)
.unwrap()
);
let failures = fixture.connected.runtime_startup_failures.lock().unwrap();
assert_eq!(failures.len(), 1, "one real factory startup, not replay");
let wire = &failures[0];
let expected =
session_registration_attachment(SessionRegistrationCause::binding_mismatch()).unwrap();
assert_eq!(wire.primary, expected);
assert_eq!(
wire.persistence,
Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted)
);
let host = error
.diagnostics
.iter()
.filter_map(|d| d.startup.as_ref())
.find(|failure| failure.primary == expected)
.expect("Remote Host retains known refusal");
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,
..
}
)),
"retention cleanup failure is separate from refusal"
);
let saved = read_metadata(&fixture.connected.runtime_root, &lease.worker.worker_id);
assert!(contains_saved_value(
&saved,
&serde_json::to_value(&expected).unwrap()
));
assert_eq!(
read_metadata(&fixture.connected.runtime_root, &lease.worker.worker_id),
saved
);
}
#[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 fixture = RegistrationFixture::new(RegistrationFault::Allow).await;
let initial = open_subjektiv_store(&fixture.connected.api)
.unwrap()
.subject_permissions(&fixture.subject)
crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs
+15 -3
View File
@@ -2366,9 +2366,21 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) {
assert_eq!(probe.activations.load(SeqCst), 0);
assert_eq!(probe.publications.load(SeqCst), 0);
assert_eq!(fixture.model.calls.load(SeqCst), model_before);
assert!(
matches!(metadata.read_by_name(&name).unwrap().unwrap().subjektiv_session_attribution,
Some(session_store::SubjektivSessionAttributionState::Failed { session_id: id, .. }) if id == session_id)
let saved_failure = metadata.read_by_name(&name).unwrap().unwrap();
// Trust rejection occurs before the Session domain handler. An unsigned
// legacy 401 attests neither a Session refusal nor absence of earlier effects.
assert!(matches!(saved_failure.subjektiv_session_attribution,
Some(session_store::SubjektivSessionAttributionState::OutcomeUnknown { session_id: id, .. }) if id == session_id));
let session_store::worker_metadata::StartupFailureRead::Present(
error_contract::DecodeFailure::Known(cause),
) = saved_failure
.decode_startup_failure::<protocol::session_failure::SessionRegistrationCause>()
else {
panic!("recovery must retain its typed Session unknown cause")
};
assert_eq!(
*cause.facts().cause(),
protocol::session_failure::SessionRegistrationCause::outcome_unknown()
);
let denied = record();
assert_eq!(