runtime: allow cross-runtime workdir sessions

crates/worker-runtime/src/runtime.rs

@@ -360,8 +360,11 @@ impl Runtime {
         self.annotate_working_directory_status(status)
     }
 
-    /// Open a fresh Workdir operation session after proving that the persisted
-    /// materialization is assigned to a Worker in the authorized workspace.
+    /// Open a fresh Workdir operation session in this Runtime's authorized Workspace.
+    ///
+    /// A same-Runtime owner Worker can be supplied as an additional persisted-binding check.
+    /// Cross-Runtime callers rely on the Runtime's Workspace capability scope and the existence
+    /// of the Runtime-owned materialization; Backend attachment remains occupancy authority.
     pub fn open_workdir_session_scoped(
         &self,
         scope: &RuntimeWorkspaceScope,
@@ -373,33 +376,34 @@ impl Runtime {
             state.ensure_running()?;
             state.ensure_workspace_owner(scope, false)?;
 
-            let owns_workdir = |worker: &WorkerRecord| {
+            if let Some(worker_ref) = owner_worker_ref {
+                let owns_workdir = state
+                    .workers
+                    .get(&worker_ref.worker_id)
+                    .is_some_and(|worker| {
                         worker.belongs_to_workspace(&scope.workspace_id)
                             && worker.working_directory.as_ref().is_some_and(|status| {
                                 status.summary.working_directory_id == working_directory_id
                             })
-            };
-            let authorized = match owner_worker_ref {
-                Some(worker_ref) => state
-                    .workers
-                    .get(&worker_ref.worker_id)
-                    .is_some_and(owns_workdir),
-                None => state.workers.values().any(owns_workdir),
-            };
-            if !authorized {
+                    });
+                if !owns_workdir {
                     return Err(RuntimeError::WorkingDirectory(
                         crate::working_directory::WorkingDirectoryDiagnostic::rejected(
                             "working_directory_not_found",
-                        "working directory was not found in the authorized workspace",
+                            "working directory was not assigned to the authorized owner Worker",
                         ),
                     ));
                 }
+            }
             state.execution_backend.clone().ok_or_else(|| {
                 RuntimeError::ExecutionBackendUnavailable {
                     message: "opening a Workdir session requires an execution backend".to_string(),
                 }
             })?
         };
+        backend
+            .working_directory(working_directory_id)
+            .map_err(RuntimeError::WorkingDirectory)?;
         backend
             .open_workdir_session(working_directory_id)
             .map_err(RuntimeError::WorkingDirectory)

crates/workspace-server/src/hosts.rs

@@ -1181,14 +1181,16 @@ impl RuntimeRegistry {
         &self,
         runtime_id: &str,
         working_directory_id: &str,
-        owner_worker_id: &str,
+        owner_worker_id: Option<&str>,
     ) -> Result<WorkdirSessionHandle, RuntimeRegistryError> {
         validate_backend_identifier("runtime_id", runtime_id)?;
         validate_backend_identifier("working_directory_id", working_directory_id)?;
+        if let Some(owner_worker_id) = owner_worker_id {
             validate_backend_identifier("owner_worker_id", owner_worker_id)?;
+        }
         let runtime = self.runtime(runtime_id)?;
         runtime
-            .open_workdir_session(working_directory_id, Some(owner_worker_id))
+            .open_workdir_session(working_directory_id, owner_worker_id)
             .await
             .map_err(|error| RuntimeRegistryError::RuntimeOperationFailed {
                 runtime_id: runtime_id.to_string(),

crates/workspace-server/src/server.rs

@@ -3166,6 +3166,14 @@ fn current_worker_session_lock(
         .clone()
 }
 
+fn runtime_local_owner_worker_id(
+    caller_runtime_id: &str,
+    target_runtime_id: &str,
+    caller_worker_id: u64,
+) -> Option<String> {
+    (caller_runtime_id == target_runtime_id).then(|| caller_worker_id.to_string())
+}
+
 async fn open_current_worker_workdir_session_locked(
     api: &WorkspaceApi,
     runtime_id: &str,
@@ -3194,10 +3202,14 @@ async fn open_current_worker_workdir_session_locked(
                 link.workdir_id
             ),
         })?;
-    let owner_worker_id = format!("{runtime_id}-{worker_id}");
+    let owner_worker_id = runtime_local_owner_worker_id(runtime_id, &workdir.runtime_id, worker_id);
     let session = api
         .runtime
-        .open_workdir_session(&workdir.runtime_id, &workdir.workdir_id, &owner_worker_id)
+        .open_workdir_session(
+            &workdir.runtime_id,
+            &workdir.workdir_id,
+            owner_worker_id.as_deref(),
+        )
         .await
         .map_err(|error| error.into_error())?;
     let key = (runtime_id.to_string(), worker_id);
@@ -11226,6 +11238,18 @@ mod tests {
         );
     }
 
+    #[test]
+    fn workdir_session_owner_is_only_sent_for_same_runtime_worker() {
+        assert_eq!(
+            runtime_local_owner_worker_id("embedded-worker-runtime", "arcadia", 5),
+            None
+        );
+        assert_eq!(
+            runtime_local_owner_worker_id("arcadia", "arcadia", 30).as_deref(),
+            Some("30")
+        );
+    }
+
     #[tokio::test]
     async fn current_worker_workdir_routes_require_a_live_runtime_worker_identity() {
         let workspace = tempfile::tempdir().unwrap();