From ffaaf2ecb0464dbf9e6fbeeb8d587499147ef6b3 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 16 Oct 2026 15:04:52 +0900 Subject: [PATCH] test: isolate local rules and reduce repeated fixture setup --- Cargo.toml | 10 + crates/api-macros/tests/renamed_dependency.rs | 51 +- crates/plugin-pdk/tests/template.rs | 75 +- crates/tools/src/view_image.rs | 46 +- crates/worker-runtime/Cargo.toml | 2 +- crates/worker-runtime/src/worker_backend.rs | 181 +++-- crates/workspace-server/src/auth.rs | 2 +- crates/workspace-server/src/authority.rs | 19 + crates/workspace-server/src/config_source.rs | 112 ++- crates/workspace-server/src/hosts.rs | 2 +- crates/workspace-server/src/retention.rs | 2 +- crates/workspace-server/src/server.rs | 392 +++++----- .../server/tests/subjektiv_journal_tests.rs | 729 +++++++++--------- .../src/server/tests/worker_drive_tests.rs | 32 +- .../server/tests/workspace_admission_tests.rs | 5 + .../src/server/workspace_admission.rs | 62 ++ crates/workspace-server/src/store.rs | 353 +++++---- .../src/store_test_support.rs | 190 +++++ .../src/ticket_item_checker.rs | 232 +++++- .../workspace-server/src/workdir_removal.rs | 2 +- .../workspace-server/src/workspace_catalog.rs | 16 +- .../src/workspace_subscription.rs | 2 +- 22 files changed, 1692 insertions(+), 825 deletions(-) create mode 100644 crates/workspace-server/src/store_test_support.rs diff --git a/Cargo.toml b/Cargo.toml index 552252a61..4b39e0fd8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -91,6 +91,16 @@ debug = "line-tables-only" [profile.dev.package."*"] debug = false +# Keep the real SQLite engine in schema/transaction tests without paying for +# an unoptimized C interpreter on every statement. Rust test assertions stay enabled. +[profile.test.package.libsqlite3-sys] +opt-level = 2 + +# Schema/authority integration tests execute substantial parsing and validation. +# Optimize that code without disabling test/debug assertions or overflow checks. +[profile.test.package.yoi-workspace-server] +opt-level = 1 + [workspace.dependencies] # Internal crates client = { path = "crates/client" } diff --git a/crates/api-macros/tests/renamed_dependency.rs b/crates/api-macros/tests/renamed_dependency.rs index 6c0c5c9bd..0b951a6f1 100644 --- a/crates/api-macros/tests/renamed_dependency.rs +++ b/crates/api-macros/tests/renamed_dependency.rs @@ -1,13 +1,38 @@ +//! Compile the real public macro through a renamed dependency in an isolated consumer. +//! Only Cargo artifacts are shared; Cargo checks the fresh consumer on every invocation. + use std::{fs, path::PathBuf, process::Command}; +struct Fixture(PathBuf); + +impl Fixture { + fn new() -> Self { + for attempt in 0..1000 { + let path = std::env::temp_dir().join(format!( + "api-macros-renamed-dependency-{}-{attempt}", + std::process::id() + )); + match fs::create_dir(&path) { + Ok(()) => return Self(path), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(error) => panic!("create renamed dependency fixture: {error}"), + } + } + panic!("could not allocate a renamed dependency fixture directory"); + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + #[test] fn macro_expansion_resolves_a_renamed_support_dependency() { - let fixture = std::env::temp_dir().join(format!( - "api-macros-renamed-dependency-{}", - std::process::id() - )); - let _ = fs::remove_dir_all(&fixture); - fs::create_dir_all(fixture.join("src")).expect("create fixture source directory"); + let fixture = Fixture::new(); + let fixture = &fixture.0; + fs::create_dir(fixture.join("src")).expect("create fixture source directory"); let package_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); let package_path = package_path @@ -49,13 +74,21 @@ pub fn operation_count() -> usize { ) .expect("write fixture source"); + // Keep nested Cargo checks off the outer build's target lock. Cargo owns cache + // invalidation and locking between concurrent consumers, not a saved + // test-success marker. + let executable = std::env::current_exe().expect("current test executable"); + let target_dir = executable + .parent() + .and_then(|deps| deps.parent()) + .expect("test executable has a profile directory") + .join("compile-tests/api-macros-renamed"); let output = Command::new(env!("CARGO")) .args(["check", "--quiet", "--offline"]) - .current_dir(&fixture) - .env("CARGO_TARGET_DIR", fixture.join("target")) + .current_dir(fixture) + .env("CARGO_TARGET_DIR", target_dir) .output() .expect("run cargo check for renamed dependency fixture"); - let _ = fs::remove_dir_all(&fixture); assert!( output.status.success(), diff --git a/crates/plugin-pdk/tests/template.rs b/crates/plugin-pdk/tests/template.rs index 29d15b684..3e0d11992 100644 --- a/crates/plugin-pdk/tests/template.rs +++ b/crates/plugin-pdk/tests/template.rs @@ -1,3 +1,6 @@ +//! Check the shipped templates against the real PDK and WIT for the guest target. +//! Each check owns its source and lockfile; Cargo alone shares and validates artifacts. + use std::fs; use std::path::Path; use std::process::Command; @@ -135,36 +138,86 @@ fn embedded_service_template_cargo_checks_for_wasm_target() { fn cargo_check_template(template: &str) { let crate_dir = Path::new(env!("CARGO_MANIFEST_DIR")); - let template_dir = crate_dir.join(format!("../../resources/plugin/templates/{template}")); - let manifest_path = template_dir.join("Cargo.toml"); - let lock_path = template_dir.join("Cargo.lock"); - let _ = fs::remove_file(&lock_path); + let templates_dir = crate_dir.join("../../resources/plugin/templates"); + let template_dir = templates_dir.join(template); - let target_dir = tempfile::tempdir().expect("temporary cargo target dir"); - let cargo = std::env::var("CARGO").unwrap_or_else(|_| "cargo".to_string()); - let output = Command::new(cargo) + // A sibling fixture preserves the shipped relative PDK and WIT paths verbatim. + // Never create or remove a lockfile in the shipped template: another invocation + // may be checking it at the same time, or a developer may own that lockfile. + let fixture = tempfile::Builder::new() + .prefix(".cargo-check-") + .tempdir_in(&templates_dir) + .expect("temporary template fixture"); + copy_template_tree(&template_dir, fixture.path(), true); + + // Both templates share the PDK dependencies. A separate persistent target + // reuses those artifacts without sharing source or the outer Cargo build lock. + // Cargo fingerprints and its target lock handle changes and concurrent checks. + let executable = std::env::current_exe().expect("current test executable"); + let target_dir = executable + .parent() + .and_then(|deps| deps.parent()) + .expect("test executable has a profile directory") + .join("compile-tests/plugin-pdk-templates"); + let output = Command::new(env!("CARGO")) .arg("check") .arg("--manifest-path") - .arg(&manifest_path) + .arg(fixture.path().join("Cargo.toml")) .arg("--target") .arg("wasm32-unknown-unknown") .arg("--offline") .arg("--target-dir") - .arg(target_dir.path()) + .arg(target_dir) + .current_dir(fixture.path()) .env("CARGO_TERM_COLOR", "never") .output() .expect("spawn cargo check for embedded template"); - let _ = fs::remove_file(&lock_path); assert!( output.status.success(), - "template cargo check failed\nstatus: {}\nstdout:\n{}\nstderr:\n{}", + "template {template} cargo check failed\nstatus: {}\nstdout:\n{}\nstderr:\n{}", output.status, String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); } +fn copy_template_tree(source: &Path, destination: &Path, is_root: bool) { + let metadata = fs::symlink_metadata(source).expect("read template directory metadata"); + assert!( + metadata.is_dir() && !metadata.file_type().is_symlink(), + "template input must be a real directory: {}", + source.display() + ); + for entry in fs::read_dir(source).expect("read template directory") { + let entry = entry.expect("read template entry"); + let path = entry.path(); + let kind = entry.file_type().expect("read template entry type"); + assert!( + !kind.is_symlink() && (kind.is_dir() || kind.is_file()), + "unsupported template input (including symlinks): {}", + path.display() + ); + + let name = entry.file_name(); + // Preserve all compiler inputs, including build scripts and nested assets. + // Only omit the root build output and lockfile: each isolated consumer must + // resolve its own lockfile, as the original unlocked template check did. + if is_root + && ((kind.is_dir() && name == "target") || (kind.is_file() && name == "Cargo.lock")) + { + continue; + } + let copied = destination.join(name); + if kind.is_dir() { + fs::create_dir(&copied).expect("create template fixture directory"); + copy_template_tree(&path, &copied, false); + } else { + fs::copy(&path, &copied).expect("copy unmodified template input"); + } + } +} + #[test] fn pdk_runtime_dependencies_are_guest_side_only() { let cargo: Value = toml::from_str(PDK_CARGO).expect("PDK Cargo.toml parses"); diff --git a/crates/tools/src/view_image.rs b/crates/tools/src/view_image.rs index d6448410e..4d2ff2214 100644 --- a/crates/tools/src/view_image.rs +++ b/crates/tools/src/view_image.rs @@ -360,10 +360,52 @@ mod tests { } #[tokio::test] - async fn bounded_binary_read_accepts_arbitrary_bytes_at_caller_limit_and_rejects_larger_files() - { + async fn bounded_binary_read_accepts_caller_limits_above_the_image_tool_limit() { + // The caller's limit is a reader policy, not a LocalWorkdir storage contract. + // Supply immutable chunks here; the separate component case exercises the + // real provider across a chunk boundary without repeatedly hashing 16 MiB. let dir = TempDir::new().unwrap(); let bytes = vec![0xff; 16 * 1024 * 1024]; + let hash = Sha256::digest(&bytes).into(); + let responses = bytes + .chunks(IMAGE_READ_CHUNK_BYTES) + .enumerate() + .map(|(index, bytes)| { + chunk( + bytes, + (index * IMAGE_READ_CHUNK_BYTES) as u64, + 16 * 1024 * 1024, + hash, + ) + }) + .collect(); + let session = binary_session(&dir, responses); + let path = WorkdirPath::new("artifact.bin").unwrap(); + assert_eq!( + read_workdir_bytes(&session, path.clone(), bytes.len()) + .await + .unwrap(), + bytes + ); + let session = binary_session( + &dir, + vec![chunk( + &bytes[..IMAGE_READ_CHUNK_BYTES], + 0, + bytes.len() as u64, + hash, + )], + ); + assert!(matches!( + read_workdir_bytes(&session, path, bytes.len() - 1).await, + Err(WorkdirError::InvalidArgument(_)) + )); + } + + #[tokio::test] + async fn bounded_binary_read_reassembles_provider_chunks_and_enforces_the_caller_limit() { + let dir = TempDir::new().unwrap(); + let bytes = vec![0xff; IMAGE_READ_CHUNK_BYTES + 1]; std::fs::write(dir.path().join("artifact.bin"), &bytes).unwrap(); let session = LocalWorkdirSession::new(Scope::writable(dir.path()).unwrap(), dir.path().into()); diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index ef2f92af8..13e94295e 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -59,5 +59,5 @@ futures.workspace = true agen.workspace = true serial_test = "3.4.0" tempfile.workspace = true -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } tokio-tungstenite.workspace = true diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 40c53c9e1..610320384 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -2341,6 +2341,7 @@ pub struct WorkerRuntimeExecutionBackend { pending_restores: Mutex>>, worker_locks: Mutex>>>, spawn_restore_timeout: Duration, + shutdown_admission_timeout: Duration, } impl WorkerRuntimeExecutionBackend @@ -2362,6 +2363,7 @@ where pending_restores: Mutex::new(HashMap::new()), worker_locks: Mutex::new(HashMap::new()), spawn_restore_timeout: SPAWN_RESTORE_TASK_TIMEOUT, + shutdown_admission_timeout: Duration::from_secs(5), }) } @@ -2881,9 +2883,10 @@ where let shutdown_requested = Arc::clone(shutdown_requested); let shutdown = shutdown.clone(); let tasks_for_join = tasks.clone(); + let admission_timeout = self.shutdown_admission_timeout; let cleanup = self.run_joined_on_adapter_runtime(async move { tasks_for_join - .request_shutdown(&handle, &shutdown_requested, Duration::from_secs(5)) + .request_shutdown(&handle, &shutdown_requested, admission_timeout) .await?; tasks_for_join.confirm_shutdown_and_join(&shutdown).await }); @@ -3170,9 +3173,10 @@ where let shutdown_requested = execution.shutdown_requested.clone(); let shutdown = execution.shutdown.clone(); let tasks = execution.tasks.clone(); + let admission_timeout = self.shutdown_admission_timeout; let cleanup = self.run_joined_on_adapter_runtime(async move { tasks - .request_shutdown(&handle, &shutdown_requested, Duration::from_secs(5)) + .request_shutdown(&handle, &shutdown_requested, admission_timeout) .await?; tasks.confirm_shutdown_and_join(&shutdown).await?; handle.delete_uncommitted_uploaded_files().map_err(|error| { @@ -4871,15 +4875,11 @@ mod tests { const SAVED_SUBJEKTIV_SYSTEM_PROMPT: &str = "Committed saved Subject Worker system prompt"; - async fn saved_subjektiv_restore_fixture( + fn saved_subjektiv_policy( root: &Path, - materialize_head: bool, - ) -> WorkerExecutionRestoreRequest { - let worker_ref = WorkerRef::new(WorkerId::now_v7()); - let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); - let aggregate_dir = root.join("workers").join(worker_ref.worker_id.to_string()); - let session_store = WorkerSessionStore::new(aggregate_dir.join("session")).unwrap(); - let store = WorkerAggregateStore::new(&aggregate_dir, &worker_name).unwrap(); + worker_ref: &WorkerRef, + ) -> (WorkerManifest, CreateWorkerRequest) { + let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(worker_ref); let mut saved = WorkerManifest::from_toml(&format!( r#" [worker] @@ -4920,6 +4920,19 @@ mod tests { language: "English".to_string(), }); bind_workspace_memory_settings(&mut saved, &request).unwrap(); + (saved, request) + } + + async fn saved_subjektiv_restore_fixture( + root: &Path, + materialize_head: bool, + ) -> WorkerExecutionRestoreRequest { + let worker_ref = WorkerRef::new(WorkerId::now_v7()); + let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); + let aggregate_dir = root.join("workers").join(worker_ref.worker_id.to_string()); + let session_store = WorkerSessionStore::new(aggregate_dir.join("session")).unwrap(); + let store = WorkerAggregateStore::new(&aggregate_dir, &worker_name).unwrap(); + let (saved, request) = saved_subjektiv_policy(root, &worker_ref); let client = MockClient::new(Vec::new()); let mut engine = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated( @@ -5785,7 +5798,12 @@ mod tests { fn duplicate_connect_timeout_retains_candidate_waiter_and_no_false_stop() { let root = tempfile::tempdir().unwrap(); let candidate_root = tempfile::tempdir().unwrap(); - let (backend, request) = profile_stop_fixture(root.path()); + let (mut backend, request) = profile_stop_fixture(root.path()); + // Only admission's deadline is shortened. The real Controller, receipt + // correlation, retained candidate and completion/join barriers stay live. + // The candidate send is gated below, so timeout is not a scheduler race. + let normal_admission_timeout = backend.shutdown_admission_timeout; + backend.shutdown_admission_timeout = Duration::from_millis(100); let primary = backend .workers .lock() @@ -5864,6 +5882,7 @@ mod tests { .unwrap() .contains_key(&request.worker_ref) ); + backend.shutdown_admission_timeout = normal_admission_timeout; release.send(()).unwrap(); let rejected = backend.stop_worker(&request.worker_ref); assert!(!rejected.is_accepted()); @@ -5878,21 +5897,16 @@ mod tests { } #[test] - #[serial_test::serial(worker_allocation)] fn pending_restore_join_failure_never_becomes_absence_on_stop_retry() { let root = tempfile::tempdir().unwrap(); - let runtime_store = root.path().join("runtime"); let backend = WorkerRuntimeExecutionBackend::new( ProfileRuntimeWorkerFactory::new(root.path()) - .with_runtime_store_dir(&runtime_store) - .with_runtime_id("runtime-subjektiv-test"), + .with_runtime_store_dir(root.path().join("runtime")), ) .unwrap(); - let request = backend - .run_on_adapter_runtime(async move { - Ok(saved_subjektiv_restore_fixture(&runtime_store, true).await) - }) - .unwrap(); + // No Controller was ever published: this is the pending-task ownership + // rule, not saved-Session restore. Do not boot a Worker to manufacture it. + let request = direct_restore_request(WorkerRef::new(WorkerId::now_v7())); let failed = backend .spawn_on_adapter_runtime(async { panic!("factory scope failed before publishing result") @@ -6295,6 +6309,41 @@ mod tests { ); } + #[test] + fn saved_manifest_policy_rejects_mismatched_or_missing_trusted_settings() { + let root = tempfile::tempdir().unwrap(); + let worker_ref = WorkerRef::new(WorkerId::now_v7()); + let (saved, request) = saved_subjektiv_policy(root.path(), &worker_ref); + let mut metadata = WorkerMetadata::new(&saved.worker.name, None); + metadata.resolved_manifest_snapshot = + Some(manifest::write_persisted_worker_manifest_snapshot(&saved).unwrap()); + let (restored, _) = + ProfileRuntimeWorkerFactory::manifest_for_restore(&metadata, &request).unwrap(); + assert_eq!( + manifest::write_persisted_worker_manifest_snapshot(&restored).unwrap(), + metadata.resolved_manifest_snapshot.clone().unwrap() + ); + let mut wrong_settings = request.clone(); + wrong_settings.memory_settings.as_mut().unwrap().language = "different language".into(); + let mut detached = request.clone(); + detached.subjektiv_attached = false; + let mut missing_settings = request.clone(); + missing_settings.memory_settings = None; + for (invalid, expected) in [ + (wrong_settings, "subjektiv settings snapshot mismatch"), + (detached, "unauthorized subjektiv attachment"), + (missing_settings, "missing its trusted settings snapshot"), + ] { + let error = ProfileRuntimeWorkerFactory::manifest_for_restore(&metadata, &invalid) + .err() + .expect("invalid saved policy must be rejected"); + assert!(error.contains(expected), "{error}"); + } + } + + // Integration: real Worker startup writes the saved Manifest/Session, then + // both factory entry points reject it before allocating any live resources. + // The local input matrix belongs to saved_manifest_policy_* above. #[tokio::test] #[serial_test::serial(worker_allocation)] async fn profile_factory_saved_manifest_mismatch_rejected_before_live_work() { @@ -6312,28 +6361,27 @@ mod tests { .as_mut() .unwrap() .language = "different language".into(); - let mut detached = request.clone(); - detached.request.subjektiv_attached = false; - let mut missing_settings = request.clone(); - missing_settings.request.memory_settings = None; - for (invalid, expected) in [ - (wrong_settings, "subjektiv settings snapshot mismatch"), - (detached, "unauthorized subjektiv attachment"), - (missing_settings, "missing its trusted settings snapshot"), - ] { - let error = factory.preflight_restore(&invalid).await.unwrap_err(); - assert!(error.contains(expected), "{error}"); - let error = match factory.restore_controller(invalid).await { - Ok(controller) => { - shutdown_profile_controller(controller).await; - panic!("invalid restore started a Controller") - } - Err(error) => error, - }; - assert!(error.contains(expected), "{error}"); - assert_eq!(persisted_files(&runtime_store), before); - assert!(factory.observation_hub.workers.lock().unwrap().is_empty()); - } + let error = factory + .preflight_restore(&wrong_settings) + .await + .unwrap_err(); + assert!( + error.contains("subjektiv settings snapshot mismatch"), + "{error}" + ); + let error = match factory.restore_controller(wrong_settings).await { + Ok(controller) => { + shutdown_profile_controller(controller).await; + panic!("invalid restore started a Controller") + } + Err(error) => error, + }; + assert!( + error.contains("subjektiv settings snapshot mismatch"), + "{error}" + ); + assert_eq!(persisted_files(&runtime_store), before); + assert!(factory.observation_hub.workers.lock().unwrap().is_empty()); let run_dir = factory .worker_aggregate_dir(&request.worker_ref) .unwrap() @@ -6770,7 +6818,7 @@ mod tests { assert!(scope.tasks.lock().unwrap().is_empty()); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn execution_scope_timeout_retains_exact_task_until_completion_is_proven() { let release = Arc::new(tokio::sync::Notify::new()); let task_release = Arc::clone(&release); @@ -7075,29 +7123,35 @@ mod tests { ); } - struct DelayedFactory { - completed: Arc, - delay: Duration, + struct PendingCreateFactory { + started: mpsc::Sender<()>, + dropped: mpsc::Sender<()>, + } + + struct FactoryFutureDrop(mpsc::Sender<()>); + + impl Drop for FactoryFutureDrop { + fn drop(&mut self) { + let _ = self.0.send(()); + } } #[async_trait] - impl RuntimeWorkerFactory for DelayedFactory { + impl RuntimeWorkerFactory for PendingCreateFactory { async fn spawn_controller( &self, _request: WorkerExecutionSpawnRequest, ) -> Result { - tokio::time::sleep(self.delay).await; - self.completed.store(true, Ordering::SeqCst); - Err("delayed factory completed".to_string()) + let _drop = FactoryFutureDrop(self.dropped.clone()); + self.started.send(()).unwrap(); + std::future::pending().await } async fn restore_controller( &self, _request: WorkerExecutionRestoreRequest, ) -> Result { - tokio::time::sleep(self.delay).await; - self.completed.store(true, Ordering::SeqCst); - Err("delayed factory completed".to_string()) + panic!("create timeout must not restore a Worker") } } @@ -7105,11 +7159,12 @@ mod tests { fn create_timeout_cancels_factory_and_removes_persisted_worker() { let root = tempfile::tempdir().unwrap(); let runtime_store_dir = root.path().join("runtime"); - let completed = Arc::new(AtomicBool::new(false)); + let (started_tx, started) = mpsc::channel(); + let (dropped_tx, dropped) = mpsc::channel(); let backend = Arc::new( - WorkerRuntimeExecutionBackend::new(DelayedFactory { - completed: completed.clone(), - delay: Duration::from_millis(200), + WorkerRuntimeExecutionBackend::new(PendingCreateFactory { + started: started_tx, + dropped: dropped_tx, }) .unwrap() .with_spawn_restore_timeout(Duration::from_millis(20)), @@ -7128,7 +7183,9 @@ mod tests { let worker_id = request.worker_id; let create_runtime = runtime.clone(); let create = std::thread::spawn(move || create_runtime.create_worker(request)); - std::thread::sleep(Duration::from_millis(5)); + started + .recv_timeout(Duration::from_secs(5)) + .expect("factory must start before observing concurrent deletion"); let delete_error = runtime .delete_worker(&crate::identity::WorkerRef::new(worker_id)) @@ -7140,11 +7197,9 @@ mod tests { crate::error::RuntimeError::WorkerNotFound { worker_id: missing } if missing == worker_id )); assert!(error.to_string().contains("was cancelled")); - std::thread::sleep(Duration::from_millis(250)); - assert!( - !completed.load(Ordering::SeqCst), - "timed out factory future must not resume after create returns" - ); + dropped + .try_recv() + .expect("cancelled factory future must be dropped before create returns"); assert!(runtime.list_workers().unwrap().is_empty()); assert!(backend.workers.lock().unwrap().is_empty()); assert!( diff --git a/crates/workspace-server/src/auth.rs b/crates/workspace-server/src/auth.rs index 3fc9eb325..fc6a302fe 100644 --- a/crates/workspace-server/src/auth.rs +++ b/crates/workspace-server/src/auth.rs @@ -212,7 +212,7 @@ mod tests { }) .unwrap(); let store = Arc::new( - SqliteWorkspaceStore::in_memory_for_test() + SqliteWorkspaceStore::in_memory_fixture() .unwrap() .with_db_executor(executor), ); diff --git a/crates/workspace-server/src/authority.rs b/crates/workspace-server/src/authority.rs index cd42b41f9..d63d22cf2 100644 --- a/crates/workspace-server/src/authority.rs +++ b/crates/workspace-server/src/authority.rs @@ -237,6 +237,25 @@ impl SqliteWorkspaceAuthority { let database_path = database_path.into(); let workspace_id = workspace_id.into(); let store = SqliteWorkspaceStore::open(&database_path)?; + Self::from_store(database_path, workspace_id, store) + } + + // Normal API fixtures retain real Ticket/MR stores, but the Workspace schema + // was verified when their pristine DB was prepared. Startup still uses new(). + #[cfg(test)] + pub(crate) fn from_fixture_store( + database_path: PathBuf, + workspace_id: String, + store: SqliteWorkspaceStore, + ) -> Result { + Self::from_store(database_path, workspace_id, store) + } + + fn from_store( + database_path: PathBuf, + workspace_id: String, + store: SqliteWorkspaceStore, + ) -> Result { let merge_request_source = Arc::new(AuthorityMergeRequestSource { store: store.clone(), }); diff --git a/crates/workspace-server/src/config_source.rs b/crates/workspace-server/src/config_source.rs index 7ac608622..4b82e93a9 100644 --- a/crates/workspace-server/src/config_source.rs +++ b/crates/workspace-server/src/config_source.rs @@ -795,7 +795,7 @@ mod tests { } async fn open_store() -> SqliteWorkspaceStore { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); store } @@ -1321,26 +1321,28 @@ mod tests { assert!(main_config_has_schema_assertion(DEFAULT_MAIN_CONFIG_SOURCE)); } - #[tokio::test] - async fn candidate_normalizes_main_and_formats_changed_decodal_sources() { - let store = open_store().await; - let current = store.load_workspace_config("w-config").unwrap().unwrap(); - let candidate = store - .evaluate_workspace_config_candidate( - "w-config", - &commit_request( - ¤t, - vec![ - update_main(¤t, "{}"), - ConfigTreeChange::Create { - path: path("module.dcdl"), - content_type: ConfigContentType::Decodal, - content: "{}".into(), - }, - ], - ), - ) - .unwrap(); + #[test] + fn candidate_normalizes_main_and_formats_changed_decodal_sources() { + let current = initial_state().unwrap(); + let changes = [ + update_main(¤t, "{}"), + ConfigTreeChange::Create { + path: path("module.dcdl"), + content_type: ConfigContentType::Decodal, + content: "{}".into(), + }, + ]; + let identity = ( + current.contract.fingerprint.clone(), + current.projection_digest.clone(), + ); + let candidate = evaluate_candidate( + current, + identity, + &changes, + WorkspaceConfigSchemaBundle::empty(), + ) + .unwrap(); assert_eq!( candidate @@ -1414,34 +1416,44 @@ mod tests { ); } - #[tokio::test] - async fn required_main_entrypoint_cannot_be_deleted_or_renamed() { - let store = open_store().await; - let current = store.load_workspace_config("w-config").unwrap().unwrap(); - let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap(); + #[test] + fn required_main_entrypoint_cannot_be_deleted_or_renamed() { for change in [ ConfigTreeChange::Delete { path: path(MAIN_CONFIG_ENTRYPOINT), - expected_digest: main.content_digest.clone(), + expected_digest: "sha256:current".into(), }, ConfigTreeChange::Rename { from: path(MAIN_CONFIG_ENTRYPOINT), to: path("other.dcdl"), - expected_digest: main.content_digest.clone(), + expected_digest: "sha256:current".into(), + }, + ConfigTreeChange::Rename { + from: path("other.dcdl"), + to: path(MAIN_CONFIG_ENTRYPOINT), + expected_digest: "sha256:current".into(), }, ] { - let error = store - .evaluate_workspace_config_candidate( - "w-config", - &commit_request(¤t, vec![change]), - ) - .unwrap_err(); + let error = reject_main_entrypoint_mutation(&[change]).unwrap_err(); assert!(error.to_string().contains("cannot be")); } } + #[test] + fn entrypoint_request_requires_exactly_the_server_owned_main() { + for entrypoints in [ + vec![], + vec![path("other.dcdl")], + vec![path(MAIN_CONFIG_ENTRYPOINT), path("other.dcdl")], + ] { + let error = validate_entrypoint_request(&entrypoints).unwrap_err(); + assert!(error.to_string().contains("must be exactly [main.dcdl]")); + } + validate_entrypoint_request(&[path(MAIN_CONFIG_ENTRYPOINT)]).unwrap(); + } + #[tokio::test] - async fn browser_cannot_replace_server_owned_entrypoint_contract() { + async fn candidate_store_rejects_entrypoint_override_and_main_deletion_without_mutation() { let store = open_store().await; let current = store.load_workspace_config("w-config").unwrap().unwrap(); let mut request = commit_request(¤t, Vec::new()); @@ -1450,11 +1462,33 @@ mod tests { .evaluate_workspace_config_candidate("w-config", &request) .unwrap_err(); assert!(error.to_string().contains("must be exactly [main.dcdl]")); + let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap(); + let error = store + .evaluate_workspace_config_candidate( + "w-config", + &commit_request( + ¤t, + vec![ConfigTreeChange::Delete { + path: path(MAIN_CONFIG_ENTRYPOINT), + expected_digest: main.content_digest.clone(), + }], + ), + ) + .unwrap_err(); + assert!(error.to_string().contains("cannot be deleted")); + assert_eq!( + store + .load_workspace_config("w-config") + .unwrap() + .unwrap() + .snapshot, + current.snapshot + ); } #[tokio::test] async fn invalid_candidate_is_never_persisted() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let main = current.snapshot.get(&path(MAIN_CONFIG_ENTRYPOINT)).unwrap(); @@ -1478,7 +1512,7 @@ mod tests { #[tokio::test] async fn valid_candidate_commits_snapshot_content_and_provenance_atomically() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let committed = store @@ -1496,7 +1530,7 @@ mod tests { #[tokio::test] async fn stale_cas_cannot_overwrite_newer_tree() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let request = commit_request(¤t, vec![update_main(¤t, "{ answer = 42; }")]); @@ -1514,7 +1548,7 @@ mod tests { #[tokio::test] async fn committed_content_remains_retrievable_after_later_commit() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store.upsert_workspace(&workspace()).await.unwrap(); let current = store.load_workspace_config("w-config").unwrap().unwrap(); let first = store diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 4bae8aa3f..cb8eaf5ee 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -6681,7 +6681,7 @@ mod tests { }) .unwrap(); let store = Arc::new( - SqliteWorkspaceStore::in_memory_for_test() + SqliteWorkspaceStore::in_memory_fixture() .unwrap() .with_db_executor(executor.clone()), ); diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index 415f444e9..b8469122e 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -1404,7 +1404,7 @@ mod tests { WorkerId::from_legacy_u64(1) } fn setup() -> SqliteWorkspaceStore { - let s = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let s = SqliteWorkspaceStore::in_memory_fixture().unwrap(); s.with_conn(|c| { c.execute( "INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) \ diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index c0b171d2c..556549a73 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3635,6 +3635,38 @@ fn embedded_runtime_request_audience(config: &ServerConfig) -> crate::Result Result { + match self { + Self::Startup => crate::SqliteWorkspaceStore::open(path), + #[cfg(test)] + Self::PreparedFixture => crate::SqliteWorkspaceStore::connect_prepared_fixture(path), + } + } + + fn open_authority( + self, + path: PathBuf, + workspace_id: String, + ) -> Result { + match self { + Self::Startup => SqliteWorkspaceAuthority::new(path, workspace_id), + #[cfg(test)] + Self::PreparedFixture => { + let store = self.open_store(&path)?; + SqliteWorkspaceAuthority::from_fixture_store(path, workspace_id, store) + } + } + } +} + impl WorkspaceApi { /// Trusted Workspace-scoped Drive access for backend adapters. This is not /// model authorization; callers must authorize requests before using it. @@ -3856,7 +3888,57 @@ impl WorkspaceApi { .await } + #[cfg(test)] + async fn new_api_fixture( + config: ServerConfig, + execution_backend: Arc, + ) -> Result { + // Only a newly copied pristine fixture skips repeated Workspace schema + // checks. Restart/recovery fixtures retain the complete startup path. + let preparation = if config.database_path.exists() { + WorkspaceDatabasePreparation::Startup + } else { + WorkspaceDatabasePreparation::PreparedFixture + }; + let store = Arc::new(crate::SqliteWorkspaceStore::open_api_fixture( + &config.database_path, + )?); + seed_test_registered_workspace(store.as_ref(), &config).await?; + Self::new_with_database_preparation( + config, + store, + execution_backend, + BackendResourceBroker::default(), + HostExecutors::default(), + None, + preparation, + ) + .await + } + async fn new_with_execution_backend_and_broker( + config: ServerConfig, + store: Arc, + execution_backend: Arc, + resource_broker: BackendResourceBroker, + host_executors: HostExecutors, + worker_remove_dispatcher: Option< + Arc, + >, + ) -> Result { + Self::new_with_database_preparation( + config, + store, + execution_backend, + resource_broker, + host_executors, + worker_remove_dispatcher, + WorkspaceDatabasePreparation::Startup, + ) + .await + } + + async fn new_with_database_preparation( mut config: ServerConfig, store: Arc, execution_backend: Arc, @@ -3865,6 +3947,7 @@ impl WorkspaceApi { worker_remove_dispatcher: Option< Arc, >, + database_preparation: WorkspaceDatabasePreparation, ) -> Result { if store.get_workspace(&config.workspace_id).await?.is_none() { return Err(crate::Error::Config(format!( @@ -3904,7 +3987,8 @@ impl WorkspaceApi { let (config_store, repository_secrets) = executor .run(move || { let config_store = Arc::new( - crate::SqliteWorkspaceStore::open(database_path.clone())? + database_preparation + .open_store(&database_path)? .with_db_executor(config_executor), ); let repository_secrets = Arc::new(RepositorySecretService::open( @@ -4005,7 +4089,7 @@ impl WorkspaceApi { .map_err(|error| { Error::Store(format!("failed to register subjektiv storage: {error}")) })?; - let authority = SqliteWorkspaceAuthority::new(database_path, workspace_id)?; + let authority = database_preparation.open_authority(database_path, workspace_id)?; Ok::<_, Error>((feature_storage, registration, authority)) }) .await?; @@ -28226,6 +28310,29 @@ async fn set_worker_retention( )) } +fn cleanup_workdir_decision( + file_status: CleanupWorkdirFileStatus, + cleanliness: CleanupWorkdirCleanliness, + running_linked: bool, + pinned_linked: bool, +) -> (CleanupTargetKind, Option) { + let action = if cleanup_workdir_file_status_is_record_only(file_status) { + CleanupTargetKind::WorkdirRecordDelete + } else if !matches!(cleanliness, CleanupWorkdirCleanliness::Dirty) { + CleanupTargetKind::WorkdirCleanCleanup + } else { + CleanupTargetKind::WorkdirDirtyDiscard + }; + let blocking_reason = if running_linked { + Some("workdir is linked to a running Worker".to_string()) + } else if pinned_linked { + Some("workdir is linked to a pinned Worker/history".to_string()) + } else { + None + }; + (action, blocking_reason) +} + fn build_runtime_cleanup_plan( api: &WorkspaceApi, runtime_id: &str, @@ -28371,20 +28478,8 @@ fn build_runtime_cleanup_plan( .get(record.workdir_id.as_str()) .map(|summary| cleanup_workdir_cleanliness_from_runtime(summary.cleanliness.as_deref())) .unwrap_or_else(|| cleanup_workdir_cleanliness_from_registry(&record.cleanliness)); - let action = if cleanup_workdir_file_status_is_record_only(file_status) { - CleanupTargetKind::WorkdirRecordDelete - } else if !matches!(cleanliness, CleanupWorkdirCleanliness::Dirty) { - CleanupTargetKind::WorkdirCleanCleanup - } else { - CleanupTargetKind::WorkdirDirtyDiscard - }; - let blocking_reason = if running_linked { - Some("workdir is linked to a running Worker".to_string()) - } else if pinned_linked { - Some("workdir is linked to a pinned Worker/history".to_string()) - } else { - None - }; + let (action, blocking_reason) = + cleanup_workdir_decision(file_status, cleanliness, running_linked, pinned_linked); let repository_key = api .store .get_repository(&api.config.workspace_id, repository_id)? @@ -47647,132 +47742,56 @@ mod tests { } #[tokio::test] - async fn ticket_checker_fake_model_regressions_preserve_classification_boundaries() { - struct FakeTicketCheckerModel; - - impl FakeTicketCheckerModel { - fn respond(request: &BackendJobRequest) -> Value { - let input = ticket_item_checker::parse_input(&request.input).unwrap(); - assert!(request.instruction.contains("writer_scope")); - assert!(request.instruction.contains("internal_inconsistency")); - assert!(request.instruction.contains("ambiguous_boundary")); - assert!( - request - .instruction - .contains("Production deployment requires approval") - ); - assert!(request.instruction.contains("background, quotations")); - if input - .body - .contains("This time I only create the Ticket; I will not implement it.") - && !input.body.starts_with("Background bug example:") - { - serde_json::json!({"findings": [{ - "category": "writer_scope", - "quote": "I will not implement it", - "reason": "The active requirement records the writer's own implementation boundary.", - "suggestion": "Remove the writer-specific action boundary." - }]}) - } else if input.body.contains("The rollout region is undecided.") - && input.body.contains("The rollout region is us-east-1.") - { - serde_json::json!({"findings": [{ - "category": "internal_inconsistency", - "quote": "The rollout region is undecided.", - "reason": "The same active text also settles the rollout region as us-east-1.", - "suggestion": "Choose either the unresolved or settled rollout-region statement." - }]}) - } else if input.body == "Implementation stops after Ticket creation." { - serde_json::json!({"findings": [{ - "category": "ambiguous_boundary", - "quote": "Implementation stops after Ticket creation.", - "reason": "The text alone does not establish whether this is a delivery constraint or the writer's current action boundary.", - "suggestion": "Confirm whether this limits the product workflow or only the current writer." - }]}) - } else { - serde_json::json!({"findings": []}) - } - } - } - - struct Case { - name: &'static str, - body: &'static str, - advisory_quote: Option<&'static str>, - } - let cases = vec![ - Case { - name: "writer-scope-positive", - body: "This time I only create the Ticket; I will not implement it.", - advisory_quote: Some("I will not implement it"), - }, - Case { - name: "production-approval-negative", - body: "Production deployment requires approval.", - advisory_quote: None, - }, - Case { - name: "quoted-background-negative", - body: "Background bug example: a generated Ticket once said “I will not implement it.” Do not reproduce that sentence in generated requirements.", - advisory_quote: None, - }, - Case { - name: "unresolved-versus-settled-positive", - body: "The rollout region is undecided. The rollout region is us-east-1.", - advisory_quote: Some("The rollout region is undecided."), - }, - Case { - name: "ambiguous-boundary-confirmation", - body: "Implementation stops after Ticket creation.", - advisory_quote: Some("Implementation stops after Ticket creation."), - }, - ]; - + async fn ticket_checker_routes_model_advisory_after_authenticated_save() { + let case = ticket_item_checker::test_support::cases() + .into_iter() + .find(|case| case.name == "writer-scope-positive") + .unwrap(); let temp = tempfile::tempdir().unwrap(); let (api, execution) = test_api_with_recording_backend(temp.path()).await; let source = spawn_ticket_check_source(&api, "ticket-check-boundary-author"); - for case in cases { - let mut input = ticket::NewTicket::new(format!("Boundary case: {}", case.name)); - input.body = MarkdownText::new(case.body); - let Json(TicketBackendOperationResult::TicketRef(reference)) = - execute_worker_ticket_test_operation( - State(api.clone()), - AxumPath(ScopedWorkspacePath { - workspace_id: TEST_WORKSPACE_ID.to_string(), - }), - ticket_check_headers(&source), - Json(TicketBackendOperation::Create { input }), - ) - .await - .unwrap() - else { - panic!("{}: unexpected Ticket create result", case.name); - }; - let job = ticket_check_job(&api, &reference.id); - let attempt = wait_for_ticket_check_attempt(&api, &job).await; - execution.take_inputs(); - api.accept_backend_job_result( - attempt.worker.as_ref().unwrap(), - &BackendJobResultSubmission { - job_id: job.request.job_id.clone(), - attempt_id: attempt.attempt_id, - input_digest: job.request.input_digest().unwrap(), - result: FakeTicketCheckerModel::respond(&job.request), - }, + let mut input = ticket::NewTicket::new(format!("Boundary case: {}", case.name)); + input.body = MarkdownText::new(case.body); + let Json(TicketBackendOperationResult::TicketRef(reference)) = + execute_worker_ticket_test_operation( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ticket_check_headers(&source), + Json(TicketBackendOperation::Create { input }), ) - .unwrap_or_else(|error| panic!("{}: {error}", case.name)); - let notifications = execution.take_inputs(); - let advisories = notifications - .iter() - .filter(|(_, content)| content.contains("[Ticket item checker advisory]")) - .collect::>(); - match case.advisory_quote { - Some(quote) => { - assert_eq!(advisories.len(), 1, "{}", case.name); - assert!(advisories[0].1.contains(quote), "{}", case.name); - } - None => assert!(advisories.is_empty(), "{}", case.name), + .await + .unwrap() + else { + panic!("{}: unexpected Ticket create result", case.name); + }; + let job = ticket_check_job(&api, &reference.id); + let attempt = wait_for_ticket_check_attempt(&api, &job).await; + execution.take_inputs(); + api.accept_backend_job_result( + attempt.worker.as_ref().unwrap(), + &BackendJobResultSubmission { + job_id: job.request.job_id.clone(), + attempt_id: attempt.attempt_id, + input_digest: job.request.input_digest().unwrap(), + result: ticket_item_checker::test_support::FakeTicketCheckerModel::respond( + &job.request, + ), + }, + ) + .unwrap_or_else(|error| panic!("{}: {error}", case.name)); + let notifications = execution.take_inputs(); + let advisories = notifications + .iter() + .filter(|(_, content)| content.contains("[Ticket item checker advisory]")) + .collect::>(); + match case.advisory_quote { + Some(quote) => { + assert_eq!(advisories.len(), 1, "{}", case.name); + assert!(advisories[0].1.contains(quote), "{}", case.name); } + None => assert!(advisories.is_empty(), "{}", case.name), } } @@ -56591,16 +56610,56 @@ mod tests { .unwrap(); } + #[tokio::test] + async fn api_startup_revalidates_workspace_schema_after_store_was_opened() { + let temp = tempfile::tempdir().unwrap(); + let config = test_server_config(temp.path()); + let store = + Arc::new(SqliteWorkspaceStore::open_api_fixture(&config.database_path).unwrap()); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + // Do not hold a live API/Runtime while testing startup of the same store: + // its exclusive Runtime store lease would reject before schema validation. + store + .with_conn(|conn| { + conn.execute_batch("DROP TRIGGER ticket_assignment_ticket_parent_tombstone;")?; + Ok(()) + }) + .unwrap(); + let error = WorkspaceApi::new_for_test(config.clone(), store) + .await + .err() + .expect("API startup must revalidate an already opened store"); + assert!( + error + .to_string() + .contains("ticket_assignment_ticket_parent_tombstone"), + "expected missing-trigger refusal from API startup, got {error:?}" + ); + let error = WorkspaceApi::new_api_fixture( + config, + Arc::new(DeterministicExecutionBackend::default()), + ) + .await + .err() + .expect("existing fixture DBs must use production startup"); + assert!( + error + .to_string() + .contains("ticket_assignment_ticket_parent_tombstone"), + "expected missing-trigger refusal reopening fixture, got {error:?}" + ); + } + async fn test_api_with_recording_backend( workspace_root: impl Into, ) -> (WorkspaceApi, Arc) { let config = test_server_config(workspace_root); - let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); let execution = Arc::new(DeterministicExecutionBackend::default()); - let api = - WorkspaceApi::new_with_execution_backend(config, Arc::new(store), execution.clone()) - .await - .unwrap(); + let api = WorkspaceApi::new_api_fixture(config, execution.clone()) + .await + .unwrap(); (api, execution) } @@ -56619,14 +56678,9 @@ mod tests { config.repositories[0].source = source; config.repositories[0].path = None; config.repositories[0].default_selector = Some("develop".to_string()); - let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); - WorkspaceApi::new_with_execution_backend( - config, - Arc::new(store), - Arc::new(DeterministicExecutionBackend::default()), - ) - .await - .unwrap() + WorkspaceApi::new_api_fixture(config, Arc::new(DeterministicExecutionBackend::default())) + .await + .unwrap() } async fn test_api_with_docs_repository(workspace_root: impl Into) -> WorkspaceApi { @@ -56648,14 +56702,9 @@ mod tests { docs.path = Some(docs_root); docs.default_selector = Some("develop".to_string()); config.repositories.push(docs); - let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); - WorkspaceApi::new_with_execution_backend( - config, - Arc::new(store), - Arc::new(DeterministicExecutionBackend::default()), - ) - .await - .unwrap() + WorkspaceApi::new_api_fixture(config, Arc::new(DeterministicExecutionBackend::default())) + .await + .unwrap() } fn test_owner_actor() -> RequestActor { @@ -59899,22 +59948,17 @@ mod tests { } } - #[tokio::test] - async fn cleanup_plan_offers_normal_revalidation_for_prior_failure_and_unknown() { - let workspace = tempfile::tempdir().unwrap(); - init_clean_git_workspace(workspace.path()); - let api = test_api(workspace.path()).await; - for (id, state) in [ - ("pending-retry", "pending"), - ("unknown-retry", "unknown"), - ("corrupt-retry", "corrupted"), - ] { - seed_cleanup_workdir(&api, id, state, "unknown"); - } - let plan = build_runtime_cleanup_plan(&api, "runtime-test").unwrap(); - for candidate in plan.workdirs { - assert_eq!(candidate.action, CleanupTargetKind::WorkdirCleanCleanup); - assert!(candidate.blocking_reason.is_none()); + #[test] + fn cleanup_plan_offers_normal_revalidation_for_prior_failure_and_unknown() { + for state in ["pending", "unknown", "corrupted"] { + let (action, blocking_reason) = cleanup_workdir_decision( + cleanup_workdir_file_status_from_registry(state), + cleanup_workdir_cleanliness_from_registry("unknown"), + false, + false, + ); + assert_eq!(action, CleanupTargetKind::WorkdirCleanCleanup, "{state}"); + assert!(blocking_reason.is_none(), "{state}"); } } diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index 8b1bb7d33..9923b06d9 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -2218,167 +2218,188 @@ impl SubjectRemovalFixture { } } -#[tokio::test] -async fn erased_archive_plan_resend_and_reopen_replan_without_republishing_sources() { - for stage in ["planned", "executing", "failed"] { - for restart in [false, true] { - let mut f = SubjectRemovalFixture::new().await; - f.disable(); - let policy = f - .api - .config_store - .worker_retention_policy(TEST_WORKSPACE_ID) - .unwrap(); - let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); - let archive_parent = f - .dir - .path() - .join(".test-embedded-runtime-store/archives/workers"); - if stage == "planned" { - conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='executing' BEGIN SELECT RAISE(ABORT,'injected execution interruption'); END;").unwrap(); - } else { - // Fail Archive publication before any receipt. Suppress recording - // failure too to model the executing-at-Host-stop durable boundary. - std::fs::create_dir_all(archive_parent.parent().unwrap()).unwrap(); - std::fs::write(&archive_parent, "publication blocked").unwrap(); - if stage == "executing" { - conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='failed' BEGIN SELECT RAISE(ABORT,'injected failure-save interruption'); END;").unwrap(); - } +// Each case owns a complete isolated fixture; do not serialize independent +// recovery paths inside one test. Transitions within each helper remain sequential. +macro_rules! journal_recovery_cases { + ($helper:ident; $($name:ident => ($($argument:expr),*)),+ $(,)?) => { + $( + #[tokio::test] + async fn $name() { + $helper($($argument),*).await; } - assert_ne!(f.remove().status, StatusCode::OK.as_u16()); - conn.execute_batch("DROP TRIGGER IF EXISTS interrupt_removal") - .unwrap(); - if stage != "planned" { - std::fs::remove_file(&archive_parent).unwrap(); - } - let old = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - assert_eq!(serde_json::to_value(old.plan.state).unwrap(), stage); - assert_eq!( - old.plan.session_disposition, - worker_runtime::retention::SessionDisposition::Archive - ); - f.api - .erase_subjektiv_journal_session( - &f.attr.session_id, - "erase-interrupted-archive-plan", - ) - .unwrap(); - if restart { - f = f.reopen().await; - } - let response = f.remove(); - assert_eq!( - response.status, - StatusCode::OK.as_u16(), - "{stage}/{restart}: {}", - response.body - ); - let saved = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - assert_eq!( - saved.plan.session_disposition, - worker_runtime::retention::SessionDisposition::Purge - ); - assert_eq!(saved.plan.reason, old.plan.reason); - assert_ne!(saved.plan.operation_id, old.plan.operation_id); - let (state, fingerprint, detail): (String, String, String) = conn.query_row( - "SELECT o.state,o.input_fingerprint,a.detail FROM worker_removal_operations o JOIN worker_retention_audit_events a USING(operation_id) WHERE o.operation_id=?1 AND a.event_kind='erased_archive_superseded'", - [&old.plan.operation_id], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).unwrap(); - assert_eq!(state, "stale"); - assert_eq!(fingerprint, old.plan.input_fingerprint); - let detail: Value = serde_json::from_str(&detail).unwrap(); - assert_eq!(detail["old_acceptance"], "unknown_archive_fenced"); - assert!(detail["source_removed"].is_null()); - f.assert_erased_removal(); - assert_eq!( - f.api - .config_store - .worker_retention_policy(TEST_WORKSPACE_ID) - .unwrap(), - policy - ); - } - } + )+ + }; } -#[tokio::test] -async fn erased_completed_archive_receipt_is_audited_without_republishing_on_resend_or_reopen() { - for restart in [false, true] { - let mut f = SubjectRemovalFixture::new().await; - f.disable(); - let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); - // Runtime has committed source_removed, but Backend commit/response is lost. - conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap(); - assert_ne!(f.remove().status, StatusCode::OK.as_u16()); - conn.execute_batch("DROP TRIGGER interrupt_removal") - .unwrap(); - let old = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - assert_eq!( - old.plan.state, - crate::retention::WorkerRemovalPlanState::Failed - ); - let receipt = f - .api - .runtime - .execute_worker_retention(&f.target, old.runtime_request.clone()) - .unwrap(); - assert!(receipt.source_removed); - let archive_id = old.plan.archive_id.as_ref().unwrap(); - let archive_source = f - .dir - .path() - .join(".test-embedded-runtime-store/archives/workers") - .join(archive_id) - .join("session"); - assert!(archive_source.exists()); - // The domain tombstone commits even though retained-source resolution fails. - // Recovery must locate only this plan's exact reserved archive, not publish it. - assert!( - f.api - .erase_subjektiv_journal_session(&f.attr.session_id, "erase-transferred-original") - .is_err() - ); - assert!(f.store.journal_session_erased(&f.host).unwrap()); - if restart { - f = f.reopen().await; +async fn assert_erased_archive_plan_replans_without_republishing_sources( + stage: &str, + restart: bool, +) { + let mut f = SubjectRemovalFixture::new().await; + f.disable(); + let policy = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + let archive_parent = f + .dir + .path() + .join(".test-embedded-runtime-store/archives/workers"); + if stage == "planned" { + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='executing' BEGIN SELECT RAISE(ABORT,'injected execution interruption'); END;").unwrap(); + } else { + // Fail Archive publication before any receipt. Suppress recording + // failure too to model the executing-at-Host-stop durable boundary. + std::fs::create_dir_all(archive_parent.parent().unwrap()).unwrap(); + std::fs::write(&archive_parent, "publication blocked").unwrap(); + if stage == "executing" { + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='failed' BEGIN SELECT RAISE(ABORT,'injected failure-save interruption'); END;").unwrap(); } - let response = f.remove(); - assert_eq!( - response.status, - StatusCode::OK.as_u16(), - "{restart}: {}", - response.body - ); - let saved = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - assert_eq!(saved.plan.operation_id, old.plan.operation_id); - assert_eq!(saved.plan.input_fingerprint, old.plan.input_fingerprint); - assert_eq!(saved.plan.session_disposition, old.plan.session_disposition); - let detail: String = conn.query_row("SELECT detail FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); - let detail: Value = serde_json::from_str(&detail).unwrap(); - assert_eq!(detail["source_removed"], true); - assert_eq!(detail["old_acceptance"], "receipt_confirmed"); - assert!(!archive_source.exists()); - f.assert_erased_removal(); } + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER IF EXISTS interrupt_removal") + .unwrap(); + if stage != "planned" { + std::fs::remove_file(&archive_parent).unwrap(); + } + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(serde_json::to_value(old.plan.state).unwrap(), stage); + assert_eq!( + old.plan.session_disposition, + worker_runtime::retention::SessionDisposition::Archive + ); + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-interrupted-archive-plan") + .unwrap(); + if restart { + f = f.reopen().await; + } + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{stage}/{restart}: {}", + response.body + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!( + saved.plan.session_disposition, + worker_runtime::retention::SessionDisposition::Purge + ); + assert_eq!(saved.plan.reason, old.plan.reason); + assert_ne!(saved.plan.operation_id, old.plan.operation_id); + let (state, fingerprint, detail): (String, String, String) = conn.query_row( + "SELECT o.state,o.input_fingerprint,a.detail FROM worker_removal_operations o JOIN worker_retention_audit_events a USING(operation_id) WHERE o.operation_id=?1 AND a.event_kind='erased_archive_superseded'", + [&old.plan.operation_id], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).unwrap(); + assert_eq!(state, "stale"); + assert_eq!(fingerprint, old.plan.input_fingerprint); + let detail: Value = serde_json::from_str(&detail).unwrap(); + assert_eq!(detail["old_acceptance"], "unknown_archive_fenced"); + assert!(detail["source_removed"].is_null()); + f.assert_erased_removal(); + assert_eq!( + f.api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(), + policy + ); +} + +journal_recovery_cases! { + assert_erased_archive_plan_replans_without_republishing_sources; + erased_archive_planned_resend_replans_without_republishing_sources => ("planned", false), + erased_archive_planned_reopen_replans_without_republishing_sources => ("planned", true), + erased_archive_executing_resend_replans_without_republishing_sources => ("executing", false), + erased_archive_executing_reopen_replans_without_republishing_sources => ("executing", true), + erased_archive_failed_resend_replans_without_republishing_sources => ("failed", false), + erased_archive_failed_reopen_replans_without_republishing_sources => ("failed", true), +} + +async fn assert_erased_completed_archive_receipt_is_audited_without_republishing(restart: bool) { + let mut f = SubjectRemovalFixture::new().await; + f.disable(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + // Runtime has committed source_removed, but Backend commit/response is lost. + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER interrupt_removal") + .unwrap(); + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!( + old.plan.state, + crate::retention::WorkerRemovalPlanState::Failed + ); + let receipt = f + .api + .runtime + .execute_worker_retention(&f.target, old.runtime_request.clone()) + .unwrap(); + assert!(receipt.source_removed); + let archive_id = old.plan.archive_id.as_ref().unwrap(); + let archive_source = f + .dir + .path() + .join(".test-embedded-runtime-store/archives/workers") + .join(archive_id) + .join("session"); + assert!(archive_source.exists()); + // The domain tombstone commits even though retained-source resolution fails. + // Recovery must locate only this plan's exact reserved archive, not publish it. + assert!( + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-transferred-original") + .is_err() + ); + assert!(f.store.journal_session_erased(&f.host).unwrap()); + if restart { + f = f.reopen().await; + } + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{restart}: {}", + response.body + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(saved.plan.operation_id, old.plan.operation_id); + assert_eq!(saved.plan.input_fingerprint, old.plan.input_fingerprint); + assert_eq!(saved.plan.session_disposition, old.plan.session_disposition); + let detail: String = conn.query_row("SELECT detail FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); + let detail: Value = serde_json::from_str(&detail).unwrap(); + assert_eq!(detail["source_removed"], true); + assert_eq!(detail["old_acceptance"], "receipt_confirmed"); + assert!(!archive_source.exists()); + f.assert_erased_removal(); +} + +journal_recovery_cases! { + assert_erased_completed_archive_receipt_is_audited_without_republishing; + erased_completed_archive_receipt_resend_is_audited_without_republishing => (false), + erased_completed_archive_receipt_reopen_is_audited_without_republishing => (true), } #[tokio::test] @@ -2438,206 +2459,206 @@ async fn erased_archive_without_receipt_or_inventory_remains_unknown_without_rep assert_eq!(count, 1); } -#[tokio::test] -async fn erased_archive_retained_diagnostics_receipt_finalizes_only_after_physical_ack_on_resend_or_reopen() - { - for restart in [false, true] { - for interrupt_erase in [false, true] { - let mut f = SubjectRemovalFixture::new().await; - f.disable(); - let policy = f - .api - .config_store - .worker_retention_policy(TEST_WORKSPACE_ID) +async fn assert_erased_archive_retained_diagnostics_receipt_finalizes_only_after_physical_ack( + restart: bool, + interrupt_erase: bool, +) { + let mut f = SubjectRemovalFixture::new().await; + f.disable(); + let policy = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let policy = f + .api + .config_store + .update_worker_retention_policy( + TEST_WORKSPACE_ID, + &policy.policy_digest, + &crate::retention::WorkerRetentionPolicyUpdate { + policy_id: "fixture-retained-diagnostics".into(), + session_disposition: policy.session_disposition, + metadata_disposition: policy.metadata_disposition, + archive_retention: policy.archive_retention, + diagnostics_disposition: worker_runtime::retention::DiagnosticsDisposition::Retain, + diagnostics_retention_seconds: Some(3600), + }, + ) + .unwrap(); + let worker_root = session_root(f.dir.path(), &f.target) + .parent() + .unwrap() + .to_path_buf(); + std::fs::create_dir_all(worker_root.join("runs/fixture-run")).unwrap(); + std::fs::write( + worker_root.join("runs/fixture-run/worker.out.log"), + "diagnostic original", + ) + .unwrap(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + // Real Runtime receipt exists; Backend commit/response is interrupted. + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER interrupt_removal") + .unwrap(); + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + let runtime_root = f.dir.path().join(".test-embedded-runtime-store"); + let receipt_path = runtime_root + .join("retention/operations") + .join(format!("{}.json", old.plan.operation_id)); + let receipt_bytes = std::fs::read(&receipt_path).unwrap(); + let receipt = f + .api + .runtime + .execute_worker_retention(&f.target, old.runtime_request.clone()) + .unwrap(); + assert!(receipt.source_removed); + assert!(receipt.diagnostics_retained); + let diagnostics_root = runtime_root + .join("archives/diagnostics") + .join(&old.plan.operation_id); + let diagnostics_manifest = diagnostics_root.join("manifest.json"); + let diagnostics_bytes = std::fs::read(&diagnostics_manifest).unwrap(); + let archive_source = runtime_root + .join("archives/workers") + .join(old.plan.archive_id.as_ref().unwrap()) + .join("session"); + assert!(diagnostics_root.exists()); + assert!(archive_source.exists()); + // Domain erasure succeeds, but no physical acknowledgement can be + // claimed from the old retained locator after aggregate removal. + assert!( + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-retained-diagnostics") + .is_err() + ); + if interrupt_erase { + // Runtime must reject mismatched diagnostics identity rather than + // acknowledge removal of an unrelated operation. No fake Host ack. + let mut damaged: Value = serde_json::from_slice(&diagnostics_bytes).unwrap(); + damaged["input_fingerprint"] = json!("unrelated-operation"); + std::fs::write(&diagnostics_manifest, serde_json::to_vec(&damaged).unwrap()).unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::CONFLICT.as_u16(), + "{}", + response.body + ); + let body: Value = serde_json::from_str(&response.body).unwrap(); + assert_eq!(body["code"], "erased_removal_reconciliation_unconfirmed"); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) .unwrap() - .unwrap(); - let policy = f - .api - .config_store - .update_worker_retention_policy( - TEST_WORKSPACE_ID, - &policy.policy_digest, - &crate::retention::WorkerRetentionPolicyUpdate { - policy_id: "fixture-retained-diagnostics".into(), - session_disposition: policy.session_disposition, - metadata_disposition: policy.metadata_disposition, - archive_retention: policy.archive_retention, - diagnostics_disposition: - worker_runtime::retention::DiagnosticsDisposition::Retain, - diagnostics_retention_seconds: Some(3600), - }, - ) - .unwrap(); - let worker_root = session_root(f.dir.path(), &f.target) - .parent() - .unwrap() - .to_path_buf(); - std::fs::create_dir_all(worker_root.join("runs/fixture-run")).unwrap(); - std::fs::write( - worker_root.join("runs/fixture-run/worker.out.log"), - "diagnostic original", + .is_some() + ); + let count: i64 = conn.query_row("SELECT COUNT(*) FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); + assert_eq!( + count, 0, + "failed acknowledgement cannot authorize finalization/audit" + ); + assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt_bytes); + let waiting = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(waiting.plan.operation_id, old.plan.operation_id); + assert_eq!(waiting.plan.input_fingerprint, old.plan.input_fingerprint); + assert_eq!( + waiting.plan.state, + crate::retention::WorkerRemovalPlanState::Executing + ); + let plans: i64 = conn + .query_row( + "SELECT COUNT(*) FROM worker_removal_operations WHERE worker_id=?1", + [&f.target.worker_id], + |r| r.get(0), ) .unwrap(); - let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); - // Real Runtime receipt exists; Backend commit/response is interrupted. - conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap(); - assert_ne!(f.remove().status, StatusCode::OK.as_u16()); - conn.execute_batch("DROP TRIGGER interrupt_removal") - .unwrap(); - let old = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - let runtime_root = f.dir.path().join(".test-embedded-runtime-store"); - let receipt_path = runtime_root - .join("retention/operations") - .join(format!("{}.json", old.plan.operation_id)); - let receipt_bytes = std::fs::read(&receipt_path).unwrap(); - let receipt = f - .api - .runtime - .execute_worker_retention(&f.target, old.runtime_request.clone()) - .unwrap(); - assert!(receipt.source_removed); - assert!(receipt.diagnostics_retained); - let diagnostics_root = runtime_root - .join("archives/diagnostics") - .join(&old.plan.operation_id); - let diagnostics_manifest = diagnostics_root.join("manifest.json"); - let diagnostics_bytes = std::fs::read(&diagnostics_manifest).unwrap(); - let archive_source = runtime_root - .join("archives/workers") - .join(old.plan.archive_id.as_ref().unwrap()) - .join("session"); - assert!(diagnostics_root.exists()); - assert!(archive_source.exists()); - // Domain erasure succeeds, but no physical acknowledgement can be - // claimed from the old retained locator after aggregate removal. - assert!( - f.api - .erase_subjektiv_journal_session( - &f.attr.session_id, - "erase-retained-diagnostics" - ) - .is_err() - ); - if interrupt_erase { - // Runtime must reject mismatched diagnostics identity rather than - // acknowledge removal of an unrelated operation. No fake Host ack. - let mut damaged: Value = serde_json::from_slice(&diagnostics_bytes).unwrap(); - damaged["input_fingerprint"] = json!("unrelated-operation"); - std::fs::write(&diagnostics_manifest, serde_json::to_vec(&damaged).unwrap()) - .unwrap(); - let response = f.remove(); - assert_eq!( - response.status, - StatusCode::CONFLICT.as_u16(), - "{}", - response.body - ); - let body: Value = serde_json::from_str(&response.body).unwrap(); - assert_eq!(body["code"], "erased_removal_reconciliation_unconfirmed"); - assert!( - f.api - .store - .get_worker_registry(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .is_some() - ); - let count: i64 = conn.query_row("SELECT COUNT(*) FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); - assert_eq!( - count, 0, - "failed acknowledgement cannot authorize finalization/audit" - ); - assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt_bytes); - let waiting = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - assert_eq!(waiting.plan.operation_id, old.plan.operation_id); - assert_eq!(waiting.plan.input_fingerprint, old.plan.input_fingerprint); - assert_eq!( - waiting.plan.state, - crate::retention::WorkerRemovalPlanState::Executing - ); - let plans: i64 = conn - .query_row( - "SELECT COUNT(*) FROM worker_removal_operations WHERE worker_id=?1", - [&f.target.worker_id], - |r| r.get(0), - ) - .unwrap(); - assert_eq!( - plans, 1, - "missing physical ack cannot authorize a replacement plan" - ); - std::fs::write(&diagnostics_manifest, &diagnostics_bytes).unwrap(); - } - if restart { - f = f.reopen().await; - } - let response = f.remove(); - assert_eq!( - response.status, - StatusCode::OK.as_u16(), - "restart={restart}, interrupted={interrupt_erase}: {}", - response.body - ); - assert!(!diagnostics_root.exists()); - assert!(!archive_source.exists()); - assert_eq!( - std::fs::read(&receipt_path).unwrap(), - receipt_bytes, - "historical Runtime receipt must not be rewritten" - ); - let saved = f - .api - .config_store - .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .unwrap(); - assert_eq!(saved.plan.operation_id, old.plan.operation_id); - assert_eq!(saved.plan.input_fingerprint, old.plan.input_fingerprint); - assert_eq!( - saved.plan.diagnostics_disposition, - old.plan.diagnostics_disposition - ); - let detail: String = conn.query_row("SELECT detail FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); - let detail: Value = serde_json::from_str(&detail).unwrap(); - assert_eq!(detail["source_removed"], true); - assert_eq!(detail["old_diagnostics_retained"], true); - assert_eq!( - detail["runtime_erasure_ack"]["session_id"], - f.attr.session_id - ); - let diagnostics_catalog: i64 = conn - .query_row( - "SELECT COUNT(*) FROM worker_diagnostics_archives WHERE operation_id=?1", - [&old.plan.operation_id], - |r| r.get(0), - ) - .unwrap(); - assert_eq!(diagnostics_catalog, 0); - let tombstone_archive: Option = conn - .query_row( - "SELECT archive_id FROM worker_tombstones WHERE operation_id=?1", - [&old.plan.operation_id], - |r| r.get(0), - ) - .unwrap(); - assert!(tombstone_archive.is_none()); - f.assert_erased_removal(); - assert_eq!( - f.api - .config_store - .worker_retention_policy(TEST_WORKSPACE_ID) - .unwrap(), - Some(policy) - ); - } + assert_eq!( + plans, 1, + "missing physical ack cannot authorize a replacement plan" + ); + std::fs::write(&diagnostics_manifest, &diagnostics_bytes).unwrap(); } + if restart { + f = f.reopen().await; + } + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "restart={restart}, interrupted={interrupt_erase}: {}", + response.body + ); + assert!(!diagnostics_root.exists()); + assert!(!archive_source.exists()); + assert_eq!( + std::fs::read(&receipt_path).unwrap(), + receipt_bytes, + "historical Runtime receipt must not be rewritten" + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(saved.plan.operation_id, old.plan.operation_id); + assert_eq!(saved.plan.input_fingerprint, old.plan.input_fingerprint); + assert_eq!( + saved.plan.diagnostics_disposition, + old.plan.diagnostics_disposition + ); + let detail: String = conn.query_row("SELECT detail FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); + let detail: Value = serde_json::from_str(&detail).unwrap(); + assert_eq!(detail["source_removed"], true); + assert_eq!(detail["old_diagnostics_retained"], true); + assert_eq!( + detail["runtime_erasure_ack"]["session_id"], + f.attr.session_id + ); + let diagnostics_catalog: i64 = conn + .query_row( + "SELECT COUNT(*) FROM worker_diagnostics_archives WHERE operation_id=?1", + [&old.plan.operation_id], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(diagnostics_catalog, 0); + let tombstone_archive: Option = conn + .query_row( + "SELECT archive_id FROM worker_tombstones WHERE operation_id=?1", + [&old.plan.operation_id], + |r| r.get(0), + ) + .unwrap(); + assert!(tombstone_archive.is_none()); + f.assert_erased_removal(); + assert_eq!( + f.api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(), + Some(policy) + ); +} + +journal_recovery_cases! { + assert_erased_archive_retained_diagnostics_receipt_finalizes_only_after_physical_ack; + erased_archive_retained_diagnostics_resend_finalizes_only_after_physical_ack => (false, false), + erased_archive_retained_diagnostics_interrupted_resend_finalizes_only_after_physical_ack => (false, true), + erased_archive_retained_diagnostics_reopen_finalizes_only_after_physical_ack => (true, false), + erased_archive_retained_diagnostics_interrupted_reopen_finalizes_only_after_physical_ack => (true, true), } diff --git a/crates/workspace-server/src/server/tests/worker_drive_tests.rs b/crates/workspace-server/src/server/tests/worker_drive_tests.rs index 481c309ad..e951a6308 100644 --- a/crates/workspace-server/src/server/tests/worker_drive_tests.rs +++ b/crates/workspace-server/src/server/tests/worker_drive_tests.rs @@ -730,13 +730,31 @@ async fn unsafe_names_and_oversized_text_cannot_publish_and_reads_truncate_at_ut #[tokio::test] async fn omitted_discovery_limits_use_backend_maximum_pages_in_tools_and_native_wip() { let f = Fixture::new(["read_write", "read_only"]).await; - let mut expected = std::collections::BTreeSet::new(); - for index in 0..201 { - let created = f - .create(&format!("default-page-{index:03}.md"), "bounded") - .await; - expected.insert(entry(&created)["node_id"].as_str().unwrap().to_owned()); - } + // This case owns discovery defaults, not repeated mutation routing. Seed through + // the real Drive owner (including blobs/receipts); reads below retain signed tools/WIP. + let drive = f.api.drive().await.unwrap(); + let expected = tokio::task::spawn_blocking(move || { + let root = drive.root().unwrap().id; + let mut expected = std::collections::BTreeSet::new(); + for index in 0..201 { + let created = drive + .mutate( + &format!("default-page-seed-{index:03}"), + "pagination-fixture", + workspace_drive::Mutation::CreateFile { + parent: root, + name: format!("default-page-{index:03}.md"), + content_type: "text/markdown".into(), + bytes: b"bounded".to_vec(), + }, + ) + .unwrap(); + expected.insert(String::from(created.node.id)); + } + expected + }) + .await + .unwrap(); let native = native_tools(&f.features[1]); let root = value( native_tool(&native, "Inspect", json!({"path":"/drive","refresh":true})) diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs index 142ae143e..54f74ebe4 100644 --- a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs @@ -401,6 +401,11 @@ async fn deletion_fences_new_requests_then_drains_without_callback_starvation() let callback = gate.admit(true).expect("parent may still need callback"); reply.send(Ok(read_result())).unwrap(); assert_json(long.await.unwrap(), StatusCode::OK).await; + // HTTP completion may precede cleanup of retained DB/Host request leases. + // Keep the callback alive while observing the real parent-drain condition. + tokio::time::timeout(BOUND, gate.wait_for_parent_requests_to_drain()) + .await + .expect("completed HTTP request retained its admission through the drain deadline"); // No new callbacks may prolong the drain once all parents have left. assert!(gate.admit(true).is_none()); assert!(futures::poll!(&mut deletion).is_pending()); diff --git a/crates/workspace-server/src/server/workspace_admission.rs b/crates/workspace-server/src/server/workspace_admission.rs index 4c13c0fe6..5168972bd 100644 --- a/crates/workspace-server/src/server/workspace_admission.rs +++ b/crates/workspace-server/src/server/workspace_admission.rs @@ -52,6 +52,28 @@ impl WorkspaceAdmission { }) } + // Observe parent cleanup without releasing a callback held by the test. + // A reply can arrive before the DB/Host job drops its retained request lease. + #[cfg(test)] + pub(super) async fn wait_for_parent_requests_to_drain(&self) { + loop { + let changed = self.changed.notified(); + tokio::pin!(changed); + changed.as_mut().enable(); + { + let state = self.state.lock().expect("Workspace admission poisoned"); + assert!( + state.deleting, + "parent drain observation requires a deletion fence" + ); + if state.requests == 0 { + return; + } + } + changed.await; + } + } + pub(super) async fn lock_deletion(self: &Arc) -> DeletionLease { let serial = self.deletion.clone().lock_owned().await; self.state @@ -135,3 +157,43 @@ impl Drop for DeletionLease { .deleting = false; } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::db_executor::DbOperationLifetime; + + #[tokio::test] + async fn parent_drain_waits_for_retained_job_cleanup_without_waiting_for_callbacks() { + let gate = Arc::new(WorkspaceAdmission::default()); + let parent = gate.admit(false).unwrap(); + let retained_job = parent.retain(); + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); + let callback = gate.admit(true).unwrap(); + drop(parent); + let mut parents_drained = Box::pin(gate.wait_for_parent_requests_to_drain()); + assert!(futures::poll!(&mut parents_drained).is_pending()); + assert!( + gate.admit(true).is_some(), + "retained parent may still need callbacks" + ); + drop(retained_job); + tokio::time::timeout(std::time::Duration::from_secs(5), parents_drained) + .await + .expect("parent cleanup did not notify the drain observer"); + assert!(gate.admit(true).is_none()); + assert!(gate.admit(false).is_none()); + assert!( + futures::poll!(&mut deletion).is_pending(), + "callback still owns its lease" + ); + drop(callback); + let deletion = tokio::time::timeout(std::time::Duration::from_secs(5), deletion) + .await + .expect("callback cleanup did not finish deletion drain"); + assert!(gate.admit(true).is_none()); + drop(deletion); + assert!(gate.admit(false).is_some()); + } +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index af1f230e5..c2e52e680 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -15616,6 +15616,10 @@ fn table_columns(conn: &Connection, table_name: &str) -> Result> { .map_err(Error::from) } +#[cfg(test)] +#[path = "store_test_support.rs"] +mod test_support; + #[cfg(test)] mod tests { include!("assignment_tracking_tests.rs"); @@ -15653,7 +15657,7 @@ mod tests { fn drive_grants_migrate_and_preserve_audit_through_revoke_worker_delete_and_restart() { let t = tempfile::tempdir().unwrap(); let path = t.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); let legacy = Connection::open_in_memory().unwrap(); configure_sqlite(&legacy).unwrap(); legacy.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY); @@ -15778,7 +15782,7 @@ mod tests { } #[test] fn drive_grants_baseline_constraints_and_worker_lifetime_fail_closed() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); let proposed = drive_grant_fixture(&store); let first = store.create_workspace_drive_grant(&proposed).unwrap(); store.with_conn(|c| { @@ -16139,7 +16143,7 @@ mod tests { #[tokio::test] async fn external_workdir_grant_persists_typed_source_and_fences_generations() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store .upsert_account(&AccountRecord { account_id: "owner".to_string(), @@ -16890,39 +16894,38 @@ mod tests { } #[test] - fn schema_v59_upgrade_adds_runtime_removal_authority_atomically() { + fn schema_v59_upgrade_adds_runtime_removal_authority_atomically() -> Result<()> { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); prepare_retained_schema(&path, 59); - let migrated = SqliteWorkspaceStore::open_for_test(&path).unwrap(); - migrated - .with_conn(|conn| { - let table_count: i64 = conn.query_row( - "SELECT COUNT(*) FROM sqlite_master \ - WHERE type = 'table' AND name = 'runtime_removal_operations'", - [], - |row| row.get(0), - )?; - let trigger_count: i64 = conn.query_row( - "SELECT COUNT(*) FROM sqlite_master \ - WHERE type = 'trigger' AND (\ - name LIKE '%_blocked_by_runtime_removal' OR \ - name LIKE 'runtime_binding_%_blocked_by_removal'\ - )", - [], - |row| row.get(0), - )?; - let foreign_key_failures: i64 = - conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { - row.get(0) - })?; - assert_eq!(table_count, 1); - assert_eq!(trigger_count, 18); - assert_eq!(foreign_key_failures, 0); - Ok(()) - }) - .unwrap(); + let conn = Connection::open(&path)?; + configure_sqlite(&conn)?; + migrate_runtime_removal_operations_v59_to_v60(&conn)?; + assert_eq!(current_schema_version(&conn)?, 60); + let table_count: i64 = conn.query_row( + "SELECT COUNT(*) FROM sqlite_master \ + WHERE type = 'table' AND name = 'runtime_removal_operations'", + [], + |row| row.get(0), + )?; + let trigger_count: i64 = conn.query_row( + "SELECT COUNT(*) FROM sqlite_master \ + WHERE type = 'trigger' AND (\ + name LIKE '%_blocked_by_runtime_removal' OR \ + name LIKE 'runtime_binding_%_blocked_by_removal'\ + )", + [], + |row| row.get(0), + )?; + let foreign_key_failures: i64 = + conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + })?; + assert_eq!(table_count, 1); + assert_eq!(trigger_count, 18); + assert_eq!(foreign_key_failures, 0); + Ok(()) } #[test] @@ -16931,12 +16934,10 @@ mod tests { let path = temp.path().join("server.db"); prepare_retained_schema(&path, 60); - drop(SqliteWorkspaceStore::open_for_test(&path).unwrap()); let conn = Connection::open(&path).unwrap(); - assert_eq!( - current_schema_version(&conn).unwrap(), - LATEST_SCHEMA_VERSION - ); + configure_sqlite(&conn).unwrap(); + migrate_worker_run_generation_v60_to_v61(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 61); let columns = conn .prepare("PRAGMA table_info(worker_removal_operations)") .unwrap() @@ -17506,34 +17507,100 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx include!("migration_entrypoint_tests.rs"); include!("legacy_evidence_column_tests.rs"); - fn prepare_retained_schema(path: &Path, version: i64) { - prepare_schema_v52(path); - let conn = Connection::open(path).unwrap(); + #[test] + fn retained_fixture_copies_isolate_seed_rows_and_keep_connection_constraints() { + let temp = tempfile::tempdir().unwrap(); + let first = temp.path().join("first.db"); + prepare_retained_schema(&first, 59); + let conn = Connection::open(first).unwrap(); + conn.execute("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) VALUES('owner','user','owner','Owner','1','1')", []).unwrap(); + let second = temp.path().join("second.db"); + prepare_retained_schema(&second, 59); + let conn = Connection::open(second).unwrap(); configure_sqlite(&conn).unwrap(); - for migration in MIGRATIONS - .iter() - .filter(|migration| migration.version > 52 && migration.version <= version) - { - (migration.apply)(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 59); + verify_schema_history(&conn, 59).unwrap(); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM accounts", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + assert_eq!( + conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0)) + .unwrap(), + "wal" + ); + assert!(conn.execute("INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('invalid','missing','Invalid','active','1','1')", []).is_err()); + } + + fn prepare_retained_schema(path: &Path, version: i64) { + // Cache only empty preconditions, built from frozen DDL and real retained + // migrations. The test's seeded rows and migration under test are never shared. + static TEMPLATES: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + assert!((52..=LATEST_SCHEMA_VERSION).contains(&version)); + let mut templates = TEMPLATES + .get_or_init(|| Mutex::new(Default::default())) + .lock() + .unwrap(); + if !templates.contains_key(&version) { + let mut source = Connection::open_in_memory().unwrap(); + let baseline = if let Some((&baseline, prior)) = templates.range(..=version).next_back() + { + Backup::new(prior, &mut source) + .unwrap() + .run_to_completion(128, Duration::from_millis(1), None) + .unwrap(); + configure_sqlite(&source).unwrap(); + baseline + } else { + build_schema_v52(&source); + 52 + }; + for migration in MIGRATIONS + .iter() + .filter(|migration| migration.version > baseline && migration.version <= version) + { + (migration.apply)(&source).unwrap(); + } + assert_eq!(current_schema_version(&source).unwrap(), version); + templates.insert(version, source); } + let mut conn = Connection::open(path).unwrap(); + Backup::new(&templates[&version], &mut conn) + .unwrap() + .run_to_completion(128, Duration::from_millis(1), None) + .unwrap(); + configure_sqlite(&conn).unwrap(); } fn prepare_schema_v52(path: &Path) { - let conn = Connection::open(path).unwrap(); - configure_sqlite(&conn).unwrap(); + prepare_retained_schema(path, 52); + } + + fn build_schema_v52(conn: &Connection) { + configure_sqlite(conn).unwrap(); + // Batch fixture DDL only, never the production migration under test. + // Commit before migration calls and foreign_keys changes below. + let fixture = conn.unchecked_transaction().unwrap(); // Retained fixtures start from genuine pre-tracking Ticket DDL; current // component 11 would introduce assignment objects before frozen host 85. conn.execute_batch(include_str!("frozen_ticket_schema_v10.sql")) .unwrap(); conn.execute_batch("CREATE TABLE ticket_schema_migrations(version INTEGER PRIMARY KEY,name TEXT NOT NULL,applied_at TEXT NOT NULL); INSERT INTO ticket_schema_migrations VALUES(10,'ticket schema baseline','fixture');").unwrap(); + fixture.commit().unwrap(); merge_request::migrate(&conn).unwrap(); conn.execute_batch("PRAGMA foreign_keys = OFF; PRAGMA legacy_alter_table = ON;") .unwrap(); + let fixture = conn.unchecked_transaction().unwrap(); create_frozen_schema_v85(&conn).unwrap(); downgrade_assignment_work_schema(&conn); downgrade_schema_66_ticket_and_workdir_authority(&conn).unwrap(); + fixture.commit().unwrap(); conn.execute_batch("PRAGMA legacy_alter_table = OFF; PRAGMA foreign_keys = ON;") .unwrap(); + let fixture = conn.unchecked_transaction().unwrap(); conn.execute_batch( r#" DROP TABLE workdir_create_credential_revision_retentions; @@ -17598,12 +17665,14 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx "#, ) .unwrap(); + fixture.commit().unwrap(); } fn prepare_schema_v50(path: &Path, workspace_id: Option<&str>) { prepare_schema_v52(path); let conn = Connection::open(path).unwrap(); configure_sqlite(&conn).unwrap(); + let fixture = conn.unchecked_transaction().unwrap(); conn.execute_batch( r#" DROP TABLE workspace_runtime_binding_audit; @@ -17654,6 +17723,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx [], ) .unwrap(); + fixture.commit().unwrap(); } #[test] @@ -18487,10 +18557,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx [], ) .unwrap(); - drop(conn); - - SqliteWorkspaceStore::migrate_database(&path).unwrap(); - let conn = Connection::open(&path).unwrap(); + migrate_workspace_runtime_bindings_v50_to_v51(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 51); let binding_workspaces = { let mut stmt = conn .prepare( @@ -18525,7 +18593,9 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx let path = temp.path().join("server.db"); prepare_schema_v50(&path, None); - let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); + let conn = Connection::open(&path).unwrap(); + configure_sqlite(&conn).unwrap(); + let plan = workspace_schema_migration_plan(&conn).unwrap(); assert_eq!(plan.current_schema_version, 50); assert_eq!(plan.target_schema_version, LATEST_SCHEMA_VERSION); @@ -18535,12 +18605,10 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx assert!(!table_exists(&unchanged, "workspace_runtime_bindings").unwrap()); drop(unchanged); - SqliteWorkspaceStore::migrate_database(&path).unwrap(); + migrate_workspace_runtime_bindings_v50_to_v51(&conn).unwrap(); + drop(conn); let migrated = Connection::open(&path).unwrap(); - assert_eq!( - current_schema_version(&migrated).unwrap(), - LATEST_SCHEMA_VERSION - ); + assert_eq!(current_schema_version(&migrated).unwrap(), 51); assert_eq!( migrated .query_row( @@ -18564,7 +18632,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx } #[test] - fn schema_v52_migrates_deletion_operations_and_normalizes_incomplete_reservations() { + fn schema_v52_migrates_deletion_operations_and_normalizes_incomplete_reservations() -> Result<()> + { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); prepare_schema_v52(&path); @@ -18590,59 +18659,51 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx );", ) .unwrap(); - drop(conn); - - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); - store - .with_conn(|conn| { - assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION); - verify_workspace_deletion_schema(conn)?; - let reservation_state: String = conn.query_row( - "SELECT state FROM worker_create_reservations WHERE workspace_id='workspace-a' AND allocation_key='allocation'", - [], - |row| row.get(0), - )?; - assert_eq!(reservation_state, "created"); - let reservation_memory = conn.query_row( - "SELECT request_fingerprint, memory_language - FROM worker_create_reservations - WHERE workspace_id='workspace-a' AND allocation_key='allocation'", - [], - |row| { - Ok(( - row.get::<_, Option>(0)?, - row.get::<_, Option>(1)?, - )) - }, - )?; - assert_eq!(reservation_memory, (None, None)); - let complete_memory = conn.query_row( - "SELECT request_fingerprint, memory_language - FROM worker_create_reservations - WHERE workspace_id='workspace-a' AND allocation_key='allocation-complete'", - [], - |row| { - Ok(( - row.get::<_, Option>(0)?, - row.get::<_, Option>(1)?, - )) - }, - )?; - assert_eq!( - complete_memory, - ( - Some("request-complete".to_string()), - Some("ja".to_string()) - ) - ); - let violations: i64 = - conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { - row.get(0) - })?; - assert_eq!(violations, 0); - Ok(()) - }) - .unwrap(); + configure_sqlite(&conn)?; + migrate_workspace_deletion_v52_to_v53(&conn)?; + assert_eq!(current_schema_version(&conn)?, 53); + verify_workspace_deletion_schema(&conn)?; + let reservation_state: String = conn.query_row( + "SELECT state FROM worker_create_reservations WHERE workspace_id='workspace-a' AND allocation_key='allocation'", + [], + |row| row.get(0), + )?; + assert_eq!(reservation_state, "created"); + let reservation_memory = conn.query_row( + "SELECT request_fingerprint, memory_language + FROM worker_create_reservations + WHERE workspace_id='workspace-a' AND allocation_key='allocation'", + [], + |row| { + Ok(( + row.get::<_, Option>(0)?, + row.get::<_, Option>(1)?, + )) + }, + )?; + assert_eq!(reservation_memory, (None, None)); + let complete_memory = conn.query_row( + "SELECT request_fingerprint, memory_language + FROM worker_create_reservations + WHERE workspace_id='workspace-a' AND allocation_key='allocation-complete'", + [], + |row| { + Ok(( + row.get::<_, Option>(0)?, + row.get::<_, Option>(1)?, + )) + }, + )?; + assert_eq!( + complete_memory, + (Some("request-complete".to_string()), Some("ja".to_string())) + ); + let violations: i64 = + conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + })?; + assert_eq!(violations, 0); + Ok(()) } #[test] @@ -18718,7 +18779,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx fn runtime_binding_identity_and_trust_uniqueness_are_workspace_scoped() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -18870,7 +18931,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx fn workspace_runtime_connection_persists_without_authentication_evidence() { let temp = tempfile::tempdir().unwrap(); let path = temp.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -19169,7 +19230,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[test] fn runtime_binding_key_mutations_are_identity_guarded_idempotent_and_audited() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -19307,7 +19368,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[test] fn embedded_runtime_binding_can_explicitly_rotate_restart_identity() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store .with_conn(|conn| { conn.execute_batch( @@ -19419,7 +19480,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn workspace_schema_rejects_cross_workspace_ticket_repository_reference_at_write_time() { let dir = tempfile::tempdir().unwrap(); let database_path = dir.path().join("workspace.sqlite"); - let store = SqliteWorkspaceStore::open_for_test(&database_path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&database_path).unwrap(); for workspace_id in ["workspace-a", "workspace-b"] { store .upsert_workspace(&WorkspaceRecord { @@ -19513,7 +19574,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn workspace_display_name_update_is_timestamp_guarded_and_preserves_identity() { let dir = tempfile::tempdir().unwrap(); let database_path = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&database_path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&database_path).unwrap(); let record = WorkspaceRecord { workspace_id: "workspace-a".to_string(), owner_account_id: "owner-account".to_string(), @@ -19566,7 +19627,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[tokio::test] async fn objective_creation_allocates_and_resolves_workspace_resource_key() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(dir.path().join("server.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".into(), @@ -19612,7 +19673,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[tokio::test] async fn worker_create_reservation_allocates_uuid_before_runtime_and_replays_exact_input() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(dir.path().join("server.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -19916,7 +19977,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn worker_singleton_restart_retry_replays_exact_workdir_intent_and_rejects_changes() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&database).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -20020,7 +20081,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn worker_singleton_ownership_is_workspace_scoped_durable_and_fences_old_generations() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&database).unwrap(); for workspace_id in ["workspace-a", "workspace-b"] { store .upsert_workspace(&WorkspaceRecord { @@ -20146,7 +20207,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn concurrent_worker_singleton_reservations_have_one_winner() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&database).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -20202,7 +20263,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx #[tokio::test] async fn workspace_flow_sources_pin_content_and_builtins_stay_resources() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(dir.path().join("server.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -21144,7 +21205,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx // boundary under test. Targets need not be materialized by a Runtime. async fn manual_worker_start_store() -> (tempfile::TempDir, SqliteWorkspaceStore) { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(dir.path().join("workspace.db")).unwrap(); + let store = + SqliteWorkspaceStore::open_api_fixture(dir.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "w".into(), @@ -21879,7 +21941,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx use ticket::TicketBackend; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "w".into(), @@ -22007,7 +22069,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx async fn ticket_worker_assignment_replaces_current_and_preserves_audit_history() { let dir = tempfile::tempdir().unwrap(); let db = dir.path().join("server.db"); - let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&db).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -22322,7 +22384,7 @@ INSERT INTO worker_registry ( async fn role_assignment_routes_and_manual_start_are_state_fenced_atomically() { let dir = tempfile::tempdir().unwrap(); let db_path = dir.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open_for_test(&db_path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&db_path).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-role".to_string(), @@ -22732,7 +22794,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn workspace_bootstrap_validates_key_and_rejects_duplicate_workspace() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store .upsert_account(&AccountRecord { account_id: "owner-account".to_string(), @@ -22864,7 +22926,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn repository_records_round_trip() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -22945,7 +23007,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -23020,7 +23082,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn legacy_memory_reset_is_atomic_scoped_idempotent_and_exact() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); for (workspace_id, display_name) in [ ("local-dev", "Local Dev"), ("other-workspace", "Other Workspace"), @@ -23187,7 +23249,7 @@ INSERT INTO worker_registry ( async fn worker_workdir_registry_round_trips_and_preserves_pinned_retention() { let temp = tempfile::tempdir().unwrap(); let db = temp.path().join("workspace.db"); - let store = SqliteWorkspaceStore::open_for_test(&db).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&db).unwrap(); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: "owner-account".to_string(), @@ -23604,7 +23666,8 @@ INSERT INTO worker_registry ( #[tokio::test] async fn worker_projection_preserves_equal_local_ids_from_distinct_runtimes() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")).unwrap(); + let store = + SqliteWorkspaceStore::open_api_fixture(temp.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -23730,7 +23793,8 @@ INSERT INTO worker_registry ( #[tokio::test] async fn worker_projection_reconciles_snapshots_and_preserves_unavailable_lifecycle() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")).unwrap(); + let store = + SqliteWorkspaceStore::open_api_fixture(temp.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -23905,7 +23969,8 @@ INSERT INTO worker_registry ( #[tokio::test] async fn late_runtime_event_does_not_recreate_retention_deleted_catalog_worker() { let temp = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open_for_test(temp.path().join("workspace.db")).unwrap(); + let store = + SqliteWorkspaceStore::open_api_fixture(temp.path().join("workspace.db")).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -24000,7 +24065,7 @@ INSERT INTO worker_registry ( async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("control-grants.db"); - let store = SqliteWorkspaceStore::open_for_test(&database).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&database).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-control".to_string(), @@ -24128,7 +24193,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn account_and_login_records_round_trip() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); assert_eq!(store.schema_version().await.unwrap(), LATEST_SCHEMA_VERSION); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { @@ -24488,7 +24553,7 @@ INSERT INTO worker_registry ( async fn backend_job_resource_reserve_converges_across_connections_without_changing_input() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("resource-jobs.sqlite"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); seed_resource_job_workspace(&store).await; let other = SqliteWorkspaceStore::open_for_test(&path).unwrap(); let requests = [ @@ -24606,7 +24671,7 @@ INSERT INTO worker_registry ( async fn backend_job_resource_retry_and_unknown_are_fenced_by_durable_cleanup() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("cleanup-jobs.sqlite"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); seed_resource_job_workspace(&store).await; let request = resource_job_request("failed", "subject-a", "candidate-1"); let first = store @@ -24772,7 +24837,7 @@ INSERT INTO worker_registry ( async fn backend_job_resource_retry_races_new_intent_atomically_and_remains_bounded() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("retry-race.sqlite"); - let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); seed_resource_job_workspace(&store).await; // Generic resources use the same common reservation API without domain grants. let mut request = resource_job_request("generic", "subject-a", "candidate-1"); @@ -24881,7 +24946,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_job_all_older_attempt_cleanups_fence_retry_and_resource_release() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); seed_resource_job_workspace(&store).await; let mut request = resource_job_request("legacy-overlap", "subject-a", "candidate-1"); request.limits.max_attempts = 3; @@ -25005,7 +25070,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_job_empty_batch_surface_refresh_locks_subject_and_defers_candidates() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); seed_resource_job_workspace(&store).await; for surface_state in ["ungenerated", "stale", "failed"] { let subject = format!("subject-{surface_state}"); @@ -25125,7 +25190,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_job_completed_resource_releases_only_after_worker_cleanup() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); seed_resource_job_workspace(&store).await; let request = resource_job_request("completed", "subject-a", "candidate-1"); let first = store @@ -25179,7 +25244,7 @@ INSERT INTO worker_registry ( #[tokio::test] async fn backend_jobs_fence_results_project_ownership_and_bound_retries() { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); let workspace_id = "workspace-job"; store .upsert_account(&AccountRecord { diff --git a/crates/workspace-server/src/store_test_support.rs b/crates/workspace-server/src/store_test_support.rs new file mode 100644 index 000000000..430c2e09c --- /dev/null +++ b/crates/workspace-server/src/store_test_support.rs @@ -0,0 +1,190 @@ +//! Ordinary store/API fixtures copy a verified empty schema into independent real DBs. +//! Each case owns its connections and executors, with production SQLite settings. +//! Additional connections within fresh API setup reuse its verified schema. Migration, +//! fresh-creation, drift detection and persisted-case reopen tests use uncached constructors. + +use super::*; + +impl SqliteWorkspaceStore { + pub(crate) fn open_api_fixture(path: impl AsRef) -> Result { + let path = path.as_ref(); + if path.exists() { + // Restart tests must reopen their persisted database, not reset it. + return Self::open_for_test(path); + } + Self::from_fixture_connection(Connection::open(path)?) + } + + pub(crate) fn in_memory_fixture() -> Result { + Self::from_fixture_connection(Connection::open_in_memory()?) + } + + // Only the fresh API fixture may use this after copying its verified template. + // Open a distinct connection without replacing the DB or sharing a transaction. + pub(crate) fn connect_prepared_fixture(path: &Path) -> Result { + let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; + configure_sqlite(&conn)?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + db_executor: crate::db_executor::DbExecutor::new( + crate::db_executor::DbExecutorLimits::default(), + )?, + }) + } + + fn from_fixture_connection(mut conn: Connection) -> Result { + static TEMPLATE: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + { + // Only initialization/copy is serialized, never a test's execution. + let mut template = TEMPLATE + .get_or_init(|| Mutex::new(None)) + .lock() + .map_err(|_| Error::Store("API fixture template lock poisoned".into()))?; + if template.is_none() { + let source = Connection::open_in_memory()?; + prepare_connection(&source)?; + *template = Some(source); + } + Backup::new(template.as_ref().unwrap(), &mut conn)?.run_to_completion( + 128, + Duration::from_millis(1), + None, + )?; + } + // Backup copies the already verified schema, not connection-local PRAGMAs. + // Existing DBs above still pass through full production startup checks. + configure_sqlite(&conn)?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + db_executor: crate::db_executor::DbExecutor::new( + crate::db_executor::DbExecutorLimits::default(), + )?, + }) + } +} + +#[test] +fn prepared_fixture_connections_observe_only_committed_writes_and_keep_foreign_keys() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("connections.db"); + assert!(SqliteWorkspaceStore::connect_prepared_fixture(&path).is_err()); + assert!(!path.exists()); + let writer = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); + let reader = SqliteWorkspaceStore::connect_prepared_fixture(&path).unwrap(); + writer.with_conn(|conn| { + let tx = conn.unchecked_transaction()?; + tx.execute("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) VALUES('owner','user','owner','Owner','1','1')", [])?; + reader.with_conn(|other| { + assert!(other.is_autocommit()); + assert_eq!(other.query_row("SELECT COUNT(*) FROM accounts", [], |row| row.get::<_, i64>(0))?, 0); + Ok(()) + })?; + tx.commit()?; + Ok(()) + }).unwrap(); + reader.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT COUNT(*) FROM accounts", [], |row| row.get::<_, i64>(0))?, 1); + assert_eq!(conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))?, "wal"); + assert!(conn.execute("INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('invalid','missing','Invalid','active','1','1')", []).is_err()); + Ok(()) + }).unwrap(); +} + +#[test] +fn memory_fixture_copies_fresh_schema_and_isolates_data_and_ddl() { + fn schema(store: &SqliteWorkspaceStore) -> Vec<(String, String, Option)> { + store + .with_conn(|conn| { + let mut statement = + conn.prepare("SELECT type, name, sql FROM sqlite_schema ORDER BY type, name")?; + let rows = statement + .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))? + .collect::>>()?; + Ok(rows) + }) + .unwrap() + } + let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let first = SqliteWorkspaceStore::in_memory_fixture().unwrap(); + assert_eq!(schema(&first), schema(&fresh)); + first.with_conn(|conn| { + assert_eq!(conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?, 1); + conn.execute_batch("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) VALUES('owner','user','owner','Owner','1','1'); DROP TRIGGER ticket_assignment_ticket_parent_tombstone;")?; + Ok(()) + }).unwrap(); + let second = SqliteWorkspaceStore::in_memory_fixture().unwrap(); + assert_eq!(schema(&second), schema(&fresh)); + second.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT COUNT(*) FROM accounts", [], |row| row.get::<_, i64>(0))?, 0); + assert!(conn.execute("INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('invalid','missing','Invalid','active','1','1')", []).is_err()); + Ok(()) + }).unwrap(); +} + +#[test] +fn api_fixture_reopen_rejects_schema_drift_instead_of_replacing_the_database() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("drift.db"); + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); + store + .with_conn(|conn| { + conn.execute_batch("DROP TRIGGER ticket_assignment_ticket_parent_tombstone;")?; + Ok(()) + }) + .unwrap(); + drop(store); + let error = SqliteWorkspaceStore::open_api_fixture(&path) + .err() + .expect("reopen must verify existing schema"); + assert!( + error + .to_string() + .contains("ticket_assignment_ticket_parent_tombstone") + ); + let conn = Connection::open(path).unwrap(); + assert_eq!(conn.query_row("SELECT COUNT(*) FROM sqlite_schema WHERE name='ticket_assignment_ticket_parent_tombstone'", [], |row| row.get::<_, i64>(0)).unwrap(), 0); +} + +#[test] +fn api_fixture_databases_isolate_writes_and_reopen_persisted_rows() { + let temp = tempfile::tempdir().unwrap(); + std::thread::scope(|scope| { + let handles: Vec<_> = ["first", "second"] + .into_iter() + .map(|name| { + let path = temp.path().join(format!("{name}.db")); + scope.spawn(move || { + let store = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); + store.with_conn(|conn| { + assert_eq!(conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?, 1); + assert_eq!(conn.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))?, "wal"); + conn.execute("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) VALUES('owner','user','owner',?1,'1','1')", [name])?; + assert!(conn.execute("INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('invalid','missing','Invalid','active','1','1')", []).is_err()); + Ok(()) + }).unwrap(); + drop(store); + let reopened = SqliteWorkspaceStore::open_api_fixture(&path).unwrap(); + reopened.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT display_name FROM accounts WHERE account_id='owner'", [], |row| row.get::<_, String>(0))?, name); + Ok(()) + }).unwrap(); + }) + }) + .collect(); + for handle in handles { + handle.join().unwrap(); + } + }); + let pristine = SqliteWorkspaceStore::open_api_fixture(temp.path().join("pristine.db")).unwrap(); + pristine + .with_conn(|conn| { + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM accounts", [], |row| row + .get::<_, i64>(0))?, + 0 + ); + Ok(()) + }) + .unwrap(); +} diff --git a/crates/workspace-server/src/ticket_item_checker.rs b/crates/workspace-server/src/ticket_item_checker.rs index 8fcd21117..254312e00 100644 --- a/crates/workspace-server/src/ticket_item_checker.rs +++ b/crates/workspace-server/src/ticket_item_checker.rs @@ -286,6 +286,94 @@ pub(crate) fn render_result(acceptance: &BackendJobResultAcceptance) -> Result Value { + let input = parse_input(&request.input).unwrap(); + assert!(request.instruction.contains("writer_scope")); + assert!(request.instruction.contains("internal_inconsistency")); + assert!(request.instruction.contains("ambiguous_boundary")); + assert!( + request + .instruction + .contains("Production deployment requires approval") + ); + assert!(request.instruction.contains("background, quotations")); + if input + .body + .contains("This time I only create the Ticket; I will not implement it.") + && !input.body.starts_with("Background bug example:") + { + serde_json::json!({"findings": [{ + "category": "writer_scope", + "quote": "I will not implement it", + "reason": "The active requirement records the writer's own implementation boundary.", + "suggestion": "Remove the writer-specific action boundary." + }]}) + } else if input.body.contains("The rollout region is undecided.") + && input.body.contains("The rollout region is us-east-1.") + { + serde_json::json!({"findings": [{ + "category": "internal_inconsistency", + "quote": "The rollout region is undecided.", + "reason": "The same active text also settles the rollout region as us-east-1.", + "suggestion": "Choose either the unresolved or settled rollout-region statement." + }]}) + } else if input.body == "Implementation stops after Ticket creation." { + serde_json::json!({"findings": [{ + "category": "ambiguous_boundary", + "quote": "Implementation stops after Ticket creation.", + "reason": "The text alone does not establish whether this is a delivery constraint or the writer's current action boundary.", + "suggestion": "Confirm whether this limits the product workflow or only the current writer." + }]}) + } else { + serde_json::json!({"findings": []}) + } + } + } + + pub(crate) struct Case { + pub(crate) name: &'static str, + pub(crate) body: &'static str, + pub(crate) advisory_quote: Option<&'static str>, + } + pub(crate) fn cases() -> Vec { + vec![ + Case { + name: "writer-scope-positive", + body: "This time I only create the Ticket; I will not implement it.", + advisory_quote: Some("I will not implement it"), + }, + Case { + name: "production-approval-negative", + body: "Production deployment requires approval.", + advisory_quote: None, + }, + Case { + name: "quoted-background-negative", + body: "Background bug example: a generated Ticket once said “I will not implement it.” Do not reproduce that sentence in generated requirements.", + advisory_quote: None, + }, + Case { + name: "unresolved-versus-settled-positive", + body: "The rollout region is undecided. The rollout region is us-east-1.", + advisory_quote: Some("The rollout region is undecided."), + }, + Case { + name: "ambiguous-boundary-confirmation", + body: "Implementation stops after Ticket creation.", + advisory_quote: Some("Implementation stops after Ticket creation."), + }, + ] + } +} + #[cfg(test)] mod tests { use super::*; @@ -301,16 +389,144 @@ mod tests { }) } + fn ticket_fixture() -> ticket::Ticket { + ticket::Ticket { + meta: ticket::TicketMeta { + id: "ticket-internal".into(), + resource_key: Some("T-42".into()), + slug: "check-this-content".into(), + title: "Check this content".into(), + status: ticket::ExtensibleTicketStatus::Open, + kind: "task".into(), + priority: "P2".into(), + labels: vec![], + created_at: None, + created_by: None, + updated_at: None, + assignee: None, + readiness: None, + risk_flags: vec![], + workflow_state: ticket::TicketState::Open, + workflow_state_explicit: true, + queued_by: None, + queued_at: None, + targets: vec![], + raw: Default::default(), + }, + document: ticket::TicketDocument { + body: ticket::MarkdownText::new("Check the saved content."), + raw_frontmatter: Default::default(), + }, + tags: vec![], + events: vec![], + artifacts: vec![], + relations: Default::default(), + resolution: None, + closure_operation_key: None, + current_closure: None, + case_history: vec![], + legacy_workflow_snapshot: None, + } + } + + fn completed_acceptance( + request: BackendJobRequest, + result: Value, + ) -> BackendJobResultAcceptance { + use crate::backend_job::{ + BackendJobAttemptRecord, BackendJobAttemptState, BackendJobRecord, BackendJobState, + }; + let digest = crate::backend_job::result_digest(&result, request.limits.max_result_bytes) + .unwrap() + .0; + let attempt = BackendJobAttemptRecord { + workspace_id: "checker-unit".into(), + job_id: request.job_id.clone(), + attempt_id: "checker-attempt".into(), + attempt: 1, + input_digest: request.input_digest().unwrap(), + state: BackendJobAttemptState::Completed, + worker: Some(RuntimeWorkerRef::new("runtime", "checker")), + runtime_run_id: Some("run".into()), + dispatched_at: Some("1".into()), + deadline_at: "3".into(), + result: Some(result.clone()), + result_digest: Some(digest.clone()), + failure_category: None, + failure_detail: None, + worker_cleanup_state: None, + worker_cleanup_failure_category: None, + worker_cleanup_failure_detail: None, + worker_cleanup_updated_at: None, + worker_cleanup_completed_at: None, + created_at: "1".into(), + updated_at: "2".into(), + completed_at: Some("2".into()), + }; + BackendJobResultAcceptance { + job: BackendJobRecord { + workspace_id: "checker-unit".into(), + intent_fingerprint: request.fingerprint().unwrap(), + request, + state: BackendJobState::Completed, + current_attempt: 1, + result: Some(result), + result_digest: Some(digest), + failure_category: None, + failure_detail: None, + created_at: "1".into(), + updated_at: "2".into(), + completed_at: Some("2".into()), + }, + attempt, + replayed: false, + } + } + + #[test] + fn supplied_model_classifications_validate_and_render_advisories_or_no_findings() { + let instruction = worker::PromptCatalog::builtins_only() + .unwrap() + .ticket_item_checker() + .unwrap(); + for case in test_support::cases() { + let mut ticket = ticket_fixture(); + ticket.meta.title = format!("Boundary case: {}", case.name); + ticket.document.body = ticket::MarkdownText::new(case.body); + let request = request( + &ticket, + TicketItemCheckEdit::Create, + RuntimeWorkerRef::new("runtime", "source"), + instruction.clone(), + ) + .unwrap(); + let result = test_support::FakeTicketCheckerModel::respond(&request); + let checked = validate_result(&request.input, &result).unwrap(); + let acceptance = completed_acceptance(request, result); + match ( + case.advisory_quote, + result_facts(&acceptance, &ticket).unwrap(), + ) { + (Some(quote), TicketItemCheckFacts::Findings(advisory)) => { + assert_eq!(checked.findings.len(), 1, "{}", case.name); + assert!( + advisory.contains("[Ticket item checker advisory]"), + "{}", + case.name + ); + assert!(advisory.contains(quote), "{}", case.name); + } + (None, TicketItemCheckFacts::NoFindings) => { + assert!(checked.findings.is_empty(), "{}", case.name); + } + (_, actual) => panic!("{}: unexpected classification {actual:?}", case.name), + } + } + } + #[test] fn checker_job_identity_reuses_exact_intent_but_separates_edits_and_sources() { - use ticket::TicketBackend; - - let dir = tempfile::tempdir().unwrap(); - let backend = - ticket::SqliteTicketBackend::open(dir.path().join("tickets.db"), "space").unwrap(); - let (_, ticket) = backend - .create_with_snapshot(ticket::NewTicket::new("Check this content")) - .unwrap(); + let ticket = ticket_fixture(); let source = RuntimeWorkerRef { runtime_id: "runtime".into(), worker_id: "source".into(), diff --git a/crates/workspace-server/src/workdir_removal.rs b/crates/workspace-server/src/workdir_removal.rs index 519541ef2..46378b3a7 100644 --- a/crates/workspace-server/src/workdir_removal.rs +++ b/crates/workspace-server/src/workdir_removal.rs @@ -939,7 +939,7 @@ mod tests { } async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) { - let store = SqliteWorkspaceStore::in_memory_for_test().unwrap(); + let store = SqliteWorkspaceStore::in_memory_fixture().unwrap(); store .upsert_account(&AccountRecord { account_id: "account-a".to_string(), diff --git a/crates/workspace-server/src/workspace_catalog.rs b/crates/workspace-server/src/workspace_catalog.rs index 49f789038..07b1059e0 100644 --- a/crates/workspace-server/src/workspace_catalog.rs +++ b/crates/workspace-server/src/workspace_catalog.rs @@ -302,7 +302,7 @@ mod tests { #[tokio::test] async fn create_is_atomic_and_exact_retries_converge() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let service = WorkspaceCatalogService::new( store.clone(), Arc::new( @@ -356,7 +356,7 @@ mod tests { #[test] fn create_rejects_noncanonical_repository_keys_without_persisting_a_workspace() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let service = WorkspaceCatalogService::new( store.clone(), Arc::new( @@ -517,7 +517,7 @@ mod tests { #[tokio::test] async fn idempotency_key_reuse_with_different_payload_is_rejected() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let owner_account_id = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store, @@ -549,7 +549,7 @@ mod tests { #[test] fn repository_free_create_initializes_authority_and_concurrent_retries_converge() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let owner = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store.clone(), @@ -628,7 +628,7 @@ mod tests { #[test] fn replay_reads_current_repository_at_initial_key_and_never_recreates_it() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let owner = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store.clone(), @@ -756,7 +756,7 @@ mod tests { #[test] fn create_rejects_non_user_account_owners() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); store .upsert_account(&AccountRecord { account_id: "organization-owner".to_string(), @@ -794,7 +794,7 @@ mod tests { #[test] fn catalog_list_is_scoped_to_the_required_owner_account() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let owner_a = owner_account(store.as_ref()); let owner_b = "account-owner-b".to_string(); store @@ -862,7 +862,7 @@ mod tests { #[test] fn remote_repository_creation_persists_typed_source_without_auth_metadata() { - let store = Arc::new(SqliteWorkspaceStore::in_memory_for_test().unwrap()); + let store = Arc::new(SqliteWorkspaceStore::in_memory_fixture().unwrap()); let owner_account_id = owner_account(store.as_ref()); let service = WorkspaceCatalogService::new( store.clone(), diff --git a/crates/workspace-server/src/workspace_subscription.rs b/crates/workspace-server/src/workspace_subscription.rs index 1994c627f..24728385f 100644 --- a/crates/workspace-server/src/workspace_subscription.rs +++ b/crates/workspace-server/src/workspace_subscription.rs @@ -747,7 +747,7 @@ mod tests { WorkspaceRecord, }; let store = std::sync::Arc::new( - SqliteWorkspaceStore::in_memory_for_test() + SqliteWorkspaceStore::in_memory_fixture() .unwrap() .with_db_executor(crate::db_executor::DbExecutor::new(Default::default()).unwrap()), );