feat(zotero): execute verified rollback receipts

crates/conceptweave-zotero/src/lib.rs

@@ -863,19 +863,60 @@ pub struct ClassificationWriteRequest {
     pub tags: Vec<ItemTag>,
 }
 
-/// A conditional inverse write created only after a verified successful write.
+/// Mutable inverse-write audit data; this value alone does not prove authority.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct ClassificationRollbackOperation {
+    /// Local API server identity that produced the state to undo.
+    pub server_id: String,
     /// Stable Zotero item key.
     pub item_key: String,
     /// Post-write item revision required by a rollback adapter.
     pub item_version: u64,
+    /// Complete post-write collection state that must still be current.
+    pub expected_collection_keys: Vec<String>,
+    /// Complete post-write typed-tag state that must still be current.
+    pub expected_tags: Vec<ItemTag>,
     /// Complete collection state to restore.
     pub collection_keys: Vec<String>,
     /// Complete typed-tag state to restore.
     pub tags: Vec<ItemTag>,
 }
 
+/// Observable result of one rollback execution attempt.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ClassificationRollbackOutcome {
+    /// Every inverse write was verified.
+    Restored,
+    /// No inverse write began because complete preflight could not be proven.
+    PreflightFailure,
+    /// An inverse write or its response failed after preflight.
+    PartialFailure,
+}
+
+/// Secret-free evidence for restored, failed, indeterminate, and pending inverse writes.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ClassificationRollbackReceipt {
+    /// Overall rollback outcome.
+    pub outcome: ClassificationRollbackOutcome,
+    /// Items whose restored state was verified, in receipt order.
+    pub restored_item_keys: Vec<String>,
+    /// First item whose preflight, write, or response failed.
+    pub failed_item_key: Option<String>,
+    /// Item whose state could not be proven after an unverifiable response.
+    pub indeterminate_item_key: Option<String>,
+    /// Complete operation retained for manual reconciliation of indeterminate state.
+    pub indeterminate_operation: Option<ClassificationRollbackOperation>,
+    /// Exact submitted inverse request; observations do not authorize its retry.
+    pub indeterminate_request: Option<ClassificationWriteRequest>,
+    /// Complete readback after failure, retained without causal completion claims.
+    pub reconciliation_observation: Option<ClassificationItemState>,
+    /// Items whose inverse write was not attempted.
+    pub not_attempted_item_keys: Vec<String>,
+    /// Untouched operations only; this audit list does not grant retry authority.
+    pub remaining_operations: Vec<ClassificationRollbackOperation>,
+}
+
 /// Observable result of a write-plan execution attempt.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]
@@ -1792,8 +1833,11 @@ pub fn execute_classification_write_plan<PreflightError, WriteError>(
         current_library_version = state.library_version;
         applied_item_keys.push(operation.item_key.clone());
         rollback_operations.push(ClassificationRollbackOperation {
+            server_id: server_id.to_owned(),
             item_key: operation.item_key.clone(),
             item_version: state.item_version,
+            expected_collection_keys: operation.after_collection_keys.clone(),
+            expected_tags: operation.after_tags.clone(),
             collection_keys: operation.rollback_collection_keys.clone(),
             tags: operation.rollback_tags.clone(),
         });
@@ -1836,6 +1880,145 @@ pub fn execute_classification_write_plan_with_zotero10(
     )
 }
 
+/// Executes caller-validated inverse operations; this primitive grants no authority.
+pub fn execute_classification_rollback<PreflightError, WriteError>(
+    operations: &[ClassificationRollbackOperation],
+    mut preflight: impl FnMut(&str) -> Result<ClassificationItemState, PreflightError>,
+    mut write_item: impl FnMut(
+        &ClassificationWriteRequest,
+    ) -> Result<ClassificationItemState, WriteError>,
+) -> ClassificationRollbackReceipt {
+    if let Some(operation) = operations
+        .windows(2)
+        .find(|pair| pair[0].server_id != pair[1].server_id)
+        .map(|pair| &pair[1])
+    {
+        return rollback_preflight_failure(operations, &operation.item_key);
+    }
+
+    let mut preflight_states = Vec::with_capacity(operations.len());
+    let mut library_version = None;
+    for operation in operations {
+        let Ok(state) = preflight(&operation.item_key) else {
+            return rollback_preflight_failure(operations, &operation.item_key);
+        };
+        let same_library = library_version.is_none_or(|version| version == state.library_version);
+        if !same_library || !matches_rollback_current(&state, operation) {
+            return rollback_preflight_failure(operations, &operation.item_key);
+        }
+        library_version = Some(state.library_version);
+        preflight_states.push(state);
+    }
+
+    let mut current_library_version = library_version.unwrap_or_default();
+    let mut restored_item_keys = Vec::new();
+    for (index, (operation, state)) in operations.iter().zip(preflight_states).enumerate() {
+        let request = ClassificationWriteRequest {
+            server_id: operation.server_id.clone(),
+            library_version: current_library_version,
+            item_key: operation.item_key.clone(),
+            item_version: state.item_version,
+            collection_keys: operation.collection_keys.clone(),
+            tags: operation.tags.clone(),
+        };
+        let verified = write_item(&request)
+            .ok()
+            .filter(|state| matches_rollback_restored(state, current_library_version, operation));
+        if let Some(state) = verified {
+            current_library_version = state.library_version;
+            restored_item_keys.push(operation.item_key.clone());
+            continue;
+        }
+
+        let reconciled = preflight(&operation.item_key).ok();
+        return ClassificationRollbackReceipt {
+            outcome: ClassificationRollbackOutcome::PartialFailure,
+            restored_item_keys,
+            failed_item_key: Some(operation.item_key.clone()),
+            indeterminate_item_key: Some(operation.item_key.clone()),
+            indeterminate_operation: Some(operation.clone()),
+            indeterminate_request: Some(request),
+            reconciliation_observation: reconciled,
+            not_attempted_item_keys: operations[index + 1..]
+                .iter()
+                .map(|operation| operation.item_key.clone())
+                .collect(),
+            remaining_operations: operations[index + 1..].to_vec(),
+        };
+    }
+    ClassificationRollbackReceipt {
+        outcome: ClassificationRollbackOutcome::Restored,
+        restored_item_keys,
+        failed_item_key: None,
+        indeterminate_item_key: None,
+        indeterminate_operation: None,
+        indeterminate_request: None,
+        reconciliation_observation: None,
+        not_attempted_item_keys: Vec::new(),
+        remaining_operations: Vec::new(),
+    }
+}
+
+/// Executes caller-validated inverses through an adapter, without issuing approval.
+pub fn execute_classification_rollback_with_zotero10(
+    operations: &[ClassificationRollbackOperation],
+    adapter: &Zotero10LocalAdapter,
+) -> ClassificationRollbackReceipt {
+    execute_classification_rollback(
+        operations,
+        |item_key| adapter.get_item(item_key),
+        |request| adapter.write_item(request),
+    )
+}
+
+fn rollback_preflight_failure(
+    operations: &[ClassificationRollbackOperation],
+    failed_item_key: &str,
+) -> ClassificationRollbackReceipt {
+    ClassificationRollbackReceipt {
+        outcome: ClassificationRollbackOutcome::PreflightFailure,
+        restored_item_keys: Vec::new(),
+        failed_item_key: Some(failed_item_key.to_owned()),
+        indeterminate_item_key: None,
+        indeterminate_operation: None,
+        indeterminate_request: None,
+        reconciliation_observation: None,
+        not_attempted_item_keys: operations
+            .iter()
+            .map(|item| item.item_key.clone())
+            .collect(),
+        remaining_operations: operations.to_vec(),
+    }
+}
+
+fn matches_rollback_current(
+    state: &ClassificationItemState,
+    operation: &ClassificationRollbackOperation,
+) -> bool {
+    normalized_metadata(&state.collection_keys, &state.tags).is_ok_and(|(collections, tags)| {
+        state.server_id == operation.server_id
+            && state.item_key == operation.item_key
+            && state.item_version == operation.item_version
+            && collections == operation.expected_collection_keys
+            && tags == operation.expected_tags
+    })
+}
+
+fn matches_rollback_restored(
+    state: &ClassificationItemState,
+    library_version: u64,
+    operation: &ClassificationRollbackOperation,
+) -> bool {
+    normalized_metadata(&state.collection_keys, &state.tags).is_ok_and(|(collections, tags)| {
+        state.server_id == operation.server_id
+            && state.library_version > library_version
+            && state.item_key == operation.item_key
+            && state.item_version > operation.item_version
+            && collections == operation.collection_keys
+            && tags == operation.tags
+    })
+}
+
 fn matches_before_state(
     state: &ClassificationItemState,
     server_id: &str,
@@ -2607,6 +2790,7 @@ mod tests {
 
     static LOCAL_API_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
 
+    #[cfg_attr(coverage_nightly, coverage(off))]
     fn serve(responses: Vec<&'static str>) -> (String, std::thread::JoinHandle<Vec<String>>) {
         let listener = TcpListener::bind("127.0.0.1:0").unwrap();
         let address = listener.local_addr().unwrap();
@@ -2617,25 +2801,27 @@ mod tests {
                     let (mut stream, _) = listener.accept().unwrap();
                     let mut bytes = Vec::new();
                     loop {
-                        let mut buffer = [0; 4096];
-                        let length = stream.read(&mut buffer).unwrap();
+                        let mut chunk = [0; 4096];
+                        let length = stream.read(&mut chunk).unwrap();
                         assert_ne!(length, 0);
-                        bytes.extend_from_slice(&buffer[..length]);
-                        if let Some(header_end) =
+                        bytes.extend_from_slice(&chunk[..length]);
+                        let Some(header_end) =
                             bytes.windows(4).position(|part| part == b"\r\n\r\n")
-                        {
-                            let headers = std::str::from_utf8(&bytes[..header_end]).unwrap();
-                            let body_length = headers
-                                .lines()
-                                .find_map(|line| {
-                                    line.to_ascii_lowercase()
-                                        .strip_prefix("content-length: ")
-                                        .map(|value| value.parse::<usize>().unwrap())
-                                })
-                                .unwrap_or(0);
-                            if bytes.len() >= header_end + 4 + body_length {
-                                break;
-                            }
+                        else {
+                            continue;
+                        };
+                        let headers = String::from_utf8_lossy(&bytes[..header_end]);
+                        let content_length = headers
+                            .lines()
+                            .find_map(|line| {
+                                let (name, value) = line.split_once(':')?;
+                                name.eq_ignore_ascii_case("content-length")
+                                    .then(|| value.trim().parse::<usize>().ok())
+                                    .flatten()
+                            })
+                            .unwrap_or_default();
+                        if bytes.len() >= header_end + 4 + content_length {
+                            break;
                         }
                     }
                     stream.write_all(response.as_bytes()).unwrap();
@@ -2792,6 +2978,42 @@ mod tests {
         assert_eq!(server.join().unwrap().len(), 4);
     }
 
+    #[test]
+    fn approved_zotero10_adapter_executes_the_rollback_boundary() {
+        let operation = ClassificationRollbackOperation {
+            server_id: "server-10".into(),
+            item_key: "ABCD2345".into(),
+            item_version: 43,
+            expected_collection_keys: vec!["CDEF4567".into()],
+            expected_tags: vec![ItemTag {
+                tag: "classified".into(),
+                tag_type: None,
+            }],
+            collection_keys: vec!["BCDE3456".into()],
+            tags: vec![ItemTag {
+                tag: "kept".into(),
+                tag_type: Some(1),
+            }],
+        };
+        let before = library_response("server-10", 44);
+        let item_body = r#"{"key":"ABCD2345","version":43,"data":{"itemType":"book","collections":["CDEF4567"],"tags":[{"tag":"classified"}]}}"#;
+        let item = raw_response(Some("server-10"), Some(43), item_body);
+        let after = library_response("server-10", 44);
+        let written = write_response("server-10", 45, 45);
+        let (base, server) = serve(vec![
+            Box::leak(before.into_boxed_str()),
+            Box::leak(item.into_boxed_str()),
+            Box::leak(after.into_boxed_str()),
+            Box::leak(written.into_boxed_str()),
+        ]);
+
+        let receipt = execute_classification_rollback_with_zotero10(&[operation], &transport(base));
+
+        assert_eq!(receipt.outcome, ClassificationRollbackOutcome::Restored);
+        assert_eq!(receipt.restored_item_keys, ["ABCD2345"]);
+        assert_eq!(server.join().unwrap().len(), 4);
+    }
+
     #[test]
     fn zotero10_authorization_uses_exact_wire_contract_and_builds_adapter() {
         let body = r#"{"key":"0123456789abcdef0123456789abcdef","remember":true}"#;