From 471db64bcc686ad02c338b3219792c3deb736729 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 26 Aug 2026 15:21:52 +0900 Subject: [PATCH] fix: preserve repository access across retries --- .../worker-runtime/src/working_directory.rs | 61 ++++-- .../workspace-server/src/repository_access.rs | 92 ++++++-- crates/workspace-server/src/server.rs | 200 +++++++++++++++++- crates/workspace-server/src/store.rs | 133 ++++++++++-- .../src/workdir_create_operations.rs | 168 +++++++++++++-- 5 files changed, 570 insertions(+), 84 deletions(-) diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index f2407a6e..79b7817b 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -402,12 +402,13 @@ impl RuntimeGitCacheMaterializer { return Ok(binding); }; validate_ssh_materialization_access(&access)?; - let agent = Arc::new(RepositorySshAgent::start( + let command_access = Arc::new(RepositoryCommandAccess::prepare_ssh( &self.runtime_root, - working_directory_id, + &format!("attachment-{working_directory_id}"), + &binding.working_directory.repository_id, &access, )?); - let weak_agent = Arc::downgrade(&agent); + let weak_access = Arc::downgrade(&command_access); let expires_at = access.expires_at_epoch_seconds; std::thread::spawn(move || { let now = SystemTime::now() @@ -417,13 +418,17 @@ impl RuntimeGitCacheMaterializer { if expires_at > now { std::thread::sleep(Duration::from_secs(expires_at - now)); } - if let Some(agent) = weak_agent.upgrade() { - agent.stop(); + if let Some(access) = weak_access.upgrade() { + access.stop(); } }); binding.command_environment.insert( "SSH_AUTH_SOCK".to_string(), - agent.socket.to_string_lossy().to_string(), + command_access.agent.socket.to_string_lossy().to_string(), + ); + binding.command_environment.insert( + "GIT_SSH_COMMAND".to_string(), + command_access.ssh_command.to_string_lossy().to_string(), ); binding.command_environment.insert( "YOI_REPOSITORY_ACCESS".to_string(), @@ -433,7 +438,7 @@ impl RuntimeGitCacheMaterializer { } .to_string(), ); - binding.session_resources.push(agent); + binding.session_resources.push(command_access); Ok(binding) } @@ -1122,6 +1127,7 @@ impl Drop for RepositorySshAgent { } } +#[derive(Debug)] struct RepositoryCommandAccess { root: PathBuf, ssh_command: PathBuf, @@ -1148,10 +1154,24 @@ impl RepositoryCommandAccess { .as_ref() .map(|materialization| materialization.operation_id.as_str()) .unwrap_or("operation"); + Ok(Some(Self::prepare_ssh( + runtime_root, + operation_id, + &request.repository.id, + ssh, + )?)) + } + + fn prepare_ssh( + runtime_root: &Path, + operation_id: &str, + repository_id: &str, + ssh: &RepositorySshMaterializationAccess, + ) -> Result { let root = runtime_root.join(REPOSITORY_ACCESS_DIR).join(format!( "{}-{}", sanitize_path_component(operation_id), - next_working_directory_id(&request.repository.id) + next_working_directory_id(repository_id) )); fs::create_dir_all(&root).map_err(|_| { WorkingDirectoryDiagnostic::new( @@ -1170,17 +1190,22 @@ impl RepositoryCommandAccess { write_owner_only(&ssh_command, script.as_bytes())?; set_file_owner_executable(&ssh_command)?; let agent = RepositorySshAgent::start(runtime_root, operation_id, ssh)?; - Ok(Some(Self { + Ok(Self { root, ssh_command, agent, - })) + }) + } + + fn stop(&self) { + self.agent.stop(); + let _ = fs::remove_dir_all(&self.root); } } impl Drop for RepositoryCommandAccess { fn drop(&mut self) { - let _ = fs::remove_dir_all(&self.root); + self.stop(); } } @@ -2006,14 +2031,18 @@ mod tests { }) .unwrap(); let binding = materializer.bind_working_directory(&id, None).unwrap(); - let socket = PathBuf::from(binding.command_environment()["SSH_AUTH_SOCK"].clone()); + let environment = binding.command_environment(); + let socket = PathBuf::from(environment["SSH_AUTH_SOCK"].clone()); + let ssh_command = PathBuf::from(environment["GIT_SSH_COMMAND"].clone()); assert!(socket.exists()); - assert_eq!( - binding.command_environment()["YOI_REPOSITORY_ACCESS"], - "read_write" - ); + assert!(ssh_command.exists()); + let ssh_policy = fs::read_to_string(&ssh_command).unwrap(); + assert!(ssh_policy.contains("StrictHostKeyChecking=yes")); + assert!(ssh_policy.contains("UserKnownHostsFile=")); + assert_eq!(environment["YOI_REPOSITORY_ACCESS"], "read_write"); drop(binding); assert!(!socket.exists()); + assert!(!ssh_command.exists()); assert_eq!( materializer .bind_working_directory(&id, None) diff --git a/crates/workspace-server/src/repository_access.rs b/crates/workspace-server/src/repository_access.rs index 9736a07c..9c642ac9 100644 --- a/crates/workspace-server/src/repository_access.rs +++ b/crates/workspace-server/src/repository_access.rs @@ -865,33 +865,73 @@ impl RepositorySecretService { binding.host_trust_id )) })?; - let (private_key, passphrase) = self.store.with_conn(|conn| { + 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 { + let (private_key, passphrase, hostname, port, host_key) = self.store.with_conn(|conn| { let private_key = read_sealed_secret( conn, workspace_id, - &binding.credential_id, - credential.current_revision, + credential_id, + credential_revision, "private_key", )? .ok_or_else(|| { Error::RegistryInconsistency(format!( - "Repository SSH credential `{}` is missing its private-key revision", - binding.credential_id + "Repository SSH credential `{credential_id}` revision {credential_revision} is unavailable" )) })?; let passphrase = read_sealed_secret( conn, workspace_id, - &binding.credential_id, - credential.current_revision, + credential_id, + credential_revision, "passphrase", )?; - Ok((private_key, 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, - &binding.credential_id, - credential.current_revision, + credential_id, + credential_revision, "private_key", private_key, )?; @@ -899,8 +939,8 @@ impl RepositorySecretService { .map(|secret| { self.unseal( workspace_id, - &binding.credential_id, - credential.current_revision, + credential_id, + credential_revision, "passphrase", secret, ) @@ -933,18 +973,18 @@ impl RepositorySecretService { let private_key = key .to_openssh(LineEnding::LF) .map_err(|_| Error::Store("Repository SSH private key encoding failed".to_string()))?; - let host = if host_trust.port == 22 { - host_trust.hostname.clone() + let host = if port == 22 { + hostname } else { - format!("[{}]:{}", host_trust.hostname, host_trust.port) + format!("[{hostname}]:{port}") }; Ok(LeasedRepositorySshAccess { - credential_id: binding.credential_id.clone(), - credential_revision: credential.current_revision, - host_trust_id: binding.host_trust_id.clone(), - host_trust_revision: host_trust.current_revision, + 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} {}\n", host_trust.host_key), + known_hosts_entry: format!("{host} {host_key}\n"), }) } @@ -1888,6 +1928,18 @@ mod tests { .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#"{ diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 660b9245..bdc953df 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -8474,6 +8474,34 @@ async fn create_workspace_working_directory( ) .into()); } + if let Some(existing) = api + .config_store + .load_workdir_create_operation(workspace_id, &operation_id)? + && let (Some(kind), Some(uri), Some(revision), Some(fingerprint)) = ( + existing.source_kind.as_deref(), + existing.source_uri, + existing.source_revision, + existing.source_fingerprint, + ) + { + let kind = match kind { + "local_path" => workspace_api::RepositorySourceKind::LocalPath, + "file" => workspace_api::RepositorySourceKind::File, + "https" => workspace_api::RepositorySourceKind::Https, + "http" => workspace_api::RepositorySourceKind::Http, + "ssh" => workspace_api::RepositorySourceKind::Ssh, + "invalid" => workspace_api::RepositorySourceKind::Invalid, + _ => { + return Err(settings_bad_request( + "working_directory_repository_source_invalid", + "persisted Workdir create Repository source kind is invalid", + )); + } + }; + working_directory_request.repository.source = workspace_api::RepositorySource { kind, uri }; + working_directory_request.repository.source_revision = revision; + working_directory_request.repository.source_fingerprint = fingerprint; + } let selector = working_directory_request .repository .selector @@ -8538,6 +8566,28 @@ async fn create_workspace_working_directory( resolved_runtime_id, config_revision: runtime_projection.config_revision, config_projection_digest: runtime_projection.projection_digest, + source_kind: Some( + working_directory_request + .repository + .source + .kind + .as_str() + .to_string(), + ), + source_uri: Some(working_directory_request.repository.source.uri.clone()), + source_revision: Some(working_directory_request.repository.source_revision), + source_fingerprint: Some( + working_directory_request + .repository + .source_fingerprint + .clone(), + ), + credential_id: None, + credential_revision: None, + host_trust_id: None, + host_trust_revision: None, + repository_access_mode: None, + cache_generation: 0, working_directory_id: next_backend_workdir_id(&request.repository_id), state: "pending".to_string(), failure: None, @@ -8628,12 +8678,10 @@ async fn create_workspace_working_directory( )); } - let repository_access_projection = active_repository_access_projection(api, workspace_id)?; - if let Err(error) = authorize_repository_materialization( + if let Err(error) = authorize_repository_materialization_operation( api, - &reserved.resolved_runtime_id, - &operation_id, - &repository_access_projection, + &reserved, + &request_fingerprint, &mut working_directory_request, ) { api.config_store.finish_workdir_create_operation( @@ -13968,6 +14016,132 @@ fn validate_working_directory_claim_for_browser( Ok(()) } +fn authorize_repository_materialization_operation( + api: &WorkspaceApi, + operation: &WorkdirCreateOperationRecord, + request_fingerprint: &str, + request: &mut WorkingDirectoryRequest, +) -> ApiResult<()> { + let context = if request.repository.source.kind == workspace_api::RepositorySourceKind::Ssh { + if let ( + Some(credential_id), + Some(credential_revision), + Some(host_trust_id), + Some(host_trust_revision), + Some(access_mode), + ) = ( + operation.credential_id.as_deref(), + operation.credential_revision, + operation.host_trust_id.as_deref(), + operation.host_trust_revision, + operation.repository_access_mode.as_deref(), + ) { + let lease = api + .repository_secrets + .lease_ssh_materialization_access_revision( + &api.config.workspace_id, + credential_id, + credential_revision, + host_trust_id, + host_trust_revision, + )?; + let access = match access_mode { + "read_only" => workspace_api::RepositoryAccessMode::ReadOnly, + "read_write" => workspace_api::RepositoryAccessMode::ReadWrite, + _ => { + return Err(settings_bad_request( + "working_directory_repository_access_invalid", + "persisted Repository access mode is invalid", + )); + } + }; + RepositoryMaterializationContext { + workspace_id: api.config.workspace_id.clone(), + runtime_id: operation.resolved_runtime_id.clone(), + operation_id: operation.operation_id.clone(), + config_revision: operation.config_revision, + config_projection_digest: operation.config_projection_digest.clone(), + cache_generation: operation.cache_generation, + ssh: Some(RepositorySshMaterializationAccess { + credential_id: lease.credential_id, + credential_revision: lease.credential_revision, + host_trust_id: lease.host_trust_id, + host_trust_revision: lease.host_trust_revision, + access, + expires_at_epoch_seconds: repository_access_expiry(), + private_key: SensitiveString::new(lease.private_key.as_str()), + known_hosts_entry: SensitiveString::new(lease.known_hosts_entry), + }), + } + } else { + let projection = active_repository_access_projection(api, &api.config.workspace_id)?; + if projection.config_revision != operation.config_revision + || projection.projection_digest != operation.config_projection_digest + { + return Err(settings_bad_request( + "working_directory_repository_access_revision_changed", + "Workspace Repository access revision changed before operation reservation was bound", + )); + } + authorize_repository_materialization( + api, + &operation.resolved_runtime_id, + &operation.operation_id, + &projection, + request, + )?; + let context = request.materialization.take().ok_or_else(|| { + settings_bad_request( + "working_directory_remote_repository_access_required", + "SSH Repository access authority is unavailable", + ) + })?; + let ssh = context.ssh.as_ref().ok_or_else(|| { + settings_bad_request( + "working_directory_remote_repository_access_required", + "SSH Repository access authority is unavailable", + ) + })?; + api.config_store.bind_workdir_create_repository_access( + &api.config.workspace_id, + &operation.operation_id, + request_fingerprint, + &ssh.credential_id, + ssh.credential_revision, + &ssh.host_trust_id, + ssh.host_trust_revision, + match ssh.access { + workspace_api::RepositoryAccessMode::ReadOnly => "read_only", + workspace_api::RepositoryAccessMode::ReadWrite => "read_write", + }, + context.cache_generation, + &now_registry_timestamp(), + )?; + context + } + } else { + RepositoryMaterializationContext { + workspace_id: api.config.workspace_id.clone(), + runtime_id: operation.resolved_runtime_id.clone(), + operation_id: operation.operation_id.clone(), + config_revision: operation.config_revision, + config_projection_digest: operation.config_projection_digest.clone(), + cache_generation: operation.cache_generation, + ssh: None, + } + }; + request.materialization = Some(context); + Ok(()) +} + +fn repository_access_expiry() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + .saturating_add(300) +} + fn authorize_repository_materialization( api: &WorkspaceApi, runtime_id: &str, @@ -13995,11 +14169,7 @@ fn authorize_repository_materialization( host_trust_id: lease.host_trust_id, host_trust_revision: lease.host_trust_revision, access: binding.access, - expires_at_epoch_seconds: std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() - .saturating_add(300), + expires_at_epoch_seconds: repository_access_expiry(), private_key: SensitiveString::new(lease.private_key.as_str()), known_hosts_entry: SensitiveString::new(lease.known_hosts_entry), }) @@ -21247,6 +21417,16 @@ mod tests { resolved_runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), config_revision: 1, config_projection_digest: "sha256:test".to_string(), + source_kind: Some("local_path".to_string()), + source_uri: Some("/tmp/repo".to_string()), + source_revision: Some(1), + source_fingerprint: Some("sha256:test".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: "workdir-provider-rejection".to_string(), state: "pending".to_string(), failure: None, diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 694ebf24..effe1a3e 100644 --- a/crates/workspace-server/src/store.rs +++ b/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, + pub source_uri: Option, + pub source_revision: Option, + pub source_fingerprint: Option, + pub credential_id: Option, + pub credential_revision: Option, + pub host_trust_id: Option, + pub host_trust_revision: Option, + pub repository_access_mode: Option, + pub cache_generation: u64, pub working_directory_id: String, pub state: String, pub failure: Option, @@ -6812,6 +6827,25 @@ 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_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#" @@ -9697,7 +9731,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 \ @@ -9775,7 +9809,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); @@ -9789,7 +9823,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(); @@ -9925,7 +9959,7 @@ mod tests { ), ] ); - assert_eq!(current_schema_version(&conn).unwrap(), 46); + assert_eq!(current_schema_version(&conn).unwrap(), 47); let foreign_key_error: Option = conn .query_row("PRAGMA foreign_key_check", [], |row| row.get(0)) .optional() @@ -10054,7 +10088,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( @@ -10172,7 +10206,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()); } @@ -10190,7 +10224,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 \ @@ -10231,7 +10265,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()); @@ -10298,7 +10332,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'", @@ -10481,7 +10515,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")) @@ -10498,7 +10532,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) @@ -11252,7 +11286,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, @@ -11609,13 +11643,13 @@ 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);", + 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 [ @@ -11647,13 +11681,23 @@ 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_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", @@ -11672,19 +11716,62 @@ 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_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}" + ); + } + } + #[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}"); } @@ -11905,7 +11992,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 = conn .query_row( "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'", @@ -12528,7 +12615,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| { @@ -12717,7 +12804,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, @@ -12795,7 +12882,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, @@ -13202,7 +13289,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(), diff --git a/crates/workspace-server/src/workdir_create_operations.rs b/crates/workspace-server/src/workdir_create_operations.rs index b8b0e25c..81e5cdbd 100644 --- a/crates/workspace-server/src/workdir_create_operations.rs +++ b/crates/workspace-server/src/workdir_create_operations.rs @@ -48,9 +48,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, @@ -61,6 +62,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, @@ -87,6 +92,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 { + 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, @@ -141,7 +221,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"#, @@ -157,11 +240,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>(11)?.map(|value| value as u64), + source_fingerprint: row.get(12)?, + credential_id: row.get(13)?, + credential_revision: row.get::<_, Option>(14)?.map(|value| value as u64), + host_trust_id: row.get(15)?, + host_trust_revision: row.get::<_, Option>(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)?, }) }, ) @@ -222,6 +315,16 @@ mod tests { 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, @@ -232,20 +335,55 @@ 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 =