Merge commit '6f5b165831a7f4ce63758c0f63027e56c5028cd8' into work/t838-source-policy-adoption

crates/protocol/src/session_failure.rs

@@ -3,6 +3,48 @@
 //! 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.
+//!
+//! # Projection boundary
+//!
+//! The same closed cause is mapped explicitly into the shared error-contract codec.
+//! The transport boundary does not convert an internal Session denial into an HTTP
+//! verdict about a Browser operator. Public consumers must resolve current viewer
+//! authority before projecting; a saved attachment cannot carry that authority.
+//!
+//! | Contract fact | Internal transfer / storage | Authorized Browser / Tool |
+//! | --- | --- | --- |
+//! | Domain, reason, principal, required operation | Typed cause | Same cause, never inferred from prose |
+//! | Explanation | Fixed local template | Fixed local template, not received text |
+//! | Operation context | Allowlisted operation/stage pairs | Same allowlist, no operation references |
+//! | Subject, Session, Worker IDs; credentials; paths; raw HTTP body | Not representable in this cause | Not representable in this cause |
+//! | Resource ownership, persistence, secondary cleanup | Separate startup result facts | Not disclosure/retry authority |
+//!
+//! A denied current visibility decision fails projection instead of substituting
+//! an invented cause. Unknown codec versions and corrupt attachments stay distinct
+//! from known causes. Missing legacy attachments do not certify legacy diagnostics.
+//!
+//! # Failure investigation
+//!
+//! A known denial identifies the operation and principal class, not an instruction
+//! to grant rights. Check the current Subject execution policy and the authenticated
+//! Runtime/Worker binding at the Session-record authorization boundary. Compare
+//! existing Session attribution and saved Worker metadata before deciding whether
+//! any state repair is needed. Operator authority and Subject execution policy are
+//! different decisions; neither can be derived from an internal HTTP status alone.
+//!
+//! Transport and unknown outcomes do not establish non-execution. Reconcile the
+//! existing Session attribution, Worker aggregate and Runtime operation result;
+//! a lost response may coexist with a committed write. A conflict invalidates the
+//! old metadata expectation. Do not retry the startup, create another Worker, or
+//! enlarge permissions solely because a failure was displayed. Inspect primary,
+//! secondary, persistence and resource facts independently: rollback failure must
+//! not replace the refusal, and persistence uncertainty must not be called saved.
+//!
+//! The real authorization/factory fixture is owned by workspace-server's normal
+//! `session_registration_connected` tests. Its signed Session callback revokes the
+//! real policy at a synchronization point; a real retention pin forces independent
+//! cleanup failure. Leaf transport/projection tests do not substitute for that
+//! product connection, and the fixture claims neither a process crash nor restore.
 
 use error_contract::{
     Browser, Category, ContractError, Disclosure, FailureCause, FailureFacts, FailureIdentity,

crates/worker/src/subjektiv.rs

@@ -897,6 +897,10 @@ pub(crate) fn test_subject_authorization_response(
     })
 }
 
+#[cfg(test)]
+#[path = "subjektiv_session_registration_tests.rs"]
+mod session_registration_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/worker/src/subjektiv_session_registration_tests.rs

@@ -0,0 +1,159 @@
+//! The real authorization/factory fixture lives in workspace-server. This leaf
+//! exercises only WorkspaceClient transfer failures and response validation; it
+//! must not substitute simulated policy decisions for that connected fixture.
+use super::*;
+use crate::{WorkspaceClientError, WorkspaceResponse};
+use protocol::session_failure::{SessionPrincipal, SessionRegistrationCause as Cause};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+#[derive(Debug)]
+struct Client {
+    reply: Reply,
+    calls: AtomicUsize,
+}
+#[derive(Debug)]
+enum Reply {
+    Response(u16, String),
+    Unavailable,
+    Request,
+}
+impl WorkspaceClient for Client {
+    fn kind(&self) -> &str {
+        "registration-transfer-test"
+    }
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace")
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        assert_eq!(request.method, WorkspaceRequestMethod::Post);
+        assert_eq!(request.path, "/api/w/workspace/subjektiv/sessions");
+        match &self.reply {
+            Reply::Response(status, body) => Ok(WorkspaceResponse {
+                status: *status,
+                body: body.clone(),
+            }),
+            Reply::Unavailable => Err(WorkspaceClientError::Unavailable(
+                "TEST_ONLY_CREDENTIAL /private/path raw_http_body".into(),
+            )),
+            Reply::Request => Err(WorkspaceClientError::Request(
+                "TEST_ONLY_CREDENTIAL /private/path raw_http_body".into(),
+            )),
+        }
+    }
+}
+fn client(reply: Reply) -> Client {
+    Client {
+        reply,
+        calls: AtomicUsize::new(0),
+    }
+}
+fn failure_body(cause: Cause) -> String {
+    let mut error = RepositoryApiError::new(
+        403,
+        "ignored",
+        "TEST_ONLY_CREDENTIAL /private/path raw_http_body",
+        vec![],
+    );
+    error.failure =
+        Some(protocol::session_failure::session_registration_attachment(cause).unwrap());
+    serde_json::to_string(&error).unwrap()
+}
+fn assert_failure(client: Client, expected: Cause) -> SubjektivHostError {
+    let error = record_backend_session(&client, "session", true).unwrap_err();
+    assert_eq!(
+        client.calls.load(Ordering::SeqCst),
+        1,
+        "classification grants no replay"
+    );
+    assert_eq!(error.session_registration_cause(), expected);
+    let attachment = protocol::session_failure::session_registration_attachment(
+        error.session_registration_cause(),
+    )
+    .unwrap();
+    let encoded = serde_json::to_string(&attachment).unwrap();
+    for secret in ["TEST_ONLY_CREDENTIAL", "/private/path", "raw_http_body"] {
+        assert!(!encoded.contains(secret));
+        assert!(!error.to_string().contains(secret));
+    }
+    error
+}
+
+#[test]
+fn session_registration_http_reuses_typed_causes_not_status_or_prose() {
+    for cause in [
+        Cause::denied(SessionPrincipal::Operator),
+        Cause::denied(SessionPrincipal::Subject),
+        Cause::attribution_missing(),
+        Cause::binding_mismatch(),
+        Cause::outcome_unknown(),
+    ] {
+        for status in [403, 409, 500] {
+            assert_failure(client(Reply::Response(status, failure_body(cause))), cause);
+        }
+    }
+    // A legacy 403 without a domain attestation is not a certified denial.
+    let body = serde_json::to_string(&RepositoryApiError::new(
+        403,
+        "permission_denied",
+        "session.record denied",
+        vec![],
+    ))
+    .unwrap();
+    assert_failure(client(Reply::Response(403, body)), Cause::outcome_unknown());
+}
+
+#[test]
+fn session_registration_transport_and_unknown_keep_error_sources_without_disclosure_or_retry() {
+    for (reply, cause) in [
+        (Reply::Unavailable, Cause::transport()),
+        (Reply::Request, Cause::outcome_unknown()),
+    ] {
+        let error = assert_failure(client(reply), cause);
+        assert!(
+            std::error::Error::source(&error)
+                .unwrap()
+                .downcast_ref::<WorkspaceClientError>()
+                .is_some()
+        );
+    }
+    let legacy =
+        SubjektivHostError::PermissionDenied("session.record denied for secret subject".into());
+    assert_eq!(
+        legacy.session_registration_cause(),
+        Cause::outcome_unknown()
+    );
+}
+
+#[test]
+fn session_registration_rejects_malformed_incomplete_and_mismatched_responses() {
+    for (status, body) in [
+        (403, "not json"),
+        (200, "not json"),
+        (200, "{}"),
+        (200, r#"{"session_id":"session","subject_id":""}"#),
+        (
+            200,
+            r#"{"session_id":"other-session","subject_id":"subject"}"#,
+        ),
+    ] {
+        assert_failure(
+            client(Reply::Response(status, body.into())),
+            Cause::invalid_response(),
+        );
+    }
+    let allowed = client(Reply::Response(
+        200,
+        r#"{"session_id":"session","subject_id":"subject"}"#.into(),
+    ));
+    let output = record_backend_session(&allowed, "session", true).unwrap();
+    assert_eq!(output.session_id, "session");
+    assert_eq!(output.subject_id, "subject");
+    assert_eq!(allowed.calls.load(Ordering::SeqCst), 1);
+}

crates/workspace-server/src/hosts.rs

@@ -6772,6 +6772,9 @@ 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();
+        }
         projection.startup = result.startup_failure.as_ref().and_then(|primary| {
             internal_startup_transfer(runtime_api::WorkerStartupFailure {
                 primary: primary.clone(),
@@ -7178,6 +7181,79 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use std::thread;
 
+    #[test]
+    fn session_registration_embedded_and_remote_host_projection_is_internal_only() {
+        use protocol::session_failure::{
+            SessionPrincipal, SessionRegistrationCause, session_registration_attachment,
+        };
+        use worker_runtime::execution::{
+            WorkerExecutionOperation, WorkerExecutionOutcome, WorkerExecutionResult,
+        };
+        let mut result = WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn);
+        result.outcome = WorkerExecutionOutcome::Errored;
+        result.startup_failure = Some(
+            session_registration_attachment(SessionRegistrationCause::denied(
+                SessionPrincipal::Subject,
+            ))
+            .unwrap(),
+        );
+        result.startup_failure_persistence =
+            Some(protocol::startup_failure::StartupPersistenceOutcome::OutcomeUnknown);
+        result.secondary_failures.push(
+            protocol::startup_failure::StartupSecondaryFailure::Persistence {
+                phase: protocol::startup_failure::StartupFailurePhase::MetadataSave,
+                outcome: protocol::startup_failure::StartupPersistenceOutcome::OutcomeUnknown,
+            },
+        );
+        let error = EmbeddedRuntimeError::WorkerExecutionRejected {
+            worker_id: EmbeddedWorkerId::now_v7(),
+            operation: result.operation,
+            outcome: result.outcome,
+            message: "TEST_ONLY_CREDENTIAL /private/path raw_http_body".into(),
+            result,
+        };
+        let embedded = embedded_runtime_diagnostic(&error);
+        let mut wire = runtime_api::RuntimeApiError::new(
+            500,
+            "worker_execution_rejected",
+            "unused remote prose",
+        );
+        wire.error.startup = embedded.startup.clone();
+        let received = serde_json::from_str(&serde_json::to_string(&wire).unwrap()).unwrap();
+        let remote = runtime_api_diagnostic(
+            "remote",
+            runtime_api::client_support::ClientError::Public {
+                status: reqwest::StatusCode::INTERNAL_SERVER_ERROR,
+                error: received,
+            },
+        );
+        assert!(embedded.startup.is_some());
+        assert_eq!(embedded.startup, remote.startup);
+        let public = serde_json::to_value(&remote).unwrap();
+        assert!(
+            public.get("startup").is_none(),
+            "no blind public/storage attachment forwarding"
+        );
+        for secret in ["TEST_ONLY_CREDENTIAL", "/private/path", "raw_http_body"] {
+            assert!(!serde_json::to_string(&embedded).unwrap().contains(secret));
+        }
+
+        let legacy = runtime_api_diagnostic(
+            "remote",
+            runtime_api::client_support::ClientError::Public {
+                status: reqwest::StatusCode::FORBIDDEN,
+                error: runtime_api::RuntimeApiError::new(403, "forbidden", "session.record denied"),
+            },
+        );
+        assert!(
+            legacy.startup.is_none(),
+            "HTTP status and prose cannot attest a Session cause"
+        );
+        let mut corrupt = embedded.startup.unwrap();
+        corrupt.primary = serde_json::from_value(json!("not an error-contract envelope")).unwrap();
+        assert!(internal_startup_transfer(corrupt).is_none());
+    }
+
     #[tokio::test]
     async fn async_workdir_authorization_uses_store_capacity_without_bearer_fallback() {
         use crate::db_executor::{DbExecutor, DbExecutorLimits};

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

@@ -199,10 +199,99 @@ fn contains_saved_value(saved: &Value, expected: &Value) -> bool {
 }
 
 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()
+    // Reopen the canonical metadata store, not the separate Runtime catalog.
+    let name = format!("worker-runtime-{worker}");
+    let store = session_store::WorkerAggregateStore::open_read_only(
+        &root.join("workers").join(worker),
+        &name,
+    )
+    .unwrap();
+    let metadata = store.read_by_name(&name).unwrap().unwrap();
+    assert!(matches!(
+        metadata.decode_startup_failure::<SessionRegistrationCause>(),
+        session_store::worker_metadata::StartupFailureRead::Present(_)
+    ));
+    serde_json::to_value(metadata).unwrap()
+}
+
+#[tokio::test]
+async fn session_registration_real_judgments_separate_operator_subject_and_attribution() {
+    // This normal boundary fixture uses the real router, Runtime signature,
+    // authority store and attribution store; its Runtime does not run a model.
+    let f = Fixture::new().await;
+    let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions");
+    let missing_session = session_store::new_session_id().to_string();
+    let request = json!({"session_id": missing_session, "create_if_missing": false});
+    let (status, body) = response(
+        &f.api,
+        human_request(&f.owner, "POST", &path, Some(request.clone())),
+    )
+    .await;
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert_registration_cause(
+        body,
+        SessionRegistrationCause::denied(SessionPrincipal::Operator),
+    );
+
+    let signed = |request: &Value| {
+        runtime_source_request(
+            &f.identity,
+            Some(&f.source.worker_id),
+            "POST",
+            &path,
+            serde_json::to_vec(request).unwrap(),
+        )
+    };
+    let (status, body) = response(&f.api, signed(&request)).await;
+    assert_eq!(status, StatusCode::CONFLICT);
+    assert_registration_cause(body, SessionRegistrationCause::attribution_missing());
+
+    let store = open_subjektiv_store(&f.api).unwrap();
+    let foreign_session = session_store::new_session_id().to_string();
+    store
+        .record_session_attribution(
+            crate::subjektiv::SubjectSessionAttribution::new(
+                &f.subject,
+                &f.source.runtime_id,
+                "foreign-worker",
+                &foreign_session,
+            )
+            .unwrap(),
+        )
+        .unwrap();
+    let (status, body) = response(
+        &f.api,
+        signed(&json!({"session_id":foreign_session, "create_if_missing":false})),
+    )
+    .await;
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert_registration_cause(body, SessionRegistrationCause::binding_mismatch());
+
+    // The fixture removes its bootstrap policy only after the real setup POST.
+    let (status, body) = response(
+        &f.api,
+        signed(&json!({"session_id":missing_session, "create_if_missing":true})),
+    )
+    .await;
+    assert_eq!(status, StatusCode::FORBIDDEN);
+    assert_registration_cause(
+        body,
+        SessionRegistrationCause::denied(SessionPrincipal::Subject),
+    );
+    assert!(
+        store
+            .session_attribution(&missing_session)
+            .unwrap()
+            .is_none()
+    );
+}
+
+fn assert_registration_cause(body: Value, cause: SessionRegistrationCause) {
+    let error: server_api::RepositoryApiError = serde_json::from_value(body).unwrap();
+    assert_eq!(
+        error.failure,
+        Some(session_registration_attachment(cause).unwrap())
+    );
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]