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.
This commit is contained in:
@@ -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<WipOperationOutput, WipOperationError> {
|
||||||
|
// 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]
|
#[async_trait]
|
||||||
impl WipOperationHandler for FileHandler {
|
impl WipOperationHandler for FileHandler {
|
||||||
@@ -628,58 +698,7 @@ impl WipOperationHandler for FileHandler {
|
|||||||
}
|
}
|
||||||
})?
|
})?
|
||||||
.map_err(map_tool_error)?;
|
.map_err(map_tool_error)?;
|
||||||
// Coordinates are not observations. Never prefetch unselected results;
|
self.publish_result(tool, result)
|
||||||
// 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))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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);
|
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<AtomicUsize>,
|
||||||
|
}
|
||||||
|
#[async_trait]
|
||||||
|
impl WipOperationHandler for DetachBeforePublication {
|
||||||
|
async fn call(
|
||||||
|
&self,
|
||||||
|
operation: &str,
|
||||||
|
arguments: &BTreeMap<String, Value>,
|
||||||
|
context: WipCallContext,
|
||||||
|
) -> Result<WipOperationOutput, WipOperationError> {
|
||||||
|
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));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -188,3 +188,25 @@ package/version set is unchanged.
|
|||||||
|
|
||||||
No full workspace test, full Nix/image build, deployment, target integration or
|
No full workspace test, full Nix/image build, deployment, target integration or
|
||||||
live dogfood update is claimed. Target movement remains Orchestrator authority.
|
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.
|
||||||
|
|||||||
Reference in New Issue
Block a user