fix(store): never full-replay under the writer on an operation conflict

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

@@ -1208,8 +1208,16 @@ pub fn commit_tx(
 
 /// Rebuild the existing desired and mission projections through exact set member references.
 pub fn project_tx(transaction: &Transaction<'_>) -> Result<(), FractalError> {
+    project_excluding_tx(transaction, &BTreeSet::new())
+}
+
+/// Leave only explicitly degraded set subjects untouched while projecting unrelated sets.
+pub fn project_excluding_tx(
+    transaction: &Transaction<'_>,
+    excluded_subjects: &BTreeSet<String>,
+) -> Result<(), FractalError> {
     for view in selected(transaction, None)? {
-        if !view.blockers.is_empty() {
+        if excluded_subjects.contains(&view.id) || !view.blockers.is_empty() {
             continue;
         } // Existing runtimes hold; callers fence actions below.
         for (s, (m, retired, implicit)) in effective_members(transaction, &view, None)? {

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

@@ -2294,7 +2294,7 @@ fn planning_replication_keeps_malformed_operation_and_unknown_kind_guards() {
             | IncrementalProjection::ReplayWithContext { reason, .. } => {
                 assert_eq!(reason, expected)
             }
-            IncrementalProjection::Projected => panic!("{kind} bypassed {expected}"),
+            IncrementalProjection::Projected | IncrementalProjection::ProjectedWithContext { .. } => panic!("{kind} bypassed {expected}"),
         }
         transaction.rollback().unwrap();
     }
@@ -2308,7 +2308,6 @@ fn full_replay_logs_every_guard_and_incremental_error_once() {
         "frontier-ahead",
         "non-incremental-kind",
         "malformed-operation",
-        "operation-conflict",
         "incremental-error:internal",
     ] {
         FULL_REPLAYS.with(|count| count.set(0));
@@ -2320,22 +2319,17 @@ fn full_replay_logs_every_guard_and_incremental_error_once() {
         {
             let mut connection = store.connection.lock().unwrap();
             let transaction = connection.transaction().unwrap();
-            let operation = json!({"id":"op/replay-test", "request_digest":"digest-a"});
             let (kind, body) = match reason {
                 "non-incremental-kind" => ("work.unknown", json!({"fields":{}})),
                 "malformed-operation" => (
                     "harness.observed",
                     json!({"fields":{"state":"ready"}, "_operation":{"id":"op/malformed"}}),
                 ),
-                "operation-conflict" => (
-                    "harness.observed",
-                    json!({"fields":{"state":"ready"}, "_operation":operation}),
-                ),
                 "incremental-error:internal" => ("intent.desired", json!({})),
                 _ => ("harness.observed", json!({"fields":{"state":"ready"}})),
             };
             // Seed historical input without pre-projecting it or requiring a current schema.
-            let claim = append_claim_record_tx(
+            append_claim_record_tx(
                 &transaction,
                 "node",
                 "agent/node.test",
@@ -2346,9 +2340,6 @@ fn full_replay_logs_every_guard_and_incremental_error_once() {
                 None,
             )
             .unwrap();
-            if reason == "operation-conflict" {
-                transaction.execute(/* sqlite-large-value: allow-for-now lv-h5c80b3d61773 owner=fractal_store::store::tests TODO=bound-or-reference-values-and-reclamation */ "INSERT INTO operations(id, request_digest, canonical_claim_id, state) VALUES ('op/replay-test', 'digest-b', ?1, 'active')", [&claim.id]).unwrap();
-            }
             match reason {
                 "missing-health" => {
                     transaction
@@ -2406,7 +2397,7 @@ fn full_replay_logs_every_guard_and_incremental_error_once() {
             .filter_map(|line| line.strip_prefix("fractal: projection failure detail "))
             .map(|line| serde_json::from_str::<Value>(line).unwrap())
             .collect::<Vec<_>>();
-        if matches!(reason, "incremental-error:internal" | "operation-conflict"
+        if matches!(reason, "incremental-error:internal"
             | "malformed-operation" | "non-incremental-kind") {
             assert_eq!(details.len(), 1, "{reason}");
             assert_eq!(details[0]["subject"], "agent/node.test");
@@ -2416,10 +2407,7 @@ fn full_replay_logs_every_guard_and_incremental_error_once() {
                     .is_some_and(|id| id.len() == 64)
             );
             assert_eq!(details[0]["error_truncated"], false);
-            if reason == "operation-conflict" {
-                assert_eq!(details[0]["operation_id"], "op/replay-test");
-                assert!(details[0]["error"].as_str().unwrap().contains("digest_matches=false"));
-            } else if reason == "incremental-error:internal" {
+            if reason == "incremental-error:internal" {
                 assert!(
                     details[0]["error"]
                         .as_str()
@@ -2543,7 +2531,7 @@ fn simple_replication_registers_unambiguous_event_operations() {
             None,
         )
         .unwrap();
-        assert!(!try_project_all_simple_replication_tx(&transaction, &store.origin).unwrap());
+        assert!(try_project_all_simple_replication_tx(&transaction, &store.origin).unwrap());
         transaction.commit().unwrap();
     }
     assert!(store.project_replication_backlog().unwrap());
@@ -4426,6 +4414,10 @@ fn work_nudged_incremental_guards_and_rollback_preserve_the_committed_frontier()
                 assert_eq!(refusal, "none");
                 assert_eq!(operation_tx(&transaction, "op/nudge-guard").unwrap().unwrap().2, "active");
             }
+            IncrementalProjection::ProjectedWithContext { code, .. } => {
+                assert_eq!(refusal, "operation-conflict");
+                assert_eq!(code, "operation-conflict");
+            }
             IncrementalProjection::Replay(reason)
             | IncrementalProjection::ReplayWithContext { reason, .. } => assert_eq!(reason, refusal),
         }