fix(store): graph-neutral operation conflicts repair without full replay (#9101)

crates/fractal-store/src/store.rs

@@ -17558,11 +17558,10 @@ pub fn try_project_simple_replication_tx(
             }
         }
     }
-    // Event-only claims do not change the structural graph. Their operation
-    // registry can advance with the frontier. Message sends reconcile an ambiguous
-    // receipt from that operation's indexed history, using the full replay's reducer.
-    // Other kinds retain the full replay guard for ambiguous IDs.
-    let mut reconciled_message_operations = BTreeSet::new();
+    // Event-only claims and message sends reconcile ambiguous receipts from that
+    // operation's indexed history, using the full replay's reducer. Structural
+    // kinds retain the full replay guard for ambiguous IDs.
+    let mut reconciled_operations = BTreeSet::new();
     for claim in &claims {
         let Some((operation_id, request_digest)) = operation_parts(&claim.body) else {
             continue;
@@ -17579,22 +17578,52 @@ pub fn try_project_simple_replication_tx(
         if repaired {
             continue;
         }
-        if claim.kind == "message.sent" && reconciled_message_operations.contains(operation_id) {
+        let graph_neutral = !replay_projects_kind(&claim.kind);
+        if graph_neutral && reconciled_operations.contains(operation_id) {
             continue;
         }
-        if let Some((stored_digest, _, state)) =
-            operation_tx(transaction, operation_id).map_err(|error| {
+        let stored_operation = operation_tx(transaction, operation_id).map_err(|error| {
+            incremental_claim_error(internal(error), claim, "operation-registry")
+        })?;
+        let conflict = stored_operation.as_ref().is_some_and(|(digest, _, state)| {
+            digest.as_str() != request_digest || state != "active"
+        });
+        // A checkpoint can leave only tombstones, with no operations row. Before
+        // folding either case, inspect only this operation's indexed history:
+        // a structural claim sharing the key needs its projections replayed too.
+        if conflict || (graph_neutral && stored_operation.is_none()) {
+            let neutral_history = graph_neutral && (|| -> rusqlite::Result<bool> {
+                let mut statement = transaction.prepare_cached(
+                    "SELECT kind FROM claims INDEXED BY claims_operation_index
+                     WHERE json_extract(body, '$._operation.id')=?1
+                     UNION ALL
+                     SELECT kind FROM checkpoint_claims INDEXED BY checkpoint_claims_operation
+                     WHERE operation_id=?1 AND operation_id IS NOT NULL",
+                )?;
+                let mut rows = statement.query([operation_id])?;
+                while let Some(row) = rows.next()? {
+                    let kind = row.get_ref(0)?;
+                    let kind = kind.as_str().map_err(|error| {
+                        rusqlite::Error::FromSqlConversionFailure(0, kind.data_type(), Box::new(error))
+                    })?;
+                    if replay_projects_kind(kind) {
+                        return Ok(false);
+                    }
+                }
+                Ok(true)
+            })().map_err(|error| {
                 incremental_claim_error(internal(error), claim, "operation-registry")
-            })?
-            && (stored_digest != request_digest || state != "active")
-        {
-            if claim.kind == "message.sent" {
+            })?;
+            if neutral_history {
                 repair_operations_tx(transaction, &[operation_id.to_owned()]).map_err(|error| {
                     incremental_claim_error(internal(error), claim, "operation-registry")
                 })?;
-                reconciled_message_operations.insert(operation_id.to_owned());
+                reconciled_operations.insert(operation_id.to_owned());
                 continue;
             }
+            let (stored_digest, state) = stored_operation.as_ref()
+                .map(|(digest, _, state)| (digest.as_str(), state.as_str()))
+                .unwrap_or(("", "missing"));
             crate::profile::note("replay: operation-conflict");
             return Ok(IncrementalProjection::ReplayWithContext {
                 reason: "operation-conflict",

crates/fractal-store/src/store/identities_tests.rs

@@ -410,3 +410,55 @@ proptest! {
         }
     }
 }
+
+#[test]
+fn mixed_structural_and_neutral_operation_history_requires_replay_in_both_orders() {
+    for neutral_first in [false, true] {
+        let source = node("node-a", None, None);
+        let store = node("node-b", None, None);
+        declare(&source, ORIGINAL);
+        sync(&source, &store);
+        store.resolve_identity(IdentityKind::Agent, ORIGINAL).unwrap();
+        let receipt = source.rename_identity(
+            IdentityKind::Agent, &request(ORIGINAL, RENAMED, "mixed-operation"), &[],
+        ).unwrap();
+        let rename = source.claim_by_id(&receipt.claim_id).unwrap().unwrap();
+        let (operation, _) = operation_parts(&rename.body).unwrap();
+        let neutral_body = json!({"fields": {}, "_operation": {
+            "id": operation, "request_digest": "invented-neutral-digest"
+        }});
+        for neutral in [neutral_first, !neutral_first] {
+            {
+                let mut connection = store.connection.write();
+                let tx = connection.transaction().unwrap();
+                let (subject, kind, actor, body) = if neutral {
+                    ("github-event/github/acme/garden/pull-request/7/checks/failed",
+                        "github.event", None, &neutral_body)
+                } else {
+                    (rename.subject.as_str(), rename.kind.as_str(), rename.actor.as_deref(), &rename.body)
+                };
+                append_claim_record_tx(&tx, source.origin(), subject, kind, actor, body, &[], None).unwrap();
+                tx.commit().unwrap();
+            }
+            if neutral != neutral_first {
+                let mut connection = store.connection.write();
+                let tx = connection.transaction().unwrap();
+                let result = try_project_simple_replication_tx(
+                    &tx, store.origin(), current_index_tx(&tx).unwrap(),
+                ).unwrap();
+                assert!(matches!(result, IncrementalProjection::ReplayWithContext {
+                    reason: "operation-conflict", ..
+                }), "neutral_first={neutral_first}");
+                tx.rollback().unwrap();
+            }
+            store.project_replication_backlog().unwrap();
+        }
+        assert_eq!(operation_tx(&store.readers.get(), operation).unwrap().unwrap().2, "conflict");
+        let error = store.resolve_identity(IdentityKind::Agent, ORIGINAL).unwrap_err();
+        assert_eq!(error.code, "lifecycle-unavailable");
+        assert_eq!(error.details["reason"], "conflict");
+        let digests = store.replication_snapshot().unwrap().projection_digests.clone();
+        store.replay_replication_graph().unwrap();
+        assert_eq!(store.replication_snapshot().unwrap().projection_digests, digests);
+    }
+}

crates/fractal-store/src/store/message_send_tests.rs

@@ -163,3 +163,101 @@ fn message_operation_conflict_receipt_cost_depends_on_its_own_history() {
     );
 }
 
+#[test]
+fn graph_neutral_operation_conflicts_match_replay_in_both_orders_and_after_checkpoint() {
+    const FLEET: &str = "018f6f0d-4a5d-7b8c-9d0e-123456789abc";
+    let subject = "github-event/github/acme/garden/pull-request/7/checks/failed";
+    let sources = [
+        Store::open_memory("node-a").unwrap(),
+        Store::open_memory("node-b").unwrap(),
+    ];
+    let claims = sources.iter().enumerate().map(|(index, source)| {
+        source.bind_fleet(FLEET).unwrap();
+        source.append_claim(&ClaimInput {
+            subject: subject.into(),
+            kind: "github.event".into(),
+            actor: None,
+            fields: BTreeMap::from([
+                ("version".into(), json!(1)),
+                ("item".into(), json!("resource/github/acme/garden/pull-request/7")),
+                ("event".into(), json!("checks-failed")),
+                ("observer".into(), json!(format!("observer/node-{}", index))),
+                ("at".into(), json!(format!("2026-11-11T00:00:0{index}Z"))),
+                ("detail".into(), json!({"check":format!("check-{index}")})),
+                ("head_sha".into(), json!(format!("{index:040x}"))),
+            ]),
+            evidence: Vec::new(),
+            expected_subject: None,
+            idempotency_key: Some(format!("github-event:{subject}")),
+        }).unwrap()
+    }).collect::<Vec<_>>();
+    let (operation, digest) = operation_parts(&claims[0].body).unwrap();
+    let (other_operation, other_digest) = operation_parts(&claims[1].body).unwrap();
+    assert_eq!(operation, other_operation);
+    assert_ne!(digest, other_digest);
+    assert!(!replay_projects_kind("github.event"));
+    assert!(!Store::simple_replication_kind("github.event"));
+    let exchanges = sources.iter().map(|source| {
+        source.export_replication_exchange(FLEET, &ReplicationInventory::default()).unwrap()
+    }).collect::<Vec<_>>();
+    assert!(exchanges.iter().all(|exchange| !exchange.envelopes.is_empty()));
+    for checkpointed in [false, true] {
+        for order in [[0, 1], [1, 0]] {
+            let store = Store::open_memory("target").unwrap();
+            store.bind_fleet(FLEET).unwrap();
+            store.project_replication_backlog().unwrap();
+            let context = format!("order={order:?} checkpointed={checkpointed}");
+            let mut admitted = [false; 2];
+            for (position, index) in order.into_iter().enumerate() {
+                if checkpointed && position == 1 {
+                    let previous = &claims[order[0]];
+                    let (_, previous_digest) = operation_parts(&previous.body).unwrap();
+                    let mut connection = store.connection.write();
+                    let tx = connection.transaction().unwrap();
+                    tx.execute(/* sqlite-large-value: allow-for-now lv-h0f43df342761 owner=fractal_store::store::message_send_tests TODO=stage-generation-and-keyed-reclamation */ "INSERT INTO checkpoint_claims(id,writer,sequence,envelope_hash,subject,kind,
+                        actor,predecessors,operation_id,request_digest,accepted_at_unix_ms,checkpoint)
+                        VALUES (?1,?2,1,'invented-envelope',?3,'github.event',NULL,'[]',
+                        ?4,?5,0,'checkpoint/2026-11-11')",
+                        params![previous.id, previous.origin, previous.subject, operation, previous_digest]).unwrap();
+                    tx.execute(/* sqlite-large-value: allow-for-now lv-h3c87df5ded70 owner=fractal_store::store::message_send_tests TODO=bound-or-reference-values-and-reclamation */ "DELETE FROM operations WHERE id=?1", [operation]).unwrap();
+                    tx.execute(/* sqlite-large-value: allow-for-now lv-hede6c7531978 owner=fractal_store::store::message_send_tests TODO=prepare-body-and-predecessor-values */ "DELETE FROM claims WHERE id=?1", [&previous.id]).unwrap();
+                    tx.commit().unwrap();
+                    assert!(operation_tx(&connection, operation).unwrap().is_none());
+                }
+                let before = FULL_REPLAYS.with(std::cell::Cell::get);
+                store.receive_replication_exchange(&claims[index].origin, FLEET, &exchanges[index]).unwrap();
+                // Receiving stores envelopes; admission materializes their claims.
+                store.validate_replication_backlog().unwrap();
+                assert!(store.claim_by_id(&claims[index].id).unwrap().is_some(),
+                    "{context} position={position}: incoming claim must be admitted");
+                admitted[index] = true;
+                let prior_state = operation_tx(&store.readers.get(), operation).unwrap()
+                    .map(|(_, _, state)| state);
+                assert_eq!(prior_state.as_deref(),
+                    if position == 0 || checkpointed { None } else { Some("active") },
+                    "{context} position={position}: repair must still be pending");
+                store.project_replication_backlog().unwrap();
+                assert_eq!(FULL_REPLAYS.with(std::cell::Cell::get), before,
+                    "order={order:?} checkpointed={checkpointed} position={position}");
+                // An admitted digest mismatch (or tombstone-only history) must
+                // actually transition to conflict here, without a replay.
+                let state = operation_tx(&store.readers.get(), operation).unwrap()
+                    .unwrap_or_else(|| panic!("{context} position={position}: missing repaired operation")).2;
+                assert_eq!(state, if position == 0 { "active" } else { "conflict" },
+                    "{context} position={position}: indexed operation repair must run");
+            }
+            // Never compare two empty projections: both claims were admitted,
+            // and the second incremental projection already produced conflict.
+            assert_eq!(admitted, [true, true], "{context}");
+            let operations = operation_rows(&store.readers.get());
+            let digests = store.replication_snapshot().unwrap().projection_digests.clone();
+            assert!(digests.contains_key("claim_sources"));
+            assert!(digests.contains_key("custom_registrations"));
+            assert!(digests.contains_key("custom_sources"));
+            assert!(digests.contains_key("custom_dependencies"));
+            store.replay_replication_graph().unwrap();
+            assert_eq!(operation_rows(&store.readers.get()), operations, "{context}");
+            assert_eq!(store.replication_snapshot().unwrap().projection_digests, digests, "{context}");
+        }
+    }
+}

crates/fractal-store/src/store/tests.rs

@@ -2328,8 +2328,10 @@ fn full_replay_logs_every_guard_and_incremental_error_once() {
                     json!({"fields":{"state":"ready"}, "_operation":{"id":"op/malformed"}}),
                 ),
                 "operation-conflict" => (
-                    "harness.observed",
-                    json!({"fields":{"state":"ready"}, "_operation":operation}),
+                    "work.nudged",
+                    json!({"fields":{"attempt":1,"agent":"agent/node.test",
+                        "message":"message/invented-nudge","idle_since_unix_ms":"1",
+                        "reason":"invented control","waits":[]}, "_operation":operation}),
                 ),
                 "incremental-error:internal" => ("intent.desired", json!({})),
                 _ => ("harness.observed", json!({"fields":{"state":"ready"}})),