From ec6c5beda090eedda4d7d3fd87a65797ee054b39 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 00:34:50 +0900 Subject: [PATCH] test: connect Drive publication to client and resource lifetimes --- .../tests/ticket_worker_launch_tests.rs | 56 ++- .../src/server/tests/worker_drive_tests.rs | 422 +++++++++++++++++- 2 files changed, 461 insertions(+), 17 deletions(-) diff --git a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs index ee8dba11d..304e18160 100644 --- a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs +++ b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs @@ -132,6 +132,16 @@ impl WorkerExecutionBackend for LaunchExecutionBackend { self.execution.create_working_directory(request) } + fn cleanup_working_directory( + &self, + id: &str, + ) -> std::result::Result< + worker_runtime::catalog::WorkingDirectoryStatus, + worker_runtime::working_directory::WorkingDirectoryDiagnostic, + > { + self.execution.cleanup_working_directory(id) + } + fn spawn_worker(&self, mut request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { self.creates.lock().unwrap().push(request.request.clone()); for claim in &request.request.workdir_attachments { @@ -164,9 +174,9 @@ impl WorkerExecutionBackend for LaunchExecutionBackend { } } -struct LaunchFixture { +pub(super) struct LaunchFixture { _temp: tempfile::TempDir, - api: WorkspaceApi, + pub(super) api: WorkspaceApi, backend: Arc, runtime_id: String, runtime_server: Option>, @@ -182,11 +192,51 @@ impl Drop for LaunchFixture { } impl LaunchFixture { + // Reuse the authenticated Runtime HTTP harness for Drive/source lifetime + // assertions without duplicating its trust/bootstrap/listener lifecycle. + pub(super) async fn with_source_workdir() -> ( + Self, + worker_runtime::working_directory::WorkingDirectoryBinding, + ) { + let fixture = Self::new(true).await; + init_clean_git_workspace(fixture._temp.path()); + let (_, Json(created)) = create_workspace_working_directory( + &fixture.api, + TEST_WORKSPACE_ID, + Some(&fixture.runtime_id), + BrowserWorkingDirectoryCreateRequest { + runtime_id: Some(fixture.runtime_id.clone()), + display_name: None, + repository_key: "test-repository".into(), + selector: Some("HEAD".into()), + operation_id: Some("drive-source-workdir".into()), + }, + ) + .await + .unwrap(); + let binding = fixture + .backend + .execution + .materializer + .bind_working_directory(&created.item.working_directory_id, None) + .unwrap(); + (fixture, binding) + } + async fn new(remote: bool) -> Self { let temp = tempfile::tempdir().unwrap(); let config = test_server_config(temp.path()); let store = SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap(); - let backend = Arc::new(LaunchExecutionBackend::default()); + let backend = Arc::new(LaunchExecutionBackend { + execution: DeterministicExecutionBackend { + materializer: worker_runtime::working_directory::RuntimeGitMaterializer::new( + temp.path() + .join(".test-embedded-runtime-store/source-workdirs"), + ), + ..Default::default() + }, + ..Default::default() + }); let api = WorkspaceApi::new_with_execution_backend(config, Arc::new(store), backend.clone()) .await 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 e951a6308..609cbd4ab 100644 --- a/crates/workspace-server/src/server/tests/worker_drive_tests.rs +++ b/crates/workspace-server/src/server/tests/worker_drive_tests.rs @@ -1,6 +1,10 @@ -//! Worker tools -> signed WorkspaceClient -> real isolated Server authority. -//! No Git checkout, live DB, network service, or DTO-only assertions. Workdir -//! import tests attach an explicitly read-scoped temporary source session. +//! Tools/native WIP -> signed WorkspaceClient -> isolated Server DB and LocalFileSystem. +//! Default fixtures have no Repository/Workdir and substitute Runtime execution. +//! Lifetime cases use real embedded Runtime stop/removal or the shared authenticated +//! Runtime HTTP harness for actual Git Workdir removal; controller execution remains +//! deterministic. Import sessions have explicit read scope (and real materializer +//! leases in the removal case). No external model, live data, or browser is used. +//! Run with the normal Server lib target and the `worker_drive_tests` filter. use super::*; use agen::tool::{Attachment, ToolError, ToolExecutionContext, ToolOutput}; use std::result::Result; @@ -148,28 +152,46 @@ struct Fixture { } impl Fixture { async fn new(access: [&str; 2]) -> Self { + Self::with_runtime(access, false).await + } + async fn with_runtime(access: [&str; 2], embedded: bool) -> Self { let temp = tempfile::tempdir().unwrap(); let mut config = test_server_config(temp.path()); config.repositories.clear(); let store = Arc::new(SqliteWorkspaceStore::open_for_test(&config.database_path).unwrap()); - let mut api = WorkspaceApi::new_with_execution_backend( - config, - store, - Arc::new(DeterministicExecutionBackend::default()), - ) - .await - .unwrap(); + let execution = Arc::new(DeterministicExecutionBackend { + materializer: worker_runtime::working_directory::RuntimeGitMaterializer::new( + temp.path().join("runtime-workdirs"), + ), + ..Default::default() + }); + let api = WorkspaceApi::new_with_execution_backend(config, store, execution) + .await + .unwrap(); + Self::on_api(api, temp, access, embedded).await + } + async fn on_api( + mut api: WorkspaceApi, + temp: tempfile::TempDir, + access: [&str; 2], + embedded: bool, + ) -> Self { let owner = seed_test_api_token(api.store.as_ref(), TEST_WORKSPACE_ID); - register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; - api.runtime - .register_or_replace(WorkdirlessFixtureRuntime::default()); + let runtime_id = if embedded { + EMBEDDED_RUNTIME_ID + } else { + register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; + api.runtime + .register_or_replace(WorkdirlessFixtureRuntime::default()); + WorkdirlessFixtureRuntime::RUNTIME_ID + }; let mut workers = Vec::new(); for name in ["Drive first", "Drive second"] { let Json(created) = create_workspace_worker( State(api.clone()), HeaderMap::new(), Json(CreateWorkspaceWorkerRequest { - runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.into(), + runtime_id: runtime_id.into(), display_name: name.into(), singleton_key: None, profile: Some("builtin:coder".into()), @@ -248,6 +270,45 @@ impl Fixture { .unwrap(), ) } + async fn upload_text(&self, name: &str, text: &str) -> Value { + use sha2::{Digest, Sha256}; + let root = self.root(0).await; + let digest: String = Sha256::digest(text.as_bytes()) + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + let query = url::form_urlencoded::Serializer::new(String::new()) + .append_pair("operation", "create") + .append_pair("request_id", &format!("user-upload-{name}")) + .append_pair("entry_workspace_id", TEST_WORKSPACE_ID) + .append_pair("parent_id", root["node_id"].as_str().unwrap()) + .append_pair("name", name) + .append_pair("content_type", "text/markdown") + .append_pair("size", &text.len().to_string()) + .append_pair("sha256", &digest) + .finish(); + let response = build_router(self.api.clone()) + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("{}/upload?{query}", base())) + .header("authorization", format!("Bearer {}", self.owner)) + .header(CONTENT_TYPE, "application/octet-stream") + .body(Body::from(text.to_owned())) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + serde_json::from_slice(&bytes).unwrap() + } fn importer(&self, worker: usize) -> DriveFeature { let root = self._temp.path().join("import-source"); fs::create_dir_all(root.join("public")).unwrap(); @@ -415,6 +476,339 @@ async fn workdirless_workers_share_markdown_through_current_read_and_write_grant ); } +#[tokio::test] +async fn user_uploaded_markdown_roundtrips_through_workdirless_tools_and_native_wip_at_one_identity() + { + let f = Fixture::new(["read_write", "read_only"]).await; + let original = "# ユーザー資料\n未処理\n"; + let updated_text = "# ユーザー資料\n処理済み ✓\n"; + for native in [false, true] { + let uploaded = f + .upload_text(if native { "native.md" } else { "tools.md" }, original) + .await; + let id = uploaded["entry"]["entry"].clone(); + let updated = if native { + let tools = native_tools(&f.features[0]); + // Resolve the production path projection, not an invented node path. + let metadata = value( + tool(&f.features[0], "DriveMetadata", json!({"entry":id})) + .await + .unwrap(), + ); + let path = metadata["path"].as_str().unwrap(); + let inspected = value( + native_tool(&tools, "Inspect", json!({"path":path,"refresh":true})) + .await + .unwrap(), + ); + let reference = inspected["interfaces"][0]["reference"].clone(); + let read = value( + native_tool( + &tools, + "Invoke", + json!({"path":path,"interface":reference,"operation":"read","arguments":{}}), + ) + .await + .unwrap(), + ); + assert_eq!(read["text"], original); + assert_eq!(read["entry"]["metadata"]["entry"], id); + assert_eq!( + read["entry"]["metadata"]["last_mutation_id"], + uploaded["entry"]["last_mutation_id"] + ); + value(native_tool(&tools, "Invoke", json!({"path":path,"interface":reference,"operation":"write","arguments":{"content":updated_text}})).await.unwrap()) + } else { + let read = value( + tool(&f.features[0], "DriveRead", json!({"entry":id})) + .await + .unwrap(), + ); + assert_eq!(read["text"], original); + assert_eq!(read["entry"]["metadata"]["entry"], id); + assert_eq!( + read["entry"]["metadata"]["last_mutation_id"], + uploaded["entry"]["last_mutation_id"] + ); + value( + tool( + &f.features[0], + "DriveWrite", + json!({"entry":id,"content":updated_text}), + ) + .await + .unwrap(), + ) + }; + assert_eq!(entry(&updated), id); + let mutation = &updated["entry"]["metadata"]["last_mutation_id"]; + assert_ne!(*mutation, uploaded["entry"]["last_mutation_id"]); + assert_eq!(*mutation, updated["request_id"]); + let independent = value( + tool(&f.features[1], "DriveRead", json!({"entry":id})) + .await + .unwrap(), + ); + assert_eq!(independent["text"], updated_text); + assert_eq!(independent["entry"], updated["entry"]); + let metadata = owner_call( + &f.api, + &f.owner, + "GET", + &format!( + "{}/metadata?entry_workspace_id={TEST_WORKSPACE_ID}&id={}", + base(), + id["node_id"].as_str().unwrap() + ), + None, + ) + .await; + assert_eq!(metadata, updated["entry"]["metadata"]); + let receipt = owner_call( + &f.api, + &f.owner, + "GET", + &format!( + "{}/requests/{}", + base(), + updated["request_id"].as_str().unwrap() + ), + None, + ) + .await; + assert_eq!(receipt["state"], "committed"); + assert_eq!(receipt["response"]["entry"], metadata); + let response = build_router(f.api.clone()).oneshot(Request::builder() + .uri(format!("{}/download?entry_workspace_id={TEST_WORKSPACE_ID}&id={}&expected_mutation_id={}", base(), id["node_id"].as_str().unwrap(), mutation.as_str().unwrap())) + .header("authorization", format!("Bearer {}", f.owner)) + .body(Body::empty()).unwrap()).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK, "native={native}"); + assert_eq!( + to_bytes(response.into_body(), 65536) + .await + .unwrap() + .as_ref(), + updated_text.as_bytes() + ); + } +} + +#[tokio::test] +async fn source_workdir_removal_keeps_published_drive_bytes_and_receipt() { + let (source, binding) = + super::ticket_worker_launch_tests::LaunchFixture::with_source_workdir().await; + let workdir_id = binding.working_directory.id.clone(); + let f = Fixture::on_api( + source.api.clone(), + tempfile::tempdir().unwrap(), + ["read_write", "read_only"], + false, + ) + .await; + let bytes = b"clean\n"; + let root_path = binding.root().to_owned(); + let materialization_root = binding.working_directory_root().to_owned(); + let scope = manifest::Scope::from_config(&manifest::ScopeConfig { + allow: vec![manifest::ScopeRule { + target: root_path.clone(), + permission: manifest::Permission::Read, + recursive: true, + symlink_policy: manifest::SymlinkPolicy::Resolved, + }], + deny: vec![], + }) + .unwrap(); + let sessions = Arc::new(WorkdirSessionRouter::default()); + sessions + .attach( + workdir::WorkdirAttachmentAlias::new("source").unwrap(), + Arc::new( + workdir::LocalWorkdirSession::materialized_bound_with_environment( + workdir::Workdir::new(workdir_id.clone()), + root_path.clone(), + root_path.clone(), + manifest::SharedScope::new(scope), + workdir::WorkdirSessionCapabilities::READ_ONLY, + binding.command_environment(), + binding.session_resources().unwrap(), + ), + ), + ) + .unwrap(); + let importer = DriveFeature::new(f.clients[0].clone(), sessions.clone()); + let root = f.root(0).await; + let saved = value(tool(&importer, "DriveSaveWorkdir", json!({"target_workdir":"source","path":"README.md","parent":root,"name":"retained.md","content_type":"text/markdown"})).await.unwrap()); + sessions.close_all().await.unwrap(); + drop(importer); + drop(sessions); + drop(binding); + assert!(root_path.join("README.md").exists()); + let api = source.api.clone(); + let removal_id = workdir_id.clone(); + let removed = tokio::task::spawn_blocking(move || { + execute_workdir_removal( + &api, + &removal_id, + "account:drive-test", + "source no longer needed", + ) + }) + .await + .unwrap() + .unwrap(); + assert_eq!( + removed.disposition, + server_api::WorkingDirectoryRemovalDisposition::Removed, + "{removed:?}" + ); + assert!(!removed.retryable); + assert!( + !materialization_root.exists(), + "actual source materialization must be gone" + ); + assert!( + f.api + .store + .get_workdir_registry(TEST_WORKSPACE_ID, &workdir_id) + .unwrap() + .is_none() + ); + let fresh = DriveFeature::new( + f.clients[1].clone(), + Arc::new(WorkdirSessionRouter::default()), + ); + let read = value( + tool(&fresh, "DriveRead", json!({"entry":entry(&saved)})) + .await + .unwrap(), + ); + assert_eq!(read["entry"], saved["entry"]); + assert_eq!(read["text"].as_str().unwrap().as_bytes(), bytes); + let receipt = value( + tool( + &fresh, + "DriveRequestStatus", + json!({"request_id":saved["request_id"]}), + ) + .await + .unwrap(), + ); + assert_eq!(receipt["state"], "committed"); + assert_eq!(receipt["response"]["entry"], saved["entry"]["metadata"]); +} + +#[tokio::test] +async fn worker_removal_preserves_drive_reference_for_explicitly_granted_replacement() { + // Real embedded Runtime lifecycle/retention; only controller execution is + // deterministic. No Workdir or external model is involved in this case. + let f = Fixture::with_runtime(["read_write", "read_only"], true).await; + let saved = f.create("retained.md", "# Beyond the Worker\n").await; + let id = entry(&saved); + let old = &f.clients[0].worker; + let worker = WorkspaceWorker::resolve_async(&f.api, &old.runtime_id, &old.worker_id) + .await + .unwrap(); + let stopped = worker + .stop( + &WorkerOperationContext::Backend, + WorkerLifecycleRequest { + reason: Some("Drive lifetime test".into()), + ticket_assignment: None, + }, + ) + .await + .unwrap(); + assert_eq!(stopped.state, InternalWorkerOperationState::Accepted); + assert_eq!(f.api.runtime.worker(old).unwrap().state, "stopped"); + seed_worker_session_for_cleanup(f._temp.path(), old); + let plan = build_runtime_cleanup_plan(&f.api, &old.runtime_id).unwrap(); + let candidate = plan + .workers + .iter() + .find(|c| c.runtime_worker_id == old.worker_id) + .unwrap(); + assert_eq!(candidate.blocking_reason, None); + worker + .delete_from_plan(&WorkerOperationContext::Backend, candidate) + .await + .unwrap(); + assert!(f.api.runtime.worker(old).is_err()); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, old) + .unwrap() + .is_none() + ); + denied(tool(&f.features[0], "DriveRead", json!({"entry":id})).await); + let Json(created) = create_workspace_worker( + State(f.api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_RUNTIME_ID.into(), + display_name: "Replacement reader".into(), + singleton_key: None, + profile: Some("builtin:coder".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: Default::default(), + control_operation_id: None, + }), + ) + .await + .unwrap(); + assert_ne!(created.worker_id, old.worker_id); + let client = Arc::new(RouterClient { + api: f.api.clone(), + identity: f.clients[0].identity.clone(), + worker: RuntimeWorkerRef::new(created.runtime_id, created.worker_id), + handle: tokio::runtime::Handle::current(), + lose_next_mutation_response: AtomicBool::new(false), + lose_next_upload_response: AtomicBool::new(false), + requests: Mutex::new(vec![]), + }); + let replacement = DriveFeature::new(client.clone(), Arc::new(WorkdirSessionRouter::default())); + denied(tool(&replacement, "DriveRead", json!({"entry":id})).await); + owner_call(&f.api, &f.owner, "POST", &format!("{}/grants", base()), Some(json!({"runtime_id":client.worker.runtime_id,"worker_id":client.worker.worker_id,"access":"read_only"}))).await; + let read = value( + tool(&replacement, "DriveRead", json!({"entry":id})) + .await + .unwrap(), + ); + assert_eq!(read["entry"], saved["entry"]); + assert_eq!(read["text"], "# Beyond the Worker\n"); + let receipt = owner_call( + &f.api, + &f.owner, + "GET", + &format!( + "{}/requests/{}", + base(), + saved["request_id"].as_str().unwrap() + ), + None, + ) + .await; + assert_eq!(receipt["state"], "committed"); + assert_eq!(receipt["response"]["entry"], saved["entry"]["metadata"]); + // Stop remaining deterministic controllers before dropping isolated resources. + for target in [&f.clients[1].worker, &client.worker] { + let stopped = f + .api + .runtime + .stop_worker( + target, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); + assert_eq!(stopped.state, InternalWorkerOperationState::Accepted); + } +} + #[tokio::test] async fn simultaneous_worker_edits_publish_exactly_one_last_mutation_id_and_preserve_the_loser_conflict() {