feat(zotero): emit reversible write execution receipts

CHANGELOG.md

@@ -6,6 +6,9 @@ All notable changes to ConceptWeave are documented here.
 
 ### Fixed
 
+- Failed write responses no longer turn later observations into proof of completion or permission to undo another change; the submitted request and unresolved result remain available for investigation.
+- Every execution outcome retains the identity of the reviewed supporting evidence.
+
 - Research change plans reject changed supporting evidence and incomplete inventories before approval, and preserve the identity of the reviewed evidence.
 
 - Duplicate review rejects incomplete source inventories and changed supporting evidence before approval, while retaining reversible identity mappings.

crates/conceptweave-zotero/src/lib.rs

@@ -373,30 +373,160 @@ pub struct ClassificationWriteOperation {
 }
 
 /// Local-only, snapshot-bound plan for reviewed Zotero classification writes.
+///
+/// Execution-critical fields are read-only outside this crate, so callers cannot
+/// mutate a verified plan before passing it to the execution boundary.
+///
+/// ```compile_fail
+/// use conceptweave_zotero::{ClassificationWritePlan, WriteMode};
+/// fn forge(plan: &mut ClassificationWritePlan) {
+///     plan.mode = WriteMode::Execute;
+/// }
+/// ```
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct ClassificationWritePlan {
     /// Requested write behavior; dry-run is the default.
-    pub mode: WriteMode,
+    mode: WriteMode,
     /// Opaque review receipt identifier.
-    pub review_id: String,
+    review_id: String,
     /// Opaque governance authority receipt.
-    pub authority_receipt: String,
+    authority_receipt: String,
     /// Exact Local API server identity.
-    pub server_id: Option<String>,
+    server_id: Option<String>,
     /// Exact Zotero version used to establish execute eligibility.
-    pub zotero_version: String,
+    zotero_version: String,
     /// Exact library-version precondition.
-    pub library_version: u64,
+    library_version: u64,
     /// Exact classifier revision.
-    pub rule_revision: String,
+    rule_revision: String,
     /// Exact raw-snapshot digest.
-    pub snapshot_digest: String,
+    snapshot_digest: String,
     /// Content identity retained from the independently verified review.
-    pub proposal_digest: String,
+    proposal_digest: String,
     /// Deterministically ordered item operations.
-    pub operations: Vec<ClassificationWriteOperation>,
+    operations: Vec<ClassificationWriteOperation>,
     /// Classification writes never delete source records or attachments.
-    pub source_records_preserved: bool,
+    source_records_preserved: bool,
+}
+
+impl ClassificationWritePlan {
+    /// Returns the requested dry-run or execute behavior.
+    pub const fn mode(&self) -> WriteMode {
+        self.mode
+    }
+
+    /// Returns the exact library revision used by every initial preflight.
+    pub const fn library_version(&self) -> u64 {
+        self.library_version
+    }
+
+    /// Returns the deterministic reviewed operations.
+    pub fn operations(&self) -> &[ClassificationWriteOperation] {
+        &self.operations
+    }
+
+    /// Confirms that the plan contains metadata changes only.
+    pub const fn source_records_preserved(&self) -> bool {
+        self.source_records_preserved
+    }
+}
+
+/// Complete item state observed at the Local API write boundary.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ClassificationItemState {
+    /// Local API server identity that produced this state.
+    pub server_id: String,
+    /// Library revision that produced this state.
+    pub library_version: u64,
+    /// Stable Zotero item key.
+    pub item_key: String,
+    /// Current optimistic item revision.
+    pub item_version: u64,
+    /// Complete collection state.
+    pub collection_keys: Vec<String>,
+    /// Complete typed-tag state.
+    pub tags: Vec<ItemTag>,
+}
+
+/// One conditional complete-state replacement passed to an authenticated adapter.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ClassificationWriteRequest {
+    /// Expected Local API server identity.
+    pub server_id: String,
+    /// Current library revision precondition.
+    pub library_version: u64,
+    /// Stable Zotero item key.
+    pub item_key: String,
+    /// Current item revision precondition.
+    pub item_version: u64,
+    /// Complete collection replacement.
+    pub collection_keys: Vec<String>,
+    /// Complete typed-tag replacement.
+    pub tags: Vec<ItemTag>,
+}
+
+/// A conditional inverse write created only after a verified successful write.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ClassificationRollbackOperation {
+    /// Stable Zotero item key.
+    pub item_key: String,
+    /// Post-write item revision required by a rollback adapter.
+    pub item_version: u64,
+    /// Complete collection state to restore.
+    pub collection_keys: Vec<String>,
+    /// Complete typed-tag state to restore.
+    pub tags: Vec<ItemTag>,
+}
+
+/// Observable result of a write-plan execution attempt.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ClassificationWriteOutcome {
+    /// The plan intentionally made no Local API calls.
+    DryRun,
+    /// Every requested item replacement was verified.
+    Applied,
+    /// No write began because a complete preflight could not be proven.
+    PreflightFailure,
+    /// A write or its response failed after preflight.
+    PartialFailure,
+}
+
+/// Secret-free evidence for applied, failed, pending, and reversible writes.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct ClassificationWriteReceipt {
+    /// Opaque review receipt identifier bound to this attempt.
+    pub review_id: String,
+    /// Opaque governance authority receipt bound to this attempt.
+    pub authority_receipt: String,
+    /// Exact Local API server identity bound to this attempt.
+    pub server_id: Option<String>,
+    /// Exact Zotero version bound to this attempt.
+    pub zotero_version: String,
+    /// Exact library-version precondition bound to this attempt.
+    pub library_version: u64,
+    /// Exact classifier revision bound to this attempt.
+    pub rule_revision: String,
+    /// Exact raw-snapshot digest bound to this attempt.
+    pub snapshot_digest: String,
+    /// Verified proposal and retained-source identity bound to this attempt.
+    pub proposal_digest: String,
+    /// Overall execution outcome.
+    pub outcome: ClassificationWriteOutcome,
+    /// Items whose post-write state was verified, in application order.
+    pub applied_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 write response.
+    pub indeterminate_item_key: Option<String>,
+    /// Exact submitted request whose completion is unknown; audit evidence, not retry authority.
+    pub indeterminate_request: Option<ClassificationWriteRequest>,
+    /// Subsequent observation, if available; it cannot prove causal completion.
+    pub reconciliation_observation: Option<ClassificationItemState>,
+    /// Items whose write was not attempted.
+    pub not_attempted_item_keys: Vec<String>,
+    /// Verified inverse operations in safe reverse application order.
+    pub rollback_operations: Vec<ClassificationRollbackOperation>,
 }
 
 /// A fail-closed reviewed write-plan contract violation.
@@ -1181,6 +1311,209 @@ where
     })
 }
 
+/// Executes a reviewed plan through caller-owned authenticated Local API functions.
+///
+/// Every item is preflighted before the first write. Adapter errors are deliberately
+/// reduced to secret-free receipt states instead of being serialized.
+pub fn execute_classification_write_plan<PreflightError, WriteError>(
+    plan: &ClassificationWritePlan,
+    mut preflight: impl FnMut(&str) -> Result<ClassificationItemState, PreflightError>,
+    mut write_item: impl FnMut(
+        &ClassificationWriteRequest,
+    ) -> Result<ClassificationItemState, WriteError>,
+) -> ClassificationWriteReceipt {
+    if plan.mode == WriteMode::DryRun {
+        return ClassificationWriteReceipt {
+            review_id: plan.review_id.clone(),
+            authority_receipt: plan.authority_receipt.clone(),
+            server_id: plan.server_id.clone(),
+            zotero_version: plan.zotero_version.clone(),
+            library_version: plan.library_version,
+            rule_revision: plan.rule_revision.clone(),
+            snapshot_digest: plan.snapshot_digest.clone(),
+            proposal_digest: plan.proposal_digest.clone(),
+            outcome: ClassificationWriteOutcome::DryRun,
+            applied_item_keys: Vec::new(),
+            failed_item_key: None,
+            indeterminate_item_key: None,
+            indeterminate_request: None,
+            reconciliation_observation: None,
+            not_attempted_item_keys: plan
+                .operations
+                .iter()
+                .map(|operation| operation.item_key.clone())
+                .collect(),
+            rollback_operations: Vec::new(),
+        };
+    }
+    let server_id = plan
+        .server_id
+        .as_deref()
+        .filter(|server_id| !server_id.trim().is_empty())
+        .expect("execute plans are built with a nonblank server identity");
+
+    for operation in &plan.operations {
+        let Ok(state) = preflight(&operation.item_key) else {
+            return preflight_failure_receipt(plan, Some(&operation.item_key));
+        };
+        if !matches_before_state(&state, server_id, plan.library_version, operation) {
+            return preflight_failure_receipt(plan, Some(&operation.item_key));
+        }
+    }
+
+    let mut current_library_version = plan.library_version;
+    let mut applied_item_keys = Vec::new();
+    let mut rollback_operations = Vec::new();
+    for (operation_index, operation) in plan.operations.iter().enumerate() {
+        let request = ClassificationWriteRequest {
+            server_id: server_id.to_owned(),
+            library_version: current_library_version,
+            item_key: operation.item_key.clone(),
+            item_version: operation.item_version,
+            collection_keys: operation.after_collection_keys.clone(),
+            tags: operation.after_tags.clone(),
+        };
+        let response = write_item(&request);
+        let verified_state = response.ok().filter(|state| {
+            matches_after_state(state, server_id, current_library_version, operation)
+        });
+        let Some(state) = verified_state else {
+            let reconciled_state = preflight(&operation.item_key).ok();
+            // A delayed request or concurrent writer can explain any observed state.
+            // Retain the observation without granting completion, retry, or inverse authority.
+            rollback_operations.reverse();
+            return partial_failure_receipt(
+                plan,
+                operation_index,
+                applied_item_keys,
+                rollback_operations,
+                request,
+                reconciled_state,
+            );
+        };
+        current_library_version = state.library_version;
+        applied_item_keys.push(operation.item_key.clone());
+        rollback_operations.push(ClassificationRollbackOperation {
+            item_key: operation.item_key.clone(),
+            item_version: state.item_version,
+            collection_keys: operation.rollback_collection_keys.clone(),
+            tags: operation.rollback_tags.clone(),
+        });
+    }
+    rollback_operations.reverse();
+    ClassificationWriteReceipt {
+        review_id: plan.review_id.clone(),
+        authority_receipt: plan.authority_receipt.clone(),
+        server_id: plan.server_id.clone(),
+        zotero_version: plan.zotero_version.clone(),
+        library_version: plan.library_version,
+        rule_revision: plan.rule_revision.clone(),
+        snapshot_digest: plan.snapshot_digest.clone(),
+        proposal_digest: plan.proposal_digest.clone(),
+        outcome: ClassificationWriteOutcome::Applied,
+        applied_item_keys,
+        failed_item_key: None,
+        indeterminate_item_key: None,
+        indeterminate_request: None,
+        reconciliation_observation: None,
+        not_attempted_item_keys: Vec::new(),
+        rollback_operations,
+    }
+}
+
+fn matches_before_state(
+    state: &ClassificationItemState,
+    server_id: &str,
+    library_version: u64,
+    operation: &ClassificationWriteOperation,
+) -> bool {
+    normalized_metadata(&state.collection_keys, &state.tags).is_ok_and(|(collections, tags)| {
+        state.server_id == server_id
+            && state.library_version == library_version
+            && state.item_key == operation.item_key
+            && state.item_version == operation.item_version
+            && collections == operation.before_collection_keys
+            && tags == operation.before_tags
+    })
+}
+
+fn matches_after_state(
+    state: &ClassificationItemState,
+    server_id: &str,
+    library_version: u64,
+    operation: &ClassificationWriteOperation,
+) -> bool {
+    normalized_metadata(&state.collection_keys, &state.tags).is_ok_and(|(collections, tags)| {
+        state.server_id == server_id
+            && state.library_version > library_version
+            && state.item_key == operation.item_key
+            && state.item_version > operation.item_version
+            && collections == operation.after_collection_keys
+            && tags == operation.after_tags
+    })
+}
+
+fn preflight_failure_receipt(
+    plan: &ClassificationWritePlan,
+    failed_item_key: Option<&str>,
+) -> ClassificationWriteReceipt {
+    ClassificationWriteReceipt {
+        review_id: plan.review_id.clone(),
+        authority_receipt: plan.authority_receipt.clone(),
+        server_id: plan.server_id.clone(),
+        zotero_version: plan.zotero_version.clone(),
+        library_version: plan.library_version,
+        rule_revision: plan.rule_revision.clone(),
+        snapshot_digest: plan.snapshot_digest.clone(),
+        proposal_digest: plan.proposal_digest.clone(),
+        outcome: ClassificationWriteOutcome::PreflightFailure,
+        applied_item_keys: Vec::new(),
+        failed_item_key: failed_item_key.map(str::to_owned),
+        indeterminate_item_key: None,
+        indeterminate_request: None,
+        reconciliation_observation: None,
+        not_attempted_item_keys: plan
+            .operations
+            .iter()
+            .map(|operation| operation.item_key.clone())
+            .collect(),
+        rollback_operations: Vec::new(),
+    }
+}
+
+fn partial_failure_receipt(
+    plan: &ClassificationWritePlan,
+    failed_index: usize,
+    applied_item_keys: Vec<String>,
+    rollback_operations: Vec<ClassificationRollbackOperation>,
+    indeterminate_request: ClassificationWriteRequest,
+    reconciliation_observation: Option<ClassificationItemState>,
+) -> ClassificationWriteReceipt {
+    ClassificationWriteReceipt {
+        review_id: plan.review_id.clone(),
+        authority_receipt: plan.authority_receipt.clone(),
+        server_id: plan.server_id.clone(),
+        zotero_version: plan.zotero_version.clone(),
+        library_version: plan.library_version,
+        rule_revision: plan.rule_revision.clone(),
+        snapshot_digest: plan.snapshot_digest.clone(),
+        proposal_digest: plan.proposal_digest.clone(),
+        outcome: ClassificationWriteOutcome::PartialFailure,
+        applied_item_keys,
+        failed_item_key: Some(plan.operations[failed_index].item_key.clone()),
+        indeterminate_item_key: Some(indeterminate_request.item_key.clone()),
+        indeterminate_request: Some(indeterminate_request),
+        reconciliation_observation,
+        not_attempted_item_keys: plan
+            .operations
+            .iter()
+            .skip(failed_index + 1)
+            .map(|operation| operation.item_key.clone())
+            .collect(),
+        rollback_operations,
+    }
+}
+
 /// Failure raised when a bounded, immutable Local API read cannot be proven.
 #[derive(Debug)]
 pub enum ReadError {