From ece6f737ee5ab46193311b7efd916a6819aa999e Mon Sep 17 00:00:00 2001 From: aoruLola Date: Sun, 26 Jul 2026 22:15:17 +0800 Subject: [PATCH 1/4] fix(server): harden high-risk guard, API key cache, and refresh rotation - high_risk: fail closed when HIGH_RISK_SIGNATURE_REQUIRED is set but no signing secret is configured, instead of falling back to a hardcoded secret shipped in source; verify signatures in constant time via blake3::Hash equality rather than hex string comparison. - auth: give the in-memory API key cache a TTL (API_KEY_CACHE_TTL_SECS, default 60s) so a DB-side revocation or permission change is honored within the TTL instead of being trusted forever; drop stale entries when the DB reports a key invalid. - auth: make refresh-token rotation atomic via a single conditional UPDATE (rotate_refresh_token); rows_affected == 0 is treated as replay and burns the token family, closing the concurrent-refresh replay window. - auth: list_keys_persistent returns only DB-backed hashed keys when a DB is configured, so raw secret prefixes are never echoed by list_keys. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/server/src/bootstrap.rs | 10 ++- crates/server/src/core/auth.rs | 108 ++++++++++++++++++++-------- crates/server/src/core/auth/repo.rs | 47 ++++++++++-- crates/server/src/core/high_risk.rs | 32 +++++++-- 4 files changed, 154 insertions(+), 43 deletions(-) diff --git a/crates/server/src/bootstrap.rs b/crates/server/src/bootstrap.rs index 602f04d..5cc771a 100644 --- a/crates/server/src/bootstrap.rs +++ b/crates/server/src/bootstrap.rs @@ -119,6 +119,14 @@ pub(crate) async fn run() { } }; + let high_risk_guard = match HighRiskGuard::from_env(db_pool.clone()) { + Ok(guard) => guard, + Err(e) => { + tracing::error!("Failed to initialize high-risk guard: {e}"); + std::process::exit(1); + } + }; + let state = AppState { lock_manager, storage_manager, @@ -129,7 +137,7 @@ pub(crate) async fn run() { audit_chain: Some(AuditChain::new(db_pool.clone())), checkpoint_service: Some(CheckpointService::new(db_pool.clone())), witness_service: Some(WitnessService::from_env(db_pool.clone())), - high_risk_guard: Some(HighRiskGuard::from_env(db_pool.clone())), + high_risk_guard: Some(high_risk_guard), replay_service: Some(ReplayService::new(db_pool.clone())), retention_policy: RetentionPolicy::from_env(), db_pool: Some(db_pool), diff --git a/crates/server/src/core/auth.rs b/crates/server/src/core/auth.rs index 230c7b6..1538dc5 100644 --- a/crates/server/src/core/auth.rs +++ b/crates/server/src/core/auth.rs @@ -99,14 +99,27 @@ impl AuthIdentity { } } +const DEFAULT_KEY_CACHE_TTL_SECS: i64 = 60; + +/// A cached API key together with the instant it was cached. Cache entries expire +/// after `key_cache_ttl_secs` so that a revocation or permission change made in the +/// database (possibly by another instance) is picked up within the TTL instead of +/// being trusted forever. +#[derive(Clone)] +struct CachedKey { + api_key: ApiKey, + cached_at: DateTime, +} + #[derive(Clone)] pub struct AuthManager { - keys: Arc>, + keys: Arc>, dev_master_key: Option, repo: Option, token_service: Option, access_token_ttl_secs: i64, refresh_token_ttl_secs: i64, + key_cache_ttl_secs: i64, } impl AuthManager { @@ -118,6 +131,7 @@ impl AuthManager { token_service: None, access_token_ttl_secs: 15 * 60, refresh_token_ttl_secs: 7 * 24 * 60 * 60, + key_cache_ttl_secs: DEFAULT_KEY_CACHE_TTL_SECS, } } @@ -130,6 +144,7 @@ impl AuthManager { token_service: None, access_token_ttl_secs: 15 * 60, refresh_token_ttl_secs: 7 * 24 * 60 * 60, + key_cache_ttl_secs: DEFAULT_KEY_CACHE_TTL_SECS, }; let dev_api_key = ApiKey { @@ -145,7 +160,7 @@ impl AuthManager { expires_at: None, revoked: false, }; - manager.keys.insert(dev_api_key.key.clone(), dev_api_key); + manager.cache_key(dev_api_key); manager } @@ -167,6 +182,11 @@ impl AuthManager { .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(7 * 24 * 60 * 60); + manager.key_cache_ttl_secs = std::env::var("API_KEY_CACHE_TTL_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|secs| *secs >= 0) + .unwrap_or(DEFAULT_KEY_CACHE_TTL_SECS); if let Some(master) = manager.dev_master_key.clone() { repo.upsert_api_key( @@ -189,10 +209,26 @@ impl AuthManager { Ok(manager) } + fn cache_key(&self, api_key: ApiKey) { + self.keys.insert( + api_key.key.clone(), + CachedKey { + api_key, + cached_at: Utc::now(), + }, + ); + } + + fn cache_entry_fresh(&self, cached_at: DateTime) -> bool { + Utc::now() < cached_at + Duration::seconds(self.key_cache_ttl_secs.max(0)) + } + pub fn validate_key(&self, key: &str) -> Option { - self.keys.get(key).and_then(|api_key| { - if api_key.is_valid() { - Some(api_key.clone()) + self.keys.get(key).and_then(|entry| { + // A stale entry is treated as a miss so callers with a DB fall through + // (validate_key_any) re-check the authoritative record. + if self.cache_entry_fresh(entry.cached_at) && entry.api_key.is_valid() { + Some(entry.api_key.clone()) } else { None } @@ -223,13 +259,13 @@ impl AuthManager { revoked: false, }; - self.keys.insert(key, api_key.clone()); + self.cache_key(api_key.clone()); api_key } pub fn revoke_key(&self, key: &str) -> bool { - if let Some(mut api_key) = self.keys.get_mut(key) { - api_key.revoked = true; + if let Some(mut entry) = self.keys.get_mut(key) { + entry.api_key.revoked = true; true } else { false @@ -237,7 +273,10 @@ impl AuthManager { } pub fn list_keys(&self) -> Vec { - self.keys.iter().map(|kv| kv.value().clone()).collect() + self.keys + .iter() + .map(|kv| kv.value().api_key.clone()) + .collect() } pub async fn validate_key_any(&self, key: &str) -> Option { @@ -263,9 +302,12 @@ impl AuthManager { revoked: stored.revoked, }; if api_key.is_valid() { - self.keys.insert(key.to_string(), api_key.clone()); + self.cache_key(api_key.clone()); Some(api_key) } else { + // Drop any stale cached copy so a key revoked in the DB stops + // authenticating from cache on this instance too. + self.keys.remove(key); None } } @@ -335,23 +377,27 @@ impl AuthManager { } pub async fn list_keys_persistent(&self) -> Result, HyperTideError> { - let mut keys = self.list_keys(); + // When a DB is configured it is the authoritative store and holds only + // hashed keys. Return those exclusively: merging the in-memory cache here + // duplicated persisted keys and, worse, exposed the raw secret bytes of + // cached keys (dev master / freshly generated) to `list_keys`. if let Some(repo) = &self.repo { let stored = repo.list_api_keys().await.map_err(|error| { HyperTideError::Persistence(format!("failed to list api keys: {error}")) })?; - for (key_hash, row) in stored { - keys.push(ApiKey { + return Ok(stored + .into_iter() + .map(|(key_hash, row)| ApiKey { key: key_hash, owner_id: row.owner_id, permissions: row.permissions, created_at: row.created_at, expires_at: row.expires_at, revoked: row.revoked, - }); - } + }) + .collect()); } - Ok(keys) + Ok(self.list_keys()) } pub async fn exchange_key_for_tokens( @@ -471,22 +517,26 @@ impl AuthManager { ) .map_err(HyperTideError::Authentication)?; - repo.insert_refresh_token( - &new_refresh_token, - &claims.sub, - &family_id, - Some(refresh_token), - Utc::now() + Duration::seconds(self.refresh_token_ttl_secs), - ) - .await - .map_err(|error| { - HyperTideError::Persistence(format!("failed to persist rotated refresh token: {error}")) - })?; - repo.mark_refresh_replaced(refresh_token, &new_refresh_token) + let rotated = repo + .rotate_refresh_token( + refresh_token, + &new_refresh_token, + &claims.sub, + &family_id, + Utc::now() + Duration::seconds(self.refresh_token_ttl_secs), + ) .await .map_err(|error| { - HyperTideError::Persistence(format!("failed to mark refresh rotation: {error}")) + HyperTideError::Persistence(format!("failed to rotate refresh token: {error}")) })?; + if !rotated { + // The token was claimed by a concurrent refresh or revoked between our + // read above and this atomic claim: treat as replay and burn the family. + let _ = repo.revoke_refresh_family(&stored.family_id).await; + return Err(HyperTideError::Authentication( + "Refresh token replay detected; family revoked".to_string(), + )); + } Ok(TokenPair { access_token, diff --git a/crates/server/src/core/auth/repo.rs b/crates/server/src/core/auth/repo.rs index 9000dde..8e525f4 100644 --- a/crates/server/src/core/auth/repo.rs +++ b/crates/server/src/core/auth/repo.rs @@ -251,25 +251,60 @@ impl AuthRepo { })) } - pub async fn mark_refresh_replaced( + /// Atomically claim `old_refresh_token` and persist its replacement in a single + /// transaction. Returns `Ok(false)` when the old token was already rotated or + /// revoked (a concurrent or replayed refresh), in which case nothing is written. + /// The conditional `UPDATE` is the serialization point: of two concurrent + /// refreshes of the same token, exactly one flips `replaced_by_token_hash` from + /// NULL and proceeds; the other matches zero rows and is rejected as replay. + pub async fn rotate_refresh_token( &self, old_refresh_token: &str, new_refresh_token: &str, + principal_id: &str, + family_id: &str, + expires_at: DateTime, ) -> Result { let old_hash = self.hash_secret(old_refresh_token); let new_hash = self.hash_secret(new_refresh_token); - let result = sqlx::query( + + let mut tx = self.pool.begin().await?; + + let claimed = sqlx::query( r#" UPDATE refresh_tokens SET replaced_by_token_hash = $2 WHERE token_hash = $1 + AND replaced_by_token_hash IS NULL + AND revoked_at IS NULL "#, ) - .bind(old_hash) - .bind(new_hash) - .execute(&self.pool) + .bind(&old_hash) + .bind(&new_hash) + .execute(&mut *tx) .await?; - Ok(result.rows_affected() > 0) + + if claimed.rows_affected() == 0 { + tx.rollback().await?; + return Ok(false); + } + + sqlx::query( + r#" + INSERT INTO refresh_tokens (token_hash, principal_id, family_id, parent_token_hash, expires_at) + VALUES ($1, $2, $3, $4, $5) + "#, + ) + .bind(&new_hash) + .bind(principal_id) + .bind(family_id) + .bind(&old_hash) + .bind(expires_at) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(true) } pub async fn revoke_refresh_token(&self, refresh_token: &str) -> Result { diff --git a/crates/server/src/core/high_risk.rs b/crates/server/src/core/high_risk.rs index 783dd73..3a6acfd 100644 --- a/crates/server/src/core/high_risk.rs +++ b/crates/server/src/core/high_risk.rs @@ -11,23 +11,35 @@ pub struct HighRiskGuard { } impl HighRiskGuard { - pub fn from_env(pool: PgPool) -> Self { + pub fn from_env(pool: PgPool) -> Result { let required = std::env::var("HIGH_RISK_SIGNATURE_REQUIRED") .ok() .map(|value| value == "1" || value.eq_ignore_ascii_case("true")) .unwrap_or(false); + // Fail closed: never fall back to a hardcoded/shipped signing secret when + // enforcement is on. A default secret in open-source code would let anyone + // forge a valid X-HT-Signature and defeat the step-up check entirely. let secret = std::env::var("HIGH_RISK_SIGNING_SECRET") - .unwrap_or_else(|_| "hypertide-dev-signing-secret".to_string()); + .ok() + .filter(|value| !value.trim().is_empty()); + if required && secret.is_none() { + return Err( + "HIGH_RISK_SIGNATURE_REQUIRED is enabled but HIGH_RISK_SIGNING_SECRET is unset \ + or empty; refusing to start with an insecure default signing secret" + .to_string(), + ); + } + let secret = secret.unwrap_or_default(); let skew_secs = std::env::var("HIGH_RISK_SIG_SKEW_SECS") .ok() .and_then(|value| value.parse::().ok()) .unwrap_or(300); - Self { + Ok(Self { pool, required, secret, skew_secs, - } + }) } pub async fn verify( @@ -72,9 +84,15 @@ impl HighRiskGuard { "{}|{}|{}|{}|{}|{}", self.secret, action, actor_id, nonce, timestamp, payload_hash ); - let expected = blake3::hash(material.as_bytes()).to_hex().to_string(); - - if expected != signature { + let expected = blake3::hash(material.as_bytes()); + // Parse the client signature into a fixed 32-byte digest and compare with + // blake3::Hash's constant-time equality, avoiding a byte-by-byte timing + // oracle on the expected MAC. + let provided = match blake3::Hash::from_hex(signature) { + Ok(hash) => hash, + Err(_) => return Err("invalid signature".to_string()), + }; + if expected != provided { return Err("invalid signature".to_string()); } From a1fb69050588e5b41f86473ac9bf148ddcb7d9b4 Mon Sep 17 00:00:00 2001 From: aoruLola Date: Sun, 26 Jul 2026 22:15:30 +0800 Subject: [PATCH 2/4] fix(server): harden locks, versioning persistence, storage dedup - versioning: allow a draft-first changeset (base=ROOT, head never advanced) to be promoted by mirroring submit's ROOT acceptance rule in promote_changeset and changeset_gate (head_accepts_base). Adds regression test. - versioning: persist each snapshot under its changeset's actual branch instead of always the default branch (snapshots are keyed by branch_name). - versioning: add an optimistic state_version guard (migration 202602260018) so a concurrent writer's update is detected as a conflict rather than silently clobbered. Single-instance behavior is unchanged. - versioning: opt-in separation of duties (HYPERTIDE_REQUIRE_SEPARATE_APPROVER) rejecting self-approve/self-promote. - lock: unlock/expired-renew use owner-scoped DB deletes and an ownership guard on renew upsert, so a stale cached view cannot delete or steal a lock that another owner re-acquired. Force-release stays owner-blind. - storage: on a dedup hit, compare object size via metadata instead of reading the whole existing object back into memory and re-hashing it; integrity is still verified on retrieve. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/server/src/api/versioning.rs | 10 + crates/server/src/core/lock.rs | 46 +- crates/server/src/core/lock/repo_pg.rs | 38 +- crates/server/src/core/storage.rs | 29 +- crates/server/src/core/versioning.rs | 3682 +++++++++-------- crates/server/src/core/versioning/repo_pg.rs | 89 +- .../202602260018_repo_state_version.down.sql | 2 + .../202602260018_repo_state_version.up.sql | 6 + 8 files changed, 2073 insertions(+), 1829 deletions(-) create mode 100644 migrations/202602260018_repo_state_version.down.sql create mode 100644 migrations/202602260018_repo_state_version.up.sql diff --git a/crates/server/src/api/versioning.rs b/crates/server/src/api/versioning.rs index b9e3566..a92eb99 100644 --- a/crates/server/src/api/versioning.rs +++ b/crates/server/src/api/versioning.rs @@ -195,6 +195,16 @@ fn map_versioning_error(error: VersioningError) -> (StatusCode, String) { StatusCode::BAD_REQUEST, format!("Invalid asset layout for {repo_id}: {message}"), ), + VersioningError::SelfApprovalForbidden { + repo_id, + changeset_id, + actor, + } => ( + StatusCode::FORBIDDEN, + format!( + "Separation of duties: {actor} cannot approve/promote their own changeset {repo_id}/{changeset_id}" + ), + ), VersioningError::Persistence { message } => ( StatusCode::INTERNAL_SERVER_ERROR, format!("Versioning persistence failed: {message}"), diff --git a/crates/server/src/core/lock.rs b/crates/server/src/core/lock.rs index 84a76a4..ca2b089 100644 --- a/crates/server/src/core/lock.rs +++ b/crates/server/src/core/lock.rs @@ -181,11 +181,19 @@ impl LockManager { } if self.is_expired(&existing) { if let Some(repo) = &self.repo { - repo.delete_lock(&existing.repo_id, &existing.scope, &existing.file_path) - .await - .map_err(|e| { - HyperTideError::Persistence(format!("failed to cleanup expired lock: {e}")) - })?; + // Owner-scoped: if the lease expired and another principal already + // re-acquired the lock in the DB, this removes nothing rather than + // deleting their valid lock. + repo.delete_lock_owned( + &existing.repo_id, + &existing.scope, + &existing.file_path, + &existing.owner_id, + ) + .await + .map_err(|e| { + HyperTideError::Persistence(format!("failed to cleanup expired lock: {e}")) + })?; } self.locks.remove(&lock_key); return Err(HyperTideError::Conflict( @@ -199,9 +207,18 @@ impl LockManager { }; if let Some(repo) = &self.repo { - repo.upsert_lock(&renewed).await.map_err(|e| { + let extended = repo.upsert_lock(&renewed).await.map_err(|e| { HyperTideError::Persistence(format!("failed to persist lock renew: {e}")) })?; + if !extended { + // The DB lock is now owned by someone else (e.g. re-acquired after + // an expiry our stale cache missed). Drop the stale entry instead of + // overwriting their lock. + self.locks.remove(&lock_key); + return Err(HyperTideError::Conflict( + "Cannot renew: lock is held by another owner".to_string(), + )); + } } self.locks.insert(lock_key, renewed.clone()); Ok(renewed) @@ -234,9 +251,24 @@ impl LockManager { } if let Some(repo) = &self.repo { - repo.delete_lock(&existing.repo_id, &existing.scope, &existing.file_path) + let removed = repo + .delete_lock_owned( + &existing.repo_id, + &existing.scope, + &existing.file_path, + &existing.owner_id, + ) .await .map_err(|e| HyperTideError::Persistence(format!("failed to delete lock: {e}")))?; + if !removed { + // Our cached view said we owned it, but the DB disagrees (lease + // expired and someone else re-acquired). Drop the stale entry and + // refuse rather than silently succeeding. + self.locks.remove(&lock_key); + return Err(HyperTideError::Conflict( + "Cannot unlock: lock is no longer held by this owner".to_string(), + )); + } } self.locks.remove(&lock_key); diff --git a/crates/server/src/core/lock/repo_pg.rs b/crates/server/src/core/lock/repo_pg.rs index 7537d12..c905aa8 100644 --- a/crates/server/src/core/lock/repo_pg.rs +++ b/crates/server/src/core/lock/repo_pg.rs @@ -48,8 +48,12 @@ impl LockRepoPg { .collect()) } - pub async fn upsert_lock(&self, lock: &FileLock) -> Result<(), sqlx::Error> { - sqlx::query( + /// Renew/insert a lock, but never steal one: on conflict the lease is only + /// extended when the existing row is still owned by the same principal. + /// Returns `false` (0 rows) when a different owner holds the DB lock, so a + /// stale in-memory view cannot overwrite the authoritative owner. + pub async fn upsert_lock(&self, lock: &FileLock) -> Result { + let result = sqlx::query( r#" INSERT INTO locks (file_path, owner_id, locked_at, lease_expires_at, force_released, repo_id, scope) VALUES ($1, $2, $3, $4, FALSE, $5, $6) @@ -59,6 +63,7 @@ impl LockRepoPg { locked_at = EXCLUDED.locked_at, lease_expires_at = EXCLUDED.lease_expires_at, force_released = FALSE + WHERE locks.owner_id = EXCLUDED.owner_id "#, ) .bind(&lock.file_path) @@ -69,7 +74,7 @@ impl LockRepoPg { .bind(&lock.scope) .execute(&self.pool) .await?; - Ok(()) + Ok(result.rows_affected() > 0) } pub async fn acquire_lock_atomic(&self, lock: &FileLock) -> Result { @@ -139,6 +144,7 @@ impl LockRepoPg { }) } + /// Admin/force release: delete regardless of owner. pub async fn delete_lock( &self, repo_id: &str, @@ -158,4 +164,30 @@ impl LockRepoPg { .await?; Ok(()) } + + /// Owner-scoped release used by `unlock`/expired-renew cleanup: only removes + /// the lock when it is still owned by `owner_id` in the database. Returns + /// `false` when no such row exists (e.g. the lease expired and another + /// principal re-acquired it), so we never delete a valid lock we no longer hold. + pub async fn delete_lock_owned( + &self, + repo_id: &str, + scope: &str, + file_path: &str, + owner_id: &str, + ) -> Result { + let result = sqlx::query( + r#" + DELETE FROM locks + WHERE repo_id = $1 AND scope = $2 AND file_path = $3 AND owner_id = $4 + "#, + ) + .bind(repo_id) + .bind(scope) + .bind(file_path) + .bind(owner_id) + .execute(&self.pool) + .await?; + Ok(result.rows_affected() > 0) + } } diff --git a/crates/server/src/core/storage.rs b/crates/server/src/core/storage.rs index 3db5cc3..b95ecce 100644 --- a/crates/server/src/core/storage.rs +++ b/crates/server/src/core/storage.rs @@ -117,15 +117,20 @@ impl StorageManager { let object_dir = self.storage_root.join("objects").join(prefix); let object_path = object_dir.join(rest); - // Check if already exists (deduplication) + // Check if already exists (deduplication). The incoming bytes already hash + // to `hash`, so a same-sized object at the CAS path is a dedup hit; trust + // the content-addressed layout instead of reading the whole existing object + // back into memory and re-hashing it (that doubled transient memory on every + // dedup hit). Object integrity is still re-verified on `retrieve`. if Self::check_path_exists(&object_path, "object existence before store") .await .map_err(HyperTideError::Persistence)? { - let existing = fs::read(&object_path).await.map_err(|error| { - HyperTideError::Persistence(format!("Failed to verify existing object: {error}")) - })?; - if Self::calculate_hash(&existing) == hash { + let same_size = match fs::metadata(&object_path).await { + Ok(metadata) => metadata.len() == size_bytes, + Err(_) => false, + }; + if same_size { return Ok(StoredFile { hash, original_path: original_path.to_string(), @@ -133,11 +138,15 @@ impl StorageManager { stored_at: chrono::Utc::now(), }); } - fs::remove_file(&object_path).await.map_err(|error| { - HyperTideError::Persistence(format!( - "Failed to replace corrupt CAS object {hash}: {error}" - )) - })?; + // Size mismatch (or unreadable metadata): the object is corrupt for this + // hash; drop it and rewrite. Tolerate a concurrent removal. + if let Err(error) = fs::remove_file(&object_path).await { + if error.kind() != std::io::ErrorKind::NotFound { + return Err(HyperTideError::Persistence(format!( + "Failed to replace corrupt CAS object {hash}: {error}" + ))); + } + } } // Create subdirectory if needed diff --git a/crates/server/src/core/versioning.rs b/crates/server/src/core/versioning.rs index dd7b814..4cfadca 100644 --- a/crates/server/src/core/versioning.rs +++ b/crates/server/src/core/versioning.rs @@ -1,1795 +1,1887 @@ -use std::collections::{BTreeSet, HashMap, HashSet}; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, RwLock}; - -use crate::core::error::HyperTideError; -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use sqlx::PgPool; -use uuid::Uuid; - -pub mod repo_pg; -use self::repo_pg::VersionRepoPg; - -pub const ROOT_BASE_CHANGESET_ID: &str = "ROOT"; - -#[cfg(windows)] -fn replace_state_file(temp_path: &Path, state_path: &Path) -> std::io::Result<()> { - use std::iter::once; - use std::os::windows::ffi::OsStrExt; - use windows_sys::Win32::Storage::FileSystem::ReplaceFileW; - - if !state_path.exists() { - return std::fs::rename(temp_path, state_path); - } - - let state_wide = state_path - .as_os_str() - .encode_wide() - .chain(once(0)) - .collect::>(); - let temp_wide = temp_path - .as_os_str() - .encode_wide() - .chain(once(0)) - .collect::>(); - let replaced = unsafe { - ReplaceFileW( - state_wide.as_ptr(), - temp_wide.as_ptr(), - std::ptr::null(), - 0, - std::ptr::null(), - std::ptr::null(), - ) - }; - if replaced == 0 { - Err(std::io::Error::last_os_error()) - } else { - Ok(()) - } -} - -#[cfg(not(windows))] -fn replace_state_file(temp_path: &Path, state_path: &Path) -> std::io::Result<()> { - std::fs::rename(temp_path, state_path) -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum ChangesetKind { - Normal, - Rollback, -} - -impl ChangesetKind { - pub fn as_str(&self) -> &'static str { - match self { - ChangesetKind::Normal => "normal", - ChangesetKind::Rollback => "rollback", - } - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum ChangesetVisibility { - Visible, - Draft, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum ChangesetStatus { - Draft, - Approved, - Visible, -} - -impl ChangesetStatus { - pub fn as_str(&self) -> &'static str { - match self { - ChangesetStatus::Draft => "draft", - ChangesetStatus::Approved => "approved", - ChangesetStatus::Visible => "visible", - } - } -} - -fn default_changeset_status() -> ChangesetStatus { - ChangesetStatus::Visible -} - -fn staging_ref(repo_id: &str, branch: &str, changeset_id: &str) -> String { - format!("refs/ht/staging/{repo_id}/{branch}/{changeset_id}") -} - -fn visible_ref(branch: &str) -> String { - format!("refs/heads/{branch}") -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct AssetDelta { - #[serde(default)] - pub asset_id: Option, - pub path: String, - #[serde(default)] - pub from_blob_hash: Option, - pub blob_hash: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ChangesetRecord { - pub changeset_id: String, - pub repo_id: String, - pub branch: String, - pub parent_changeset_id: Option, - pub base_changeset_id: Option, - pub kind: ChangesetKind, - pub rollback_of: Option, - pub author: String, - pub message: String, - pub created_at: DateTime, - #[serde(default = "default_changeset_status")] - pub status: ChangesetStatus, - pub approved_by: Option, - pub approved_at: Option>, - pub promoted_at: Option>, - #[serde(default)] - pub staging_ref: Option, - #[serde(default)] - pub visible_ref: Option, - #[serde(default)] - pub intent_id: Option, - #[serde(default)] - pub task_id: Option, - #[serde(default)] - pub agent_run_id: Option, - #[serde(default)] - pub session_id: Option, - #[serde(default)] - pub parent_checkpoint_id: Option, - #[serde(default)] - pub risk_level: Option, - #[serde(default)] - pub semantic_summary: Option, - pub assets: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct BranchRecord { - pub name: String, - pub created_by: String, - pub created_at: DateTime, - pub is_default: bool, - pub head_changeset_id: Option, -} - -#[derive(Debug, Clone)] -pub struct SubmitChangesetInput { - pub repo_id: String, - pub branch: String, - pub base_changeset_id: Option, - pub kind: ChangesetKind, - pub rollback_of: Option, - pub author: String, - pub message: String, - pub visibility: ChangesetVisibility, - pub intent_id: Option, - pub task_id: Option, - pub agent_run_id: Option, - pub session_id: Option, - pub parent_checkpoint_id: Option, - pub risk_level: Option, - pub semantic_summary: Option, - pub assets: Vec, -} - -#[derive(Debug, Clone, Serialize)] -pub struct HistoryPage { - pub items: Vec, - pub next_cursor: Option, -} - -#[derive(Debug, Clone, Serialize)] -pub struct ChangesetGate { - pub repo_id: String, - pub changeset_id: String, - pub branch: String, - pub status: ChangesetStatus, - pub required_state: &'static str, - pub can_promote: bool, - pub blocking_reason: Option, - pub base_changeset_id: Option, - pub branch_head_changeset_id: Option, - pub staging_ref: Option, - pub visible_ref: Option, -} - -#[derive(Debug, Clone, Serialize)] -pub struct SyncSnapshot { - pub repo_id: String, - pub branch: String, - pub changeset_id: Option, - pub assets: Vec, -} - -#[derive(Debug, Clone, Serialize)] -pub struct RepoSummary { - pub repo_id: String, - pub default_branch: String, - pub branch_count: usize, - pub default_head_changeset_id: Option, -} - -#[derive(Debug, Clone, Serialize)] -pub struct RepoInfo { - pub repo_id: String, - pub default_branch: String, - pub branch_count: usize, - pub default_head_changeset_id: Option, - pub branches: Vec, -} - -#[derive(Debug, Clone, Serialize)] -pub struct SnapshotEntry { - pub asset_id: String, - pub path: String, - pub blob_hash: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(super) struct SnapshotAsset { - pub asset_id: String, - pub path: String, - pub blob_hash: String, -} - -#[derive(Debug, Clone)] -pub struct RollbackPlan { - pub repo_id: String, - pub branch: String, - pub base_changeset_id: String, - pub target_changeset_id: String, - pub assets: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum VersioningError { - RepoAlreadyExists { - repo_id: String, - }, - RepoNotFound { - repo_id: String, - }, - BranchNotFound { - repo_id: String, - branch: String, - }, - BranchAlreadyExists { - repo_id: String, - branch: String, - }, - ChangesetNotFound { - repo_id: String, - changeset_id: String, - }, - BaseChangesetRequired, - BaseChangesetMismatch { - repo_id: String, - branch: String, - expected: Option, - got: Option, - }, - InvalidRollbackTarget { - repo_id: String, - branch: String, - target_changeset_id: String, - }, - InvalidChangesetState { - repo_id: String, - changeset_id: String, - status: ChangesetStatus, - expected: &'static str, - }, - InvalidAssetLayout { - repo_id: String, - message: String, - }, - Persistence { - message: String, - }, -} - -#[derive(Clone)] -pub struct VersionManager { - repos: Arc>>, - persistence_path: Option, - repo_pg: Option, - mutation_lock: Arc>, -} - -impl VersionManager { - pub fn new() -> Self { - Self { - repos: Arc::new(RwLock::new(HashMap::new())), - persistence_path: None, - repo_pg: None, - mutation_lock: Arc::new(tokio::sync::Mutex::new(())), - } - } - - pub fn with_persistence(path: impl AsRef) -> Self { - let persistence_path = path.as_ref().to_path_buf(); - let repos = match Self::load_repos(&persistence_path) { - Ok(repos) => repos, - Err(error) => { - tracing::warn!( - "versioning persistence load failed at {}: {}", - persistence_path.display(), - error - ); - HashMap::new() - } - }; - - Self { - repos: Arc::new(RwLock::new(repos)), - persistence_path: Some(persistence_path), - repo_pg: None, - mutation_lock: Arc::new(tokio::sync::Mutex::new(())), - } - } - - pub async fn with_pg(pool: PgPool) -> Result { - let repo_pg = VersionRepoPg::new(pool); - let repos = repo_pg.load_repos().await.map_err(|error| { - HyperTideError::Persistence(format!("failed to load versioning state from db: {error}")) - })?; - Ok(Self { - repos: Arc::new(RwLock::new(repos)), - persistence_path: None, - repo_pg: Some(repo_pg), - mutation_lock: Arc::new(tokio::sync::Mutex::new(())), - }) - } - - pub async fn create_repo( - &self, - repo_id: &str, - default_branch: &str, - created_by: &str, - ) -> Result { - let _mutation = self.mutation_lock.lock().await; - let (info, snapshot) = { - let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); - if snapshot.contains_key(repo_id) { - return Err(VersioningError::RepoAlreadyExists { - repo_id: repo_id.to_string(), - }); - } - - let repo = RepoState::new_with_default(default_branch, created_by); - snapshot.insert(repo_id.to_string(), repo); - let info = - Self::repo_info_from_state(repo_id, snapshot.get(repo_id).expect("repo exists")); - (info, snapshot) - }; - self.persist_repo(repo_id, &snapshot) - .await - .map_err(|message| VersioningError::Persistence { message })?; - *self.repos.write().expect("versioning lock poisoned") = snapshot; - Ok(info) - } - - pub fn list_repos(&self) -> Vec { - let repos = self.repos.read().expect("versioning lock poisoned"); - let mut items: Vec = repos - .iter() - .map(|(repo_id, repo)| Self::repo_summary_from_state(repo_id, repo)) - .collect(); - items.sort_by(|a, b| a.repo_id.cmp(&b.repo_id)); - items - } - - pub fn get_repo_info(&self, repo_id: &str) -> Result { - let repos = self.repos.read().expect("versioning lock poisoned"); - let repo = repos - .get(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - Ok(Self::repo_info_from_state(repo_id, repo)) - } - - pub async fn create_branch( - &self, - repo_id: &str, - branch: &str, - from_changeset_id: Option<&str>, - created_by: &str, - ) -> Result { - let _mutation = self.mutation_lock.lock().await; - let (record, snapshot) = { - let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); - let repo = snapshot - .entry(repo_id.to_string()) - .or_insert_with(|| RepoState::new(created_by)); - repo.ensure_default_branch(created_by); - - if repo.branches.contains_key(branch) { - return Err(VersioningError::BranchAlreadyExists { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - }); - } - - let head = if let Some(id) = from_changeset_id { - if !repo.changesets.contains_key(id) { - return Err(VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: id.to_string(), - }); - } - Some(id.to_string()) - } else { - repo.default_head() - }; - - let history = if let Some(ref head_id) = head { - repo.lineage_to(head_id) - .ok_or_else(|| VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: head_id.clone(), - })? - } else { - Vec::new() - }; - - let record = BranchRecord { - name: branch.to_string(), - created_by: created_by.to_string(), - created_at: Utc::now(), - is_default: false, - head_changeset_id: head.clone(), - }; - - repo.branches.insert( - branch.to_string(), - BranchState { - record: record.clone(), - history, - }, - ); - - (record, snapshot) - }; - self.persist_repo(repo_id, &snapshot) - .await - .map_err(|message| VersioningError::Persistence { message })?; - *self.repos.write().expect("versioning lock poisoned") = snapshot; - Ok(record) - } - - fn repo_summary_from_state(repo_id: &str, repo: &RepoState) -> RepoSummary { - RepoSummary { - repo_id: repo_id.to_string(), - default_branch: repo.default_branch.clone(), - branch_count: repo.branches.len(), - default_head_changeset_id: repo.default_head(), - } - } - - fn repo_info_from_state(repo_id: &str, repo: &RepoState) -> RepoInfo { - let mut branches: Vec = - repo.branches.values().map(|b| b.record.clone()).collect(); - branches.sort_by(|a, b| a.name.cmp(&b.name)); - RepoInfo { - repo_id: repo_id.to_string(), - default_branch: repo.default_branch.clone(), - branch_count: branches.len(), - default_head_changeset_id: repo.default_head(), - branches, - } - } - - pub fn list_branches(&self, repo_id: &str) -> Result, VersioningError> { - let repos = self.repos.read().expect("versioning lock poisoned"); - let repo = repos - .get(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - - let mut items: Vec = - repo.branches.values().map(|b| b.record.clone()).collect(); - items.sort_by(|a, b| a.name.cmp(&b.name)); - Ok(items) - } - - pub async fn submit_changeset( - &self, - input: SubmitChangesetInput, - ) -> Result { - let _mutation = self.mutation_lock.lock().await; - let repo_id = input.repo_id.clone(); - let (record, snapshot) = { - let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); - let repo = snapshot - .entry(input.repo_id.clone()) - .or_insert_with(|| RepoState::new(&input.author)); - repo.ensure_default_branch(&input.author); - let record = Self::submit_internal(repo, input)?; - (record, snapshot) - }; - self.persist_repo(&repo_id, &snapshot) - .await - .map_err(|message| VersioningError::Persistence { message })?; - *self.repos.write().expect("versioning lock poisoned") = snapshot; - Ok(record) - } - - pub async fn approve_changeset( - &self, - repo_id: &str, - changeset_id: &str, - approver: &str, - ) -> Result { - let _mutation = self.mutation_lock.lock().await; - let (record, snapshot) = { - let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); - let repo = snapshot - .get_mut(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - let record = repo.changesets.get_mut(changeset_id).ok_or_else(|| { - VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - } - })?; - - match record.status { - ChangesetStatus::Draft => { - record.status = ChangesetStatus::Approved; - record.approved_by = Some(approver.to_string()); - record.approved_at = Some(Utc::now()); - } - status => { - return Err(VersioningError::InvalidChangesetState { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - status, - expected: "draft", - }); - } - } - - (record.clone(), snapshot) - }; - self.persist_repo(repo_id, &snapshot) - .await - .map_err(|message| VersioningError::Persistence { message })?; - *self.repos.write().expect("versioning lock poisoned") = snapshot; - Ok(record) - } - - pub async fn promote_changeset( - &self, - repo_id: &str, - changeset_id: &str, - promoter: &str, - ) -> Result { - let _mutation = self.mutation_lock.lock().await; - let (record, snapshot) = { - let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); - let repo = snapshot - .get_mut(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - - let record_view = repo.changesets.get(changeset_id).ok_or_else(|| { - VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - } - })?; - if record_view.status != ChangesetStatus::Approved { - return Err(VersioningError::InvalidChangesetState { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - status: record_view.status, - expected: "approved", - }); - } - - let branch = record_view.branch.clone(); - let base = record_view.base_changeset_id.clone(); - let branch_state = - repo.branches - .get_mut(&branch) - .ok_or_else(|| VersioningError::BranchNotFound { - repo_id: repo_id.to_string(), - branch: branch.clone(), - })?; - let expected_head = branch_state.record.head_changeset_id.clone(); - if expected_head != base { - return Err(VersioningError::BaseChangesetMismatch { - repo_id: repo_id.to_string(), - branch, - expected: expected_head, - got: base, - }); - } - - branch_state.record.head_changeset_id = Some(changeset_id.to_string()); - if !branch_state.history.iter().any(|id| id == changeset_id) { - branch_state.history.push(changeset_id.to_string()); - } - - let record = repo.changesets.get_mut(changeset_id).ok_or_else(|| { - VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - } - })?; - record.status = ChangesetStatus::Visible; - if record.approved_by.is_none() { - record.approved_by = Some(promoter.to_string()); - record.approved_at = Some(Utc::now()); - } - record.promoted_at = Some(Utc::now()); - record.visible_ref = Some(visible_ref(&record.branch)); - - (record.clone(), snapshot) - }; - self.persist_repo(repo_id, &snapshot) - .await - .map_err(|message| VersioningError::Persistence { message })?; - *self.repos.write().expect("versioning lock poisoned") = snapshot; - Ok(record) - } - - pub fn changeset_gate( - &self, - repo_id: &str, - changeset_id: &str, - ) -> Result { - let repos = self.repos.read().expect("versioning lock poisoned"); - let repo = repos - .get(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - let record = repo.changesets.get(changeset_id).ok_or_else(|| { - VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - } - })?; - let branch_state = - repo.branches - .get(&record.branch) - .ok_or_else(|| VersioningError::BranchNotFound { - repo_id: repo_id.to_string(), - branch: record.branch.clone(), - })?; - let current_head = branch_state.record.head_changeset_id.clone(); - let base = record.base_changeset_id.clone(); - - let (can_promote, blocking_reason) = if record.status != ChangesetStatus::Approved { - ( - false, - Some(format!( - "changeset status is {}, expected approved", - record.status.as_str() - )), - ) - } else if current_head != base { - ( - false, - Some(format!( - "branch head mismatch: current={current_head:?}, base={base:?}" - )), - ) - } else { - (true, None) - }; - - Ok(ChangesetGate { - repo_id: repo_id.to_string(), - changeset_id: changeset_id.to_string(), - branch: record.branch.clone(), - status: record.status, - required_state: "approved", - can_promote, - blocking_reason, - base_changeset_id: base, - branch_head_changeset_id: current_head, - staging_ref: record.staging_ref.clone(), - visible_ref: record.visible_ref.clone(), - }) - } - - pub fn history( - &self, - repo_id: &str, - branch: &str, - limit: usize, - cursor: usize, - ) -> Result { - let repos = self.repos.read().expect("versioning lock poisoned"); - let repo = repos - .get(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - let branch_state = - repo.branches - .get(branch) - .ok_or_else(|| VersioningError::BranchNotFound { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - })?; - - let total = branch_state.history.len(); - let max_limit = limit.clamp(1, 200); - let items: Vec = branch_state - .history - .iter() - .rev() - .skip(cursor) - .take(max_limit) - .filter_map(|id| repo.changesets.get(id).cloned()) - .collect(); - - let consumed = cursor + items.len(); - let next_cursor = if consumed < total { - Some(consumed) - } else { - None - }; - Ok(HistoryPage { items, next_cursor }) - } - - pub fn build_rollback_plan( - &self, - repo_id: &str, - branch: &str, - target_changeset_id: &str, - ) -> Result { - let repos = self.repos.read().expect("versioning lock poisoned"); - let repo = repos - .get(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - let branch_state = - repo.branches - .get(branch) - .ok_or_else(|| VersioningError::BranchNotFound { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - })?; - - let head_id = branch_state - .record - .head_changeset_id - .clone() - .ok_or_else(|| VersioningError::InvalidRollbackTarget { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - target_changeset_id: target_changeset_id.to_string(), - })?; - - if head_id == target_changeset_id { - return Err(VersioningError::InvalidRollbackTarget { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - target_changeset_id: target_changeset_id.to_string(), - }); - } - - if !branch_state - .history - .iter() - .any(|id| id == target_changeset_id) - { - return Err(VersioningError::InvalidRollbackTarget { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - target_changeset_id: target_changeset_id.to_string(), - }); - } - - let current = repo.snapshots.get(&head_id).cloned().unwrap_or_default(); - let target = repo - .snapshots - .get(target_changeset_id) - .cloned() - .ok_or_else(|| VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: target_changeset_id.to_string(), - })?; - - let mut asset_ids = BTreeSet::new(); - current.keys().for_each(|k| { - asset_ids.insert(k.clone()); - }); - target.keys().for_each(|k| { - asset_ids.insert(k.clone()); - }); - - let mut assets = Vec::new(); - for asset_id in asset_ids { - let current_asset = current.get(&asset_id); - let target_asset = target.get(&asset_id); - let current_hash = current_asset.map(|asset| asset.blob_hash.as_str()); - let target_hash = target_asset.map(|asset| asset.blob_hash.as_str()); - if current_hash == target_hash { - continue; - } - assets.push(AssetDelta { - asset_id: Some(asset_id.clone()), - path: target_asset - .map(|asset| asset.path.clone()) - .or_else(|| current_asset.map(|asset| asset.path.clone())) - .unwrap_or(asset_id), - from_blob_hash: current_asset.map(|asset| asset.blob_hash.clone()), - blob_hash: target_asset.map(|asset| asset.blob_hash.clone()), - }); - } - - Ok(RollbackPlan { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - base_changeset_id: head_id, - target_changeset_id: target_changeset_id.to_string(), - assets, - }) - } - - pub fn sync_snapshot( - &self, - repo_id: &str, - branch: &str, - to_changeset_id: Option<&str>, - ) -> Result { - let repos = self.repos.read().expect("versioning lock poisoned"); - let repo = repos - .get(repo_id) - .ok_or_else(|| VersioningError::RepoNotFound { - repo_id: repo_id.to_string(), - })?; - let branch_state = - repo.branches - .get(branch) - .ok_or_else(|| VersioningError::BranchNotFound { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - })?; - - let chosen = if let Some(id) = to_changeset_id { - if !branch_state.history.iter().any(|entry| entry == id) { - return Err(VersioningError::ChangesetNotFound { - repo_id: repo_id.to_string(), - changeset_id: id.to_string(), - }); - } - Some(id.to_string()) - } else { - branch_state.record.head_changeset_id.clone() - }; - - let snapshot_map = chosen - .as_ref() - .and_then(|id| repo.snapshots.get(id)) - .cloned() - .unwrap_or_default(); - let mut assets: Vec = snapshot_map - .into_iter() - .map(|(asset_id, asset)| SnapshotEntry { - asset_id, - path: asset.path, - blob_hash: asset.blob_hash, - }) - .collect(); - assets.sort_by(|a, b| { - a.path - .cmp(&b.path) - .then_with(|| a.asset_id.cmp(&b.asset_id)) - }); - - Ok(SyncSnapshot { - repo_id: repo_id.to_string(), - branch: branch.to_string(), - changeset_id: chosen, - assets, - }) - } - - fn submit_internal( - repo: &mut RepoState, - input: SubmitChangesetInput, - ) -> Result { - let SubmitChangesetInput { - repo_id, - branch, - base_changeset_id, - kind, - rollback_of, - author, - message, - visibility, - intent_id, - task_id, - agent_run_id, - session_id, - parent_checkpoint_id, - risk_level, - semantic_summary, - assets, - } = input; - - if base_changeset_id.is_none() { - return Err(VersioningError::BaseChangesetRequired); - } - - let branch_state = - repo.branches - .get_mut(&branch) - .ok_or_else(|| VersioningError::BranchNotFound { - repo_id: repo_id.clone(), - branch: branch.clone(), - })?; - - let expected = branch_state.record.head_changeset_id.clone(); - if expected.is_none() { - if base_changeset_id.as_deref() != Some(ROOT_BASE_CHANGESET_ID) { - return Err(VersioningError::BaseChangesetMismatch { - repo_id, - branch, - expected, - got: base_changeset_id, - }); - } - } else if base_changeset_id != expected { - return Err(VersioningError::BaseChangesetMismatch { - repo_id, - branch, - expected, - got: base_changeset_id, - }); - } - - let parent_changeset_id = branch_state.record.head_changeset_id.clone(); - let mut new_snapshot = parent_changeset_id - .as_ref() - .and_then(|id| repo.snapshots.get(id)) - .cloned() - .unwrap_or_default(); - - let mut normalized_assets = Vec::with_capacity(assets.len()); - for mut asset in assets { - let asset_id = asset.asset_id.clone().unwrap_or_else(|| asset.path.clone()); - asset.asset_id = Some(asset_id.clone()); - asset.from_blob_hash = new_snapshot - .get(&asset_id) - .map(|snapshot_asset| snapshot_asset.blob_hash.clone()); - - if let Some(hash) = &asset.blob_hash { - new_snapshot.insert( - asset_id.clone(), - SnapshotAsset { - asset_id, - path: asset.path.clone(), - blob_hash: hash.clone(), - }, - ); - } else { - new_snapshot.remove(&asset_id); - } - normalized_assets.push(asset); - } - Self::validate_snapshot_layout(&repo_id, &new_snapshot)?; - - let changeset_id = Uuid::new_v4().to_string(); - let status = match visibility { - ChangesetVisibility::Visible => ChangesetStatus::Visible, - ChangesetVisibility::Draft => ChangesetStatus::Draft, - }; - let staging_ref_value = if status == ChangesetStatus::Draft { - Some(staging_ref(&repo_id, &branch, &changeset_id)) - } else { - None - }; - let visible_ref_value = if status == ChangesetStatus::Visible { - Some(visible_ref(&branch)) - } else { - None - }; - let record = ChangesetRecord { - changeset_id: changeset_id.clone(), - repo_id, - branch: branch.clone(), - parent_changeset_id, - base_changeset_id, - kind, - rollback_of, - author, - message, - created_at: Utc::now(), - status, - approved_by: None, - approved_at: None, - promoted_at: None, - staging_ref: staging_ref_value, - visible_ref: visible_ref_value, - intent_id, - task_id, - agent_run_id, - session_id, - parent_checkpoint_id, - risk_level, - semantic_summary, - assets: normalized_assets, - }; - - repo.snapshots.insert(changeset_id.clone(), new_snapshot); - repo.changesets.insert(changeset_id.clone(), record.clone()); - if record.status == ChangesetStatus::Visible { - branch_state.record.head_changeset_id = Some(changeset_id.clone()); - branch_state.history.push(changeset_id); - } - - Ok(record) - } - - fn validate_snapshot_layout( - repo_id: &str, - snapshot: &HashMap, - ) -> Result<(), VersioningError> { - let mut paths = HashSet::with_capacity(snapshot.len()); - for asset in snapshot.values() { - let normalized = asset.path.replace('\\', "/"); - if !paths.insert(normalized.clone()) { - return Err(VersioningError::InvalidAssetLayout { - repo_id: repo_id.to_string(), - message: format!("duplicate asset path: {}", asset.path), - }); - } - } - for path in &paths { - for (index, byte) in path.bytes().enumerate() { - if byte == b'/' && paths.contains(&path[..index]) { - return Err(VersioningError::InvalidAssetLayout { - repo_id: repo_id.to_string(), - message: format!( - "asset path conflicts with parent asset: {} and {}", - &path[..index], - path - ), - }); - } - } - } - Ok(()) - } - - fn load_repos(path: &Path) -> Result, String> { - if !path.exists() { - return Ok(HashMap::new()); - } - - let bytes = std::fs::read(path) - .map_err(|error| format!("failed to read state file {}: {error}", path.display()))?; - serde_json::from_slice::>(&bytes) - .map_err(|error| format!("failed to parse state file {}: {error}", path.display())) - } - - async fn persist_repo( - &self, - repo_id: &str, - repos: &HashMap, - ) -> Result<(), String> { - if let Some(repo_pg) = &self.repo_pg { - if let Some(state) = repos.get(repo_id) { - repo_pg - .replace_repo_state(repo_id, state) - .await - .map_err(|error| format!("db persistence failed: {error}"))?; - } - return Ok(()); - } - - self.persist_repos_file(repos) - } - - fn persist_repos_file(&self, repos: &HashMap) -> Result<(), String> { - let Some(path) = self.persistence_path.as_ref() else { - return Ok(()); - }; - - if let Some(parent) = path.parent() { - if let Err(error) = std::fs::create_dir_all(parent) { - return Err(format!( - "failed to create versioning state dir {}: {}", - parent.display(), - error - )); - } - } - - let payload = match serde_json::to_vec_pretty(repos) { - Ok(payload) => payload, - Err(error) => return Err(format!("failed to serialize versioning state: {error}")), - }; - - let temp_path = path.with_extension("tmp"); - if let Err(error) = std::fs::write(&temp_path, payload) { - return Err(format!( - "failed to write versioning temp state {}: {}", - temp_path.display(), - error - )); - } - - if let Err(error) = replace_state_file(&temp_path, path) { - return Err(format!( - "failed to atomically replace versioning state {}: {}", - path.display(), - error - )); - } - Ok(()) - } -} - -impl Default for VersionManager { - fn default() -> Self { - Self::new() - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(super) struct BranchState { - record: BranchRecord, - history: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(super) struct RepoState { - default_branch: String, - branches: HashMap, - changesets: HashMap, - snapshots: HashMap>, -} - -impl RepoState { - fn new(created_by: &str) -> Self { - Self::new_with_default("main", created_by) - } - - fn new_with_default(default_branch: &str, created_by: &str) -> Self { - let mut repo = Self { - default_branch: default_branch.to_string(), - branches: HashMap::new(), - changesets: HashMap::new(), - snapshots: HashMap::new(), - }; - repo.ensure_default_branch(created_by); - repo - } - - fn ensure_default_branch(&mut self, created_by: &str) { - if self.branches.contains_key(&self.default_branch) { - return; - } - let record = BranchRecord { - name: self.default_branch.clone(), - created_by: created_by.to_string(), - created_at: Utc::now(), - is_default: true, - head_changeset_id: None, - }; - self.branches.insert( - self.default_branch.clone(), - BranchState { - record, - history: Vec::new(), - }, - ); - } - - fn default_head(&self) -> Option { - self.branches - .get(&self.default_branch) - .and_then(|branch| branch.record.head_changeset_id.clone()) - } - - fn lineage_to(&self, changeset_id: &str) -> Option> { - let mut chain = Vec::new(); - let mut current = Some(changeset_id.to_string()); - while let Some(id) = current { - let node = self.changesets.get(&id)?; - chain.push(id); - current = node.parent_changeset_id.clone(); - } - chain.reverse(); - Some(chain) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn submit_with_head_match_advances_branch_head() { - let manager = VersionManager::new(); - - let c1 = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-a".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "first".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![AssetDelta { - asset_id: None, - path: "a.txt".to_string(), - from_blob_hash: None, - blob_hash: Some("hash-1".to_string()), - }], - }) - .await - .expect("first commit should succeed"); - - let c2 = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-a".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(c1.changeset_id.clone()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "second".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![AssetDelta { - asset_id: None, - path: "a.txt".to_string(), - from_blob_hash: None, - blob_hash: Some("hash-2".to_string()), - }], - }) - .await - .expect("second commit should succeed"); - - let sync = manager - .sync_snapshot("repo-a", "main", None) - .expect("snapshot should exist"); - assert_eq!(sync.changeset_id, Some(c2.changeset_id)); - assert_eq!(sync.assets.len(), 1); - assert_eq!(sync.assets[0].blob_hash, "hash-2"); - } - - #[tokio::test] - async fn submit_rejects_conflicting_snapshot_paths() { - let manager = VersionManager::new(); - let error = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-layout".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "invalid layout".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![ - AssetDelta { - asset_id: Some("asset-parent".to_string()), - path: "Content".to_string(), - from_blob_hash: None, - blob_hash: Some("hash-parent".to_string()), - }, - AssetDelta { - asset_id: Some("asset-child".to_string()), - path: "Content/A.uasset".to_string(), - from_blob_hash: None, - blob_hash: Some("hash-child".to_string()), - }, - ], - }) - .await - .expect_err("conflicting paths must be rejected"); - - assert!(matches!(error, VersioningError::InvalidAssetLayout { .. })); - assert!(manager.list_repos().is_empty()); - } - - #[tokio::test] - async fn create_repo_creates_default_branch_and_rejects_duplicates() { - let manager = VersionManager::new(); - - let repo = manager - .create_repo("repo-explicit", "main", "alice") - .await - .expect("repo should be created"); - - assert_eq!(repo.repo_id, "repo-explicit"); - assert_eq!(repo.default_branch, "main"); - assert_eq!(repo.branch_count, 1); - assert_eq!(repo.default_head_changeset_id, None); - - let duplicate = manager - .create_repo("repo-explicit", "main", "alice") - .await - .expect_err("duplicate repo should fail"); - - assert_eq!( - duplicate, - VersioningError::RepoAlreadyExists { - repo_id: "repo-explicit".to_string(), - } - ); - } - - #[tokio::test] - async fn list_and_get_repo_info_return_default_branch() { - let manager = VersionManager::new(); - - manager - .create_repo("repo-info", "main", "alice") - .await - .expect("repo should be created"); - - let repos = manager.list_repos(); - assert_eq!(repos.len(), 1); - assert_eq!(repos[0].repo_id, "repo-info"); - assert_eq!(repos[0].default_branch, "main"); - - let info = manager - .get_repo_info("repo-info") - .expect("repo info should exist"); - assert_eq!(info.branches.len(), 1); - assert_eq!(info.branches[0].name, "main"); - assert!(info.branches[0].is_default); - } - - #[tokio::test] - async fn stale_base_is_rejected() { - let manager = VersionManager::new(); - - let c1 = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-b".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "first".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![], - }) - .await - .expect("first should succeed"); - - let c2 = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-b".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "invalid".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![], - }) - .await - .expect_err("stale base must fail"); - - assert_eq!( - c2, - VersioningError::BaseChangesetMismatch { - repo_id: "repo-b".to_string(), - branch: "main".to_string(), - expected: Some(c1.changeset_id), - got: Some(ROOT_BASE_CHANGESET_ID.to_string()), - } - ); - } - - #[tokio::test] - async fn rollback_plan_targets_existing_history() { - let manager = VersionManager::new(); - let c1 = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-c".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "first".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![AssetDelta { - asset_id: None, - path: "a".to_string(), - from_blob_hash: None, - blob_hash: Some("h1".to_string()), - }], - }) - .await - .expect("first commit"); - - let c2 = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-c".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(c1.changeset_id.clone()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "second".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![AssetDelta { - asset_id: None, - path: "a".to_string(), - from_blob_hash: None, - blob_hash: Some("h2".to_string()), - }], - }) - .await - .expect("second commit"); - - let plan = manager - .build_rollback_plan("repo-c", "main", &c1.changeset_id) - .expect("rollback plan"); - assert_eq!(plan.base_changeset_id, c2.changeset_id.clone()); - assert_eq!(plan.assets.len(), 1); - assert_eq!(plan.assets[0].blob_hash.as_deref(), Some("h1")); - - manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-c".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(plan.base_changeset_id.clone()), - kind: ChangesetKind::Rollback, - rollback_of: Some(plan.target_changeset_id), - author: "alice".to_string(), - message: "rollback".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: plan.assets, - }) - .await - .expect("rollback commit should be accepted"); - - let sync = manager - .sync_snapshot("repo-c", "main", None) - .expect("snapshot"); - assert_eq!(sync.assets[0].blob_hash, "h1"); - } - - #[tokio::test] - async fn draft_changeset_uses_staging_ref_and_promote_sets_visible_ref() { - let manager = VersionManager::new(); - - let base = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-gate".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "base".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![], - }) - .await - .expect("base changeset"); - - let draft = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-gate".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(base.changeset_id.clone()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "draft".to_string(), - visibility: ChangesetVisibility::Draft, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![], - }) - .await - .expect("draft changeset"); - - assert!(draft.staging_ref.is_some()); - assert_eq!(draft.visible_ref, None); - - let approved = manager - .approve_changeset("repo-gate", &draft.changeset_id, "reviewer") - .await - .expect("approve draft"); - assert_eq!(approved.visible_ref, None); - - let promoted = manager - .promote_changeset("repo-gate", &draft.changeset_id, "release-bot") - .await - .expect("promote approved"); - assert_eq!(promoted.visible_ref.as_deref(), Some("refs/heads/main")); - assert!(promoted.staging_ref.is_some()); - } - - #[tokio::test] - async fn changeset_gate_requires_approved_before_promote() { - let manager = VersionManager::new(); - - let base = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-gate-2".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "base".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![], - }) - .await - .expect("base changeset"); - - let draft = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-gate-2".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(base.changeset_id.clone()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "draft".to_string(), - visibility: ChangesetVisibility::Draft, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![], - }) - .await - .expect("draft changeset"); - - let gate_before = manager - .changeset_gate("repo-gate-2", &draft.changeset_id) - .expect("gate for draft"); - assert!(!gate_before.can_promote); - assert_eq!(gate_before.required_state, "approved"); - - manager - .approve_changeset("repo-gate-2", &draft.changeset_id, "reviewer") - .await - .expect("approve draft"); - - let gate_after = manager - .changeset_gate("repo-gate-2", &draft.changeset_id) - .expect("gate for approved"); - assert!(gate_after.can_promote); - assert_eq!(gate_after.required_state, "approved"); - } - - #[tokio::test] - async fn submit_preserves_agent_session_metadata() { - let manager = VersionManager::new(); - - let changeset = manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-agent-meta".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "agent-a".to_string(), - message: "draft from checkpoint".to_string(), - visibility: ChangesetVisibility::Draft, - intent_id: Some("intent-1".to_string()), - task_id: Some("task-1".to_string()), - agent_run_id: Some("run-1".to_string()), - session_id: Some("session-1".to_string()), - parent_checkpoint_id: Some("checkpoint-1".to_string()), - risk_level: Some("local".to_string()), - semantic_summary: Some("inventory implementation draft".to_string()), - assets: vec![], - }) - .await - .expect("draft changeset"); - - assert_eq!(changeset.status, ChangesetStatus::Draft); - assert_eq!(changeset.intent_id.as_deref(), Some("intent-1")); - assert_eq!(changeset.task_id.as_deref(), Some("task-1")); - assert_eq!(changeset.agent_run_id.as_deref(), Some("run-1")); - assert_eq!(changeset.session_id.as_deref(), Some("session-1")); - assert_eq!( - changeset.parent_checkpoint_id.as_deref(), - Some("checkpoint-1") - ); - assert_eq!(changeset.risk_level.as_deref(), Some("local")); - assert_eq!( - changeset.semantic_summary.as_deref(), - Some("inventory implementation draft") - ); - } - - #[tokio::test] - async fn persists_state_across_manager_restarts() { - let state_file = - std::env::temp_dir().join(format!("hypertide-versioning-{}.json", Uuid::new_v4())); - - let first_manager = VersionManager::with_persistence(&state_file); - first_manager - .submit_changeset(SubmitChangesetInput { - repo_id: "repo-p".to_string(), - branch: "main".to_string(), - base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), - kind: ChangesetKind::Normal, - rollback_of: None, - author: "alice".to_string(), - message: "first".to_string(), - visibility: ChangesetVisibility::Visible, - intent_id: None, - task_id: None, - agent_run_id: None, - session_id: None, - parent_checkpoint_id: None, - risk_level: None, - semantic_summary: None, - assets: vec![AssetDelta { - asset_id: None, - path: "env/config.json".to_string(), - from_blob_hash: None, - blob_hash: Some("blob-v1".to_string()), - }], - }) - .await - .expect("submit should persist"); - - let second_manager = VersionManager::with_persistence(&state_file); - let snapshot = second_manager - .sync_snapshot("repo-p", "main", None) - .expect("snapshot should load from persistence"); - assert_eq!(snapshot.assets.len(), 1); - assert_eq!(snapshot.assets[0].path, "env/config.json"); - assert_eq!(snapshot.assets[0].blob_hash, "blob-v1"); - - let _ = std::fs::remove_file(state_file); - } - - #[tokio::test] - async fn persistence_failure_does_not_publish_in_memory_state() { - let blocker = - std::env::temp_dir().join(format!("hypertide-versioning-blocker-{}", Uuid::new_v4())); - std::fs::write(&blocker, b"not-a-directory").expect("create blocker"); - let manager = VersionManager::with_persistence(blocker.join("state.json")); - - let error = manager - .create_repo("repo-not-persisted", "main", "alice") - .await - .expect_err("persistence must fail"); - - assert!(matches!(error, VersioningError::Persistence { .. })); - assert!(manager.list_repos().is_empty()); - let _ = std::fs::remove_file(blocker); - } - - #[tokio::test] - async fn file_persistence_supports_consecutive_mutations() { - let state_file = - std::env::temp_dir().join(format!("hypertide-versioning-{}.json", Uuid::new_v4())); - let manager = VersionManager::with_persistence(&state_file); - - manager - .create_repo("repo-p", "main", "alice") - .await - .expect("first persistence write"); - manager - .create_branch("repo-p", "feature", None, "alice") - .await - .expect("replacement persistence write"); - - let reloaded = VersionManager::with_persistence(&state_file); - let branches = reloaded - .list_branches("repo-p") - .expect("load persisted repo"); - assert_eq!(branches.len(), 2); - assert!(branches.iter().any(|branch| branch.name == "feature")); - - let _ = std::fs::remove_file(state_file); - } -} +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, RwLock}; + +use crate::core::error::HyperTideError; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use sqlx::PgPool; +use uuid::Uuid; + +pub mod repo_pg; +use self::repo_pg::VersionRepoPg; + +pub const ROOT_BASE_CHANGESET_ID: &str = "ROOT"; + +/// Returns true when a changeset's `base` is a valid predecessor for the current +/// branch `head`. Mirrors the acceptance rule in `submit_internal`: an empty head +/// (no commits yet) accepts the `ROOT` sentinel, otherwise the base must equal the +/// current head. Used by both promote and the changeset gate so a draft-first +/// changeset (which never advanced the head) can still be promoted. +fn head_accepts_base(head: &Option, base: &Option) -> bool { + match head { + None => base.as_deref() == Some(ROOT_BASE_CHANGESET_ID), + Some(_) => head == base, + } +} + +#[cfg(windows)] +fn replace_state_file(temp_path: &Path, state_path: &Path) -> std::io::Result<()> { + use std::iter::once; + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::ReplaceFileW; + + if !state_path.exists() { + return std::fs::rename(temp_path, state_path); + } + + let state_wide = state_path + .as_os_str() + .encode_wide() + .chain(once(0)) + .collect::>(); + let temp_wide = temp_path + .as_os_str() + .encode_wide() + .chain(once(0)) + .collect::>(); + let replaced = unsafe { + ReplaceFileW( + state_wide.as_ptr(), + temp_wide.as_ptr(), + std::ptr::null(), + 0, + std::ptr::null(), + std::ptr::null(), + ) + }; + if replaced == 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(()) + } +} + +#[cfg(not(windows))] +fn replace_state_file(temp_path: &Path, state_path: &Path) -> std::io::Result<()> { + std::fs::rename(temp_path, state_path) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ChangesetKind { + Normal, + Rollback, +} + +impl ChangesetKind { + pub fn as_str(&self) -> &'static str { + match self { + ChangesetKind::Normal => "normal", + ChangesetKind::Rollback => "rollback", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ChangesetVisibility { + Visible, + Draft, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ChangesetStatus { + Draft, + Approved, + Visible, +} + +impl ChangesetStatus { + pub fn as_str(&self) -> &'static str { + match self { + ChangesetStatus::Draft => "draft", + ChangesetStatus::Approved => "approved", + ChangesetStatus::Visible => "visible", + } + } +} + +fn default_changeset_status() -> ChangesetStatus { + ChangesetStatus::Visible +} + +fn staging_ref(repo_id: &str, branch: &str, changeset_id: &str) -> String { + format!("refs/ht/staging/{repo_id}/{branch}/{changeset_id}") +} + +fn visible_ref(branch: &str) -> String { + format!("refs/heads/{branch}") +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AssetDelta { + #[serde(default)] + pub asset_id: Option, + pub path: String, + #[serde(default)] + pub from_blob_hash: Option, + pub blob_hash: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ChangesetRecord { + pub changeset_id: String, + pub repo_id: String, + pub branch: String, + pub parent_changeset_id: Option, + pub base_changeset_id: Option, + pub kind: ChangesetKind, + pub rollback_of: Option, + pub author: String, + pub message: String, + pub created_at: DateTime, + #[serde(default = "default_changeset_status")] + pub status: ChangesetStatus, + pub approved_by: Option, + pub approved_at: Option>, + pub promoted_at: Option>, + #[serde(default)] + pub staging_ref: Option, + #[serde(default)] + pub visible_ref: Option, + #[serde(default)] + pub intent_id: Option, + #[serde(default)] + pub task_id: Option, + #[serde(default)] + pub agent_run_id: Option, + #[serde(default)] + pub session_id: Option, + #[serde(default)] + pub parent_checkpoint_id: Option, + #[serde(default)] + pub risk_level: Option, + #[serde(default)] + pub semantic_summary: Option, + pub assets: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BranchRecord { + pub name: String, + pub created_by: String, + pub created_at: DateTime, + pub is_default: bool, + pub head_changeset_id: Option, +} + +#[derive(Debug, Clone)] +pub struct SubmitChangesetInput { + pub repo_id: String, + pub branch: String, + pub base_changeset_id: Option, + pub kind: ChangesetKind, + pub rollback_of: Option, + pub author: String, + pub message: String, + pub visibility: ChangesetVisibility, + pub intent_id: Option, + pub task_id: Option, + pub agent_run_id: Option, + pub session_id: Option, + pub parent_checkpoint_id: Option, + pub risk_level: Option, + pub semantic_summary: Option, + pub assets: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct HistoryPage { + pub items: Vec, + pub next_cursor: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ChangesetGate { + pub repo_id: String, + pub changeset_id: String, + pub branch: String, + pub status: ChangesetStatus, + pub required_state: &'static str, + pub can_promote: bool, + pub blocking_reason: Option, + pub base_changeset_id: Option, + pub branch_head_changeset_id: Option, + pub staging_ref: Option, + pub visible_ref: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct SyncSnapshot { + pub repo_id: String, + pub branch: String, + pub changeset_id: Option, + pub assets: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct RepoSummary { + pub repo_id: String, + pub default_branch: String, + pub branch_count: usize, + pub default_head_changeset_id: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct RepoInfo { + pub repo_id: String, + pub default_branch: String, + pub branch_count: usize, + pub default_head_changeset_id: Option, + pub branches: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct SnapshotEntry { + pub asset_id: String, + pub path: String, + pub blob_hash: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct SnapshotAsset { + pub asset_id: String, + pub path: String, + pub blob_hash: String, +} + +#[derive(Debug, Clone)] +pub struct RollbackPlan { + pub repo_id: String, + pub branch: String, + pub base_changeset_id: String, + pub target_changeset_id: String, + pub assets: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum VersioningError { + RepoAlreadyExists { + repo_id: String, + }, + RepoNotFound { + repo_id: String, + }, + BranchNotFound { + repo_id: String, + branch: String, + }, + BranchAlreadyExists { + repo_id: String, + branch: String, + }, + ChangesetNotFound { + repo_id: String, + changeset_id: String, + }, + BaseChangesetRequired, + BaseChangesetMismatch { + repo_id: String, + branch: String, + expected: Option, + got: Option, + }, + InvalidRollbackTarget { + repo_id: String, + branch: String, + target_changeset_id: String, + }, + InvalidChangesetState { + repo_id: String, + changeset_id: String, + status: ChangesetStatus, + expected: &'static str, + }, + InvalidAssetLayout { + repo_id: String, + message: String, + }, + SelfApprovalForbidden { + repo_id: String, + changeset_id: String, + actor: String, + }, + Persistence { + message: String, + }, +} + +/// Whether approve/promote must be performed by someone other than the changeset +/// author (four-eyes). Opt-in and off by default so existing single-user flows are +/// unaffected; operators enable it with `HYPERTIDE_REQUIRE_SEPARATE_APPROVER=1`. +fn separate_approver_required() -> bool { + std::env::var("HYPERTIDE_REQUIRE_SEPARATE_APPROVER") + .ok() + .map(|value| value == "1" || value.eq_ignore_ascii_case("true")) + .unwrap_or(false) +} + +#[derive(Clone)] +pub struct VersionManager { + repos: Arc>>, + persistence_path: Option, + repo_pg: Option, + mutation_lock: Arc>, +} + +impl VersionManager { + pub fn new() -> Self { + Self { + repos: Arc::new(RwLock::new(HashMap::new())), + persistence_path: None, + repo_pg: None, + mutation_lock: Arc::new(tokio::sync::Mutex::new(())), + } + } + + pub fn with_persistence(path: impl AsRef) -> Self { + let persistence_path = path.as_ref().to_path_buf(); + let repos = match Self::load_repos(&persistence_path) { + Ok(repos) => repos, + Err(error) => { + tracing::warn!( + "versioning persistence load failed at {}: {}", + persistence_path.display(), + error + ); + HashMap::new() + } + }; + + Self { + repos: Arc::new(RwLock::new(repos)), + persistence_path: Some(persistence_path), + repo_pg: None, + mutation_lock: Arc::new(tokio::sync::Mutex::new(())), + } + } + + pub async fn with_pg(pool: PgPool) -> Result { + let repo_pg = VersionRepoPg::new(pool); + let repos = repo_pg.load_repos().await.map_err(|error| { + HyperTideError::Persistence(format!("failed to load versioning state from db: {error}")) + })?; + Ok(Self { + repos: Arc::new(RwLock::new(repos)), + persistence_path: None, + repo_pg: Some(repo_pg), + mutation_lock: Arc::new(tokio::sync::Mutex::new(())), + }) + } + + pub async fn create_repo( + &self, + repo_id: &str, + default_branch: &str, + created_by: &str, + ) -> Result { + let _mutation = self.mutation_lock.lock().await; + let (info, snapshot) = { + let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); + if snapshot.contains_key(repo_id) { + return Err(VersioningError::RepoAlreadyExists { + repo_id: repo_id.to_string(), + }); + } + + let repo = RepoState::new_with_default(default_branch, created_by); + snapshot.insert(repo_id.to_string(), repo); + let info = + Self::repo_info_from_state(repo_id, snapshot.get(repo_id).expect("repo exists")); + (info, snapshot) + }; + self.persist_repo(repo_id, &snapshot) + .await + .map_err(|message| VersioningError::Persistence { message })?; + *self.repos.write().expect("versioning lock poisoned") = snapshot; + Ok(info) + } + + pub fn list_repos(&self) -> Vec { + let repos = self.repos.read().expect("versioning lock poisoned"); + let mut items: Vec = repos + .iter() + .map(|(repo_id, repo)| Self::repo_summary_from_state(repo_id, repo)) + .collect(); + items.sort_by(|a, b| a.repo_id.cmp(&b.repo_id)); + items + } + + pub fn get_repo_info(&self, repo_id: &str) -> Result { + let repos = self.repos.read().expect("versioning lock poisoned"); + let repo = repos + .get(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + Ok(Self::repo_info_from_state(repo_id, repo)) + } + + pub async fn create_branch( + &self, + repo_id: &str, + branch: &str, + from_changeset_id: Option<&str>, + created_by: &str, + ) -> Result { + let _mutation = self.mutation_lock.lock().await; + let (record, snapshot) = { + let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); + let repo = snapshot + .entry(repo_id.to_string()) + .or_insert_with(|| RepoState::new(created_by)); + repo.ensure_default_branch(created_by); + + if repo.branches.contains_key(branch) { + return Err(VersioningError::BranchAlreadyExists { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + }); + } + + let head = if let Some(id) = from_changeset_id { + if !repo.changesets.contains_key(id) { + return Err(VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: id.to_string(), + }); + } + Some(id.to_string()) + } else { + repo.default_head() + }; + + let history = if let Some(ref head_id) = head { + repo.lineage_to(head_id) + .ok_or_else(|| VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: head_id.clone(), + })? + } else { + Vec::new() + }; + + let record = BranchRecord { + name: branch.to_string(), + created_by: created_by.to_string(), + created_at: Utc::now(), + is_default: false, + head_changeset_id: head.clone(), + }; + + repo.branches.insert( + branch.to_string(), + BranchState { + record: record.clone(), + history, + }, + ); + + (record, snapshot) + }; + self.persist_repo(repo_id, &snapshot) + .await + .map_err(|message| VersioningError::Persistence { message })?; + *self.repos.write().expect("versioning lock poisoned") = snapshot; + Ok(record) + } + + fn repo_summary_from_state(repo_id: &str, repo: &RepoState) -> RepoSummary { + RepoSummary { + repo_id: repo_id.to_string(), + default_branch: repo.default_branch.clone(), + branch_count: repo.branches.len(), + default_head_changeset_id: repo.default_head(), + } + } + + fn repo_info_from_state(repo_id: &str, repo: &RepoState) -> RepoInfo { + let mut branches: Vec = + repo.branches.values().map(|b| b.record.clone()).collect(); + branches.sort_by(|a, b| a.name.cmp(&b.name)); + RepoInfo { + repo_id: repo_id.to_string(), + default_branch: repo.default_branch.clone(), + branch_count: branches.len(), + default_head_changeset_id: repo.default_head(), + branches, + } + } + + pub fn list_branches(&self, repo_id: &str) -> Result, VersioningError> { + let repos = self.repos.read().expect("versioning lock poisoned"); + let repo = repos + .get(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + + let mut items: Vec = + repo.branches.values().map(|b| b.record.clone()).collect(); + items.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(items) + } + + pub async fn submit_changeset( + &self, + input: SubmitChangesetInput, + ) -> Result { + let _mutation = self.mutation_lock.lock().await; + let repo_id = input.repo_id.clone(); + let (record, snapshot) = { + let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); + let repo = snapshot + .entry(input.repo_id.clone()) + .or_insert_with(|| RepoState::new(&input.author)); + repo.ensure_default_branch(&input.author); + let record = Self::submit_internal(repo, input)?; + (record, snapshot) + }; + self.persist_repo(&repo_id, &snapshot) + .await + .map_err(|message| VersioningError::Persistence { message })?; + *self.repos.write().expect("versioning lock poisoned") = snapshot; + Ok(record) + } + + pub async fn approve_changeset( + &self, + repo_id: &str, + changeset_id: &str, + approver: &str, + ) -> Result { + let _mutation = self.mutation_lock.lock().await; + let (record, snapshot) = { + let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); + let repo = snapshot + .get_mut(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + let record = repo.changesets.get_mut(changeset_id).ok_or_else(|| { + VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + } + })?; + + if separate_approver_required() && record.author == approver { + return Err(VersioningError::SelfApprovalForbidden { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + actor: approver.to_string(), + }); + } + + match record.status { + ChangesetStatus::Draft => { + record.status = ChangesetStatus::Approved; + record.approved_by = Some(approver.to_string()); + record.approved_at = Some(Utc::now()); + } + status => { + return Err(VersioningError::InvalidChangesetState { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + status, + expected: "draft", + }); + } + } + + (record.clone(), snapshot) + }; + self.persist_repo(repo_id, &snapshot) + .await + .map_err(|message| VersioningError::Persistence { message })?; + *self.repos.write().expect("versioning lock poisoned") = snapshot; + Ok(record) + } + + pub async fn promote_changeset( + &self, + repo_id: &str, + changeset_id: &str, + promoter: &str, + ) -> Result { + let _mutation = self.mutation_lock.lock().await; + let (record, snapshot) = { + let mut snapshot = self.repos.read().expect("versioning lock poisoned").clone(); + let repo = snapshot + .get_mut(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + + let record_view = repo.changesets.get(changeset_id).ok_or_else(|| { + VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + } + })?; + if record_view.status != ChangesetStatus::Approved { + return Err(VersioningError::InvalidChangesetState { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + status: record_view.status, + expected: "approved", + }); + } + if separate_approver_required() && record_view.author == promoter { + return Err(VersioningError::SelfApprovalForbidden { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + actor: promoter.to_string(), + }); + } + + let branch = record_view.branch.clone(); + let base = record_view.base_changeset_id.clone(); + let branch_state = + repo.branches + .get_mut(&branch) + .ok_or_else(|| VersioningError::BranchNotFound { + repo_id: repo_id.to_string(), + branch: branch.clone(), + })?; + let expected_head = branch_state.record.head_changeset_id.clone(); + if !head_accepts_base(&expected_head, &base) { + return Err(VersioningError::BaseChangesetMismatch { + repo_id: repo_id.to_string(), + branch, + expected: expected_head, + got: base, + }); + } + + branch_state.record.head_changeset_id = Some(changeset_id.to_string()); + if !branch_state.history.iter().any(|id| id == changeset_id) { + branch_state.history.push(changeset_id.to_string()); + } + + let record = repo.changesets.get_mut(changeset_id).ok_or_else(|| { + VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + } + })?; + record.status = ChangesetStatus::Visible; + if record.approved_by.is_none() { + record.approved_by = Some(promoter.to_string()); + record.approved_at = Some(Utc::now()); + } + record.promoted_at = Some(Utc::now()); + record.visible_ref = Some(visible_ref(&record.branch)); + + (record.clone(), snapshot) + }; + self.persist_repo(repo_id, &snapshot) + .await + .map_err(|message| VersioningError::Persistence { message })?; + *self.repos.write().expect("versioning lock poisoned") = snapshot; + Ok(record) + } + + pub fn changeset_gate( + &self, + repo_id: &str, + changeset_id: &str, + ) -> Result { + let repos = self.repos.read().expect("versioning lock poisoned"); + let repo = repos + .get(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + let record = repo.changesets.get(changeset_id).ok_or_else(|| { + VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + } + })?; + let branch_state = + repo.branches + .get(&record.branch) + .ok_or_else(|| VersioningError::BranchNotFound { + repo_id: repo_id.to_string(), + branch: record.branch.clone(), + })?; + let current_head = branch_state.record.head_changeset_id.clone(); + let base = record.base_changeset_id.clone(); + + let (can_promote, blocking_reason) = if record.status != ChangesetStatus::Approved { + ( + false, + Some(format!( + "changeset status is {}, expected approved", + record.status.as_str() + )), + ) + } else if !head_accepts_base(¤t_head, &base) { + ( + false, + Some(format!( + "branch head mismatch: current={current_head:?}, base={base:?}" + )), + ) + } else { + (true, None) + }; + + Ok(ChangesetGate { + repo_id: repo_id.to_string(), + changeset_id: changeset_id.to_string(), + branch: record.branch.clone(), + status: record.status, + required_state: "approved", + can_promote, + blocking_reason, + base_changeset_id: base, + branch_head_changeset_id: current_head, + staging_ref: record.staging_ref.clone(), + visible_ref: record.visible_ref.clone(), + }) + } + + pub fn history( + &self, + repo_id: &str, + branch: &str, + limit: usize, + cursor: usize, + ) -> Result { + let repos = self.repos.read().expect("versioning lock poisoned"); + let repo = repos + .get(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + let branch_state = + repo.branches + .get(branch) + .ok_or_else(|| VersioningError::BranchNotFound { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + })?; + + let total = branch_state.history.len(); + let max_limit = limit.clamp(1, 200); + let items: Vec = branch_state + .history + .iter() + .rev() + .skip(cursor) + .take(max_limit) + .filter_map(|id| repo.changesets.get(id).cloned()) + .collect(); + + let consumed = cursor + items.len(); + let next_cursor = if consumed < total { + Some(consumed) + } else { + None + }; + Ok(HistoryPage { items, next_cursor }) + } + + pub fn build_rollback_plan( + &self, + repo_id: &str, + branch: &str, + target_changeset_id: &str, + ) -> Result { + let repos = self.repos.read().expect("versioning lock poisoned"); + let repo = repos + .get(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + let branch_state = + repo.branches + .get(branch) + .ok_or_else(|| VersioningError::BranchNotFound { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + })?; + + let head_id = branch_state + .record + .head_changeset_id + .clone() + .ok_or_else(|| VersioningError::InvalidRollbackTarget { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + target_changeset_id: target_changeset_id.to_string(), + })?; + + if head_id == target_changeset_id { + return Err(VersioningError::InvalidRollbackTarget { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + target_changeset_id: target_changeset_id.to_string(), + }); + } + + if !branch_state + .history + .iter() + .any(|id| id == target_changeset_id) + { + return Err(VersioningError::InvalidRollbackTarget { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + target_changeset_id: target_changeset_id.to_string(), + }); + } + + let current = repo.snapshots.get(&head_id).cloned().unwrap_or_default(); + let target = repo + .snapshots + .get(target_changeset_id) + .cloned() + .ok_or_else(|| VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: target_changeset_id.to_string(), + })?; + + let mut asset_ids = BTreeSet::new(); + current.keys().for_each(|k| { + asset_ids.insert(k.clone()); + }); + target.keys().for_each(|k| { + asset_ids.insert(k.clone()); + }); + + let mut assets = Vec::new(); + for asset_id in asset_ids { + let current_asset = current.get(&asset_id); + let target_asset = target.get(&asset_id); + let current_hash = current_asset.map(|asset| asset.blob_hash.as_str()); + let target_hash = target_asset.map(|asset| asset.blob_hash.as_str()); + if current_hash == target_hash { + continue; + } + assets.push(AssetDelta { + asset_id: Some(asset_id.clone()), + path: target_asset + .map(|asset| asset.path.clone()) + .or_else(|| current_asset.map(|asset| asset.path.clone())) + .unwrap_or(asset_id), + from_blob_hash: current_asset.map(|asset| asset.blob_hash.clone()), + blob_hash: target_asset.map(|asset| asset.blob_hash.clone()), + }); + } + + Ok(RollbackPlan { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + base_changeset_id: head_id, + target_changeset_id: target_changeset_id.to_string(), + assets, + }) + } + + pub fn sync_snapshot( + &self, + repo_id: &str, + branch: &str, + to_changeset_id: Option<&str>, + ) -> Result { + let repos = self.repos.read().expect("versioning lock poisoned"); + let repo = repos + .get(repo_id) + .ok_or_else(|| VersioningError::RepoNotFound { + repo_id: repo_id.to_string(), + })?; + let branch_state = + repo.branches + .get(branch) + .ok_or_else(|| VersioningError::BranchNotFound { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + })?; + + let chosen = if let Some(id) = to_changeset_id { + if !branch_state.history.iter().any(|entry| entry == id) { + return Err(VersioningError::ChangesetNotFound { + repo_id: repo_id.to_string(), + changeset_id: id.to_string(), + }); + } + Some(id.to_string()) + } else { + branch_state.record.head_changeset_id.clone() + }; + + let snapshot_map = chosen + .as_ref() + .and_then(|id| repo.snapshots.get(id)) + .cloned() + .unwrap_or_default(); + let mut assets: Vec = snapshot_map + .into_iter() + .map(|(asset_id, asset)| SnapshotEntry { + asset_id, + path: asset.path, + blob_hash: asset.blob_hash, + }) + .collect(); + assets.sort_by(|a, b| { + a.path + .cmp(&b.path) + .then_with(|| a.asset_id.cmp(&b.asset_id)) + }); + + Ok(SyncSnapshot { + repo_id: repo_id.to_string(), + branch: branch.to_string(), + changeset_id: chosen, + assets, + }) + } + + fn submit_internal( + repo: &mut RepoState, + input: SubmitChangesetInput, + ) -> Result { + let SubmitChangesetInput { + repo_id, + branch, + base_changeset_id, + kind, + rollback_of, + author, + message, + visibility, + intent_id, + task_id, + agent_run_id, + session_id, + parent_checkpoint_id, + risk_level, + semantic_summary, + assets, + } = input; + + if base_changeset_id.is_none() { + return Err(VersioningError::BaseChangesetRequired); + } + + let branch_state = + repo.branches + .get_mut(&branch) + .ok_or_else(|| VersioningError::BranchNotFound { + repo_id: repo_id.clone(), + branch: branch.clone(), + })?; + + let expected = branch_state.record.head_changeset_id.clone(); + if expected.is_none() { + if base_changeset_id.as_deref() != Some(ROOT_BASE_CHANGESET_ID) { + return Err(VersioningError::BaseChangesetMismatch { + repo_id, + branch, + expected, + got: base_changeset_id, + }); + } + } else if base_changeset_id != expected { + return Err(VersioningError::BaseChangesetMismatch { + repo_id, + branch, + expected, + got: base_changeset_id, + }); + } + + let parent_changeset_id = branch_state.record.head_changeset_id.clone(); + let mut new_snapshot = parent_changeset_id + .as_ref() + .and_then(|id| repo.snapshots.get(id)) + .cloned() + .unwrap_or_default(); + + let mut normalized_assets = Vec::with_capacity(assets.len()); + for mut asset in assets { + let asset_id = asset.asset_id.clone().unwrap_or_else(|| asset.path.clone()); + asset.asset_id = Some(asset_id.clone()); + asset.from_blob_hash = new_snapshot + .get(&asset_id) + .map(|snapshot_asset| snapshot_asset.blob_hash.clone()); + + if let Some(hash) = &asset.blob_hash { + new_snapshot.insert( + asset_id.clone(), + SnapshotAsset { + asset_id, + path: asset.path.clone(), + blob_hash: hash.clone(), + }, + ); + } else { + new_snapshot.remove(&asset_id); + } + normalized_assets.push(asset); + } + Self::validate_snapshot_layout(&repo_id, &new_snapshot)?; + + let changeset_id = Uuid::new_v4().to_string(); + let status = match visibility { + ChangesetVisibility::Visible => ChangesetStatus::Visible, + ChangesetVisibility::Draft => ChangesetStatus::Draft, + }; + let staging_ref_value = if status == ChangesetStatus::Draft { + Some(staging_ref(&repo_id, &branch, &changeset_id)) + } else { + None + }; + let visible_ref_value = if status == ChangesetStatus::Visible { + Some(visible_ref(&branch)) + } else { + None + }; + let record = ChangesetRecord { + changeset_id: changeset_id.clone(), + repo_id, + branch: branch.clone(), + parent_changeset_id, + base_changeset_id, + kind, + rollback_of, + author, + message, + created_at: Utc::now(), + status, + approved_by: None, + approved_at: None, + promoted_at: None, + staging_ref: staging_ref_value, + visible_ref: visible_ref_value, + intent_id, + task_id, + agent_run_id, + session_id, + parent_checkpoint_id, + risk_level, + semantic_summary, + assets: normalized_assets, + }; + + repo.snapshots.insert(changeset_id.clone(), new_snapshot); + repo.changesets.insert(changeset_id.clone(), record.clone()); + if record.status == ChangesetStatus::Visible { + branch_state.record.head_changeset_id = Some(changeset_id.clone()); + branch_state.history.push(changeset_id); + } + + Ok(record) + } + + fn validate_snapshot_layout( + repo_id: &str, + snapshot: &HashMap, + ) -> Result<(), VersioningError> { + let mut paths = HashSet::with_capacity(snapshot.len()); + for asset in snapshot.values() { + let normalized = asset.path.replace('\\', "/"); + if !paths.insert(normalized.clone()) { + return Err(VersioningError::InvalidAssetLayout { + repo_id: repo_id.to_string(), + message: format!("duplicate asset path: {}", asset.path), + }); + } + } + for path in &paths { + for (index, byte) in path.bytes().enumerate() { + if byte == b'/' && paths.contains(&path[..index]) { + return Err(VersioningError::InvalidAssetLayout { + repo_id: repo_id.to_string(), + message: format!( + "asset path conflicts with parent asset: {} and {}", + &path[..index], + path + ), + }); + } + } + } + Ok(()) + } + + fn load_repos(path: &Path) -> Result, String> { + if !path.exists() { + return Ok(HashMap::new()); + } + + let bytes = std::fs::read(path) + .map_err(|error| format!("failed to read state file {}: {error}", path.display()))?; + serde_json::from_slice::>(&bytes) + .map_err(|error| format!("failed to parse state file {}: {error}", path.display())) + } + + async fn persist_repo( + &self, + repo_id: &str, + repos: &HashMap, + ) -> Result<(), String> { + if let Some(repo_pg) = &self.repo_pg { + if let Some(state) = repos.get(repo_id) { + repo_pg + .replace_repo_state(repo_id, state) + .await + .map_err(|error| format!("db persistence failed: {error}"))?; + } + return Ok(()); + } + + self.persist_repos_file(repos) + } + + fn persist_repos_file(&self, repos: &HashMap) -> Result<(), String> { + let Some(path) = self.persistence_path.as_ref() else { + return Ok(()); + }; + + if let Some(parent) = path.parent() { + if let Err(error) = std::fs::create_dir_all(parent) { + return Err(format!( + "failed to create versioning state dir {}: {}", + parent.display(), + error + )); + } + } + + let payload = match serde_json::to_vec_pretty(repos) { + Ok(payload) => payload, + Err(error) => return Err(format!("failed to serialize versioning state: {error}")), + }; + + let temp_path = path.with_extension("tmp"); + if let Err(error) = std::fs::write(&temp_path, payload) { + return Err(format!( + "failed to write versioning temp state {}: {}", + temp_path.display(), + error + )); + } + + if let Err(error) = replace_state_file(&temp_path, path) { + return Err(format!( + "failed to atomically replace versioning state {}: {}", + path.display(), + error + )); + } + Ok(()) + } +} + +impl Default for VersionManager { + fn default() -> Self { + Self::new() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct BranchState { + record: BranchRecord, + history: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct RepoState { + default_branch: String, + branches: HashMap, + changesets: HashMap, + snapshots: HashMap>, +} + +impl RepoState { + fn new(created_by: &str) -> Self { + Self::new_with_default("main", created_by) + } + + fn new_with_default(default_branch: &str, created_by: &str) -> Self { + let mut repo = Self { + default_branch: default_branch.to_string(), + branches: HashMap::new(), + changesets: HashMap::new(), + snapshots: HashMap::new(), + }; + repo.ensure_default_branch(created_by); + repo + } + + fn ensure_default_branch(&mut self, created_by: &str) { + if self.branches.contains_key(&self.default_branch) { + return; + } + let record = BranchRecord { + name: self.default_branch.clone(), + created_by: created_by.to_string(), + created_at: Utc::now(), + is_default: true, + head_changeset_id: None, + }; + self.branches.insert( + self.default_branch.clone(), + BranchState { + record, + history: Vec::new(), + }, + ); + } + + fn default_head(&self) -> Option { + self.branches + .get(&self.default_branch) + .and_then(|branch| branch.record.head_changeset_id.clone()) + } + + fn lineage_to(&self, changeset_id: &str) -> Option> { + let mut chain = Vec::new(); + let mut current = Some(changeset_id.to_string()); + while let Some(id) = current { + let node = self.changesets.get(&id)?; + chain.push(id); + current = node.parent_changeset_id.clone(); + } + chain.reverse(); + Some(chain) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn submit_with_head_match_advances_branch_head() { + let manager = VersionManager::new(); + + let c1 = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-a".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "first".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![AssetDelta { + asset_id: None, + path: "a.txt".to_string(), + from_blob_hash: None, + blob_hash: Some("hash-1".to_string()), + }], + }) + .await + .expect("first commit should succeed"); + + let c2 = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-a".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(c1.changeset_id.clone()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "second".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![AssetDelta { + asset_id: None, + path: "a.txt".to_string(), + from_blob_hash: None, + blob_hash: Some("hash-2".to_string()), + }], + }) + .await + .expect("second commit should succeed"); + + let sync = manager + .sync_snapshot("repo-a", "main", None) + .expect("snapshot should exist"); + assert_eq!(sync.changeset_id, Some(c2.changeset_id)); + assert_eq!(sync.assets.len(), 1); + assert_eq!(sync.assets[0].blob_hash, "hash-2"); + } + + #[tokio::test] + async fn submit_rejects_conflicting_snapshot_paths() { + let manager = VersionManager::new(); + let error = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-layout".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "invalid layout".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![ + AssetDelta { + asset_id: Some("asset-parent".to_string()), + path: "Content".to_string(), + from_blob_hash: None, + blob_hash: Some("hash-parent".to_string()), + }, + AssetDelta { + asset_id: Some("asset-child".to_string()), + path: "Content/A.uasset".to_string(), + from_blob_hash: None, + blob_hash: Some("hash-child".to_string()), + }, + ], + }) + .await + .expect_err("conflicting paths must be rejected"); + + assert!(matches!(error, VersioningError::InvalidAssetLayout { .. })); + assert!(manager.list_repos().is_empty()); + } + + #[tokio::test] + async fn create_repo_creates_default_branch_and_rejects_duplicates() { + let manager = VersionManager::new(); + + let repo = manager + .create_repo("repo-explicit", "main", "alice") + .await + .expect("repo should be created"); + + assert_eq!(repo.repo_id, "repo-explicit"); + assert_eq!(repo.default_branch, "main"); + assert_eq!(repo.branch_count, 1); + assert_eq!(repo.default_head_changeset_id, None); + + let duplicate = manager + .create_repo("repo-explicit", "main", "alice") + .await + .expect_err("duplicate repo should fail"); + + assert_eq!( + duplicate, + VersioningError::RepoAlreadyExists { + repo_id: "repo-explicit".to_string(), + } + ); + } + + #[tokio::test] + async fn list_and_get_repo_info_return_default_branch() { + let manager = VersionManager::new(); + + manager + .create_repo("repo-info", "main", "alice") + .await + .expect("repo should be created"); + + let repos = manager.list_repos(); + assert_eq!(repos.len(), 1); + assert_eq!(repos[0].repo_id, "repo-info"); + assert_eq!(repos[0].default_branch, "main"); + + let info = manager + .get_repo_info("repo-info") + .expect("repo info should exist"); + assert_eq!(info.branches.len(), 1); + assert_eq!(info.branches[0].name, "main"); + assert!(info.branches[0].is_default); + } + + #[tokio::test] + async fn stale_base_is_rejected() { + let manager = VersionManager::new(); + + let c1 = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-b".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "first".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect("first should succeed"); + + let c2 = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-b".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "invalid".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect_err("stale base must fail"); + + assert_eq!( + c2, + VersioningError::BaseChangesetMismatch { + repo_id: "repo-b".to_string(), + branch: "main".to_string(), + expected: Some(c1.changeset_id), + got: Some(ROOT_BASE_CHANGESET_ID.to_string()), + } + ); + } + + #[tokio::test] + async fn rollback_plan_targets_existing_history() { + let manager = VersionManager::new(); + let c1 = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-c".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "first".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![AssetDelta { + asset_id: None, + path: "a".to_string(), + from_blob_hash: None, + blob_hash: Some("h1".to_string()), + }], + }) + .await + .expect("first commit"); + + let c2 = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-c".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(c1.changeset_id.clone()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "second".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![AssetDelta { + asset_id: None, + path: "a".to_string(), + from_blob_hash: None, + blob_hash: Some("h2".to_string()), + }], + }) + .await + .expect("second commit"); + + let plan = manager + .build_rollback_plan("repo-c", "main", &c1.changeset_id) + .expect("rollback plan"); + assert_eq!(plan.base_changeset_id, c2.changeset_id.clone()); + assert_eq!(plan.assets.len(), 1); + assert_eq!(plan.assets[0].blob_hash.as_deref(), Some("h1")); + + manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-c".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(plan.base_changeset_id.clone()), + kind: ChangesetKind::Rollback, + rollback_of: Some(plan.target_changeset_id), + author: "alice".to_string(), + message: "rollback".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: plan.assets, + }) + .await + .expect("rollback commit should be accepted"); + + let sync = manager + .sync_snapshot("repo-c", "main", None) + .expect("snapshot"); + assert_eq!(sync.assets[0].blob_hash, "h1"); + } + + #[tokio::test] + async fn draft_changeset_uses_staging_ref_and_promote_sets_visible_ref() { + let manager = VersionManager::new(); + + let base = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-gate".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "base".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect("base changeset"); + + let draft = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-gate".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(base.changeset_id.clone()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "draft".to_string(), + visibility: ChangesetVisibility::Draft, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect("draft changeset"); + + assert!(draft.staging_ref.is_some()); + assert_eq!(draft.visible_ref, None); + + let approved = manager + .approve_changeset("repo-gate", &draft.changeset_id, "reviewer") + .await + .expect("approve draft"); + assert_eq!(approved.visible_ref, None); + + let promoted = manager + .promote_changeset("repo-gate", &draft.changeset_id, "release-bot") + .await + .expect("promote approved"); + assert_eq!(promoted.visible_ref.as_deref(), Some("refs/heads/main")); + assert!(promoted.staging_ref.is_some()); + } + + #[tokio::test] + async fn changeset_gate_requires_approved_before_promote() { + let manager = VersionManager::new(); + + let base = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-gate-2".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "base".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect("base changeset"); + + let draft = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-gate-2".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(base.changeset_id.clone()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "draft".to_string(), + visibility: ChangesetVisibility::Draft, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect("draft changeset"); + + let gate_before = manager + .changeset_gate("repo-gate-2", &draft.changeset_id) + .expect("gate for draft"); + assert!(!gate_before.can_promote); + assert_eq!(gate_before.required_state, "approved"); + + manager + .approve_changeset("repo-gate-2", &draft.changeset_id, "reviewer") + .await + .expect("approve draft"); + + let gate_after = manager + .changeset_gate("repo-gate-2", &draft.changeset_id) + .expect("gate for approved"); + assert!(gate_after.can_promote); + assert_eq!(gate_after.required_state, "approved"); + } + + #[tokio::test] + async fn draft_first_changeset_can_be_promoted() { + // Regression: the very first changeset on a branch, submitted as a draft + // (base=ROOT), never advances the branch head. Promote/gate must still + // accept ROOT against an empty head, otherwise it is permanently stuck. + let manager = VersionManager::new(); + + let draft = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-draft-first".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "draft-first".to_string(), + visibility: ChangesetVisibility::Draft, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![], + }) + .await + .expect("draft-first changeset"); + + manager + .approve_changeset("repo-draft-first", &draft.changeset_id, "reviewer") + .await + .expect("approve draft-first"); + + let gate = manager + .changeset_gate("repo-draft-first", &draft.changeset_id) + .expect("gate for approved draft-first"); + assert!( + gate.can_promote, + "approved draft-first should be promotable" + ); + + let promoted = manager + .promote_changeset("repo-draft-first", &draft.changeset_id, "release-bot") + .await + .expect("promote draft-first should succeed"); + assert_eq!(promoted.status, ChangesetStatus::Visible); + assert_eq!(promoted.visible_ref.as_deref(), Some("refs/heads/main")); + } + + #[tokio::test] + async fn submit_preserves_agent_session_metadata() { + let manager = VersionManager::new(); + + let changeset = manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-agent-meta".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "agent-a".to_string(), + message: "draft from checkpoint".to_string(), + visibility: ChangesetVisibility::Draft, + intent_id: Some("intent-1".to_string()), + task_id: Some("task-1".to_string()), + agent_run_id: Some("run-1".to_string()), + session_id: Some("session-1".to_string()), + parent_checkpoint_id: Some("checkpoint-1".to_string()), + risk_level: Some("local".to_string()), + semantic_summary: Some("inventory implementation draft".to_string()), + assets: vec![], + }) + .await + .expect("draft changeset"); + + assert_eq!(changeset.status, ChangesetStatus::Draft); + assert_eq!(changeset.intent_id.as_deref(), Some("intent-1")); + assert_eq!(changeset.task_id.as_deref(), Some("task-1")); + assert_eq!(changeset.agent_run_id.as_deref(), Some("run-1")); + assert_eq!(changeset.session_id.as_deref(), Some("session-1")); + assert_eq!( + changeset.parent_checkpoint_id.as_deref(), + Some("checkpoint-1") + ); + assert_eq!(changeset.risk_level.as_deref(), Some("local")); + assert_eq!( + changeset.semantic_summary.as_deref(), + Some("inventory implementation draft") + ); + } + + #[tokio::test] + async fn persists_state_across_manager_restarts() { + let state_file = + std::env::temp_dir().join(format!("hypertide-versioning-{}.json", Uuid::new_v4())); + + let first_manager = VersionManager::with_persistence(&state_file); + first_manager + .submit_changeset(SubmitChangesetInput { + repo_id: "repo-p".to_string(), + branch: "main".to_string(), + base_changeset_id: Some(ROOT_BASE_CHANGESET_ID.to_string()), + kind: ChangesetKind::Normal, + rollback_of: None, + author: "alice".to_string(), + message: "first".to_string(), + visibility: ChangesetVisibility::Visible, + intent_id: None, + task_id: None, + agent_run_id: None, + session_id: None, + parent_checkpoint_id: None, + risk_level: None, + semantic_summary: None, + assets: vec![AssetDelta { + asset_id: None, + path: "env/config.json".to_string(), + from_blob_hash: None, + blob_hash: Some("blob-v1".to_string()), + }], + }) + .await + .expect("submit should persist"); + + let second_manager = VersionManager::with_persistence(&state_file); + let snapshot = second_manager + .sync_snapshot("repo-p", "main", None) + .expect("snapshot should load from persistence"); + assert_eq!(snapshot.assets.len(), 1); + assert_eq!(snapshot.assets[0].path, "env/config.json"); + assert_eq!(snapshot.assets[0].blob_hash, "blob-v1"); + + let _ = std::fs::remove_file(state_file); + } + + #[tokio::test] + async fn persistence_failure_does_not_publish_in_memory_state() { + let blocker = + std::env::temp_dir().join(format!("hypertide-versioning-blocker-{}", Uuid::new_v4())); + std::fs::write(&blocker, b"not-a-directory").expect("create blocker"); + let manager = VersionManager::with_persistence(blocker.join("state.json")); + + let error = manager + .create_repo("repo-not-persisted", "main", "alice") + .await + .expect_err("persistence must fail"); + + assert!(matches!(error, VersioningError::Persistence { .. })); + assert!(manager.list_repos().is_empty()); + let _ = std::fs::remove_file(blocker); + } + + #[tokio::test] + async fn file_persistence_supports_consecutive_mutations() { + let state_file = + std::env::temp_dir().join(format!("hypertide-versioning-{}.json", Uuid::new_v4())); + let manager = VersionManager::with_persistence(&state_file); + + manager + .create_repo("repo-p", "main", "alice") + .await + .expect("first persistence write"); + manager + .create_branch("repo-p", "feature", None, "alice") + .await + .expect("replacement persistence write"); + + let reloaded = VersionManager::with_persistence(&state_file); + let branches = reloaded + .list_branches("repo-p") + .expect("load persisted repo"); + assert_eq!(branches.len(), 2); + assert!(branches.iter().any(|branch| branch.name == "feature")); + + let _ = std::fs::remove_file(state_file); + } +} diff --git a/crates/server/src/core/versioning/repo_pg.rs b/crates/server/src/core/versioning/repo_pg.rs index 06752ef..48d7258 100644 --- a/crates/server/src/core/versioning/repo_pg.rs +++ b/crates/server/src/core/versioning/repo_pg.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; +use std::sync::Arc; use chrono::{DateTime, Utc}; +use dashmap::DashMap; use sqlx::{FromRow, PgPool}; use crate::core::versioning::{ @@ -11,12 +13,17 @@ use crate::core::versioning::{ #[derive(Clone)] pub struct VersionRepoPg { pool: PgPool, + /// Per-repo `state_version` last observed by this process. Used as the expected + /// value in the optimistic-concurrency guard so a concurrent writer's update is + /// detected instead of silently overwritten. + versions: Arc>, } #[derive(Debug, FromRow)] struct RepoRow { repo_id: String, created_by: String, + state_version: i64, } #[derive(Debug, FromRow)] @@ -75,7 +82,10 @@ struct SnapshotRow { impl VersionRepoPg { pub fn new(pool: PgPool) -> Self { - Self { pool } + Self { + pool, + versions: Arc::new(DashMap::new()), + } } pub(super) async fn load_repos(&self) -> Result, sqlx::Error> { @@ -83,7 +93,7 @@ impl VersionRepoPg { let repo_rows = sqlx::query_as::<_, RepoRow>( r#" - SELECT repo_id, created_by + SELECT repo_id, created_by, state_version FROM repos ORDER BY created_at ASC "#, @@ -92,6 +102,8 @@ impl VersionRepoPg { .await?; for repo_row in repo_rows { + self.versions + .insert(repo_row.repo_id.clone(), repo_row.state_version); let mut repo = RepoState { default_branch: "main".to_string(), branches: HashMap::new(), @@ -269,17 +281,57 @@ impl VersionRepoPg { }) .unwrap_or("system"); - sqlx::query( - r#" - INSERT INTO repos (repo_id, created_by) - VALUES ($1, $2) - ON CONFLICT (repo_id) DO UPDATE SET created_by = EXCLUDED.created_by - "#, - ) - .bind(repo_id) - .bind(created_by) - .execute(&mut *tx) - .await?; + // Optimistic-concurrency guard. `expected` is the version this process last + // observed for the repo; the guarded write only succeeds if the DB still + // holds that version, so a concurrent writer (e.g. another instance) that + // advanced the repo is detected here instead of being silently clobbered. + let expected_version = self.versions.get(repo_id).map(|entry| *entry); + let new_version = match expected_version { + Some(expected) => { + let updated = sqlx::query( + r#" + UPDATE repos + SET created_by = $2, state_version = state_version + 1 + WHERE repo_id = $1 AND state_version = $3 + "#, + ) + .bind(repo_id) + .bind(created_by) + .bind(expected) + .execute(&mut *tx) + .await?; + if updated.rows_affected() == 0 { + tx.rollback().await?; + return Err(sqlx::Error::Protocol(format!( + "concurrent modification of repo {repo_id}: expected state_version {expected}" + ))); + } + expected + 1 + } + None => { + let inserted = sqlx::query( + r#" + INSERT INTO repos (repo_id, created_by, state_version) + VALUES ($1, $2, 0) + ON CONFLICT (repo_id) DO NOTHING + "#, + ) + .bind(repo_id) + .bind(created_by) + .execute(&mut *tx) + .await?; + if inserted.rows_affected() == 0 { + // The repo already exists in the DB but this process never loaded + // or persisted it: another writer owns it. Refuse rather than + // overwrite an unknown state. + tx.rollback().await?; + return Err(sqlx::Error::Protocol(format!( + "concurrent creation of repo {repo_id} by another writer" + ))); + } + 0 + } + }; sqlx::query("DELETE FROM branches WHERE repo_id = $1") .bind(repo_id) @@ -344,6 +396,14 @@ impl VersionRepoPg { } for (changeset_id, snapshot) in &repo.snapshots { + // Persist each snapshot under the branch its changeset actually belongs + // to. Binding the default branch unconditionally mislabeled every + // non-default-branch snapshot (the table is keyed by branch_name). + let branch_name = repo + .changesets + .get(changeset_id) + .map(|changeset| changeset.branch.as_str()) + .unwrap_or(repo.default_branch.as_str()); for (asset_id, snapshot_asset) in snapshot { sqlx::query( r#" @@ -352,7 +412,7 @@ impl VersionRepoPg { "#, ) .bind(repo_id) - .bind(&repo.default_branch) + .bind(branch_name) .bind(changeset_id) .bind(asset_id) .bind(&snapshot_asset.path) @@ -380,6 +440,7 @@ impl VersionRepoPg { } tx.commit().await?; + self.versions.insert(repo_id.to_string(), new_version); Ok(()) } } diff --git a/migrations/202602260018_repo_state_version.down.sql b/migrations/202602260018_repo_state_version.down.sql new file mode 100644 index 0000000..5c0aa27 --- /dev/null +++ b/migrations/202602260018_repo_state_version.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE repos + DROP COLUMN IF EXISTS state_version; diff --git a/migrations/202602260018_repo_state_version.up.sql b/migrations/202602260018_repo_state_version.up.sql new file mode 100644 index 0000000..2f84a48 --- /dev/null +++ b/migrations/202602260018_repo_state_version.up.sql @@ -0,0 +1,6 @@ +-- Optimistic concurrency guard for repo state persistence. +-- Lets replace_repo_state reject a write when another writer advanced the repo +-- since this process last persisted it, turning a silent lost update into a +-- detectable conflict. +ALTER TABLE repos + ADD COLUMN IF NOT EXISTS state_version BIGINT NOT NULL DEFAULT 0; From 7a7068d09346c3e8a4fd9866fe1551a96843ccda Mon Sep 17 00:00:00 2001 From: aoruLola Date: Sun, 26 Jul 2026 22:15:41 +0800 Subject: [PATCH 3/4] fix(server): verify witness quorum and fix trust/replay correctness - witness: summary() recomputes each receipt's HMAC over the referenced checkpoint material (constant-time verify_slice) and counts only configured, signature-verified witnesses toward quorum. Previously quorum was a raw row count, so forged receipts could fake it. - checkpoint: build generate_checkpoint inside one REPEATABLE READ transaction holding the audit advisory lock, so log head/size and the table counts describe a single consistent moment (removes a TOCTOU). - replay: key the replayed lock set by (repo_id, file_path) to match DB uniqueness; always full-scan events (the checkpoint marker has no state snapshot, so a suffix replay produced spurious mismatches). - audit_chain: report `checked` from a running verified counter instead of deriving it from seq, which has gaps from rolled-back appends. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/server/src/core/audit_chain.rs | 10 ++-- crates/server/src/core/checkpoint.rs | 27 +++++++--- crates/server/src/core/replay.rs | 31 +++++++---- crates/server/src/core/witness.rs | 76 +++++++++++++++++++++++---- 4 files changed, 113 insertions(+), 31 deletions(-) diff --git a/crates/server/src/core/audit_chain.rs b/crates/server/src/core/audit_chain.rs index e19e2c7..f96998c 100644 --- a/crates/server/src/core/audit_chain.rs +++ b/crates/server/src/core/audit_chain.rs @@ -124,11 +124,14 @@ impl AuditChain { .await?; let mut expected_prev = "GENESIS".to_string(); + // Count verified rows directly rather than deriving from `seq`, which has + // gaps whenever a BIGSERIAL value is consumed by a rolled-back append. + let mut checked = 0i64; for row in &rows { if row.prev_hash != expected_prev { return Ok(AuditVerifyResult { valid: false, - checked: row.seq.saturating_sub(1), + checked, broken_at_seq: Some(row.seq), reason: Some("prev_hash mismatch".to_string()), }); @@ -151,18 +154,19 @@ impl AuditChain { if row.entry_hash != expected_hash { return Ok(AuditVerifyResult { valid: false, - checked: row.seq.saturating_sub(1), + checked, broken_at_seq: Some(row.seq), reason: Some("entry_hash mismatch".to_string()), }); } expected_prev = row.entry_hash.clone(); + checked += 1; } Ok(AuditVerifyResult { valid: true, - checked: rows.len() as i64, + checked, broken_at_seq: None, reason: None, }) diff --git a/crates/server/src/core/checkpoint.rs b/crates/server/src/core/checkpoint.rs index eeb1b6d..b7411f4 100644 --- a/crates/server/src/core/checkpoint.rs +++ b/crates/server/src/core/checkpoint.rs @@ -24,6 +24,18 @@ impl CheckpointService { } pub async fn generate_checkpoint(&self) -> Result { + // Take a consistent point-in-time snapshot: run every read inside one + // REPEATABLE READ transaction and hold the same advisory lock the audit + // appender uses, so log_head/log_size and the table counts can't describe + // different moments (a TOCTOU that witnesses would then attest). + let mut tx = self.pool.begin().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") + .execute(&mut *tx) + .await?; + sqlx::query("SELECT pg_advisory_xact_lock(92426001)") + .execute(&mut *tx) + .await?; + let log_head_hash = sqlx::query_scalar::<_, Option>( r#" SELECT entry_hash @@ -32,29 +44,29 @@ impl CheckpointService { LIMIT 1 "#, ) - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await? .unwrap_or_else(|| "GENESIS".to_string()); let log_size = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM audit_chain_entries") - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await?; let locks_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM locks WHERE force_released = FALSE") - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await .unwrap_or(0); let changesets_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM changesets") - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await .unwrap_or(0); let manifests_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM manifests") - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await .unwrap_or(0); let chunks_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM chunks") - .fetch_one(&self.pool) + .fetch_one(&mut *tx) .await .unwrap_or(0); @@ -93,9 +105,10 @@ impl CheckpointService { .bind(checkpoint.log_size) .bind(&checkpoint.state_root) .bind(checkpoint.created_at) - .execute(&self.pool) + .execute(&mut *tx) .await?; + tx.commit().await?; Ok(checkpoint) } diff --git a/crates/server/src/core/replay.rs b/crates/server/src/core/replay.rs index 1592a26..5cb024f 100644 --- a/crates/server/src/core/replay.rs +++ b/crates/server/src/core/replay.rs @@ -85,13 +85,13 @@ impl ReplayAccumulator { "LOCK_ACQUIRED" => { self.summary.lock_acquired += 1; if let Some(path) = extract_file_path(payload) { - self.current_locks.insert(path.to_string()); + self.current_locks.insert(lock_key(repo_id, path)); } } "LOCK_RELEASED" | "LOCK_FORCE_RELEASED" => { self.summary.lock_released += 1; if let Some(path) = extract_file_path(payload) { - self.current_locks.remove(path); + self.current_locks.remove(&lock_key(repo_id, path)); } } "CHANGESET_VISIBLE" | "ROLLBACK_VISIBLE" => { @@ -139,6 +139,13 @@ fn extract_file_path(payload: Option<&Value>) -> Option<&str> { payload?.get("file_path")?.as_str() } +/// Key locks by `(repo_id, file_path)` to mirror the DB's uniqueness. Keying by +/// path alone collapsed identical paths across repos into one replay entry, +/// producing a false mismatch against `SELECT COUNT(*) FROM locks`. +fn lock_key(repo_id: Option<&str>, path: &str) -> String { + format!("{}::{}", repo_id.unwrap_or(""), path) +} + fn extract_branch(payload: Option<&Value>) -> Option<&str> { payload?.get("branch")?.as_str() } @@ -181,22 +188,24 @@ impl ReplayService { &self, from_checkpoint: Option<&str>, ) -> Result { - let start_seq = if let Some(cp_id) = from_checkpoint { + // Always replay from the beginning. `replay_checkpoints` records only an + // `event_seq` marker with no accumulated state snapshot, so replaying just + // the suffix after a checkpoint into a fresh accumulator cannot reproduce + // full state and would report spurious mismatches against the absolute DB + // counts below. We still validate the checkpoint exists to preserve the + // API contract, but a correct result requires a full scan. + if let Some(cp_id) = from_checkpoint { let seq: Option = sqlx::query_scalar( "SELECT event_seq FROM replay_checkpoints WHERE checkpoint_id = $1", ) .bind(cp_id) .fetch_optional(&self.pool) .await?; - match seq { - Some(s) => s, - None => { - return Err(sqlx::Error::RowNotFound); - } + if seq.is_none() { + return Err(sqlx::Error::RowNotFound); } - } else { - 0 - }; + } + let start_seq = 0i64; let events = sqlx::query_as::<_, EventRow>( r#" diff --git a/crates/server/src/core/witness.rs b/crates/server/src/core/witness.rs index 1b5406b..14afe5b 100644 --- a/crates/server/src/core/witness.rs +++ b/crates/server/src/core/witness.rs @@ -211,6 +211,38 @@ impl WitnessService { }) } + /// Recompute a receipt's HMAC over the referenced checkpoint material and + /// verify it in constant time. Returns false for receipts from unconfigured + /// witnesses or with a malformed/invalid signature, so forged rows cannot count. + fn verify_receipt_signature( + &self, + checkpoint: &CheckpointRecord, + receipt: &WitnessReceipt, + ) -> bool { + use hmac::{Hmac, Mac}; + use sha2::Sha256; + type HmacSha256 = Hmac; + + let Some(witness) = self.witnesses.iter().find(|w| w.id == receipt.witness_id) else { + return false; + }; + let Ok(provided) = hex::decode(&receipt.signature) else { + return false; + }; + let material = format!( + "{}|{}|{}|{}", + checkpoint.checkpoint_id, + checkpoint.log_head_hash, + checkpoint.log_size, + checkpoint.state_root + ); + let Ok(mut mac) = HmacSha256::new_from_slice(witness.secret.as_bytes()) else { + return false; + }; + mac.update(material.as_bytes()); + mac.verify_slice(&provided).is_ok() + } + pub async fn summary(&self, checkpoint_id: &str) -> Result { let receipts = sqlx::query_as::<_, WitnessReceipt>( r#" @@ -225,25 +257,49 @@ impl WitnessService { .await .map_err(|error| format!("failed to query witness receipts: {error}"))?; + // Quorum must be established by cryptographically verified receipts from + // configured witnesses, not by counting rows: a DB-write attacker could + // otherwise insert junk receipts to fake a quorum. Fetch the referenced + // checkpoint and re-verify each receipt's signature against its material. + let checkpoint = sqlx::query_as::<_, CheckpointRecord>( + r#" + SELECT checkpoint_id, log_head_hash, log_size, state_root, created_at + FROM trust_checkpoints + WHERE checkpoint_id = $1 + "#, + ) + .bind(checkpoint_id) + .fetch_optional(&self.pool) + .await + .map_err(|error| format!("failed to query checkpoint: {error}"))?; + + let mut verified_witnesses = HashSet::new(); let mut scopes = HashSet::new(); - for receipt in &receipts { - if let Some(scope) = self - .witnesses - .iter() - .find(|w| w.id == receipt.witness_id) - .map(|w| w.scope.clone()) - { - scopes.insert(scope); + if let Some(checkpoint) = &checkpoint { + for receipt in &receipts { + if !self.verify_receipt_signature(checkpoint, receipt) { + continue; + } + verified_witnesses.insert(receipt.witness_id.clone()); + if let Some(scope) = self + .witnesses + .iter() + .find(|w| w.id == receipt.witness_id) + .map(|w| w.scope.clone()) + { + scopes.insert(scope); + } } } let mut distinct_scopes = scopes.into_iter().collect::>(); distinct_scopes.sort(); + let quorum_met = verified_witnesses.len() >= self.quorum; Ok(WitnessSummary { checkpoint_id: checkpoint_id.to_string(), quorum: self.quorum, - quorum_met: receipts.len() >= self.quorum, - cross_scope_quorum_met: receipts.len() >= self.quorum && distinct_scopes.len() >= 2, + quorum_met, + cross_scope_quorum_met: quorum_met && distinct_scopes.len() >= 2, distinct_scopes, receipts, }) From 7812c58c71c201c765bac098a87f5d0871115961 Mon Sep 17 00:00:00 2001 From: aoruLola Date: Sun, 26 Jul 2026 22:15:52 +0800 Subject: [PATCH 4/4] fix(cli): protect local work and credentials from silent loss - workspace: write state atomically (temp file + rename) with 0600 file / 0700 dir permissions on Unix, and auto-create .hypertide/.gitignore so plaintext credentials can't be world-readable or accidentally committed. - checkpoint restore/branch: run the same pre-flight as checkout (staged / local-modification / untracked-collision checks) and add --force, instead of overwriting local files unconditionally. - sync: materialize snapshot content and update checked_out_assets when advancing the base pointer, and refuse (or --force) on local divergence, fixing a silent lost update where a later submit discarded intervening changes. - confirm_dangerous: return a non-zero error on cancel and hard-fail in a non-TTY without --yes, instead of exiting 0 as a silent no-op. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/cli/src/cmd/checkpoint.rs | 8 +- crates/cli/src/cmd/sync.rs | 135 ++++++++++++++++++++++++++++--- crates/cli/src/utils.rs | 90 ++++++++++++++++++++- crates/cli/src/workspace.rs | 55 ++++++++++++- 4 files changed, 269 insertions(+), 19 deletions(-) diff --git a/crates/cli/src/cmd/checkpoint.rs b/crates/cli/src/cmd/checkpoint.rs index 9f30cfb..94c61df 100644 --- a/crates/cli/src/cmd/checkpoint.rs +++ b/crates/cli/src/cmd/checkpoint.rs @@ -37,6 +37,8 @@ pub(crate) struct CheckpointCreateArgs { pub(crate) struct CheckpointRestoreArgs { #[arg(long)] pub id: String, + #[arg(long, help = "Force restore, overwriting local modifications")] + pub force: bool, } #[derive(Debug, Args)] @@ -45,6 +47,8 @@ pub(crate) struct CheckpointBranchArgs { pub id: String, #[arg(long)] pub name: String, + #[arg(long, help = "Force materialization, overwriting local modifications")] + pub force: bool, } #[derive(Debug, Args)] @@ -105,7 +109,7 @@ async fn checkpoint_restore(args: CheckpointRestoreArgs) -> Result<()> { let mut profile = load_profile()?; let client = reqwest::Client::new(); let snapshot = fetch_checkpoint_snapshot(&client, &mut profile, &args.id).await?; - materialize_checkpoint_snapshot(&client, &mut profile, &snapshot).await?; + materialize_checkpoint_snapshot(&client, &mut profile, &snapshot, args.force).await?; println!( "checkpoint restored: checkpoint_id={} session_id={} repo_id={} branch={} asset_count={}", snapshot.checkpoint_id, @@ -140,7 +144,7 @@ async fn checkpoint_branch(args: CheckpointBranchArgs) -> Result<()> { "create branch failed" ))); } - materialize_checkpoint_snapshot(&client, &mut profile, &snapshot).await?; + materialize_checkpoint_snapshot(&client, &mut profile, &snapshot, args.force).await?; profile.current_repo = Some(snapshot.repo_id.clone()); profile.current_branch = args.name.clone(); save_profile(&profile)?; diff --git a/crates/cli/src/cmd/sync.rs b/crates/cli/src/cmd/sync.rs index c753857..9d7d018 100644 --- a/crates/cli/src/cmd/sync.rs +++ b/crates/cli/src/cmd/sync.rs @@ -1,4 +1,6 @@ -use anyhow::Result; +use std::{collections::HashSet, fs, path::Path}; + +use anyhow::{anyhow, Context, Result}; use clap::Args; use crate::utils::*; @@ -11,6 +13,8 @@ pub(crate) struct SyncArgs { pub branch: Option, #[arg(long = "to", help = "Optional changeset id to sync to")] pub to_changeset_id: Option, + #[arg(long, help = "Force sync, overwriting local modifications")] + pub force: bool, } pub(crate) async fn execute(args: SyncArgs) -> Result<()> { @@ -19,6 +23,7 @@ pub(crate) async fn execute(args: SyncArgs) -> Result<()> { let branch = args .branch .unwrap_or_else(|| profile.current_branch.clone()); + let workspace_root = std::env::current_dir()?; let client = reqwest::Client::new(); let snapshot = fetch_snapshot( &client, @@ -29,23 +34,129 @@ pub(crate) async fn execute(args: SyncArgs) -> Result<()> { ) .await?; - // Preserve existing stage assets — only update base_changeset_id - let mut stage = load_stage().unwrap_or_else(|_| StageFile::default_for_branch(&branch)); - stage.base_changeset_id = snapshot.changeset_id; - save_stage(&stage)?; - if let Ok(mut workspace) = load_workspace() { - if workspace.repo_id == repo && workspace.branch == branch { - workspace.base_changeset_id = stage.base_changeset_id.clone(); - workspace.last_synced_at = now_unix(); - save_workspace(&workspace)?; + let existing_workspace = load_workspace().ok().filter(|workspace| { + workspace.repo_id == repo + && workspace.branch == branch + && Path::new(&workspace.workspace_root) == workspace_root + }); + + // A base-pointer advance without reconciling file content silently discards + // intervening changes on the next submit. Refuse when local work would be + // clobbered so the user resolves it (submit / --force) rather than losing it. + if !args.force { + if let Ok(stage) = load_stage() { + if !stage.assets.is_empty() { + return Err(anyhow!( + "workspace has {} staged change(s); submit them before syncing or use --force", + stage.assets.len() + )); + } + } + if let Some(workspace) = &existing_workspace { + let conflicts = detect_local_modifications(workspace)?; + if !conflicts.is_empty() { + eprintln!( + "error: workspace has {} uncommitted modification(s); sync would overwrite:", + conflicts.len() + ); + for conflict in &conflicts { + eprintln!(" {}", conflict.path); + } + eprintln!("submit your changes, or re-run with --force to overwrite."); + return Err(anyhow!("sync refused to overwrite local changes")); + } + } + } + + let snapshot_paths = snapshot + .assets + .iter() + .map(|asset| asset.path.as_str()) + .collect::>(); + + // Guard against overwriting untracked local files that collide with the snapshot. + let tracked_paths = existing_workspace + .as_ref() + .map(|workspace| { + workspace + .checked_out_assets + .iter() + .map(|asset| asset.path.as_str()) + .collect::>() + }) + .unwrap_or_default(); + if !args.force { + for asset in &snapshot.assets { + if tracked_paths.contains(asset.path.as_str()) { + continue; + } + let target = resolve_workspace_target(&workspace_root, &asset.path)?; + if target.exists() + && (target.is_dir() + || hash_local_asset(&workspace_root, &asset.path)?.as_deref() + != Some(asset.blob_hash.as_str())) + { + return Err(anyhow!( + "sync would overwrite untracked local file {}; use --force", + asset.path + )); + } + } + } + + // Remove tracked files that no longer exist in the new snapshot. + if let Some(workspace) = &existing_workspace { + for asset in &workspace.checked_out_assets { + if snapshot_paths.contains(asset.path.as_str()) { + continue; + } + let target = resolve_workspace_target(&workspace_root, &asset.path)?; + if target.is_file() { + fs::remove_file(&target) + .with_context(|| format!("failed to delete {}", target.display()))?; + } + } + } + + // Materialize snapshot content so recorded hashes and on-disk files agree with + // the advanced base pointer. + let mut checked_out_assets = Vec::with_capacity(snapshot.assets.len()); + for asset in &snapshot.assets { + let target = resolve_workspace_target(&workspace_root, &asset.path)?; + let bytes = fetch_blob_bytes(&client, &mut profile, &asset.blob_hash).await?; + if let Some(parent) = target.parent() { + fs::create_dir_all(parent)?; } + fs::write(&target, &bytes) + .with_context(|| format!("failed to write {}", target.display()))?; + checked_out_assets.push(WorkspaceFile { + path: asset.path.clone(), + blob_hash: asset.blob_hash.clone(), + asset_id: asset.asset_id.clone(), + }); } + + let workspace = WorkspaceState { + repo_id: repo.clone(), + branch: branch.clone(), + workspace_root: workspace_root.to_string_lossy().to_string(), + base_changeset_id: snapshot.changeset_id.clone(), + checked_out_assets, + last_synced_at: now_unix(), + }; + save_workspace(&workspace)?; + + // Advance the base pointer; a clean workspace now has no staged assets. + let mut stage = StageFile::default_for_branch(&branch); + stage.base_changeset_id = snapshot.changeset_id.clone(); + save_stage(&stage)?; + println!( "synced {}@{} to {} ({} assets)", repo, branch, - stage - .base_changeset_id + snapshot + .changeset_id .clone() .unwrap_or_else(|| "ROOT".to_string()), snapshot.assets.len() diff --git a/crates/cli/src/utils.rs b/crates/cli/src/utils.rs index b80ea24..b6f937a 100644 --- a/crates/cli/src/utils.rs +++ b/crates/cli/src/utils.rs @@ -603,15 +603,26 @@ pub(crate) fn normalize_asset_path(path: &Path) -> String { } pub(crate) fn confirm_dangerous(action: &str, yes: bool) -> Result<()> { + use std::io::IsTerminal; + if yes { return Ok(()); } + // Never silently "cancel" (as success) when there is no interactive terminal to + // prompt: automation that forgot --yes must get a hard error, not a no-op exit 0. + if !std::io::stdin().is_terminal() { + return Err(anyhow!( + "refusing dangerous operation ({action}) without confirmation; \ + re-run with --yes to proceed non-interactively" + )); + } eprint!("dangerous operation: {}. confirm? [y/N] ", action); let mut input = String::new(); - std::io::stdin().read_line(&mut input)?; - if input.trim().to_lowercase() != "y" { - eprintln!("cancelled."); - std::process::exit(0); + let read = std::io::stdin().read_line(&mut input)?; + if read == 0 || input.trim().to_lowercase() != "y" { + // Return an error so a declined operation exits non-zero instead of + // reporting success to any calling script. + return Err(anyhow!("operation cancelled by user")); } Ok(()) } @@ -1692,8 +1703,10 @@ pub(crate) async fn materialize_checkpoint_snapshot( client: &reqwest::Client, profile: &mut CliProfile, snapshot: &CheckpointSnapshot, + force: bool, ) -> Result<()> { let workspace_root = std::env::current_dir()?; + guard_checkpoint_overwrite(&workspace_root, snapshot, force)?; let mut checked_out_assets = Vec::with_capacity(snapshot.assets.len()); for asset in &snapshot.assets { let target = resolve_workspace_target(&workspace_root, &asset.path)?; @@ -1723,6 +1736,75 @@ pub(crate) async fn materialize_checkpoint_snapshot( Ok(()) } +/// Refuse to overwrite local work when restoring/branching from a checkpoint, +/// mirroring the pre-flight in `ht checkout`. Bypassed only with `force`. +fn guard_checkpoint_overwrite( + workspace_root: &Path, + snapshot: &CheckpointSnapshot, + force: bool, +) -> Result<()> { + if force { + return Ok(()); + } + + if let Ok(stage) = load_stage() { + if !stage.assets.is_empty() { + return Err(anyhow!( + "workspace has {} staged change(s); submit them or use --force", + stage.assets.len() + )); + } + } + + let existing_workspace = load_workspace().ok(); + let matching_workspace = existing_workspace.as_ref().filter(|workspace| { + workspace.repo_id == snapshot.repo_id + && Path::new(&workspace.workspace_root) == workspace_root + }); + + let mut tracked_paths = std::collections::HashSet::new(); + if let Some(workspace) = matching_workspace { + let conflicts = detect_local_modifications(workspace)?; + if !conflicts.is_empty() { + eprintln!( + "error: workspace has {} uncommitted modification(s), restore would overwrite:", + conflicts.len() + ); + for conflict in &conflicts { + eprintln!(" {}", conflict.path); + } + eprintln!("commit/submit your changes, or re-run with --force to overwrite."); + return Err(anyhow!( + "checkpoint restore refused to overwrite local changes" + )); + } + tracked_paths.extend( + workspace + .checked_out_assets + .iter() + .map(|asset| asset.path.as_str()), + ); + } + + for asset in &snapshot.assets { + if tracked_paths.contains(asset.path.as_str()) { + continue; + } + let target = resolve_workspace_target(workspace_root, &asset.path)?; + if target.exists() + && (target.is_dir() + || hash_local_asset(workspace_root, &asset.path)?.as_deref() + != Some(asset.blob_hash.as_str())) + { + return Err(anyhow!( + "checkpoint restore would overwrite untracked local file {}; use --force", + asset.path + )); + } + } + Ok(()) +} + // ── Lock helper ── pub(crate) async fn send_lock_path_request( diff --git a/crates/cli/src/workspace.rs b/crates/cli/src/workspace.rs index 8c7697f..76ebe0f 100644 --- a/crates/cli/src/workspace.rs +++ b/crates/cli/src/workspace.rs @@ -28,12 +28,42 @@ pub fn ensure_state_dirs(paths: &StatePaths) -> Result<()> { if !paths.state_dir.exists() { fs::create_dir_all(&paths.state_dir)?; } + // Restrict the state directory to the owner: it holds credentials. + harden_dir_permissions(&paths.state_dir); + // Never let the local state (including plaintext credentials) be committed. + ensure_state_gitignore(&paths.state_dir); if !paths.cache_dir.exists() { fs::create_dir_all(&paths.cache_dir)?; } Ok(()) } +fn ensure_state_gitignore(state_dir: &Path) { + let gitignore = state_dir.join(".gitignore"); + if !gitignore.exists() { + // Ignore everything under .hypertide/, including this file itself. + let _ = fs::write(&gitignore, "*\n"); + } +} + +#[cfg(unix)] +fn harden_dir_permissions(dir: &Path) { + use std::os::unix::fs::PermissionsExt; + let _ = fs::set_permissions(dir, fs::Permissions::from_mode(0o700)); +} + +#[cfg(not(unix))] +fn harden_dir_permissions(_dir: &Path) {} + +#[cfg(unix)] +fn harden_file_permissions(path: &Path) { + use std::os::unix::fs::PermissionsExt; + let _ = fs::set_permissions(path, fs::Permissions::from_mode(0o600)); +} + +#[cfg(not(unix))] +fn harden_file_permissions(_path: &Path) {} + pub fn load_json(path: &Path) -> Result { let content = fs::read_to_string(path).with_context(|| format!("failed to read {}", path.display()))?; @@ -44,10 +74,33 @@ pub fn save_json(path: &Path, value: &T) -> Result<()> { if let Some(parent) = path.parent() { fs::create_dir_all(parent)?; } - fs::write(path, serde_json::to_vec_pretty(value)?)?; + let bytes = serde_json::to_vec_pretty(value)?; + // Atomic write: serialize to a sibling temp file, tighten permissions before + // it holds any data, then rename over the target so a crash/IO error mid-write + // can never truncate or corrupt existing state (e.g. profile.json credentials). + let temp_path = temp_sibling(path); + fs::write(&temp_path, &bytes) + .with_context(|| format!("failed to write {}", temp_path.display()))?; + harden_file_permissions(&temp_path); + if let Err(err) = fs::rename(&temp_path, path) { + let _ = fs::remove_file(&temp_path); + return Err(err).with_context(|| format!("failed to replace {}", path.display())); + } Ok(()) } +fn temp_sibling(path: &Path) -> PathBuf { + let file_name = path + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| "state".to_string()); + let temp_name = format!(".{}.tmp.{}", file_name, std::process::id()); + match path.parent() { + Some(parent) => parent.join(temp_name), + None => PathBuf::from(temp_name), + } +} + pub fn cache_object_path(paths: &StatePaths, hash: &str) -> PathBuf { paths.cache_dir.join(hash) }