fix(client): preserve normalized request ownership

crates/wip-client/src/lib.rs

@@ -397,6 +397,7 @@ struct EntryRecord {
 #[derive(Debug, Clone)]
 struct ObjectRecord {
     object: Object,
+    revision: RequestId,
     touched: u64,
 }
 
@@ -675,7 +676,18 @@ impl Client {
         let id = self.allocate_request_id();
         let maximum = self.limits.objects_per_session;
         let state = self.session_mut(session)?;
+        ensure_entry_slot(state, &request.path, maximum)?;
         ensure_tree_slot(state, &request.path, maximum)?;
+        let entry = state
+            .entries
+            .entry(request.path.clone())
+            .or_insert(EntryRecord {
+                object_key: None,
+                revision: id,
+                state: ObservationState::Loading(id),
+            });
+        entry.revision = id;
+        entry.state = ObservationState::Loading(id);
         let tree = state
             .trees
             .entry(request.path.clone())
@@ -1068,11 +1080,12 @@ impl Client {
                 let maximum = self.limits.objects_per_session;
                 let mut candidate = self.session(session)?.clone();
                 if let Err(error) = apply_tree(&mut candidate, &request.path, tree, maximum, id) {
-                    self.session_mut(session)?
+                    set_tree_subject_state(
-                        .trees
+                        self.session_mut(session)?,
-                        .get_mut(&request.path)
+                        &request.path,
-                        .expect("loading tree")
+                        id,
-                        .state = ObservationState::Error(error.clone());
+                        ObservationState::Error(error.clone()),
+                    );
                     self.push_failure_diagnostic(session, id, &error)?;
                     return Err(error);
                 }
@@ -1085,11 +1098,7 @@ impl Client {
                 status_mismatch,
             }) => {
                 self.record_status_mismatch(session, id, status_mismatch)?;
-                self.session_mut(session)?
+                let observation_state = if error.code == ProtocolErrorCode::NotFound {
-                    .trees
-                    .get_mut(&request.path)
-                    .expect("loading tree")
-                    .state = if error.code == ProtocolErrorCode::NotFound {
                     ObservationState::Stale
                 } else {
                     ObservationState::Error(ClientError::new(
@@ -1097,6 +1106,12 @@ impl Client {
                         error.to_string(),
                     ))
                 };
+                set_tree_subject_state(
+                    self.session_mut(session)?,
+                    &request.path,
+                    id,
+                    observation_state,
+                );
                 Ok(Completion::ProtocolFailure(error))
             }
             Err(error) => self.retrieval_decode_error(
@@ -1206,6 +1221,7 @@ impl Client {
                 {
                     if let Some(object) = state.objects.get_mut(&context.object_key) {
                         object.object.validator = value.validator.clone();
+                        object.revision = id;
                     }
                     state.mark_object_stale(context.object_key);
                 }
@@ -1320,11 +1336,12 @@ impl Client {
                     .state = ObservationState::Error(error.clone());
             }
             RetrievalSubject::Tree(path) => {
-                self.session_mut(session)?
+                set_tree_subject_state(
-                    .trees
+                    self.session_mut(session)?,
-                    .get_mut(&path)
+                    &path,
-                    .expect("tree exists")
+                    id,
-                    .state = ObservationState::Error(error.clone());
+                    ObservationState::Error(error.clone()),
+                );
             }
             RetrievalSubject::Interface(reference) => {
                 self.session_mut(session)?
@@ -1341,7 +1358,7 @@ impl Client {
 
     fn mark_retrieval_error(
         &mut self,
-        _id: RequestId,
+        id: RequestId,
         pending: Pending,
         error: ClientError,
     ) -> Result<(), ClientError> {
@@ -1354,11 +1371,12 @@ impl Client {
                     .state = ObservationState::Error(error);
             }
             Pending::Tree { session, request } => {
-                self.session_mut(&session)?
+                set_tree_subject_state(
-                    .trees
+                    self.session_mut(&session)?,
-                    .get_mut(&request.path)
+                    &request.path,
-                    .expect("tree exists")
+                    id,
-                    .state = ObservationState::Error(error);
+                    ObservationState::Error(error),
+                );
             }
             Pending::Interface { session, request } => {
                 self.session_mut(&session)?
@@ -1527,6 +1545,24 @@ fn classify_response_error(error: ClientResponseError) -> ClientError {
     }
 }
 
+fn set_tree_subject_state(
+    state: &mut Session,
+    path: &str,
+    revision: RequestId,
+    observation_state: ObservationState,
+) {
+    if let Some(tree) = state.trees.get_mut(path)
+        && tree.revision == revision
+    {
+        tree.state = observation_state.clone();
+    }
+    if let Some(entry) = state.entries.get_mut(path)
+        && entry.revision == revision
+    {
+        entry.state = observation_state;
+    }
+}
+
 fn ensure_interface_slot(
     state: &mut Session,
     reference: &str,
@@ -1614,6 +1650,16 @@ fn remove_entry(state: &mut Session, path: &str) {
     }
 }
 
+fn object_ownership_revision(state: &Session, key: u64) -> Option<RequestId> {
+    state
+        .entries
+        .values()
+        .filter(|entry| entry.object_key == Some(key))
+        .map(|entry| entry.revision)
+        .chain(state.objects.get(&key).map(|record| record.revision))
+        .max()
+}
+
 fn insert_object(
     state: &mut Session,
     path: &str,
@@ -1632,6 +1678,7 @@ fn insert_object(
     let key = if let Some(reference) = &object.r#ref {
         if let Some(key) = state.refs.get(reference).copied() {
             if let Some(existing) = state.objects.get(&key)
+                && object_ownership_revision(state, key).is_none_or(|owner| owner <= revision)
                 && existing.object.validator.is_some()
                 && existing.object.validator == object.validator
                 && existing.object != object
@@ -1669,7 +1716,23 @@ fn insert_object(
         state.refs.insert(reference.clone(), key);
     }
     let touched = state.tick();
-    state.objects.insert(key, ObjectRecord { object, touched });
+    let ownership_revision = object_ownership_revision(state, key).unwrap_or(revision);
+    match state.objects.get_mut(&key) {
+        Some(existing) if ownership_revision > revision => {
+            existing.revision = ownership_revision;
+            existing.touched = touched;
+        }
+        _ => {
+            state.objects.insert(
+                key,
+                ObjectRecord {
+                    object,
+                    revision,
+                    touched,
+                },
+            );
+        }
+    }
     state.entries.insert(
         path.to_owned(),
         EntryRecord {

crates/wip-client/tests/runtime.rs

@@ -202,6 +202,94 @@ fn normalized_object_cache_rejects_conflicts_and_evicts_to_its_bound() {
     assert!(client.object(&session, "/third").is_some());
 }
 
+#[test]
+fn normalized_aliases_preserve_newer_request_ownership() {
+    let mut client = client(8, 8, 8, 8);
+    let session = client
+        .open_session("https://example.test/wip", SecurityContext::new("subject"))
+        .unwrap();
+    observe_object(
+        &mut client,
+        &session,
+        "/a/shared",
+        object("shared", Some("shared-ref"), b"initial"),
+    );
+    observe_object(
+        &mut client,
+        &session,
+        "/b/shared",
+        object("shared", Some("shared-ref"), b"initial"),
+    );
+    let request = FetchTreeRequest {
+        path: "/a".into(),
+        depth: 1,
+        limit: 4,
+    };
+    let older_tree_response = || FetchTreeResponse {
+        root: ObjectTreeNode {
+            object: object("a", None, b"root"),
+            children: vec![ObjectTreeNode {
+                object: object("shared", Some("shared-ref"), b"older"),
+                children: vec![],
+            }],
+        },
+        truncated: false,
+    };
+
+    let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap();
+    let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap();
+    client
+        .complete(
+            newer_alias.id,
+            fetch_response("/b/shared", &object("shared", Some("shared-ref"), b"newer")),
+        )
+        .unwrap();
+    client
+        .complete(
+            older_tree.id,
+            encode_fetch_tree_response(&request, &older_tree_response(), wire_limits()).unwrap(),
+        )
+        .unwrap();
+    assert_eq!(
+        client.object(&session, "/a/shared").unwrap().validator,
+        Some(b"newer".to_vec())
+    );
+    assert_eq!(
+        client.object(&session, "/b/shared").unwrap().validator,
+        Some(b"newer".to_vec())
+    );
+
+    let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap();
+    let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap();
+    client
+        .complete(
+            older_tree.id,
+            encode_fetch_tree_response(&request, &older_tree_response(), wire_limits()).unwrap(),
+        )
+        .unwrap();
+    assert!(matches!(
+        client.object(&session, "/b/shared").unwrap().state,
+        ObservationState::Loading(id) if id == newer_alias.id
+    ));
+    assert_eq!(
+        client.object(&session, "/b/shared").unwrap().validator,
+        Some(b"newer".to_vec())
+    );
+    client
+        .complete(
+            newer_alias.id,
+            fetch_response(
+                "/b/shared",
+                &object("shared", Some("shared-ref"), b"newest"),
+            ),
+        )
+        .unwrap();
+    assert_eq!(
+        client.object(&session, "/a/shared").unwrap().validator,
+        Some(b"newest".to_vec())
+    );
+}
+
 #[test]
 fn tree_expansion_is_lazy_truncated_and_refreshes_edges() {
     let mut client = client(16, 8, 8, 8);
@@ -595,6 +683,15 @@ fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
         client.tree(&session, "/child").unwrap().state,
         ObservationState::Loading(id) if id == newer_child_tree.id
     ));
+    let child_observation = client.object(&session, "/child").unwrap();
+    assert!(matches!(
+        child_observation.state,
+        ObservationState::Loading(id) if id == newer_child_tree.id
+    ));
+    assert_eq!(
+        child_observation.validator,
+        Some(b"newer-than-omission".to_vec())
+    );
     let child_request = FetchTreeRequest {
         path: "/child".into(),
         depth: 0,
@@ -857,7 +954,6 @@ fn failed_entry_and_tree_observation_metadata_is_bounded() {
         );
     }
     assert!(client.tree(&session, "/alpha").is_none());
-    assert!(client.tree(&session, "/beta").is_some());
     assert!(client.tree(&session, "/gamma").is_some());
 }