feat: integrate runtime Git cache materialization

crates/workdir/src/lib.rs

@@ -28,7 +28,9 @@ pub use fs_operation::{
     GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult,
     ReadRequest, ReadResult, StatRequest, StatResult, WriteRequest, WriteResult,
 };
-pub use local::{LocalWorkdirSession, SymlinkInfo, direct_symlink, first_symlink};
+pub use local::{
+    LocalWorkdirSession, SymlinkInfo, WorkdirSessionResource, direct_symlink, first_symlink,
+};
 pub use operation::*;
 
 /// Persistent, opaque identity of one materialized Workdir.

crates/workdir/src/local.rs

@@ -8,7 +8,8 @@
 //! `LocalWorkdirSession` is cheap to clone (`Arc` inside). Tool-specific session
 //! state, such as read-before-edit tracking, remains owned by the tool layer.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
+use std::fmt::Debug;
 #[cfg(test)]
 use std::io::Write as _;
 use std::io::{Read as _, Seek as _, SeekFrom};
@@ -228,6 +229,8 @@ struct LocalWorkdirSessionInner {
     next_command_id: AtomicU64,
     commands: Mutex<HashMap<String, LocalCommand>>,
     command_telemetry: CommandTelemetry,
+    command_environment: BTreeMap<String, String>,
+    resources: StdMutex<Vec<Arc<dyn WorkdirSessionResource>>>,
 }
 
 impl Drop for LocalWorkdirSessionInner {
@@ -242,6 +245,9 @@ impl Drop for LocalWorkdirSessionInner {
     }
 }
 
+pub trait WorkdirSessionResource: Debug + Send + Sync {}
+impl<T> WorkdirSessionResource for T where T: Debug + Send + Sync {}
+
 /// Scope-aware filesystem handle. Clone-cheap (`Arc` inside).
 ///
 /// The wrapped [`SharedScope`] is shared with every clone of this
@@ -318,6 +324,26 @@ impl LocalWorkdirSession {
         cwd: PathBuf,
         scope: SharedScope,
         capabilities: WorkdirSessionCapabilities,
+    ) -> Self {
+        Self::materialized_bound_with_environment(
+            workdir,
+            root,
+            cwd,
+            scope,
+            capabilities,
+            BTreeMap::new(),
+            Vec::new(),
+        )
+    }
+
+    pub fn materialized_bound_with_environment(
+        workdir: Workdir,
+        root: PathBuf,
+        cwd: PathBuf,
+        scope: SharedScope,
+        capabilities: WorkdirSessionCapabilities,
+        command_environment: BTreeMap<String, String>,
+        resources: Vec<Arc<dyn WorkdirSessionResource>>,
     ) -> Self {
         Self {
             inner: Arc::new(LocalWorkdirSessionInner {
@@ -331,6 +357,8 @@ impl LocalWorkdirSession {
                 next_command_id: AtomicU64::new(1),
                 commands: Mutex::new(HashMap::new()),
                 command_telemetry: CommandTelemetry::new(),
+                command_environment,
+                resources: StdMutex::new(resources),
             }),
         }
     }
@@ -669,9 +697,18 @@ impl WorkdirSession for LocalWorkdirSession {
         let (completion_tx, completion) = watch::channel(false);
         let command_id = handle.0.clone();
         let telemetry = self.inner.command_telemetry.clone();
+        let command_environment = self.inner.command_environment.clone();
         let (cancel, cancel_rx) = watch::channel(false);
         let task = tokio::spawn(async move {
-            let output = run_command(cwd, request, command_id, telemetry, cancel_rx).await;
+            let output = run_command(
+                cwd,
+                request,
+                command_id,
+                telemetry,
+                command_environment,
+                cancel_rx,
+            )
+            .await;
             let _ = completion_tx.send(true);
             output
         });
@@ -840,6 +877,9 @@ impl WorkdirSession for LocalWorkdirSession {
                 LocalCommand::Completed(_) => {}
             }
         }
+        if let Ok(mut resources) = self.inner.resources.lock() {
+            resources.clear();
+        }
         Ok(())
     }
 }
@@ -909,6 +949,7 @@ async fn run_command(
     request: CommandRequest,
     command_id: String,
     telemetry: CommandTelemetry,
+    command_environment: BTreeMap<String, String>,
     mut cancel: watch::Receiver<bool>,
 ) -> Result<CommandOutput, WorkdirError> {
     let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
@@ -925,6 +966,7 @@ async fn run_command(
         .arg("-c")
         .arg(&request.command)
         .current_dir(&cwd)
+        .envs(command_environment)
         .stdin(Stdio::null())
         .stdout(Stdio::from(stdout_file))
         .stderr(Stdio::from(stderr_file))
@@ -2319,6 +2361,31 @@ mod tests {
         assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None)));
     }
 
+    #[tokio::test]
+    async fn closing_session_releases_runtime_resources() {
+        #[derive(Debug)]
+        struct Resource(Arc<AtomicBool>);
+        impl Drop for Resource {
+            fn drop(&mut self) {
+                self.0.store(true, Ordering::Release);
+            }
+        }
+
+        let dir = TempDir::new().unwrap();
+        let released = Arc::new(AtomicBool::new(false));
+        let session = LocalWorkdirSession::materialized_bound_with_environment(
+            Workdir::new("resource-session"),
+            dir.path().to_path_buf(),
+            dir.path().to_path_buf(),
+            SharedScope::new(Scope::writable(dir.path()).unwrap()),
+            WorkdirSessionCapabilities::ALL,
+            BTreeMap::from([("SSH_AUTH_SOCK".to_string(), "test-socket".to_string())]),
+            vec![Arc::new(Resource(released.clone()))],
+        );
+        WorkdirSession::close(&session).await.unwrap();
+        assert!(released.load(Ordering::Acquire));
+    }
+
     #[tokio::test]
     async fn provider_cancels_active_command() {
         let dir = TempDir::new().unwrap();

crates/workdir/src/workspace.rs

@@ -30,6 +30,8 @@ impl RuntimeWorkerRef {
 #[serde(rename_all = "snake_case")]
 pub enum MaterializerKind {
     #[default]
+    RuntimeGitCache,
+    /// Legacy persisted value from the pre-cache local `git worktree` materializer.
     LocalGitWorktree,
 }
 
@@ -109,6 +111,8 @@ pub struct WorkingDirectoryProvenance {
     pub creation_selector: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub creation_ref: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub creation_tree: Option<String>,
     pub materializer_kind: MaterializerKind,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cleanup_target: Option<WorkingDirectoryCleanupTarget>,
@@ -122,6 +126,10 @@ pub struct WorkingDirectoryCurrentObservation {
     pub current_selector: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub current_ref: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub current_tree: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub observed_at_epoch_seconds: Option<u64>,
     pub status: WorkingDirectoryStatusKind,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cleanliness: Option<String>,
@@ -141,9 +149,15 @@ pub struct WorkingDirectorySummary {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub creation_ref: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub creation_tree: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub current_selector: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub current_ref: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub current_tree: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub observed_at_epoch_seconds: Option<u64>,
     pub materializer_kind: MaterializerKind,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cleanup_target: Option<WorkingDirectoryCleanupTarget>,
@@ -166,6 +180,7 @@ impl WorkingDirectorySummary {
         WorkingDirectoryProvenance {
             creation_selector: self.creation_selector.clone(),
             creation_ref: self.creation_ref.clone(),
+            creation_tree: self.creation_tree.clone(),
             materializer_kind: self.materializer_kind.clone(),
             cleanup_target: self.cleanup_target.clone(),
         }
@@ -175,6 +190,8 @@ impl WorkingDirectorySummary {
         WorkingDirectoryCurrentObservation {
             current_selector: self.current_selector.clone(),
             current_ref: self.current_ref.clone(),
+            current_tree: self.current_tree.clone(),
+            observed_at_epoch_seconds: self.observed_at_epoch_seconds,
             status: self.status.clone(),
             cleanliness: self.cleanliness.clone(),
             primary_worker_id: self.primary_worker_id.clone(),
@@ -247,8 +264,11 @@ mod tests {
                     repository_id: "repo".to_string(),
                     creation_selector: Some("develop".to_string()),
                     creation_ref: Some("abc123".to_string()),
+                    creation_tree: Some("tree123".to_string()),
                     current_selector: Some("work/ticket".to_string()),
                     current_ref: Some("def456".to_string()),
+                    current_tree: Some("tree456".to_string()),
+                    observed_at_epoch_seconds: Some(1_777_777_777),
                     materializer_kind: MaterializerKind::LocalGitWorktree,
                     cleanup_target: Some(WorkingDirectoryCleanupTarget {
                         kind: "git_worktree".to_string(),
@@ -269,8 +289,11 @@ mod tests {
                     repository_id: "repo".to_string(),
                     creation_selector: None,
                     creation_ref: None,
+                    creation_tree: None,
                     current_selector: None,
                     current_ref: Some("987fed".to_string()),
+                    current_tree: None,
+                    observed_at_epoch_seconds: None,
                     materializer_kind: MaterializerKind::LocalGitWorktree,
                     cleanup_target: None,
                     status: WorkingDirectoryStatusKind::Active,

crates/worker-runtime/Cargo.toml

@@ -43,6 +43,7 @@ tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }
 toml.workspace = true
 url.workspace = true
 uuid = { workspace = true, features = ["v7"] }
+zeroize.workspace = true
 tower = { workspace = true, features = ["util"], optional = true }
 worker.workspace = true
 workspace-api = { path = "../workspace-api" }

crates/worker-runtime/src/catalog.rs

@@ -97,6 +97,74 @@ pub use workdir::workspace::{
     WorkingDirectorySummary,
 };
 
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SensitiveString(String);
+
+impl SensitiveString {
+    pub fn new(value: impl Into<String>) -> Self {
+        Self(value.into())
+    }
+
+    pub fn expose(&self) -> &str {
+        &self.0
+    }
+}
+
+impl Drop for SensitiveString {
+    fn drop(&mut self) {
+        zeroize::Zeroize::zeroize(&mut self.0);
+    }
+}
+
+impl Default for SensitiveString {
+    fn default() -> Self {
+        Self(String::new())
+    }
+}
+
+impl std::fmt::Debug for SensitiveString {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str("[REDACTED]")
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RepositorySshMaterializationAccess {
+    pub credential_id: String,
+    pub credential_revision: u64,
+    pub host_trust_id: String,
+    pub host_trust_revision: u64,
+    pub access: workspace_api::RepositoryAccessMode,
+    pub expires_at_epoch_seconds: u64,
+    pub repository_id: String,
+    pub repository_source_fingerprint: String,
+    pub repository_uri: String,
+    pub secret_resource: crate::resource::BackendResourceHandle,
+    #[serde(skip, default)]
+    pub private_key: SensitiveString,
+    #[serde(skip, default)]
+    pub known_hosts_entry: SensitiveString,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RepositoryMaterializationContext {
+    pub workspace_id: String,
+    pub runtime_id: String,
+    pub operation_id: String,
+    pub config_revision: u64,
+    pub config_projection_digest: String,
+    #[serde(default)]
+    pub cache_generation: u64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ssh: Option<RepositorySshMaterializationAccess>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WorkingDirectoryRepositoryAccessRequest {
+    pub working_directory_id: String,
+    pub materialization: RepositoryMaterializationContext,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct WorkingDirectoryRequest {
     pub repository: WorkingDirectoryRepository,
@@ -106,6 +174,9 @@ pub struct WorkingDirectoryRequest {
     /// Backend can create canonical registry rows before materialization.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub backend_workdir_id: Option<String>,
+    /// Backend-authored, operation-scoped repository access and cache identity.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub materialization: Option<RepositoryMaterializationContext>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

crates/worker-runtime/src/execution.rs

@@ -1,4 +1,6 @@
-use crate::catalog::{WorkingDirectoryRequest, WorkingDirectoryStatus};
+use crate::catalog::{
+    WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
+};
 use crate::config_bundle::ConfigBundle;
 use crate::error::RuntimeError;
 use crate::identity::WorkerRef;
@@ -319,6 +321,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
         ))
     }
 
+    fn authorize_working_directory_repository_access(
+        &self,
+        _request: &WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), WorkingDirectoryDiagnostic> {
+        Err(WorkingDirectoryDiagnostic::rejected(
+            "working_directory_repository_access_unsupported",
+            "Worker execution backend does not support Repository access authorization",
+        ))
+    }
+
     fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
         Vec::new()
     }
@@ -454,6 +466,14 @@ impl WorkerExecutionBackendRef {
         self.backend.create_working_directory(request)
     }
 
+    pub(crate) fn authorize_working_directory_repository_access(
+        &self,
+        request: &WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), WorkingDirectoryDiagnostic> {
+        self.backend
+            .authorize_working_directory_repository_access(request)
+    }
+
     pub(crate) fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
         self.backend.list_working_directories()
     }

crates/worker-runtime/src/http_server.rs

@@ -12,7 +12,8 @@ use crate::auth::{
 };
 use crate::catalog::{
     ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
-    WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef,
+    WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
+    WorkspaceApiRef,
 };
 use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
 use crate::error::RuntimeError;
@@ -203,6 +204,10 @@ fn runtime_http_router_with_optional_auth(
             "/v1/working-directories",
             get(list_working_directories).post(create_working_directory),
         )
+        .route(
+            "/v1/working-directories/repository-access",
+            post(authorize_working_directory_repository_access),
+        )
         .route(
             "/v1/working-directories/{working_directory_id}/sessions",
             post(open_workdir_session),
@@ -335,6 +340,11 @@ pub struct RuntimeHttpWorkingDirectoriesResponse {
     pub working_directories: Vec<WorkingDirectoryStatus>,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RuntimeHttpRepositoryAccessResponse {
+    pub authorized: bool,
+}
+
 /// Working directory response used by create/detail/delete endpoints.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RuntimeHttpWorkingDirectoryResponse {
@@ -513,6 +523,29 @@ async fn list_workers(
     Ok(Json(RuntimeHttpWorkersResponse { workers }))
 }
 
+async fn authorize_working_directory_repository_access(
+    State(state): State<RuntimeHttpState>,
+    Extension(auth): Extension<RuntimeAuthContext>,
+    body: Result<Json<WorkingDirectoryRepositoryAccessRequest>, JsonRejection>,
+) -> RestResult<RuntimeHttpRepositoryAccessResponse> {
+    let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
+    if request.materialization.workspace_id != auth.workspace_id {
+        return Err(RuntimeHttpRestError::new(
+            StatusCode::FORBIDDEN,
+            "working_directory_materialization_workspace_mismatch",
+            "Repository access authority does not match the authenticated Workspace",
+        ));
+    }
+    state
+        .runtime
+        .authorize_working_directory_repository_access_from_resource(request)
+        .await
+        .map_err(RuntimeHttpRestError::runtime)?;
+    Ok(Json(RuntimeHttpRepositoryAccessResponse {
+        authorized: true,
+    }))
+}
+
 async fn list_working_directories(
     State(state): State<RuntimeHttpState>,
 ) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
@@ -527,12 +560,23 @@ async fn list_working_directories(
 
 async fn create_working_directory(
     State(state): State<RuntimeHttpState>,
+    Extension(auth): Extension<RuntimeAuthContext>,
     body: Result<Json<WorkingDirectoryRequest>, JsonRejection>,
 ) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
     let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
+    if let Some(materialization) = request.materialization.as_ref()
+        && materialization.workspace_id != auth.workspace_id
+    {
+        return Err(RuntimeHttpRestError::new(
+            StatusCode::FORBIDDEN,
+            "working_directory_materialization_workspace_mismatch",
+            "Repository materialization authority does not match the authenticated Workspace",
+        ));
+    }
     let working_directory = state
         .runtime
-        .create_working_directory(request)
+        .create_working_directory_from_resource(request)
+        .await
         .map_err(RuntimeHttpRestError::runtime)?;
     Ok(Json(RuntimeHttpWorkingDirectoryResponse {
         working_directory,
@@ -1559,6 +1603,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
     if path == "/v1/workers" && *method == Method::POST {
         return Some("workers:create");
     }
+    if path == "/v1/working-directories/repository-access" && *method == Method::POST {
+        return Some("workdirs:operate");
+    }
     if path.starts_with("/v1/workdir-sessions")
         || (path.starts_with("/v1/working-directories/") && path.ends_with("/sessions"))
     {
@@ -2220,6 +2267,10 @@ mod tests {
 
     #[test]
     fn workdir_routes_require_dedicated_operation_permission() {
+        assert_eq!(
+            required_runtime_permission(&Method::POST, "/v1/working-directories/repository-access",),
+            Some("workdirs:operate")
+        );
         assert_eq!(
             required_runtime_permission(&Method::POST, "/v1/working-directories/wd-1/sessions"),
             Some("workdirs:operate")

crates/worker-runtime/src/main.rs

@@ -23,10 +23,21 @@ use worker_runtime::http_server::{
     RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection,
 };
 use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend};
-use worker_runtime::working_directory::LocalGitWorktreeMaterializer;
+use worker_runtime::working_directory::RuntimeGitCacheMaterializer;
 use worker_runtime::{Runtime, RuntimeOptions};
 
 fn main() -> ExitCode {
+    let mut arguments = std::env::args().skip(1).collect::<Vec<_>>();
+    if arguments.first().map(String::as_str) == Some("__repository-ssh") {
+        arguments.remove(0);
+        return match worker_runtime::working_directory::run_repository_ssh_client(&arguments) {
+            Ok(status) => ExitCode::from(u8::try_from(status).unwrap_or(1)),
+            Err(error) => {
+                eprintln!("{error}");
+                ExitCode::from(1)
+            }
+        };
+    }
     match run() {
         Ok(()) => ExitCode::SUCCESS,
         Err(error) => {
@@ -169,6 +180,9 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
             factory = factory.with_remote_worker_mutation_identity(identity);
         }
     }
+    let mut backend_resource_client: Option<
+        Arc<dyn worker_runtime::resource::BackendResourceClient>,
+    > = None;
     if let Some(endpoint) = config.backend_resource_endpoint.clone() {
         let identity = runtime_auth.identity.as_ref().ok_or_else(|| {
             ProcessError::Auth(
@@ -181,26 +195,28 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
                     .to_owned(),
             ));
         };
-        factory = factory.with_resource_client(Arc::new(
+        let client = Arc::new(
             worker_runtime::resource::HttpBackendResourceClient::new(
                 endpoint,
                 config.backend_resource_token.clone(),
             )
             .with_runtime_request_source(identity, trusted_server.server_id.clone()),
-        ));
+        );
+        factory = factory.with_resource_client(client.clone());
+        backend_resource_client = Some(client);
     }
     let backend = Arc::new(
         WorkerRuntimeExecutionBackend::new(factory)
             .map_err(ProcessError::WorkerAdapter)?
-            .with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
+            .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
                 fs_paths.workdir_target.clone(),
             )),
     );
 
-    match &config.http.store {
+    let runtime = match &config.http.store {
         RuntimeHttpStoreSelection::Memory => {
             Runtime::with_execution_backend(runtime_options_from_http(&config.http), backend)
-                .map_err(ProcessError::Runtime)
+                .map_err(ProcessError::Runtime)?
         }
         RuntimeHttpStoreSelection::Fs { root } => {
             let mut options = FsRuntimeStoreOptions::new(root.clone()).with_runtime_id(
@@ -213,12 +229,20 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
             );
             options.display_name = config.http.display_name.clone();
             Runtime::with_fs_store_and_execution_backend(options, backend)
-                .map_err(ProcessError::Runtime)
+                .map_err(ProcessError::Runtime)?
         }
-        _ => Err(ProcessError::usage(
-            "unsupported Runtime catalog store selection".to_string(),
-        )),
+        _ => {
+            return Err(ProcessError::usage(
+                "unsupported Runtime catalog store selection".to_string(),
+            ));
+        }
+    };
+    if let Some(client) = backend_resource_client {
+        runtime
+            .install_backend_resource_client(client)
+            .map_err(ProcessError::Runtime)?;
     }
+    Ok(runtime)
 }
 
 fn runtime_options_from_http(config: &RuntimeHttpServerConfig) -> RuntimeOptions {

crates/worker-runtime/src/resource.rs

@@ -11,18 +11,46 @@ use std::sync::Mutex;
 
 pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str =
     "application/vnd.yoi.profile-source-archive+tar";
+pub const REPOSITORY_SSH_ACCESS_CONTENT_TYPE: &str =
+    "application/vnd.yoi.repository-ssh-access+json";
 pub const DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES: u64 = 2 * 1024 * 1024;
+pub const DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES: u64 = 64 * 1024;
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct RepositorySshAccessSecret {
+    pub private_key: String,
+    pub known_hosts_entry: String,
+}
+
+impl Drop for RepositorySshAccessSecret {
+    fn drop(&mut self) {
+        zeroize::Zeroize::zeroize(&mut self.private_key);
+        zeroize::Zeroize::zeroize(&mut self.known_hosts_entry);
+    }
+}
+
+impl std::fmt::Debug for RepositorySshAccessSecret {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("RepositorySshAccessSecret")
+            .field("private_key", &"[REDACTED]")
+            .field("known_hosts_entry", &"[REDACTED]")
+            .finish()
+    }
+}
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum BackendResourceKind {
     ProfileSourceArchive,
+    RepositorySshAccess,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum BackendResourceOperation {
     FetchArchive,
+    FetchOnce,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -66,7 +94,7 @@ pub struct BackendResourceFetchRequest {
     pub audit_correlation_id: String,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BackendResourceFetchResponse {
     pub kind: BackendResourceKind,
     pub resource_id: String,
@@ -76,6 +104,29 @@ pub struct BackendResourceFetchResponse {
     pub audit_correlation_id: String,
 }
 
+impl std::fmt::Debug for BackendResourceFetchResponse {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("BackendResourceFetchResponse")
+            .field("kind", &self.kind)
+            .field("resource_id", &self.resource_id)
+            .field("digest", &self.digest)
+            .field("content_type", &self.content_type)
+            .field(
+                "bytes",
+                &format_args!("[REDACTED; {} bytes]", self.bytes.len()),
+            )
+            .field("audit_correlation_id", &self.audit_correlation_id)
+            .finish()
+    }
+}
+
+impl Drop for BackendResourceFetchResponse {
+    fn drop(&mut self) {
+        self.bytes.fill(0);
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
 #[serde(tag = "code", rename_all = "snake_case")]
 pub enum BackendResourceError {
@@ -248,7 +299,7 @@ pub fn build_profile_source_archive_fetch_request(
 
 pub fn profile_source_archive_from_response(
     handle: &BackendResourceHandle,
-    response: BackendResourceFetchResponse,
+    mut response: BackendResourceFetchResponse,
 ) -> Result<ProfileSourceArchive, BackendResourceError> {
     if handle.kind != BackendResourceKind::ProfileSourceArchive
         || response.kind != BackendResourceKind::ProfileSourceArchive
@@ -263,7 +314,7 @@ pub fn profile_source_archive_from_response(
     if response.content_type != handle.content_type {
         return Err(BackendResourceError::ContentTypeMismatch {
             expected: handle.content_type.clone(),
-            actual: response.content_type,
+            actual: response.content_type.clone(),
         });
     }
     let actual_bytes = response.bytes.len() as u64;
@@ -278,7 +329,7 @@ pub fn profile_source_archive_from_response(
         return Err(BackendResourceError::DigestMismatch {
             expected: handle.digest.clone(),
             actual: if response.digest != handle.digest {
-                response.digest
+                response.digest.clone()
             } else {
                 actual_digest
             },
@@ -296,7 +347,7 @@ pub fn profile_source_archive_from_response(
                 }
             })?,
         },
-        content: response.bytes,
+        content: std::mem::take(&mut response.bytes),
     })
 }
 

crates/worker-runtime/src/runtime.rs

@@ -1,6 +1,6 @@
 use crate::catalog::{
     ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck,
-    WorkerStatus, WorkerSummary, WorkingDirectoryRequest,
+    WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
     WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef,
 };
 use crate::config_bundle::{
@@ -26,6 +26,10 @@ use crate::management::{
 };
 #[cfg(feature = "ws-server")]
 use crate::observation::{WorkerObservationCursor, WorkerObservationEvent};
+use crate::resource::{
+    BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceKind,
+    REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret,
+};
 #[cfg(feature = "fs-store")]
 use crate::retention::{
     FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult,
@@ -172,6 +176,14 @@ impl Runtime {
         Ok(runtime)
     }
 
+    pub fn install_backend_resource_client(
+        &self,
+        client: Arc<dyn BackendResourceClient>,
+    ) -> Result<(), RuntimeError> {
+        self.lock()?.backend_resource_client = Some(BackendResourceClientRef(client));
+        Ok(())
+    }
+
     /// Create or restore a filesystem-backed Runtime.
     ///
     /// The store is scoped by `options.root`; if the directory already exists,
@@ -366,6 +378,103 @@ impl Runtime {
             .map_err(RuntimeError::from)
     }
 
+    pub async fn create_working_directory_from_resource(
+        &self,
+        mut request: WorkingDirectoryRequest,
+    ) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
+        if let Some(ssh) = request
+            .materialization
+            .as_mut()
+            .and_then(|materialization| materialization.ssh.as_mut())
+        {
+            self.resolve_repository_access_resource(ssh).await?;
+        }
+        self.create_working_directory(request)
+    }
+
+    pub fn authorize_working_directory_repository_access(
+        &self,
+        request: WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), RuntimeError> {
+        let backend = {
+            let state = self.lock()?;
+            state.ensure_running()?;
+            state.execution_backend.clone().ok_or_else(|| {
+                RuntimeError::ExecutionBackendUnavailable {
+                    message: "working directory Repository access requires an execution backend"
+                        .to_string(),
+                }
+            })?
+        };
+        backend
+            .authorize_working_directory_repository_access(&request)
+            .map_err(RuntimeError::from)
+    }
+
+    async fn resolve_repository_access_resource(
+        &self,
+        ssh: &mut crate::catalog::RepositorySshMaterializationAccess,
+    ) -> Result<(), RuntimeError> {
+        if !ssh.private_key.expose().is_empty() && !ssh.known_hosts_entry.expose().is_empty() {
+            return Ok(());
+        }
+        let (client, runtime_id) = {
+            let state = self.lock()?;
+            let client = state.backend_resource_client.clone().ok_or_else(|| {
+                RuntimeError::InvalidRequest(
+                    "Backend Repository access resource client is unavailable".to_string(),
+                )
+            })?;
+            let runtime_id = state.runtime_identity.clone().ok_or_else(|| {
+                RuntimeError::InvalidRequest("Runtime identity is unavailable".to_string())
+            })?;
+            (client, runtime_id)
+        };
+        let mut response = client
+            .0
+            .fetch_resource(BackendResourceFetchRequest {
+                handle: ssh.secret_resource.clone(),
+                runtime_id,
+                worker_id: None,
+                audit_correlation_id: ssh.secret_resource.audit_correlation_id.clone(),
+            })
+            .await
+            .map_err(repository_resource_error)?;
+        if response.kind != BackendResourceKind::RepositorySshAccess
+            || response.content_type != REPOSITORY_SSH_ACCESS_CONTENT_TYPE
+            || response.resource_id != ssh.secret_resource.resource_id
+            || response.digest != ssh.secret_resource.digest
+            || response.bytes.len() as u64 > ssh.secret_resource.max_bytes
+        {
+            return Err(RuntimeError::InvalidRequest(
+                "Backend Repository SSH access resource response was invalid".to_string(),
+            ));
+        }
+        let secret = serde_json::from_slice::<RepositorySshAccessSecret>(&response.bytes);
+        response.bytes.fill(0);
+        let mut secret = secret.map_err(|_| {
+            RuntimeError::InvalidRequest(
+                "Backend Repository SSH access resource payload was invalid".to_string(),
+            )
+        })?;
+        ssh.private_key =
+            crate::catalog::SensitiveString::new(std::mem::take(&mut secret.private_key));
+        ssh.known_hosts_entry =
+            crate::catalog::SensitiveString::new(std::mem::take(&mut secret.known_hosts_entry));
+        Ok(())
+    }
+
+    pub async fn authorize_working_directory_repository_access_from_resource(
+        &self,
+        mut request: WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), RuntimeError> {
+        let ssh = request.materialization.ssh.as_mut().ok_or_else(|| {
+            RuntimeError::InvalidRequest("Repository SSH access metadata is missing".to_string())
+        })?;
+        self.resolve_repository_access_resource(ssh).await?;
+        self.authorize_working_directory_repository_access(request)
+    }
+
     /// List Runtime-owned working directories through the attached execution backend.
     pub fn list_working_directories(
         &self,
@@ -566,12 +675,13 @@ impl Runtime {
             let worker_id = request.worker_id;
             let worker_ref = WorkerRef::new(worker_id);
 
+            let durable_request = durable_create_worker_request(&request);
             let record = WorkerRecord {
                 worker_ref: worker_ref.clone(),
                 worker_id: worker_id.clone(),
                 status: WorkerStatus::Stopped,
                 workspace_id: scope.map(|scope| scope.workspace_id.clone()),
-                request: request.clone(),
+                request: durable_request,
                 run_generation: 1,
                 working_directory: None,
                 execution_handle: None,
@@ -1842,6 +1952,15 @@ struct SubscriptionSink {
     lagged: Arc<AtomicBool>,
 }
 
+#[derive(Clone)]
+struct BackendResourceClientRef(Arc<dyn BackendResourceClient>);
+
+impl std::fmt::Debug for BackendResourceClientRef {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str("BackendResourceClientRef(..)")
+    }
+}
+
 #[derive(Debug)]
 struct RuntimeState {
     display_name: Option<String>,
@@ -1853,6 +1972,7 @@ struct RuntimeState {
     persistence: RuntimePersistence,
     status: RuntimeStatus,
     execution_backend: Option<WorkerExecutionBackendRef>,
+    backend_resource_client: Option<BackendResourceClientRef>,
     #[cfg(feature = "fs-store")]
     next_diagnostic_id: u64,
     workers: BTreeMap<WorkerId, WorkerRecord>,
@@ -1880,6 +2000,7 @@ impl RuntimeState {
             persistence: RuntimePersistence::Memory,
             status: RuntimeStatus::Running,
             execution_backend: None,
+            backend_resource_client: None,
             #[cfg(feature = "fs-store")]
             next_diagnostic_id: 1,
             workers: BTreeMap::new(),
@@ -1908,6 +2029,7 @@ impl RuntimeState {
             persistence: RuntimePersistence::Fs(store),
             status: RuntimeStatus::Running,
             execution_backend: None,
+            backend_resource_client: None,
             #[cfg(feature = "fs-store")]
             next_diagnostic_id: 1,
             workers: BTreeMap::new(),
@@ -1959,6 +2081,7 @@ impl RuntimeState {
             persistence: RuntimePersistence::Fs(store),
             status: persisted.status,
             execution_backend: None,
+            backend_resource_client: None,
             next_diagnostic_id,
             workers,
             config_bundles: BTreeMap::new(),
@@ -2714,6 +2837,33 @@ fn worker_status_from_run_state(run_state: WorkerExecutionRunState) -> WorkerSta
     }
 }
 
+fn repository_resource_error(error: BackendResourceError) -> RuntimeError {
+    let category = match error {
+        BackendResourceError::Expired => "expired",
+        BackendResourceError::Unauthorized { .. } => "unauthorized",
+        BackendResourceError::UnsupportedKind => "unsupported_kind",
+        BackendResourceError::MissingResource => "missing_resource",
+        BackendResourceError::Oversized { .. } => "oversized",
+        BackendResourceError::DigestMismatch { .. } => "digest_mismatch",
+        BackendResourceError::ContentTypeMismatch { .. } => "content_type_mismatch",
+        BackendResourceError::InvalidResponse { .. } => "invalid_response",
+        BackendResourceError::Transport { .. } => "transport",
+    };
+    RuntimeError::InvalidRequest(format!(
+        "Backend Repository SSH access resource fetch failed: {category}"
+    ))
+}
+
+fn durable_create_worker_request(request: &CreateWorkerRequest) -> CreateWorkerRequest {
+    let mut durable = request.clone();
+    if let Some(working_directory) = durable.working_directory_request.as_mut()
+        && let Some(materialization) = working_directory.materialization.as_mut()
+    {
+        materialization.ssh = None;
+    }
+    durable
+}
+
 fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> {
     request
         .working_directory
@@ -2884,7 +3034,9 @@ fn subscription_worker_state(status: WorkerStatus) -> SubscriptionWorkerState {
 mod tests {
     use super::*;
     use crate::catalog::{
-        ConfigBundleRef, ProfileSelector, WorkingDirectoryClaim, WorkspaceApiRef,
+        ConfigBundleRef, MaterializerKind, ProfileSelector, RepositoryMaterializationContext,
+        RepositorySshMaterializationAccess, SensitiveString, WorkingDirectoryClaim,
+        WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
     };
     use crate::config_bundle::{
         ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
@@ -2894,6 +3046,8 @@ mod tests {
         WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle,
         WorkerExecutionRestoreRequest, WorkerExecutionRunState,
     };
+    use crate::working_directory::WorkingDirectoryDiagnostic;
+    use async_trait::async_trait;
     use std::collections::BTreeMap;
     #[cfg(feature = "fs-store")]
     use std::sync::atomic::{AtomicU64, Ordering};
@@ -3115,6 +3269,243 @@ mod tests {
         }
     }
 
+    #[test]
+    fn durable_worker_request_omits_repository_credentials() {
+        let mut request = task_request("worker-secret-redaction");
+        request.working_directory_request = Some(WorkingDirectoryRequest {
+            repository: WorkingDirectoryRepository {
+                id: "repository-1".to_string(),
+                provider: "git".to_string(),
+                source: workspace_api::RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::Ssh,
+                    uri: "ssh://git@example.test/repo.git".to_string(),
+                },
+                source_revision: 1,
+                source_fingerprint: "sha256:source".to_string(),
+                selector: None,
+            },
+            materializer: MaterializerKind::RuntimeGitCache,
+            backend_workdir_id: Some("working-directory-1".to_string()),
+            materialization: Some(RepositoryMaterializationContext {
+                workspace_id: "workspace-1".to_string(),
+                runtime_id: "runtime-1".to_string(),
+                operation_id: "operation-1".to_string(),
+                config_revision: 1,
+                config_projection_digest: "sha256:projection".to_string(),
+                cache_generation: 0,
+                ssh: Some(RepositorySshMaterializationAccess {
+                    credential_id: "credential-1".to_string(),
+                    credential_revision: 1,
+                    host_trust_id: "host-trust-1".to_string(),
+                    host_trust_revision: 1,
+                    access: workspace_api::RepositoryAccessMode::ReadOnly,
+                    expires_at_epoch_seconds: u64::MAX,
+                    repository_id: "repository-1".to_string(),
+                    repository_source_fingerprint: "sha256:source".to_string(),
+                    repository_uri: "ssh://git@example.test/repo.git".to_string(),
+                    secret_resource: repository_resource_handle(),
+                    private_key: SensitiveString::new("private-key-bytes"),
+                    known_hosts_entry: SensitiveString::new("known-hosts-entry"),
+                }),
+            }),
+        });
+
+        let durable = durable_create_worker_request(&request);
+
+        assert!(
+            request
+                .working_directory_request
+                .as_ref()
+                .and_then(|working_directory| working_directory.materialization.as_ref())
+                .and_then(|materialization| materialization.ssh.as_ref())
+                .is_some()
+        );
+        assert!(
+            durable
+                .working_directory_request
+                .as_ref()
+                .and_then(|working_directory| working_directory.materialization.as_ref())
+                .and_then(|materialization| materialization.ssh.as_ref())
+                .is_none()
+        );
+        let serialized = serde_json::to_string(&durable).unwrap();
+        assert!(!serialized.contains("private-key-bytes"));
+        assert!(!serialized.contains("known-hosts-entry"));
+    }
+
+    fn repository_resource_handle() -> crate::resource::BackendResourceHandle {
+        crate::resource::BackendResourceHandle {
+            kind: crate::resource::BackendResourceKind::RepositorySshAccess,
+            workspace_id: "workspace-1".to_string(),
+            scope_id: Some("repository-ssh-access".to_string()),
+            runtime_id: Some("runtime-1".to_string()),
+            worker_id: None,
+            resource_id: "repository-access-1".to_string(),
+            digest: "opaque:repository-access-1".to_string(),
+            operation: crate::resource::BackendResourceOperation::FetchOnce,
+            expires_at_unix_seconds: i64::MAX,
+            nonce: "repository-access-1".to_string(),
+            revision: "1".to_string(),
+            generation: None,
+            max_bytes: crate::resource::DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES,
+            content_type: crate::resource::REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
+            redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
+            audit_correlation_id: "repository-access-1".to_string(),
+            profile_source_graph: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn repository_access_resource_is_fetched_before_provider_authorization() {
+        let (runtime, backend) = runtime_and_backend();
+        backend
+            .repository_access_available
+            .store(true, Ordering::SeqCst);
+        runtime.bind_runtime_identity("runtime-1").unwrap();
+        let handle = repository_resource_handle();
+        runtime
+            .install_backend_resource_client(Arc::new(TestRepositoryResourceClient {
+                response: Mutex::new(Some(crate::resource::BackendResourceFetchResponse {
+                    kind: crate::resource::BackendResourceKind::RepositorySshAccess,
+                    resource_id: handle.resource_id.clone(),
+                    digest: handle.digest.clone(),
+                    content_type: crate::resource::REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
+                    bytes: serde_json::to_vec(&RepositorySshAccessSecret {
+                        private_key: "private-key-bytes".to_string(),
+                        known_hosts_entry: "known-hosts-entry".to_string(),
+                    })
+                    .unwrap(),
+                    audit_correlation_id: handle.audit_correlation_id.clone(),
+                })),
+            }))
+            .unwrap();
+        let request = WorkingDirectoryRepositoryAccessRequest {
+            working_directory_id: "working-directory-1".to_string(),
+            materialization: RepositoryMaterializationContext {
+                workspace_id: "workspace-1".to_string(),
+                runtime_id: "runtime-1".to_string(),
+                operation_id: "operation-1".to_string(),
+                config_revision: 1,
+                config_projection_digest: "sha256:projection".to_string(),
+                cache_generation: 0,
+                ssh: Some(RepositorySshMaterializationAccess {
+                    credential_id: "credential-1".to_string(),
+                    credential_revision: 1,
+                    host_trust_id: "host-trust-1".to_string(),
+                    host_trust_revision: 1,
+                    access: workspace_api::RepositoryAccessMode::ReadOnly,
+                    expires_at_epoch_seconds: u64::MAX,
+                    repository_id: "repository-1".to_string(),
+                    repository_source_fingerprint: "sha256:source".to_string(),
+                    repository_uri: "ssh://git@example.test/repo.git".to_string(),
+                    secret_resource: handle,
+                    private_key: SensitiveString::default(),
+                    known_hosts_entry: SensitiveString::default(),
+                }),
+            },
+        };
+
+        let replay = request.clone();
+        runtime
+            .authorize_working_directory_repository_access_from_resource(request)
+            .await
+            .unwrap();
+        assert!(
+            runtime
+                .authorize_working_directory_repository_access_from_resource(replay)
+                .await
+                .is_err()
+        );
+
+        let accesses = backend.repository_accesses.lock().unwrap();
+        assert_eq!(accesses.len(), 1);
+        let access = accesses[0].materialization.ssh.as_ref().unwrap();
+        assert_eq!(access.private_key.expose(), "private-key-bytes");
+        assert_eq!(access.known_hosts_entry.expose(), "known-hosts-entry");
+    }
+
+    #[tokio::test]
+    async fn working_directory_create_fetches_repository_access_before_provider_call() {
+        let (runtime, backend) = runtime_and_backend();
+        backend
+            .repository_access_available
+            .store(true, Ordering::SeqCst);
+        runtime.bind_runtime_identity("runtime-1").unwrap();
+        let handle = repository_resource_handle();
+        runtime
+            .install_backend_resource_client(Arc::new(TestRepositoryResourceClient {
+                response: Mutex::new(Some(crate::resource::BackendResourceFetchResponse {
+                    kind: crate::resource::BackendResourceKind::RepositorySshAccess,
+                    resource_id: handle.resource_id.clone(),
+                    digest: handle.digest.clone(),
+                    content_type: crate::resource::REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
+                    bytes: serde_json::to_vec(&RepositorySshAccessSecret {
+                        private_key: "create-private-key-bytes".to_string(),
+                        known_hosts_entry: "create-known-hosts-entry".to_string(),
+                    })
+                    .unwrap(),
+                    audit_correlation_id: handle.audit_correlation_id.clone(),
+                })),
+            }))
+            .unwrap();
+        let request = WorkingDirectoryRequest {
+            repository: WorkingDirectoryRepository {
+                id: "repository-1".to_string(),
+                provider: "git".to_string(),
+                source: workspace_api::RepositorySource {
+                    kind: workspace_api::RepositorySourceKind::Ssh,
+                    uri: "ssh://git@example.test/repo.git".to_string(),
+                },
+                source_revision: 1,
+                source_fingerprint: "sha256:source".to_string(),
+                selector: None,
+            },
+            materializer: MaterializerKind::RuntimeGitCache,
+            backend_workdir_id: Some("working-directory-1".to_string()),
+            materialization: Some(RepositoryMaterializationContext {
+                workspace_id: "workspace-1".to_string(),
+                runtime_id: "runtime-1".to_string(),
+                operation_id: "operation-create".to_string(),
+                config_revision: 1,
+                config_projection_digest: "sha256:projection".to_string(),
+                cache_generation: 0,
+                ssh: Some(RepositorySshMaterializationAccess {
+                    credential_id: "credential-1".to_string(),
+                    credential_revision: 1,
+                    host_trust_id: "host-trust-1".to_string(),
+                    host_trust_revision: 1,
+                    access: workspace_api::RepositoryAccessMode::ReadOnly,
+                    expires_at_epoch_seconds: u64::MAX,
+                    repository_id: "repository-1".to_string(),
+                    repository_source_fingerprint: "sha256:source".to_string(),
+                    repository_uri: "ssh://git@example.test/repo.git".to_string(),
+                    secret_resource: handle,
+                    private_key: SensitiveString::default(),
+                    known_hosts_entry: SensitiveString::default(),
+                }),
+            }),
+        };
+
+        assert!(
+            runtime
+                .create_working_directory_from_resource(request)
+                .await
+                .is_err()
+        );
+
+        let requests = backend.working_directory_requests.lock().unwrap();
+        let access = requests[0]
+            .materialization
+            .as_ref()
+            .and_then(|materialization| materialization.ssh.as_ref())
+            .unwrap();
+        assert_eq!(access.private_key.expose(), "create-private-key-bytes");
+        assert_eq!(
+            access.known_hosts_entry.expose(),
+            "create-known-hosts-entry"
+        );
+    }
+
     fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest {
         let mut request = task_request(objective);
         request.workspace_api = Some(WorkspaceApiRef {
@@ -3196,6 +3587,9 @@ mod tests {
         config_bundles: Mutex<Vec<Option<ConfigBundle>>>,
         contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
         dispatched_inputs: Mutex<Vec<WorkerInput>>,
+        repository_accesses: Mutex<Vec<WorkingDirectoryRepositoryAccessRequest>>,
+        repository_access_available: AtomicBool,
+        working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
         preserve_commit_ack_submission_id: AtomicBool,
         #[cfg(feature = "ws-server")]
         snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
@@ -3236,6 +3630,38 @@ mod tests {
             "test-execution-backend"
         }
 
+        fn create_working_directory(
+            &self,
+            request: &WorkingDirectoryRequest,
+        ) -> Result<CatalogWorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
+            self.working_directory_requests
+                .lock()
+                .unwrap()
+                .push(request.clone());
+            Err(WorkingDirectoryDiagnostic::rejected(
+                "working_directory_unsupported",
+                "Worker execution backend does not support working directory materialization",
+            ))
+        }
+
+        fn authorize_working_directory_repository_access(
+            &self,
+            request: &WorkingDirectoryRepositoryAccessRequest,
+        ) -> Result<(), WorkingDirectoryDiagnostic> {
+            self.repository_accesses
+                .lock()
+                .unwrap()
+                .push(request.clone());
+            if self.repository_access_available.load(Ordering::SeqCst) {
+                Ok(())
+            } else {
+                Err(WorkingDirectoryDiagnostic::rejected(
+                    "working_directory_repository_access_unsupported",
+                    "Worker execution backend does not support Repository access authorization",
+                ))
+            }
+        }
+
         fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
             self.run_generations
                 .lock()
@@ -3343,6 +3769,24 @@ mod tests {
         }
     }
 
+    struct TestRepositoryResourceClient {
+        response: Mutex<Option<crate::resource::BackendResourceFetchResponse>>,
+    }
+
+    #[async_trait]
+    impl BackendResourceClient for TestRepositoryResourceClient {
+        async fn fetch_resource(
+            &self,
+            _request: BackendResourceFetchRequest,
+        ) -> Result<crate::resource::BackendResourceFetchResponse, BackendResourceError> {
+            self.response
+                .lock()
+                .unwrap()
+                .take()
+                .ok_or(BackendResourceError::MissingResource)
+        }
+    }
+
     fn runtime_with_backend() -> Runtime {
         let runtime = Runtime::with_execution_backend(
             RuntimeOptions::default(),

crates/worker-runtime/src/worker_backend.rs

@@ -20,7 +20,7 @@ use crate::auth::{
 };
 use crate::catalog::{
     CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource,
-    WorkingDirectoryRequest, WorkingDirectoryStatus,
+    WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
 };
 use crate::execution::{
     WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation,
@@ -706,13 +706,17 @@ fn runtime_local_workdir_session(
     root: &Path,
     cwd: &Path,
     scope: manifest::SharedScope,
+    command_environment: std::collections::BTreeMap<String, String>,
+    resources: Vec<Arc<dyn workdir::WorkdirSessionResource>>,
 ) -> WorkdirSessionHandle {
-    Arc::new(LocalWorkdirSession::materialized_bound(
+    Arc::new(LocalWorkdirSession::materialized_bound_with_environment(
         Workdir::new(workdir_id),
         root.to_path_buf(),
         cwd.to_path_buf(),
         scope,
         WorkdirSessionCapabilities::ALL,
+        command_environment,
+        resources,
     ))
 }
 
@@ -893,6 +897,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                 binding.root(),
                 binding.cwd(),
                 worker.scope().clone(),
+                binding.command_environment(),
+                binding.session_resources(),
             )));
         } else {
             worker.bind_workdir_session(None);
@@ -1071,6 +1077,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                 binding.root(),
                 binding.cwd(),
                 worker.scope().clone(),
+                binding.command_environment(),
+                binding.session_resources(),
             )));
         } else {
             worker.bind_workdir_session(None);
@@ -1582,6 +1590,19 @@ where
         Ok(materializer.create(request)?.status())
     }
 
+    fn authorize_working_directory_repository_access(
+        &self,
+        request: &WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), WorkingDirectoryDiagnostic> {
+        let materializer = self.working_directory_materializer.as_ref().ok_or_else(|| {
+            WorkingDirectoryDiagnostic::rejected(
+                "working_directory_materializer_unavailable",
+                "working directory Repository access requested, but no materializer is configured for this runtime backend",
+            )
+        })?;
+        materializer.authorize_repository_access(request)
+    }
+
     fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
         self.working_directory_materializer
             .as_ref()
@@ -1624,6 +1645,8 @@ where
             binding.root(),
             binding.cwd(),
             manifest::SharedScope::new(scope),
+            binding.command_environment(),
+            binding.session_resources(),
         ))
     }
 
@@ -2142,7 +2165,7 @@ mod tests {
     use crate::identity::WorkerRef;
     use crate::management::RuntimeOptions;
     use crate::observation::WorkerObservationCursor;
-    use crate::working_directory::LocalGitWorktreeMaterializer;
+    use crate::working_directory::RuntimeGitCacheMaterializer;
     use agen::Engine;
     use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
     use agen::llm_client::{ClientError, LlmClient, Request};
@@ -2728,8 +2751,9 @@ mod tests {
                 source_fingerprint: "sha256:test".to_string(),
                 selector: Some(RepositorySelector::from("HEAD")),
             },
-            materializer: MaterializerKind::LocalGitWorktree,
+            materializer: MaterializerKind::RuntimeGitCache,
             backend_workdir_id: None,
+            materialization: None,
         }
     }
 
@@ -2853,12 +2877,16 @@ mod tests {
             root.path(),
             root.path(),
             manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
+            Default::default(),
+            Vec::new(),
         );
         let restored = runtime_local_workdir_session(
             "working-directory-42",
             root.path(),
             root.path(),
             manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
+            Default::default(),
+            Vec::new(),
         );
 
         assert_eq!(spawned.workdir().id().as_str(), "working-directory-42");
@@ -3330,7 +3358,7 @@ mod tests {
         };
         let backend = WorkerRuntimeExecutionBackend::new(factory)
             .unwrap()
-            .with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
+            .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
                 runtime_base.path(),
             ));
         let runtime =
@@ -3485,7 +3513,7 @@ mod tests {
         };
         let backend = WorkerRuntimeExecutionBackend::new(factory)
             .unwrap()
-            .with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
+            .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
                 runtime_base.path(),
             ));
         let runtime =
@@ -3524,7 +3552,7 @@ mod tests {
         let repo = create_clean_repo();
         let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
             .unwrap()
-            .with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
+            .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
                 runtime_base.path(),
             ));
         let runtime =
@@ -3560,7 +3588,7 @@ mod tests {
         let repo = create_clean_repo();
         let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
             .unwrap()
-            .with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
+            .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
                 runtime_base.path(),
             ));
         let runtime =
@@ -3574,9 +3602,15 @@ mod tests {
 
         assert!(format!("{error:?}").contains("spawn failed"));
         let working_directories_root = runtime_base.path();
-        let remaining_entries = fs::read_dir(working_directories_root)
-            .map(|entries| entries.count())
+        let remaining_workdirs = fs::read_dir(working_directories_root)
+            .map(|entries| {
+                entries
+                    .flatten()
+                    .filter(|entry| !entry.file_name().to_string_lossy().starts_with('.'))
+                    .count()
+            })
             .unwrap_or(0);
-        assert_eq!(remaining_entries, 0);
+        assert_eq!(remaining_workdirs, 0);
+        assert!(working_directories_root.join(".repository-cache").is_dir());
     }
 }

crates/workspace-server/Cargo.toml

@@ -48,6 +48,7 @@ tracing.workspace = true
 ts-rs = { version = "12.0.1", optional = true }
 url.workspace = true
 uuid = { workspace = true, features = ["v7"] }
+zeroize.workspace = true
 webauthn-rs = { workspace = true }
 
 [dev-dependencies]

crates/workspace-server/src/hosts.rs

@@ -25,8 +25,9 @@ use worker_runtime::auth::{CapabilityTokenSigner, capability_claims};
 use worker_runtime::catalog::{
     ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef,
     ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail,
-    WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRequest,
-    WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef,
+    WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim,
+    WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
+    WorkingDirectorySummary, WorkspaceApiRef,
 };
 use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
 #[cfg(test)]
@@ -39,11 +40,11 @@ use worker_runtime::execution::WorkerExecutionRunState;
 use worker_runtime::fs_store::FsRuntimeStoreOptions;
 use worker_runtime::http_server::{
     RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
-    RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerCompletionsRequest,
-    RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse,
-    RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
-    RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
-    RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
+    RuntimeHttpErrorResponse, RuntimeHttpRepositoryAccessResponse, RuntimeHttpSummaryResponse,
+    RuntimeHttpWorkerCompletionsRequest, RuntimeHttpWorkerCompletionsResponse,
+    RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse,
+    RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse,
+    RuntimeHttpWorkerResponse, RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
     RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
     RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
 };
@@ -818,6 +819,16 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
         }
     }
 
+    fn authorize_working_directory_repository_access(
+        &self,
+        _request: WorkingDirectoryRepositoryAccessRequest,
+    ) -> std::result::Result<(), Error> {
+        Err(Error::InvalidInput(
+            "Runtime does not support working directory Repository access authorization"
+                .to_string(),
+        ))
+    }
+
     fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
         RuntimeList::new(Vec::new(), Vec::new())
     }
@@ -1391,6 +1402,23 @@ impl RuntimeRegistry {
         Ok(runtime.create_working_directory(request))
     }
 
+    pub fn authorize_working_directory_repository_access(
+        &self,
+        runtime_id: &str,
+        request: WorkingDirectoryRepositoryAccessRequest,
+    ) -> Result<(), RuntimeRegistryError> {
+        validate_backend_identifier("runtime_id", runtime_id)?;
+        validate_backend_identifier("working_directory_id", &request.working_directory_id)?;
+        let runtime = self.runtime(runtime_id)?;
+        runtime
+            .authorize_working_directory_repository_access(request)
+            .map_err(|error| RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: runtime_id.to_string(),
+                code: "working_directory_repository_access_failed".to_string(),
+                message: error.to_string(),
+            })
+    }
+
     pub fn list_working_directories(
         &self,
         runtime_id: &str,
@@ -3179,6 +3207,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         }
     }
 
+    fn authorize_working_directory_repository_access(
+        &self,
+        request: WorkingDirectoryRepositoryAccessRequest,
+    ) -> std::result::Result<(), Error> {
+        self.post_json::<_, RuntimeHttpRepositoryAccessResponse>(
+            "/v1/working-directories/repository-access",
+            &request,
+        )
+        .map(|_| ())
+        .map_err(|diagnostic| Error::RegistryInconsistency(diagnostic.message))
+    }
+
     fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
         match self.get_json::<RuntimeHttpWorkingDirectoriesResponse>("/v1/working-directories") {
             Ok(response) => RuntimeList::new(response.working_directories, Vec::new()),
@@ -4386,7 +4426,7 @@ mod tests {
         let handle = bundle.profile_source_archive_handle.as_ref().unwrap();
         assert!(bundle.profile_source_archive.is_none());
         let response = broker
-            .fetch_profile_source_archive(worker_runtime::resource::BackendResourceFetchRequest {
+            .fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
                 handle: handle.clone(),
                 runtime_id: runtime_id.to_string(),
                 worker_id: None,

crates/workspace-server/src/repository_access.rs

@@ -11,7 +11,7 @@ use ring::rand::{SecureRandom, SystemRandom};
 use rusqlite::{OptionalExtension, TransactionBehavior, params};
 use serde::Deserialize;
 use sha2::{Digest, Sha256};
-use ssh_key::{Algorithm, HashAlg, PrivateKey, PublicKey};
+use ssh_key::{Algorithm, HashAlg, LineEnding, PrivateKey, PublicKey};
 use workspace_api::{
     CreateRepositorySshCredentialRequest, DeleteRepositorySshCredentialRequest,
     DeleteRepositorySshHostTrustRequest, PutRepositorySshHostTrustRequest, RepositoryAccessMode,
@@ -59,7 +59,6 @@ impl WorkspaceConfigSchemaProvider for RepositoryAccessConfigSchemaProvider {
 }
 
 #[derive(Debug, Default, Deserialize)]
-#[serde(deny_unknown_fields)]
 struct VirtualWorkspaceConfig {
     #[serde(default)]
     repository_access: BTreeMap<String, VirtualRepositoryAccess>,
@@ -219,6 +218,16 @@ fn project_repository_access_evaluation(
     })
 }
 
+#[derive(Clone)]
+pub struct LeasedRepositorySshAccess {
+    pub credential_id: String,
+    pub credential_revision: u64,
+    pub host_trust_id: String,
+    pub host_trust_revision: u64,
+    pub private_key: zeroize::Zeroizing<String>,
+    pub known_hosts_entry: String,
+}
+
 #[derive(Clone)]
 pub struct RepositorySecretService {
     store: Arc<SqliteWorkspaceStore>,
@@ -829,6 +838,184 @@ impl RepositorySecretService {
         })
     }
 
+    pub fn lease_ssh_materialization_access(
+        &self,
+        workspace_id: &str,
+        binding: &RepositorySshAccessBinding,
+    ) -> Result<LeasedRepositorySshAccess> {
+        let credential = self
+            .get_credential(workspace_id, &binding.credential_id, &[])?
+            .ok_or_else(|| {
+                Error::InvalidInput(format!(
+                    "unknown Repository SSH credential `{}`",
+                    binding.credential_id
+                ))
+            })?;
+        if credential.status != "active" {
+            return Err(Error::InvalidInput(format!(
+                "Repository SSH credential `{}` is not active",
+                binding.credential_id
+            )));
+        }
+        let host_trust = self
+            .get_host_trust(workspace_id, &binding.host_trust_id, &[])?
+            .ok_or_else(|| {
+                Error::InvalidInput(format!(
+                    "unknown Repository SSH host trust `{}`",
+                    binding.host_trust_id
+                ))
+            })?;
+        self.lease_ssh_materialization_access_revision(
+            workspace_id,
+            &binding.credential_id,
+            credential.current_revision,
+            &binding.host_trust_id,
+            host_trust.current_revision,
+        )
+    }
+
+    pub fn lease_ssh_materialization_access_revision(
+        &self,
+        workspace_id: &str,
+        credential_id: &str,
+        credential_revision: u64,
+        host_trust_id: &str,
+        host_trust_revision: u64,
+    ) -> Result<LeasedRepositorySshAccess> {
+        let (private_key, passphrase, hostname, port, host_key) = self.store.with_conn(|conn| {
+            let private_key = read_sealed_secret(
+                conn,
+                workspace_id,
+                credential_id,
+                credential_revision,
+                "private_key",
+            )?
+            .ok_or_else(|| {
+                Error::RegistryInconsistency(format!(
+                    "Repository SSH credential `{credential_id}` revision {credential_revision} is unavailable"
+                ))
+            })?;
+            let passphrase = read_sealed_secret(
+                conn,
+                workspace_id,
+                credential_id,
+                credential_revision,
+                "passphrase",
+            )?;
+            let (hostname, port, host_key) = conn
+                .query_row(
+                    r#"SELECT h.hostname, h.port, v.host_key
+                       FROM repository_ssh_host_trusts h
+                       JOIN repository_ssh_host_trust_revisions v
+                         ON v.workspace_id = h.workspace_id
+                        AND v.host_trust_id = h.host_trust_id
+                       WHERE h.workspace_id = ?1 AND h.host_trust_id = ?2
+                         AND v.revision = ?3"#,
+                    params![workspace_id, host_trust_id, host_trust_revision as i64],
+                    |row| {
+                        Ok((
+                            row.get::<_, String>(0)?,
+                            row.get::<_, i64>(1)? as u16,
+                            row.get::<_, String>(2)?,
+                        ))
+                    },
+                )
+                .optional()?
+                .ok_or_else(|| {
+                    Error::RegistryInconsistency(format!(
+                        "Repository SSH host trust `{host_trust_id}` revision {host_trust_revision} is unavailable"
+                    ))
+                })?;
+            Ok((private_key, passphrase, hostname, port, host_key))
+        })?;
+        let private_key = self.unseal(
+            workspace_id,
+            credential_id,
+            credential_revision,
+            "private_key",
+            private_key,
+        )?;
+        let passphrase = passphrase
+            .map(|secret| {
+                self.unseal(
+                    workspace_id,
+                    credential_id,
+                    credential_revision,
+                    "passphrase",
+                    secret,
+                )
+            })
+            .transpose()?;
+        let private_key =
+            zeroize::Zeroizing::new(String::from_utf8(private_key).map_err(|_| {
+                Error::Store("Repository SSH private key plaintext is invalid".to_string())
+            })?);
+        let passphrase = passphrase
+            .map(|value| {
+                String::from_utf8(value)
+                    .map(zeroize::Zeroizing::new)
+                    .map_err(|_| {
+                        Error::Store("Repository SSH passphrase plaintext is invalid".to_string())
+                    })
+            })
+            .transpose()?;
+        let key = PrivateKey::from_openssh(private_key.as_str()).map_err(|_| {
+            Error::Store("Repository SSH private key plaintext is invalid".to_string())
+        })?;
+        let key = if key.is_encrypted() {
+            key.decrypt(passphrase.as_deref().ok_or_else(|| {
+                Error::Store("Repository SSH passphrase revision is unavailable".to_string())
+            })?)
+            .map_err(|_| Error::Store("Repository SSH private key decryption failed".to_string()))?
+        } else {
+            key
+        };
+        let private_key = key
+            .to_openssh(LineEnding::LF)
+            .map_err(|_| Error::Store("Repository SSH private key encoding failed".to_string()))?;
+        let host = if port == 22 {
+            hostname
+        } else {
+            format!("[{hostname}]:{port}")
+        };
+        Ok(LeasedRepositorySshAccess {
+            credential_id: credential_id.to_string(),
+            credential_revision,
+            host_trust_id: host_trust_id.to_string(),
+            host_trust_revision,
+            private_key,
+            known_hosts_entry: format!("{host} {host_key}\n"),
+        })
+    }
+
+    fn unseal(
+        &self,
+        workspace_id: &str,
+        credential_id: &str,
+        revision: u64,
+        purpose: &str,
+        secret: SealedSecret,
+    ) -> Result<Vec<u8>> {
+        let master_key = self.master_key.as_ref().ok_or_else(|| {
+            Error::Store("Repository secret encryption authority is unavailable".to_string())
+        })?;
+        let unbound = UnboundKey::new(&AES_256_GCM, master_key.as_slice())
+            .map_err(|_| Error::Store("Repository secret encryption key is invalid".to_string()))?;
+        let key = LessSafeKey::new(unbound);
+        let mut plaintext = secret.ciphertext;
+        let aad = secret_aad(workspace_id, credential_id, revision, purpose);
+        let plaintext_len = key
+            .open_in_place(
+                Nonce::assume_unique_for_key(secret.nonce),
+                Aad::from(aad.as_bytes()),
+                &mut plaintext,
+            )
+            .map_err(|_| Error::Store("Repository secret decryption failed".to_string()))?
+            .len();
+        plaintext.truncate(plaintext_len);
+        Ok(plaintext)
+    }
+
     fn seal(
         &self,
         workspace_id: &str,
@@ -968,6 +1155,45 @@ fn insert_secret(
     Ok(())
 }
 
+fn read_sealed_secret(
+    conn: &rusqlite::Connection,
+    workspace_id: &str,
+    credential_id: &str,
+    revision: u64,
+    purpose: &str,
+) -> Result<Option<SealedSecret>> {
+    let row = conn
+        .query_row(
+            r#"SELECT encryption_algorithm, nonce, ciphertext
+               FROM server_secret_versions
+               WHERE workspace_id = ?1 AND secret_id = ?2
+                 AND revision = ?3 AND purpose = ?4"#,
+            params![workspace_id, credential_id, revision, purpose],
+            |row| {
+                Ok((
+                    row.get::<_, String>(0)?,
+                    row.get::<_, Vec<u8>>(1)?,
+                    row.get::<_, Vec<u8>>(2)?,
+                ))
+            },
+        )
+        .optional()?;
+    let Some((algorithm, nonce, ciphertext)) = row else {
+        return Ok(None);
+    };
+    if algorithm != "aes-256-gcm-v1" || nonce.len() != NONCE_BYTES {
+        return Err(Error::RegistryInconsistency(
+            "Repository secret envelope is invalid".to_string(),
+        ));
+    }
+    let mut nonce_bytes = [0u8; NONCE_BYTES];
+    nonce_bytes.copy_from_slice(&nonce);
+    Ok(Some(SealedSecret {
+        nonce: nonce_bytes,
+        ciphertext,
+    }))
+}
+
 fn replay_credential_operation(
     tx: &rusqlite::Transaction<'_>,
     workspace_id: &str,
@@ -1691,6 +1917,29 @@ mod tests {
             projection.bindings[0].access,
             RepositoryAccessMode::ReadOnly
         );
+        let lease = service
+            .lease_ssh_materialization_access("workspace-a", &projection.bindings[0])
+            .unwrap();
+        assert_eq!(lease.credential_revision, 1);
+        assert_eq!(lease.host_trust_revision, 1);
+        assert!(lease.private_key.contains("BEGIN OPENSSH PRIVATE KEY"));
+        assert!(
+            lease
+                .known_hosts_entry
+                .starts_with("example.test ssh-ed25519 ")
+        );
+        let exact = service
+            .lease_ssh_materialization_access_revision(
+                "workspace-a",
+                "deploy",
+                lease.credential_revision,
+                "example",
+                lease.host_trust_revision,
+            )
+            .unwrap();
+        assert_eq!(exact.credential_revision, lease.credential_revision);
+        assert_eq!(exact.host_trust_revision, lease.host_trust_revision);
+        assert_eq!(exact.known_hosts_entry, lease.known_hosts_entry);
 
         let unknown = config_state(
             r#"{

crates/workspace-server/src/resource_broker.rs

@@ -9,7 +9,8 @@ use worker_runtime::resource::{
     BackendResourceClient, BackendResourceError, BackendResourceFetchRequest,
     BackendResourceFetchResponse, BackendResourceHandle, BackendResourceKind,
     BackendResourceOperation, DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
-    PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE, ResourceRedactionPolicy,
+    DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE,
+    REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret, ResourceRedactionPolicy,
 };
 
 #[derive(Clone, Default)]
@@ -29,7 +30,34 @@ struct StoredResource {
     runtime_id: Option<String>,
     worker: Option<RuntimeWorkerRef>,
     handle: BackendResourceHandle,
-    archive: ProfileSourceArchive,
+    bytes: Vec<u8>,
+    archive: Option<ProfileSourceArchive>,
+    one_shot: bool,
+}
+
+impl StoredResource {
+    fn byte_len(&self) -> usize {
+        self.archive
+            .as_ref()
+            .map(|archive| archive.content.len())
+            .unwrap_or_else(|| self.bytes.len())
+    }
+
+    fn take_bytes(&mut self) -> Vec<u8> {
+        self.archive
+            .as_mut()
+            .map(|archive| std::mem::take(&mut archive.content))
+            .unwrap_or_else(|| std::mem::take(&mut self.bytes))
+    }
+}
+
+impl Drop for StoredResource {
+    fn drop(&mut self) {
+        self.bytes.fill(0);
+        if let Some(archive) = self.archive.as_mut() {
+            archive.content.fill(0);
+        }
+    }
 }
 
 impl BackendResourceBroker {
@@ -73,7 +101,9 @@ impl BackendResourceBroker {
             runtime_id,
             worker,
             handle: handle.clone(),
-            archive,
+            bytes: Vec::new(),
+            archive: Some(archive),
+            one_shot: false,
         };
         if let Ok(mut resources) = self.resources.lock() {
             resources.insert(nonce, stored);
@@ -81,6 +111,84 @@ impl BackendResourceBroker {
         handle
     }
 
+    pub fn issue_repository_ssh_access_handle(
+        &self,
+        workspace_id: impl Into<String>,
+        runtime_id: &str,
+        resource_id: impl Into<String>,
+        revision: impl Into<String>,
+        expires_at_unix_seconds: i64,
+        secret: RepositorySshAccessSecret,
+    ) -> Result<BackendResourceHandle, BackendResourceError> {
+        let bytes =
+            serde_json::to_vec(&secret).map_err(|error| BackendResourceError::InvalidResponse {
+                message: error.to_string(),
+            })?;
+        if bytes.len() as u64 > DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES {
+            return Err(BackendResourceError::Oversized {
+                max_bytes: DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES,
+                actual_bytes: bytes.len() as u64,
+            });
+        }
+        let workspace_id = workspace_id.into();
+        let resource_id = resource_id.into();
+        let revision = revision.into();
+        let nonce = Uuid::now_v7().to_string();
+        let handle = BackendResourceHandle {
+            kind: BackendResourceKind::RepositorySshAccess,
+            workspace_id,
+            scope_id: Some("repository-ssh-access".to_string()),
+            runtime_id: Some(runtime_id.to_string()),
+            worker_id: None,
+            resource_id,
+            digest: format!("opaque:{nonce}"),
+            operation: BackendResourceOperation::FetchOnce,
+            expires_at_unix_seconds,
+            nonce: nonce.clone(),
+            revision,
+            generation: None,
+            max_bytes: DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES,
+            content_type: REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
+            redaction: ResourceRedactionPolicy::RuntimeInternalOnly,
+            audit_correlation_id: format!("repository-ssh-access-{nonce}"),
+            profile_source_graph: None,
+        };
+        let stored = StoredResource {
+            runtime_id: Some(runtime_id.to_string()),
+            worker: None,
+            handle: handle.clone(),
+            bytes,
+            archive: None,
+            one_shot: true,
+        };
+        let resource_key = nonce.clone();
+        self.resources
+            .lock()
+            .map_err(|_| BackendResourceError::Transport {
+                message: "resource broker lock poisoned".to_string(),
+            })?
+            .insert(resource_key.clone(), stored);
+        if expires_at_unix_seconds != i64::MAX {
+            let resources = self.resources.clone();
+            std::thread::spawn(move || {
+                let now = Utc::now().timestamp();
+                if expires_at_unix_seconds > now {
+                    std::thread::sleep(std::time::Duration::from_secs(
+                        (expires_at_unix_seconds - now) as u64,
+                    ));
+                }
+                if let Ok(mut resources) = resources.lock()
+                    && resources
+                        .get(&resource_key)
+                        .is_some_and(|stored| stored.handle.nonce == resource_key)
+                {
+                    resources.remove(&resource_key);
+                }
+            });
+        }
+        Ok(handle)
+    }
+
     pub fn profile_source_archive(
         &self,
         digest: &str,
@@ -90,20 +198,21 @@ impl BackendResourceBroker {
             .ok()?
             .values()
             .find(|resource| resource.handle.digest == digest)
-            .map(|resource| resource.archive.clone())
+            .and_then(|resource| resource.archive.clone())
     }
 
-    pub fn fetch_profile_source_archive(
+    pub fn fetch_resource(
         &self,
         request: BackendResourceFetchRequest,
     ) -> Result<BackendResourceFetchResponse, BackendResourceError> {
         verify_handle_shape(&request.handle)?;
-        let stored = self
+        let mut resources = self
             .resources
             .lock()
             .map_err(|_| BackendResourceError::Transport {
                 message: "resource broker lock poisoned".to_string(),
-            })?
+            })?;
+        let mut stored = resources
             .get(&request.handle.nonce)
             .cloned()
             .ok_or(BackendResourceError::MissingResource)?;
@@ -111,7 +220,7 @@ impl BackendResourceBroker {
         if stored.handle.expires_at_unix_seconds < Utc::now().timestamp() {
             return Err(BackendResourceError::Expired);
         }
-        let actual_bytes = stored.archive.content.len() as u64;
+        let actual_bytes = stored.byte_len() as u64;
         if actual_bytes > stored.handle.max_bytes {
             return Err(BackendResourceError::Oversized {
                 max_bytes: stored.handle.max_bytes,
@@ -139,12 +248,15 @@ impl BackendResourceBroker {
                 });
             }
         }
+        if stored.one_shot {
+            resources.remove(&request.handle.nonce);
+        }
         Ok(BackendResourceFetchResponse {
-            kind: BackendResourceKind::ProfileSourceArchive,
-            resource_id: stored.archive.reference.id,
-            digest: stored.archive.reference.digest,
-            content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
-            bytes: stored.archive.content,
+            kind: stored.handle.kind.clone(),
+            resource_id: stored.handle.resource_id.clone(),
+            digest: stored.handle.digest.clone(),
+            content_type: stored.handle.content_type.clone(),
+            bytes: stored.take_bytes(),
             audit_correlation_id: request.audit_correlation_id,
         })
     }
@@ -156,24 +268,38 @@ impl BackendResourceClient for BackendResourceBroker {
         &self,
         request: BackendResourceFetchRequest,
     ) -> Result<BackendResourceFetchResponse, BackendResourceError> {
-        self.fetch_profile_source_archive(request)
+        self.fetch_resource(request)
     }
 }
 
 fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendResourceError> {
-    if handle.kind != BackendResourceKind::ProfileSourceArchive {
-        return Err(BackendResourceError::UnsupportedKind);
-    }
-    if handle.operation != BackendResourceOperation::FetchArchive {
-        return Err(BackendResourceError::Unauthorized {
-            message: "resource handle operation is not fetch_archive".to_string(),
-        });
-    }
-    if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE {
-        return Err(BackendResourceError::ContentTypeMismatch {
-            expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
-            actual: handle.content_type.clone(),
-        });
+    match handle.kind {
+        BackendResourceKind::ProfileSourceArchive => {
+            if handle.operation != BackendResourceOperation::FetchArchive {
+                return Err(BackendResourceError::Unauthorized {
+                    message: "resource handle operation is not fetch_archive".to_string(),
+                });
+            }
+            if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE {
+                return Err(BackendResourceError::ContentTypeMismatch {
+                    expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+                    actual: handle.content_type.clone(),
+                });
+            }
+        }
+        BackendResourceKind::RepositorySshAccess => {
+            if handle.operation != BackendResourceOperation::FetchOnce {
+                return Err(BackendResourceError::Unauthorized {
+                    message: "resource handle operation is not fetch_once".to_string(),
+                });
+            }
+            if handle.content_type != REPOSITORY_SSH_ACCESS_CONTENT_TYPE {
+                return Err(BackendResourceError::ContentTypeMismatch {
+                    expected: REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
+                    actual: handle.content_type.clone(),
+                });
+            }
+        }
     }
     Ok(())
 }
@@ -249,7 +375,7 @@ mod tests {
             archive(),
         );
         let response = broker
-            .fetch_profile_source_archive(BackendResourceFetchRequest {
+            .fetch_resource(BackendResourceFetchRequest {
                 handle: handle.clone(),
                 runtime_id: runtime_id.to_string(),
                 worker_id: None,
@@ -260,6 +386,46 @@ mod tests {
         assert_eq!(response.content_type, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE);
     }
 
+    #[test]
+    fn repository_ssh_access_resource_is_runtime_bound_and_one_shot() {
+        let broker = BackendResourceBroker::default();
+        let handle = broker
+            .issue_repository_ssh_access_handle(
+                "workspace-test",
+                "runtime-test",
+                "repository-access-test",
+                "1",
+                i64::MAX,
+                RepositorySshAccessSecret {
+                    private_key: "private-key-bytes".to_string(),
+                    known_hosts_entry: "known-hosts-entry".to_string(),
+                },
+            )
+            .unwrap();
+        let unauthorized = broker
+            .fetch_resource(request(handle.clone(), "runtime-other", None))
+            .unwrap_err();
+        assert!(matches!(
+            unauthorized,
+            BackendResourceError::Unauthorized { .. }
+        ));
+
+        let response = broker
+            .fetch_resource(request(handle.clone(), "runtime-test", None))
+            .unwrap();
+        assert_eq!(response.kind, BackendResourceKind::RepositorySshAccess);
+        assert_eq!(response.content_type, REPOSITORY_SSH_ACCESS_CONTENT_TYPE);
+        let debug = format!("{response:?}");
+        assert!(!debug.contains("private-key-bytes"));
+        assert!(debug.contains("REDACTED"));
+        let secret: RepositorySshAccessSecret = serde_json::from_slice(&response.bytes).unwrap();
+        assert_eq!(secret.private_key, "private-key-bytes");
+        assert!(matches!(
+            broker.fetch_resource(request(handle, "runtime-test", None)),
+            Err(BackendResourceError::MissingResource)
+        ));
+    }
+
     #[test]
     fn broker_rejects_runtime_mismatch() {
         let broker = BackendResourceBroker::default();
@@ -270,7 +436,7 @@ mod tests {
             archive(),
         );
         let err = broker
-            .fetch_profile_source_archive(request(handle, "runtime-b", None))
+            .fetch_resource(request(handle, "runtime-b", None))
             .unwrap_err();
         assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
     }
@@ -287,7 +453,7 @@ mod tests {
             archive(),
         );
         let err = broker
-            .fetch_profile_source_archive(request(handle, runtime_id, Some(&worker_b.worker_id)))
+            .fetch_resource(request(handle, runtime_id, Some(&worker_b.worker_id)))
             .unwrap_err();
         assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
     }
@@ -312,7 +478,7 @@ mod tests {
         let mut extended = handle;
         extended.expires_at_unix_seconds = 4_102_444_800;
         let err = broker
-            .fetch_profile_source_archive(request(extended, &runtime_id, None))
+            .fetch_resource(request(extended, &runtime_id, None))
             .unwrap_err();
         assert!(matches!(err, BackendResourceError::Expired));
     }
@@ -328,7 +494,7 @@ mod tests {
         );
         handle.scope_id = Some("tampered-scope".to_string());
         let err = broker
-            .fetch_profile_source_archive(request(handle, &runtime_id, None))
+            .fetch_resource(request(handle, &runtime_id, None))
             .unwrap_err();
         assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
     }
@@ -345,7 +511,7 @@ mod tests {
         );
         handle.max_bytes = DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1024;
         let err = broker
-            .fetch_profile_source_archive(request(handle, &runtime_id, None))
+            .fetch_resource(request(handle, &runtime_id, None))
             .unwrap_err();
         assert!(matches!(err, BackendResourceError::Oversized { .. }));
     }

crates/workspace-server/src/store.rs

@@ -257,6 +257,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "create Workspace Repository SSH secret authority",
         apply: create_repository_ssh_secret_authority,
     },
+    Migration {
+        version: 47,
+        name: "bind Workdir create repository access evidence",
+        apply: bind_workdir_create_repository_access_evidence,
+    },
 ];
 
 struct Migration {
@@ -590,6 +595,16 @@ pub struct WorkdirCreateOperationRecord {
     pub resolved_runtime_id: String,
     pub config_revision: u64,
     pub config_projection_digest: String,
+    pub source_kind: Option<String>,
+    pub source_uri: Option<String>,
+    pub source_revision: Option<u64>,
+    pub source_fingerprint: Option<String>,
+    pub credential_id: Option<String>,
+    pub credential_revision: Option<u64>,
+    pub host_trust_id: Option<String>,
+    pub host_trust_revision: Option<u64>,
+    pub repository_access_mode: Option<String>,
+    pub cache_generation: u64,
     pub working_directory_id: String,
     pub state: String,
     pub failure: Option<String>,
@@ -605,8 +620,11 @@ pub struct WorkdirRegistryRecord {
     pub repository_id: String,
     pub creation_selector: Option<String>,
     pub creation_ref: Option<String>,
+    pub creation_tree: Option<String>,
     pub current_selector: Option<String>,
     pub current_ref: Option<String>,
+    pub current_tree: Option<String>,
+    pub observed_at_epoch_seconds: Option<u64>,
     pub materialization_status: String,
     pub cleanliness: String,
     pub created_at: String,
@@ -4639,16 +4657,20 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
             conn.execute(
                 r#"INSERT INTO workdir_registry (
                     workspace_id, workdir_id, runtime_id, repository_id,
-                    creation_selector, creation_ref, current_selector, current_ref,
+                    creation_selector, creation_ref, creation_tree,
+                    current_selector, current_ref, current_tree, observed_at_epoch_seconds,
                     materialization_status, cleanliness, created_at, updated_at
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)
                 ON CONFLICT(workspace_id, workdir_id) DO UPDATE SET
                     runtime_id = excluded.runtime_id,
                     repository_id = excluded.repository_id,
                     creation_selector = excluded.creation_selector,
                     creation_ref = excluded.creation_ref,
+                    creation_tree = excluded.creation_tree,
                     current_selector = excluded.current_selector,
                     current_ref = excluded.current_ref,
+                    current_tree = excluded.current_tree,
+                    observed_at_epoch_seconds = excluded.observed_at_epoch_seconds,
                     materialization_status = excluded.materialization_status,
                     cleanliness = excluded.cleanliness,
                     updated_at = excluded.updated_at"#,
@@ -4659,8 +4681,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                     record.repository_id,
                     record.creation_selector,
                     record.creation_ref,
+                    record.creation_tree,
                     record.current_selector,
                     record.current_ref,
+                    record.current_tree,
+                    record.observed_at_epoch_seconds.map(|value| value as i64),
                     record.materialization_status,
                     record.cleanliness,
                     record.created_at,
@@ -5880,7 +5905,8 @@ fn require_expected_ticket_assignment(
 fn workdir_registry_select_sql(where_clause: &str) -> String {
     format!(
         "SELECT workspace_id, workdir_id, runtime_id, repository_id, \
-         creation_selector, creation_ref, current_selector, current_ref, \
+         creation_selector, creation_ref, creation_tree, \
+         current_selector, current_ref, current_tree, observed_at_epoch_seconds, \
          materialization_status, cleanliness, created_at, updated_at \
          FROM workdir_registry {where_clause}"
     )
@@ -5896,12 +5922,15 @@ fn read_workdir_registry_record(
         repository_id: row.get(3)?,
         creation_selector: row.get(4)?,
         creation_ref: row.get(5)?,
-        current_selector: row.get(6)?,
-        current_ref: row.get(7)?,
-        materialization_status: row.get(8)?,
-        cleanliness: row.get(9)?,
-        created_at: row.get(10)?,
-        updated_at: row.get(11)?,
+        creation_tree: row.get(6)?,
+        current_selector: row.get(7)?,
+        current_ref: row.get(8)?,
+        current_tree: row.get(9)?,
+        observed_at_epoch_seconds: row.get::<_, Option<i64>>(10)?.map(|value| value as u64),
+        materialization_status: row.get(11)?,
+        cleanliness: row.get(12)?,
+        created_at: row.get(13)?,
+        updated_at: row.get(14)?,
     })
 }
 
@@ -6839,6 +6868,28 @@ fn create_repository_ssh_secret_authority(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+fn bind_workdir_create_repository_access_evidence(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+        ALTER TABLE workdir_registry ADD COLUMN creation_tree TEXT;
+        ALTER TABLE workdir_registry ADD COLUMN current_tree TEXT;
+        ALTER TABLE workdir_registry ADD COLUMN observed_at_epoch_seconds INTEGER;
+        ALTER TABLE workdir_create_operations ADD COLUMN source_kind TEXT;
+        ALTER TABLE workdir_create_operations ADD COLUMN source_uri TEXT;
+        ALTER TABLE workdir_create_operations ADD COLUMN source_revision INTEGER;
+        ALTER TABLE workdir_create_operations ADD COLUMN source_fingerprint TEXT;
+        ALTER TABLE workdir_create_operations ADD COLUMN credential_id TEXT;
+        ALTER TABLE workdir_create_operations ADD COLUMN credential_revision INTEGER;
+        ALTER TABLE workdir_create_operations ADD COLUMN host_trust_id TEXT;
+        ALTER TABLE workdir_create_operations ADD COLUMN host_trust_revision INTEGER;
+        ALTER TABLE workdir_create_operations ADD COLUMN repository_access_mode TEXT;
+        ALTER TABLE workdir_create_operations
+            ADD COLUMN cache_generation INTEGER NOT NULL DEFAULT 0;
+        "#,
+    )?;
+    Ok(())
+}
+
 fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> {
     conn.execute_batch(
         r#"
@@ -9724,7 +9775,7 @@ mod tests {
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         let remote = conn
             .query_row(
                 "SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \
@@ -9802,7 +9853,7 @@ mod tests {
         let before = std::fs::read(&path).unwrap();
         let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
         assert_eq!(plan.current_schema_version, 36);
-        assert_eq!(plan.target_schema_version, 46);
+        assert_eq!(plan.target_schema_version, 47);
         assert!(plan.migration_required);
         assert_eq!(plan.worker_count, 1);
         assert_eq!(plan.mappings[0].legacy_worker_id, 7);
@@ -9816,7 +9867,7 @@ mod tests {
         store
             .with_conn(|conn| {
                 assert!(table_exists(conn, "worker_diagnostics_archives")?);
-                assert_eq!(current_schema_version(conn)?, 46);
+                assert_eq!(current_schema_version(conn)?, 47);
                 Ok(())
             })
             .unwrap();
@@ -9952,7 +10003,7 @@ mod tests {
                 ),
             ]
         );
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         let foreign_key_error: Option<String> = conn
             .query_row("PRAGMA foreign_key_check", [], |row| row.get(0))
             .optional()
@@ -10081,7 +10132,7 @@ INSERT INTO worker_orphan_diagnostics (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
         let controller_worker_id: String = conn
             .query_row(
@@ -10199,7 +10250,7 @@ INSERT INTO worker_orphan_diagnostics (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
     }
 
@@ -10217,7 +10268,7 @@ INSERT INTO worker_orphan_diagnostics (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         let settings = conn
             .query_row(
                 "SELECT settings_revision, language FROM workspace_memory_settings \
@@ -10258,7 +10309,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         assert!(table_exists(&conn, "flow_sources").unwrap());
         assert!(table_exists(&conn, "flow_source_revisions").unwrap());
         assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -10325,7 +10376,7 @@ INSERT INTO worker_workdir_attachment_reservations (
 
         apply_migrations(&conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         let repositories_sql: String = conn
             .query_row(
                 "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -10508,7 +10559,7 @@ INSERT INTO workdir_registry (
         let db = dir.path().join("control-plane.sqlite");
         let store = SqliteWorkspaceStore::open(&db).unwrap();
 
-        assert_eq!(store.schema_version().await.unwrap(), 46);
+        assert_eq!(store.schema_version().await.unwrap(), 47);
         assert!(
             !store
                 .with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -10525,7 +10576,7 @@ INSERT INTO workdir_registry (
         store.upsert_workspace(&record).await.unwrap();
 
         let reopened = SqliteWorkspaceStore::open(&db).unwrap();
-        assert_eq!(reopened.schema_version().await.unwrap(), 46);
+        assert_eq!(reopened.schema_version().await.unwrap(), 47);
         assert_eq!(
             reopened.get_workspace("local-dev").await.unwrap(),
             Some(record)
@@ -11290,7 +11341,7 @@ INSERT INTO worker_registry (
         let migrated = SqliteWorkspaceStore::open(&db_path).unwrap();
         migrated
             .with_conn(|conn| {
-                assert_eq!(current_schema_version(conn)?, 46);
+                assert_eq!(current_schema_version(conn)?, 47);
                 assert_eq!(
                     conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?,
                     1,
@@ -11647,13 +11698,16 @@ INSERT INTO worker_registry (
              DROP TABLE repository_ssh_host_trust_revisions;
              DROP TABLE repository_ssh_host_trusts;
              DROP TABLE workdir_create_operations;
-             DELETE FROM __yoi_schema_migrations WHERE version IN (45, 46);",
+             ALTER TABLE workdir_registry DROP COLUMN creation_tree;
+             ALTER TABLE workdir_registry DROP COLUMN current_tree;
+             ALTER TABLE workdir_registry DROP COLUMN observed_at_epoch_seconds;
+             DELETE FROM __yoi_schema_migrations WHERE version IN (45, 46, 47);",
         )
         .unwrap();
         assert_eq!(current_schema_version(&conn).unwrap(), 44);
 
         apply_migrations(&conn).unwrap();
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         assert!(table_exists(&conn, "workdir_create_operations").unwrap());
         let columns = table_columns(&conn, "workdir_create_operations").unwrap();
         for required in [
@@ -11685,13 +11739,26 @@ INSERT INTO worker_registry (
              DROP TABLE repository_ssh_credentials;
              DROP TABLE repository_ssh_host_trust_revisions;
              DROP TABLE repository_ssh_host_trusts;
-             DELETE FROM __yoi_schema_migrations WHERE version = 46;",
+             ALTER TABLE workdir_registry DROP COLUMN creation_tree;
+             ALTER TABLE workdir_registry DROP COLUMN current_tree;
+             ALTER TABLE workdir_registry DROP COLUMN observed_at_epoch_seconds;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_kind;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_uri;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_revision;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_fingerprint;
+             ALTER TABLE workdir_create_operations DROP COLUMN credential_id;
+             ALTER TABLE workdir_create_operations DROP COLUMN credential_revision;
+             ALTER TABLE workdir_create_operations DROP COLUMN host_trust_id;
+             ALTER TABLE workdir_create_operations DROP COLUMN host_trust_revision;
+             ALTER TABLE workdir_create_operations DROP COLUMN repository_access_mode;
+             ALTER TABLE workdir_create_operations DROP COLUMN cache_generation;
+             DELETE FROM __yoi_schema_migrations WHERE version IN (46, 47);",
         )
         .unwrap();
         assert_eq!(current_schema_version(&conn).unwrap(), 45);
 
         apply_migrations(&conn).unwrap();
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         for table in [
             "repository_ssh_credentials",
             "repository_ssh_credential_revisions",
@@ -11710,19 +11777,72 @@ INSERT INTO worker_registry (
         assert!(foreign_key_error.is_none());
     }
 
+    #[test]
+    fn schema_v47_binds_workdir_create_repository_access_evidence() {
+        let conn = Connection::open_in_memory().unwrap();
+        configure_sqlite(&conn).unwrap();
+        apply_migrations(&conn).unwrap();
+        conn.execute_batch(
+            "ALTER TABLE workdir_registry DROP COLUMN creation_tree;
+             ALTER TABLE workdir_registry DROP COLUMN current_tree;
+             ALTER TABLE workdir_registry DROP COLUMN observed_at_epoch_seconds;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_kind;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_uri;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_revision;
+             ALTER TABLE workdir_create_operations DROP COLUMN source_fingerprint;
+             ALTER TABLE workdir_create_operations DROP COLUMN credential_id;
+             ALTER TABLE workdir_create_operations DROP COLUMN credential_revision;
+             ALTER TABLE workdir_create_operations DROP COLUMN host_trust_id;
+             ALTER TABLE workdir_create_operations DROP COLUMN host_trust_revision;
+             ALTER TABLE workdir_create_operations DROP COLUMN repository_access_mode;
+             ALTER TABLE workdir_create_operations DROP COLUMN cache_generation;
+             DELETE FROM __yoi_schema_migrations WHERE version = 47;",
+        )
+        .unwrap();
+        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+
+        apply_migrations(&conn).unwrap();
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
+        let columns = table_columns(&conn, "workdir_create_operations").unwrap();
+        for required in [
+            "source_kind",
+            "source_uri",
+            "source_revision",
+            "source_fingerprint",
+            "credential_id",
+            "credential_revision",
+            "host_trust_id",
+            "host_trust_revision",
+            "repository_access_mode",
+            "cache_generation",
+        ] {
+            assert!(
+                columns.iter().any(|column| column == required),
+                "missing column {required}"
+            );
+        }
+        let workdir_columns = table_columns(&conn, "workdir_registry").unwrap();
+        for required in ["creation_tree", "current_tree", "observed_at_epoch_seconds"] {
+            assert!(
+                workdir_columns.iter().any(|column| column == required),
+                "missing column {required}"
+            );
+        }
+    }
+
     #[test]
     fn server_refuses_a_database_from_a_newer_schema_generation() {
         let conn = Connection::open_in_memory().unwrap();
         configure_sqlite(&conn).unwrap();
         apply_migrations(&conn).unwrap();
         conn.execute(
-            "INSERT INTO __yoi_schema_migrations (version, name) VALUES (47, 'future')",
+            "INSERT INTO __yoi_schema_migrations (version, name) VALUES (48, 'future')",
             [],
         )
         .unwrap();
 
         let error = apply_migrations(&conn).unwrap_err().to_string();
-        assert!(error.contains("schema version 47 is newer"), "{error}");
+        assert!(error.contains("schema version 48 is newer"), "{error}");
         assert!(error.contains("refusing to serve"), "{error}");
     }
 
@@ -11943,7 +12063,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
 
         apply_migrations(&mut conn).unwrap();
 
-        assert_eq!(current_schema_version(&conn).unwrap(), 46);
+        assert_eq!(current_schema_version(&conn).unwrap(), 47);
         let workspace_id: Option<String> = conn
             .query_row(
                 "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'",
@@ -12488,6 +12608,9 @@ WHERE workspace_id = 'workspace-a'
                 "updated_at",
                 "current_selector",
                 "current_ref",
+                "creation_tree",
+                "current_tree",
+                "observed_at_epoch_seconds",
             ],
         );
         assert_columns(
@@ -12566,7 +12689,7 @@ WHERE workspace_id = 'workspace-a'
         .unwrap();
 
         let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 46);
+        assert_eq!(store.schema_version().await.unwrap(), 47);
 
         store
             .with_conn(|conn| {
@@ -12755,7 +12878,7 @@ CREATE TABLE ticket_assignment_operations (
     #[tokio::test]
     async fn repository_records_round_trip() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 46);
+        assert_eq!(store.schema_version().await.unwrap(), 47);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -12833,7 +12956,7 @@ CREATE TABLE ticket_assignment_operations (
     #[tokio::test]
     async fn memory_authority_records_round_trip_and_close_staging() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 46);
+        assert_eq!(store.schema_version().await.unwrap(), 47);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -12979,8 +13102,11 @@ CREATE TABLE ticket_assignment_operations (
             repository_id: "repo".to_string(),
             creation_selector: Some("develop".to_string()),
             creation_ref: Some("abcdef".to_string()),
+            creation_tree: Some("tree-creation".to_string()),
             current_selector: None,
             current_ref: Some("abcdef".to_string()),
+            current_tree: Some("tree-current".to_string()),
+            observed_at_epoch_seconds: Some(1_777_777_777),
             materialization_status: "not_found".to_string(),
             cleanliness: "clean".to_string(),
             created_at: "2".to_string(),
@@ -12994,8 +13120,11 @@ CREATE TABLE ticket_assignment_operations (
             repository_id: "repo".to_string(),
             creation_selector: Some("feature".to_string()),
             creation_ref: Some("123456".to_string()),
+            creation_tree: None,
             current_selector: Some("feature".to_string()),
             current_ref: Some("123456".to_string()),
+            current_tree: None,
+            observed_at_epoch_seconds: None,
             materialization_status: "present".to_string(),
             cleanliness: "unknown".to_string(),
             created_at: "3".to_string(),
@@ -13240,7 +13369,7 @@ CREATE TABLE ticket_assignment_operations (
     #[tokio::test]
     async fn account_and_login_records_round_trip() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 46);
+        assert_eq!(store.schema_version().await.unwrap(), 47);
         let now = "2026-07-22T00:00:00Z".to_string();
         let account = AccountRecord {
             account_id: "acct-user-alice".to_string(),

crates/workspace-server/src/workdir_create_operations.rs

@@ -4,13 +4,31 @@ use sha2::{Digest, Sha256};
 use crate::store::WorkdirCreateOperationRecord;
 use crate::{Error, Result, SqliteWorkspaceStore};
 
+pub fn selector_for_retry(
+    explicit_selector: Option<&str>,
+    persisted_selector: Option<&str>,
+    current_default_selector: Option<&str>,
+) -> Option<String> {
+    explicit_selector
+        .or(persisted_selector)
+        .or(current_default_selector)
+        .map(str::to_string)
+}
+
 pub fn request_fingerprint(
     repository_id: &str,
     selector: Option<&str>,
     requested_runtime_id: Option<&str>,
+    repository_source_fingerprint: &str,
+    repository_source_revision: u64,
 ) -> String {
     let mut hasher = Sha256::new();
-    for value in [Some(repository_id), selector, requested_runtime_id] {
+    for value in [
+        Some(repository_id),
+        selector,
+        requested_runtime_id,
+        Some(repository_source_fingerprint),
+    ] {
         match value {
             Some(value) => {
                 hasher.update([1]);
@@ -20,6 +38,7 @@ pub fn request_fingerprint(
             None => hasher.update([0]),
         }
     }
+    hasher.update(repository_source_revision.to_be_bytes());
     let digest = hasher.finalize();
     let mut encoded = String::with_capacity(digest.len() * 2);
     for byte in digest {
@@ -40,9 +59,10 @@ impl SqliteWorkspaceStore {
                 r#"INSERT OR IGNORE INTO workdir_create_operations (
                     workspace_id, operation_id, request_fingerprint, repository_id, selector,
                     requested_runtime_id, resolved_runtime_id, config_revision,
-                    config_projection_digest, working_directory_id, state, failure,
+                    config_projection_digest, source_kind, source_uri, source_revision,
+                    source_fingerprint, working_directory_id, state, failure,
                     created_at, updated_at
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#,
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)"#,
                 params![
                     record.workspace_id,
                     record.operation_id,
@@ -53,6 +73,10 @@ impl SqliteWorkspaceStore {
                     record.resolved_runtime_id,
                     record.config_revision as i64,
                     record.config_projection_digest,
+                    record.source_kind,
+                    record.source_uri,
+                    record.source_revision.map(|revision| revision as i64),
+                    record.source_fingerprint,
                     record.working_directory_id,
                     record.state,
                     record.failure,
@@ -79,6 +103,81 @@ impl SqliteWorkspaceStore {
         })
     }
 
+    pub fn bind_workdir_create_repository_access(
+        &self,
+        workspace_id: &str,
+        operation_id: &str,
+        request_fingerprint: &str,
+        credential_id: &str,
+        credential_revision: u64,
+        host_trust_id: &str,
+        host_trust_revision: u64,
+        repository_access_mode: &str,
+        cache_generation: u64,
+        now: &str,
+    ) -> Result<WorkdirCreateOperationRecord> {
+        self.with_conn_mut(|conn| {
+            let operation = read_workdir_create_operation(conn, workspace_id, operation_id)?
+                .ok_or_else(|| {
+                    Error::RegistryInconsistency(format!(
+                        "Workdir create operation `{operation_id}` disappeared before Repository access binding"
+                    ))
+                })?;
+            if operation.request_fingerprint != request_fingerprint {
+                return Err(Error::InvalidInput(format!(
+                    "Workdir create operation `{operation_id}` was reused with different input"
+                )));
+            }
+            if let Some(existing) = operation.credential_id.as_deref() {
+                if existing != credential_id
+                    || operation.credential_revision != Some(credential_revision)
+                    || operation.host_trust_id.as_deref() != Some(host_trust_id)
+                    || operation.host_trust_revision != Some(host_trust_revision)
+                    || operation.repository_access_mode.as_deref()
+                        != Some(repository_access_mode)
+                    || operation.cache_generation != cache_generation
+                {
+                    return Err(Error::InvalidInput(format!(
+                        "Workdir create operation `{operation_id}` Repository access evidence changed"
+                    )));
+                }
+                return Ok(operation);
+            }
+            conn.execute(
+                r#"UPDATE workdir_create_operations
+                   SET credential_id = ?4, credential_revision = ?5,
+                       host_trust_id = ?6, host_trust_revision = ?7,
+                       repository_access_mode = ?8, cache_generation = ?9,
+                       updated_at = ?10
+                   WHERE workspace_id = ?1 AND operation_id = ?2
+                     AND request_fingerprint = ?3 AND credential_id IS NULL"#,
+                params![
+                    workspace_id,
+                    operation_id,
+                    request_fingerprint,
+                    credential_id,
+                    i64::try_from(credential_revision).map_err(|_| Error::InvalidInput(
+                        "credential revision is out of range".to_string()
+                    ))?,
+                    host_trust_id,
+                    i64::try_from(host_trust_revision).map_err(|_| Error::InvalidInput(
+                        "host-trust revision is out of range".to_string()
+                    ))?,
+                    repository_access_mode,
+                    i64::try_from(cache_generation).map_err(|_| Error::InvalidInput(
+                        "cache generation is out of range".to_string()
+                    ))?,
+                    now,
+                ],
+            )?;
+            read_workdir_create_operation(conn, workspace_id, operation_id)?.ok_or_else(|| {
+                Error::RegistryInconsistency(format!(
+                    "Workdir create operation `{operation_id}` disappeared after Repository access binding"
+                ))
+            })
+        })
+    }
+
     pub fn finish_workdir_create_operation(
         &self,
         workspace_id: &str,
@@ -133,7 +232,10 @@ fn read_workdir_create_operation(
     conn.query_row(
         r#"SELECT workspace_id, operation_id, request_fingerprint, repository_id, selector,
                   requested_runtime_id, resolved_runtime_id, config_revision,
-                  config_projection_digest, working_directory_id, state, failure,
+                  config_projection_digest, source_kind, source_uri, source_revision,
+                  source_fingerprint, credential_id, credential_revision,
+                  host_trust_id, host_trust_revision, repository_access_mode,
+                  cache_generation, working_directory_id, state, failure,
                   created_at, updated_at
            FROM workdir_create_operations
            WHERE workspace_id = ?1 AND operation_id = ?2"#,
@@ -149,11 +251,21 @@ fn read_workdir_create_operation(
                 resolved_runtime_id: row.get(6)?,
                 config_revision: row.get::<_, i64>(7)? as u64,
                 config_projection_digest: row.get(8)?,
-                working_directory_id: row.get(9)?,
-                state: row.get(10)?,
-                failure: row.get(11)?,
-                created_at: row.get(12)?,
-                updated_at: row.get(13)?,
+                source_kind: row.get(9)?,
+                source_uri: row.get(10)?,
+                source_revision: row.get::<_, Option<i64>>(11)?.map(|value| value as u64),
+                source_fingerprint: row.get(12)?,
+                credential_id: row.get(13)?,
+                credential_revision: row.get::<_, Option<i64>>(14)?.map(|value| value as u64),
+                host_trust_id: row.get(15)?,
+                host_trust_revision: row.get::<_, Option<i64>>(16)?.map(|value| value as u64),
+                repository_access_mode: row.get(17)?,
+                cache_generation: row.get::<_, i64>(18)? as u64,
+                working_directory_id: row.get(19)?,
+                state: row.get(20)?,
+                failure: row.get(21)?,
+                created_at: row.get(22)?,
+                updated_at: row.get(23)?,
             })
         },
     )
@@ -166,6 +278,22 @@ mod tests {
     use super::*;
     use crate::store::{ControlPlaneStore, RepositoryRecord, WorkspaceRecord};
 
+    #[test]
+    fn retry_selector_keeps_persisted_default_but_honors_explicit_input() {
+        assert_eq!(
+            selector_for_retry(None, Some("develop"), Some("main")),
+            Some("develop".to_string())
+        );
+        assert_eq!(
+            selector_for_retry(Some("release"), Some("develop"), Some("main")),
+            Some("release".to_string())
+        );
+        assert_eq!(
+            selector_for_retry(None, None, Some("main")),
+            Some("main".to_string())
+        );
+    }
+
     #[test]
     fn retry_keeps_resolved_config_evidence_and_rejects_changed_input() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
@@ -201,13 +329,29 @@ mod tests {
         let record = WorkdirCreateOperationRecord {
             workspace_id: "workspace".to_string(),
             operation_id: "call-1".to_string(),
-            request_fingerprint: request_fingerprint("main", Some("develop"), None),
+            request_fingerprint: request_fingerprint(
+                "main",
+                Some("develop"),
+                None,
+                "sha256:test",
+                1,
+            ),
             repository_id: "main".to_string(),
             selector: Some("develop".to_string()),
             requested_runtime_id: None,
             resolved_runtime_id: "arcadia".to_string(),
             config_revision: 7,
             config_projection_digest: "sha256:projection".to_string(),
+            source_kind: Some("local_path".to_string()),
+            source_uri: Some("/tmp/repo".to_string()),
+            source_revision: Some(1),
+            source_fingerprint: Some("sha256:source".to_string()),
+            credential_id: None,
+            credential_revision: None,
+            host_trust_id: None,
+            host_trust_revision: None,
+            repository_access_mode: None,
+            cache_generation: 0,
             working_directory_id: "wd-1".to_string(),
             state: "pending".to_string(),
             failure: None,
@@ -218,23 +362,59 @@ mod tests {
             store.reserve_workdir_create_operation(&record).unwrap(),
             record
         );
+        let bound = store
+            .bind_workdir_create_repository_access(
+                "workspace",
+                "call-1",
+                &record.request_fingerprint,
+                "credential-1",
+                3,
+                "trust-1",
+                5,
+                "read_only",
+                2,
+                "2026-08-24T00:00:01Z",
+            )
+            .unwrap();
+        assert_eq!(bound.credential_id.as_deref(), Some("credential-1"));
+        assert_eq!(bound.credential_revision, Some(3));
+        assert_eq!(bound.host_trust_revision, Some(5));
+        assert_eq!(bound.cache_generation, 2);
+        assert!(
+            store
+                .bind_workdir_create_repository_access(
+                    "workspace",
+                    "call-1",
+                    &record.request_fingerprint,
+                    "credential-1",
+                    4,
+                    "trust-1",
+                    5,
+                    "read_only",
+                    2,
+                    "2026-08-24T00:00:02Z",
+                )
+                .is_err()
+        );
         let mut changed_resolution = record.clone();
         changed_resolution.resolved_runtime_id = "other".to_string();
         changed_resolution.config_revision = 8;
-        assert_eq!(
-            store
-                .reserve_workdir_create_operation(&changed_resolution)
-                .unwrap(),
-            record
-        );
+        changed_resolution.source_uri = Some("ssh://git@other.test/repo.git".to_string());
+        changed_resolution.source_revision = Some(9);
+        let replayed = store
+            .reserve_workdir_create_operation(&changed_resolution)
+            .unwrap();
+        assert_eq!(replayed, bound);
+        assert_eq!(replayed.source_uri.as_deref(), Some("/tmp/repo"));
         assert_eq!(
             store
                 .load_workdir_create_operation("workspace", "call-1")
                 .unwrap(),
-            Some(record.clone())
+            Some(bound.clone())
         );
         let mut changed_input = record.clone();
-        changed_input.request_fingerprint = request_fingerprint("main", Some("main"), None);
+        changed_input.request_fingerprint =
+            request_fingerprint("main", Some("main"), None, "sha256:test", 1);
         assert!(
             store
                 .reserve_workdir_create_operation(&changed_input)