From b8faa83c471ca7c94e8340830d41d42addccac3c Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 9 Oct 2026 15:30:21 +0900 Subject: [PATCH] fix: preserve unknown mutation outcomes after connection loss Address POST_COMMIT_CONNECTION_FENCE from independent review of af655ae6. Post-success publication must not claim pre-effect rejection if Write/Edit/Create may have committed before attachment loss. Keep read-only validator mismatch and lazy entry mapping unchanged. Add deterministic real checked provider/Tools commit -> detach -> production publication -> WIP Invoke/Client tests for all three mutations, asserting committed effects, OutcomeUnknown audit and exactly one dispatch. Narrow race9, Worker903+110, root check, dependent Server config19/proxy1, SDK resolution and fmt/diff checks passed. --- crates/worker/src/checkout.rs | 123 ++++++----- crates/worker/src/checkout_race_tests.rs | 197 ++++++++++++++++++ .../development/wip-integration-validation.md | 22 ++ 3 files changed, 290 insertions(+), 52 deletions(-) diff --git a/crates/worker/src/checkout.rs b/crates/worker/src/checkout.rs index 7035e782e..adba8a547 100644 --- a/crates/worker/src/checkout.rs +++ b/crates/worker/src/checkout.rs @@ -503,6 +503,76 @@ impl FileHandler { )) } } + /// Classify post-success publication failures separately from pre-effect + /// provider errors; result coordinates remain lazy for all operations. + fn publish_result( + &self, + tool: &str, + result: tools::CheckoutToolOutput, + ) -> Result { + // Coordinates are not observations. Never prefetch unselected results; + // a vanished entry cannot fail a successful discovery Operation. + self.check_connection().map_err(|error| { + if matches!(tool, "Write" | "Edit" | "Create") { + // The provider has already succeeded: connection loss while + // publishing cannot prove that a committed effect was rejected. + WipOperationError::OutcomeUnknown( + "completed file operation lost its attachment before result publication".into(), + ) + } else { + error + } + })?; + let mut fields = BTreeMap::from([("summary".into(), Value::String(result.output.summary))]); + if let Some(content) = result.output.content { + fields.insert("content".into(), Value::String(content)); + } + let items = if let Some(listing) = result.listing { + fields.insert("truncated".into(), Value::Boolean(listing.truncated)); + if let Some(after) = listing.next_after { + fields.insert("after".into(), cursor_value(&self.alias, &after)); + } + listing + .entries + .into_iter() + .map(|item| { + Value::Record(BTreeMap::from([ + ( + "entry".into(), + Value::String(object_path(&self.alias, &item.path)), + ), + ("kind".into(), Value::String(kind_name(item.kind).into())), + ( + "size".into(), + Value::Integer(i64::try_from(item.size).unwrap_or(i64::MAX)), + ), + ])) + }) + .collect() + } else { + result + .paths + .iter() + .map(|path| { + Value::Record(BTreeMap::from([( + "entry".into(), + Value::String(object_path(&self.alias, path)), + )])) + }) + .collect() + }; + fields.insert("items".into(), Value::List(items)); + let value = Value::Record(fields); + if let Some(raw) = result.validator { + Ok(WipOperationOutput::native_with_validator( + value, + self.provider + .wrap_validator(&self.alias, self.generation, &self.workdir, &raw), + )) + } else { + Ok(WipOperationOutput::native(value)) + } + } } #[async_trait] impl WipOperationHandler for FileHandler { @@ -628,58 +698,7 @@ impl WipOperationHandler for FileHandler { } })? .map_err(map_tool_error)?; - // Coordinates are not observations. Never prefetch unselected results; - // a vanished entry cannot fail a successful discovery Operation. - self.check_connection()?; - let mut fields = BTreeMap::from([("summary".into(), Value::String(result.output.summary))]); - if let Some(content) = result.output.content { - fields.insert("content".into(), Value::String(content)); - } - let items = if let Some(listing) = result.listing { - fields.insert("truncated".into(), Value::Boolean(listing.truncated)); - if let Some(after) = listing.next_after { - fields.insert("after".into(), cursor_value(&self.alias, &after)); - } - listing - .entries - .into_iter() - .map(|item| { - Value::Record(BTreeMap::from([ - ( - "entry".into(), - Value::String(object_path(&self.alias, &item.path)), - ), - ("kind".into(), Value::String(kind_name(item.kind).into())), - ( - "size".into(), - Value::Integer(i64::try_from(item.size).unwrap_or(i64::MAX)), - ), - ])) - }) - .collect() - } else { - result - .paths - .iter() - .map(|path| { - Value::Record(BTreeMap::from([( - "entry".into(), - Value::String(object_path(&self.alias, path)), - )])) - }) - .collect() - }; - fields.insert("items".into(), Value::List(items)); - let value = Value::Record(fields); - if let Some(raw) = result.validator { - Ok(WipOperationOutput::native_with_validator( - value, - self.provider - .wrap_validator(&self.alias, self.generation, &self.workdir, &raw), - )) - } else { - Ok(WipOperationOutput::native(value)) - } + self.publish_result(tool, result) } } diff --git a/crates/worker/src/checkout_race_tests.rs b/crates/worker/src/checkout_race_tests.rs index 15e1c6ade..da58fc64c 100644 --- a/crates/worker/src/checkout_race_tests.rs +++ b/crates/worker/src/checkout_race_tests.rs @@ -570,3 +570,200 @@ async fn checkout_content_children_never_call_provider_list_even_for_deep_paths( } assert_eq!(session.searches.load(Ordering::SeqCst), 0); } + +/// Deterministic adapter fixture for the production post-success publication +/// boundary. Execute through the real provider/Tools, then detach after the +/// routed operation guard has finished, before publishing the committed result. +/// No race timing or production-only test hook is needed. +struct DetachBeforePublication { + file: FileHandler, + calls: Arc, +} +#[async_trait] +impl WipOperationHandler for DetachBeforePublication { + async fn call( + &self, + operation: &str, + arguments: &BTreeMap, + context: WipCallContext, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + let tool = match operation { + "write" => "Write", + "edit" => "Edit", + "create_file" => "Create", + _ => panic!("mutation fixture only"), + }; + let result = tools::execute_checkout_tool( + self.file.provider.router.clone(), + self.file.provider.tracker.clone(), + &self.file.alias, + self.file.generation, + self.file.target.clone(), + self.file.observation.validator.clone(), + tool, + wip_to_json(&Value::Record(arguments.clone())).unwrap(), + context.execution, + ) + .await + .map_err(map_tool_error)?; + self.file + .provider + .router + .detach(&WorkdirAttachmentAlias::new(&self.file.alias).unwrap()) + .await + .unwrap(); + self.file.publish_result(tool, result) + } +} + +#[tokio::test] +async fn committed_checkout_mutation_losing_connection_is_unknown_not_rejected_or_replayed() { + use crate::wip::{WipAuditOutcome, WipRuntime}; + for (operation, arguments, target, result_path, expected) in [ + ( + "write", + json!({"content":"saved\n"}), + "a.txt", + "a.txt", + "saved\n", + ), + ( + "edit", + json!({"old_string":"old", "new_string":"changed"}), + "a.txt", + "a.txt", + "changed\n", + ), + ( + "create_file", + json!({"path":"new.txt", "content":"created\n"}), + "dir", + "dir/new.txt", + "created\n", + ), + ] { + let dir = TempDir::new().unwrap(); + std::fs::create_dir(dir.path().join("dir")).unwrap(); + std::fs::write(dir.path().join("a.txt"), "old\n").unwrap(); + let router = Arc::new(WorkdirSessionRouter::new()); + router + .attach( + WorkdirAttachmentAlias::new("main").unwrap(), + HookedSession::new(&dir), + ) + .unwrap(); + let p = provider(router.clone()); + if operation != "create_file" { + p.node("/checkouts/main/a.txt") + .await + .unwrap() + .unwrap() + .handler + .call("read", &BTreeMap::new(), context()) + .await + .unwrap_or_else(|_| panic!("prior Read failed")); + } + let selected = router.resolve(Some("main")).unwrap(); + let target = WorkdirPath::new(target).unwrap(); + let observation = selected + .session + .checkout_observe(target.clone()) + .await + .unwrap(); + let file = FileHandler { + provider: p.clone(), + alias: "main".into(), + generation: selected.generation, + workdir: selected.session.workdir().id().as_str().into(), + target, + observation, + }; + drop(selected); + let calls = Arc::new(AtomicUsize::new(0)); + let route = object_path("main", &file.target); + let mut projection = p.node(&route).await.unwrap().unwrap(); + let interface = projection.interface.clone(); + projection.handler = Arc::new(DetachBeforePublication { + file, + calls: calls.clone(), + }); + let mut registry = WipMountRegistry::new(); + registry + .allocate_namespace("checkout", "checkouts") + .unwrap(); + // Static observation metadata isolates the response boundary; real + // execution still uses checked provider authority and prior Read. + registry + .mount(p.node("/checkouts/main").await.unwrap().unwrap()) + .unwrap(); + registry.mount(projection).unwrap(); + let runtime = WipRuntime::from_mounts(registry, "post-commit-test".into()).unwrap(); + let error = runtime + .invoke( + route, + interface, + operation.into(), + arguments, + Default::default(), + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("outcome unknown"), + "{operation}: {error}" + ); + assert_eq!( + runtime.audit().last().unwrap().outcome, + WipAuditOutcome::OutcomeUnknown + ); + assert_eq!( + std::fs::read_to_string(dir.path().join(result_path)).unwrap(), + expected + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "completed mutation must never replay" + ); + assert_eq!(runtime.metrics().operation_round_trips, 1); + } +} + +#[tokio::test] +async fn readonly_checkout_result_losing_connection_keeps_validator_mismatch() { + let dir = TempDir::new().unwrap(); + let router = Arc::new(WorkdirSessionRouter::new()); + let alias = WorkdirAttachmentAlias::new("main").unwrap(); + router + .attach(alias.clone(), HookedSession::new(&dir)) + .unwrap(); + let selected = router.resolve(Some("main")).unwrap(); + let file = FileHandler { + provider: provider(router.clone()), + alias: "main".into(), + generation: selected.generation, + workdir: selected.session.workdir().id().as_str().into(), + target: WorkdirPath::root(), + observation: selected + .session + .checkout_observe(WorkdirPath::root()) + .await + .unwrap(), + }; + drop(selected); + router.detach(&alias).await.unwrap(); + for tool in ["Read", "List", "Glob", "Grep"] { + let result = tools::CheckoutToolOutput { + output: agen::tool::ToolOutput { + summary: "finished".into(), + content: None, + attachments: vec![], + }, + paths: vec![], + listing: None, + validator: None, + }; + assert_validator_mismatch(file.publish_result(tool, result)); + } +} diff --git a/docs/development/wip-integration-validation.md b/docs/development/wip-integration-validation.md index 5dfb7ce5e..7b55c86e6 100644 --- a/docs/development/wip-integration-validation.md +++ b/docs/development/wip-integration-validation.md @@ -188,3 +188,25 @@ package/version set is unchanged. No full workspace test, full Nix/image build, deployment, target integration or live dogfood update is claimed. Target movement remains Orchestrator authority. + +### Post-success connection-loss review regression + +The first independent source review identified `POST_COMMIT_CONNECTION_FENCE`: +lazy result publication had propagated a connection mismatch as a pre-effect +rejection even after successful Write/Edit/Create. The production result +publication boundary now maps post-success connection loss for mutations to +`OutcomeUnknown`, while Read/List/Glob/Grep keep their validator mismatch. Lazy +entry mapping is unchanged; there is still no result-wide acquisition. + +`committed_checkout_mutation_losing_connection_is_unknown_not_rejected_or_replayed` +executes real checked Tools/provider mutations, deterministically detaches after +the routed execution guard finishes but before the same production publication +boundary, then passes the error through WIP Invoke/Client. It checks committed +file contents, unknown audit (not rejection), and exactly one dispatch for each +Write/Edit/Create. `readonly_checkout_result_losing_connection_keeps_validator_mismatch` +protects the non-mutating error policy. No sleep or production test hook is used. + +After this fix, all **9** checkout race tests and the complete Worker suite +(**903 unit + 110 integration**) passed, as did root Cargo check, formatting, +diff checks and released SDK resolution verification. The earlier **901** Worker +count above describes the pre-review combined-crate run.