test: retain real Session denial alongside metadata save failure

crates/workspace-server/src/server/tests/session_registration_connected_tests.rs

@@ -36,8 +36,15 @@ struct RegistrationFixture {
     model: UnusedModel,
 }
 
+#[derive(Clone, Copy)]
+enum RegistrationFault {
+    Allow,
+    Deny,
+    DenyWithMetadataReadFailure,
+}
+
 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.
@@ -63,6 +70,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();
@@ -111,6 +121,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 +159,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(),
@@ -296,7 +319,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 +414,81 @@ 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(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)