diff --git a/crates/worker/src/feature/builtin/drive/backend.rs b/crates/worker/src/feature/builtin/drive/backend.rs index cae780da9..24b4ec93e 100644 --- a/crates/worker/src/feature/builtin/drive/backend.rs +++ b/crates/worker/src/feature/builtin/drive/backend.rs @@ -172,7 +172,12 @@ impl DriveBackend { }, ) .await?; - self.validate_page(&result, limit.unwrap_or(100).min(DRIVE_PAGE_MAX_LIMIT))?; + self.validate_page( + &result, + limit + .unwrap_or(DRIVE_PAGE_MAX_LIMIT) + .min(DRIVE_PAGE_MAX_LIMIT), + )?; if result .entries .iter() @@ -186,7 +191,10 @@ impl DriveBackend { let result: DriveListResponse = self.get("/search", &query).await?; self.validate_page( &result, - query.limit.unwrap_or(100).min(DRIVE_SEARCH_MAX_LIMIT), + query + .limit + .unwrap_or(DRIVE_SEARCH_MAX_LIMIT) + .min(DRIVE_SEARCH_MAX_LIMIT), )?; Ok(result) } @@ -276,25 +284,37 @@ impl DriveBackend { id: &str, result: Result, ) -> Result { - let response = match result { + // A parseable success can still be an invalid completion. Classify it + // before reconciliation so it also gets one original-request inquiry. + let result = result.and_then(|response| { + self.validate_completion(id, &response) + .map_err(|_| DriveError::OutcomeUnknown(id.into()))?; + Ok(response) + }); + match result { Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await { Ok(DriveRequestStatusResponse { state: DriveRequestState::Committed, response: Some(response), .. - }) => response, - _ => return Err(DriveError::OutcomeUnknown(id.into())), + }) => Ok(response), // status validates both IDs and entry metadata + _ => Err(DriveError::OutcomeUnknown(id.into())), }, - other => other?, - }; + other => other, + } + } + fn validate_completion( + &self, + id: &str, + response: &DriveMutationResponse, + ) -> Result<(), DriveError> { if response.request_id != id { - return Err(DriveError::OutcomeUnknown(id.into())); + return Err(DriveError::Unavailable); } if let Some(entry) = &response.entry { - self.validate_entry(entry) - .map_err(|_| DriveError::OutcomeUnknown(id.into()))?; + self.validate_entry(entry)?; } - Ok(response) + Ok(()) } pub async fn mutate( &self, @@ -371,12 +391,7 @@ impl DriveBackend { return Err(DriveError::Unavailable); } if let Some(result) = &response.response { - if result.request_id != request_id { - return Err(DriveError::Unavailable); - } - if let Some(entry) = &result.entry { - self.validate_entry(entry)?; - } + self.validate_completion(request_id, result)?; } Ok(response) } diff --git a/crates/worker/src/feature/builtin/drive/tests.rs b/crates/worker/src/feature/builtin/drive/tests.rs index 6aacbed2d..8107139c6 100644 --- a/crates/worker/src/feature/builtin/drive/tests.rs +++ b/crates/worker/src/feature/builtin/drive/tests.rs @@ -294,6 +294,143 @@ async fn uncommitted_receipt_snapshot_does_not_turn_unknown_into_failure_or_retr assert_eq!(client.requests.lock().unwrap().len(), 2); } +fn invalid_completions(id: &str) -> Vec { + let mut invalid_entry = entry("2", "2"); + invalid_entry.revision = "not-a-revision".into(); + vec![ + DriveMutationResponse { + request_id: "another-request".into(), + entry: Some(entry("2", "2")), + }, + DriveMutationResponse { + request_id: id.into(), + entry: Some(invalid_entry), + }, + ] +} + +#[tokio::test] +async fn invalid_mutation_completion_recovers_from_original_receipt_without_resending() { + let context = ToolExecutionContext::new("write", "batch", 0); + let client = Client::new(vec![]); + let feature = feature(client.clone()); + let id = feature.request_id(&context); + for completion in invalid_completions(&id) { + let committed = DriveMutationResponse { + request_id: id.clone(), + entry: Some(entry("2", "2")), + }; + client.requests.lock().unwrap().clear(); + client.responses.lock().unwrap().extend([ + Ok(response(completion)), + Ok(response(DriveRequestStatusResponse { + request_id: id.clone(), + state: DriveRequestState::Committed, + response: Some(committed.clone()), + })), + ]); + feature.observe(entry("2", "1")).unwrap(); + let output = feature + .execute( + "DriveWrite", + &json!({"entry":entry("2", "1").entry,"content":"new"}).to_string(), + &context, + None, + ) + .await + .unwrap(); + assert_eq!(output.value["request_id"], id); + assert_eq!(output.value["entry"]["metadata"]["revision"], "2"); + let requests = client.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!( + requests[0].method, + crate::worker::WorkspaceRequestMethod::Post + ); + assert_eq!( + requests[1].method, + crate::worker::WorkspaceRequestMethod::Get + ); + assert!(requests[1].path.ends_with(&format!("/requests/{id}"))); + let body: Value = serde_json::from_str(requests[0].body.as_ref().unwrap()).unwrap(); + assert_eq!(body["request_id"], id); + assert_eq!(body["mutation"]["expected_revision"], "1"); + } +} + +#[tokio::test] +async fn invalid_completion_and_unresolved_or_invalid_receipt_stay_unknown_after_one_query() { + let context = ToolExecutionContext::new("write", "batch", 0); + let client = Client::new(vec![]); + let feature = feature(client.clone()); + let id = feature.request_id(&context); + let committed = DriveMutationResponse { + request_id: id.clone(), + entry: Some(entry("2", "2")), + }; + let mut receipts = vec![ + response(DriveRequestStatusResponse { + request_id: id.clone(), + state: DriveRequestState::Uncommitted, + response: None, + }), + response(DriveRequestStatusResponse { + request_id: id.clone(), + state: DriveRequestState::Committed, + response: None, + }), + response(DriveRequestStatusResponse { + request_id: "another-request".into(), + state: DriveRequestState::Committed, + response: Some(committed), + }), + WorkspaceResponse { + status: 200, + body: "invalid-json".into(), + }, + ]; + receipts.extend(invalid_completions(&id).into_iter().map(|completion| { + response(DriveRequestStatusResponse { + request_id: id.clone(), + state: DriveRequestState::Committed, + response: Some(completion), + }) + })); + for completion in invalid_completions(&id) { + for receipt in &receipts { + client.requests.lock().unwrap().clear(); + client + .responses + .lock() + .unwrap() + .extend([Ok(response(&completion)), Ok(receipt.clone())]); + feature.observe(entry("2", "1")).unwrap(); + let result = feature + .execute( + "DriveWrite", + &json!({"entry":entry("2", "1").entry,"content":"new"}).to_string(), + &context, + None, + ) + .await; + assert!( + matches!(result, Err(DriveError::OutcomeUnknown(ref observed)) if observed == &id) + ); + let requests = client.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!( + requests[0].method, + crate::worker::WorkspaceRequestMethod::Post + ); + assert_eq!( + requests[1].method, + crate::worker::WorkspaceRequestMethod::Get + ); + assert!(requests[1].path.ends_with(&format!("/requests/{id}"))); + } + } +} + #[tokio::test] async fn observation_eviction_never_substitutes_another_entry() { let feature = feature(Client::new(Vec::new())); 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 683f89d1f..4bcd6af78 100644 --- a/crates/workspace-server/src/server/tests/worker_drive_tests.rs +++ b/crates/workspace-server/src/server/tests/worker_drive_tests.rs @@ -726,6 +726,80 @@ 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()); + } + let native = native_tools(&f.features[1]); + let root = value( + native_tool(&native, "Inspect", json!({"path":"/drive","refresh":true})) + .await + .unwrap(), + ); + let reference = root["interfaces"][0]["reference"].clone(); + for use_native in [false, true] { + for (name, operation, mut args, maximum) in [ + ( + "DriveList", + "list", + json!({}), + server_api::DRIVE_PAGE_MAX_LIMIT as usize, + ), + ( + "DriveSearch", + "search", + json!({"query":"default-page-"}), + server_api::DRIVE_SEARCH_MAX_LIMIT as usize, + ), + ] { + let mut actual = Vec::new(); + for page_index in 0..2 { + let output = if use_native { + native_tool(&native, "Invoke", json!({"path":"/drive","interface":reference,"operation":operation,"arguments":args})).await + } else { + tool(&f.features[1], name, args.clone()).await + }; + let page = value(output.unwrap()); + let entries = page["entries"].as_array().unwrap(); + assert_eq!( + entries.len(), + if page_index == 0 { + maximum + } else { + 201 - maximum + }, + "native={use_native} operation={operation}" + ); + actual.extend(entries.iter().map(|e| { + e["metadata"]["entry"]["node_id"] + .as_str() + .unwrap() + .to_owned() + })); + if page_index == 0 { + assert!(page["next_after"].is_string()); + args["after"] = page["next_after"].clone(); + } else { + assert!(page["next_after"].is_null()); + } + } + assert_eq!(actual.len(), expected.len()); + assert_eq!( + actual + .into_iter() + .collect::>(), + expected + ); + } + } +} + #[tokio::test] async fn list_and_search_pages_are_bounded_disjoint_and_cursors_follow_backend_scope() { let f = Fixture::new(["read_write", "read_only"]).await; diff --git a/docs/development/worker-drive.md b/docs/development/worker-drive.md index b16f983f7..3099d73c2 100644 --- a/docs/development/worker-drive.md +++ b/docs/development/worker-drive.md @@ -63,7 +63,9 @@ authenticated latest URLs. Deleting/recreating the same name allocates a new ID. | `DriveSaveWorkdir` | explicit `target_workdir` alias, relative `path`, destination `parent`, `name`, optional `content_type` (octet-stream by default) | | `DriveRequestStatus` | `request_id` from an uncertain mutation | -List/search are paged discovery only, not implicit mutation observations. Read or +List/search are paged discovery only, not implicit mutation observations. Omitting +`limit` uses the Backend maximum (200 for list, 128 for search) in both Tools and +native WIP; use the returned `next_after` to continue. Read or metadata-inspect an entry before replacing, editing, moving or deleting it. The Tools adapter retains at most 256 observations, scoped to its injected Workspace. Eviction requires a new explicit observation; it never substitutes another entry. @@ -120,9 +122,11 @@ latest and retry. Observe explicitly, reconsider the edit and submit a deliberat new operation. Native Client stale handling likewise does not replay dispatch. Timeout, lost response, oversized/invalid mutation response or invalid completion -may mean **OutcomeUnknown**. The adapter queries the *same* request receipt once; -if committed it returns that confirmed DB result. Otherwise it preserves unknown -and the `request_id`. Query `DriveRequestStatus`; an `uncommitted` snapshot is not +may mean **OutcomeUnknown**, including a completion with a mismatched request ID +or invalid entry metadata. The adapter queries the *same* request receipt once; +only a committed receipt with matching IDs and valid metadata returns confirmed +DB success. Otherwise it preserves unknown and the `request_id`. Query +`DriveRequestStatus`; an `uncommitted` snapshot is not proof that an in-flight request cannot later commit. Do not retry blindly with a fresh request ID. Backend grants also protect receipt lookup/replay. Safe typed Denied/NotFound/Conflict/Invalid/Limit/StorageUnavailable/OutcomeUnknown errors are