fix(github): per-observation operation keys for github.event (#9101 producer side)

crates/fractal-github/src/idle_holds.rs

@@ -30,9 +30,12 @@ const QUEUE_EXIT_STATE_SQL: &str = "SELECT json_extract(body, '$.fields'), accep
              ORDER BY length(accepted_at_unix_ms) DESC,accepted_at_unix_ms DESC,store_index DESC
              LIMIT ?2";
 
-/// When a thread's `merge-queue-left` events were written: one subject per exit.
+/// One canonical first observation per exit: another poller's observation is not a new exit.
 const QUEUE_EXIT_EVENT_SQL: &str = "SELECT accepted_at_unix_ms FROM claims INDEXED BY claims_subject_kind_accepted_index
-             WHERE subject>=?1 AND subject<?2 AND kind='github.event'";
+             WHERE subject>=?1 AND subject<?2 AND kind='github.event'
+               AND id=(SELECT request.id FROM claims request
+                       WHERE request.subject=claims.subject AND request.kind='github.event'
+                       ORDER BY CANONICAL_ASC(request) LIMIT 1)";
 
 // Each mailbox alias contributes at most one row; the aggregate never sorts retained wakes.
 const LAST_WAKE_SQL: &str = "SELECT MAX(sent_at) FROM (
@@ -198,7 +201,7 @@ impl IdleHoldsStore for Store {
         let exited = fractal_schema::github_plugin::event_subject(&item, "merge-queue-left")
             .map(|subject| -> Result<Option<u128>> {
                 let upper = format!("{subject}0");
-                let mut events = connection.prepare_cached(/* sqlite-large-value: allow-for-now lv-h13f5ed40c752 owner=fractal_github::idle_holds TODO=trace-statement-and-value-representation */ QUEUE_EXIT_EVENT_SQL)?;
+                let mut events = connection.prepare_cached(/* sqlite-large-value: allow-for-now lv-haa1264e7b371 owner=fractal_github::idle_holds TODO=trace-statement-and-value-representation */ &smallclaims::store::canonical::canonical_sql(QUEUE_EXIT_EVENT_SQL))?;
                 let times = events
                     .query_map(params![subject, upper], |row| row.get::<_, String>(0))?
                     .collect::<rusqlite::Result<Vec<_>>>()?;
@@ -409,6 +412,28 @@ mod tests {
             "at": "2026-10-30T12:00:00.000Z"}));
         assert_eq!(store.thread_left_merge_queue(&thread, 0).unwrap(), Some(event), "the newest exit event");
         assert_eq!(store.thread_left_merge_queue(&thread, event).unwrap(), None);
+        // A second observation can arrive by replication despite local Once admission.
+        // Simulate that raw history with the append primitive, preserving the old payload.
+        {
+            let mut connection = store.connection.write();
+            let transaction = connection.transaction().unwrap();
+            smallclaims::store::append_claim_tx(
+                &transaction, &store.origin,
+                "github-event/github/acme/garden/pull-request/12/merge-queue-left/2",
+                "github.event", None,
+                &json!({"fields": {
+                    "version": 1, "item": item, "event": "merge-queue-left",
+                    "observer": "observer/another-poller", "at": "2026-11-11T10:00:00Z",
+                    "head_sha": "another-head", "detail": {}
+                }}),
+                &[], None,
+            ).unwrap();
+            transaction.commit().unwrap();
+        }
+        assert_eq!(store.thread_left_merge_queue(&thread, 0).unwrap(), Some(event),
+            "another observation must not move the canonical exit");
+        assert_eq!(store.thread_left_merge_queue(&thread, event).unwrap(), None,
+            "another observation must not wake an idle holder again");
         append(&store, item, "vcs.item.state", compact(json!({"state": "closed", "merged": true})));
         assert_eq!(store.thread_left_merge_queue(&thread, 0).unwrap(), None, "a merged thread did not stall");
     }

crates/fractal-reconcile/src/reconcile.rs

@@ -19409,14 +19409,15 @@ fn mirror_observation(
             .admit(subject, fractal_schema::github_plugin::EVENT, &fields)
             .map_err(anyhow::Error::new)
             .and_then(|()| {
+                let key = github_events::observation_key(subject, &fields)?;
                 store.append_claim(&ClaimInput {
                     subject: subject.to_owned(),
                     kind: fractal_schema::github_plugin::EVENT.into(),
                     actor: None,
                     fields,
                     evidence: Vec::new(),
                     expected_subject: None,
-                    idempotency_key: Some(format!("github-event:{subject}")),
+                    idempotency_key: Some(key),
                 }).map_err(|error| anyhow::anyhow!(error.message))
             });
         if let Err(error) = written {
@@ -43106,6 +43107,8 @@ subscription "issues" { observer "observer/repo"; on "issues"; to "agent/example
             }
         }
         let store = Arc::new(Store::open_memory("node").unwrap());
+        const EVENT_FLEET: &str = "018f6f0d-4a5d-7b8c-9d0e-123456789abc";
+        store.bind_fleet(EVENT_FLEET).unwrap();
         apply_source(
             &store,
             r#"version 2
@@ -43163,6 +43166,36 @@ subscription "comments" { observer "observer/repo"; on "comments"; to "agent/exa
         poll(plugin.clone(), listing("open", "Orchid", &[1]), "c2").await;
         assert_eq!(delivered(), 1);
         assert_eq!(count("resource.observed"), snapshots + 1, "a comment writes a snapshot");
+        // An older poller raced this look, before receiving its event. Its global operation
+        // and node-local payload must not cause another listing row or subscription delivery.
+        let old = Store::open_memory("old-poller").unwrap();
+        old.bind_fleet(EVENT_FLEET).unwrap();
+        let subject = "github-event/repo/issue/1/comment/1";
+        old.append_claim(&ClaimInput {
+            subject: subject.into(), kind: fractal_schema::github_plugin::EVENT.into(),
+            actor: None,
+            fields: BTreeMap::from([
+                ("version".into(), serde_json::json!(1)),
+                ("item".into(), serde_json::json!(item)),
+                ("event".into(), serde_json::json!("comment")),
+                ("observer".into(), serde_json::json!("observer/old")),
+                ("at".into(), serde_json::json!("2026-11-11T10:00:01Z")),
+                ("detail".into(), serde_json::json!({"author": "fern"})),
+            ]),
+            evidence: Vec::new(), expected_subject: None,
+            idempotency_key: Some(format!("github-event:{subject}")),
+        }).unwrap();
+        let exchange = old.export_replication_exchange_answering(
+            EVENT_FLEET, &store.replication_inventory().unwrap(),
+            &store.replication_signature_requests().unwrap(),
+        ).unwrap();
+        store.receive_replication_exchange(&old.origin, EVENT_FLEET, &exchange).unwrap();
+        store.validate_replication_backlog().unwrap();
+        store.project_replication_backlog().unwrap();
+        assert_eq!(store.claims_for(subject, None).unwrap().len(), 2);
+        assert_eq!(store.github_events("resource/repo", Some(1), 10).unwrap().len(), 1);
+        assert_eq!(store.idempotency_conflicts(10).unwrap().0, 0);
+        assert_eq!(delivered(), 1, "event provenance does not deliver mail");
         poll(plugin.clone(), listing("open", "Orchid", &[1]), "c3").await;
         assert_eq!(delivered(), 1, "a known comment is not delivered again");
 

crates/fractal-reconcile/src/reconcile/github_events.rs

@@ -5,6 +5,17 @@
 //! mirror is new, wakes nobody.
 
 use serde_json::{Map, Value, json};
+use std::collections::BTreeMap;
+
+/// The operation names this observation, not the global event subject. Keep every payload
+/// field in the fingerprint: even the same observer on two pollers can see different facts.
+pub(super) fn observation_key(subject: &str, fields: &BTreeMap<String, Value>) -> anyhow::Result<String> {
+    let fields = fields.iter()
+        .map(|(name, value)| (name, smallclaims::hash::canonical_json_value(value)))
+        .collect::<BTreeMap<_, _>>();
+    let digest = smallclaims::hash::canonical_hash(&("github-event-observation.v1", subject, fields))?;
+    Ok(format!("github-event-observation:v1:{digest}"))
+}
 
 /// One thing that happened to one item.
 #[derive(Clone, Debug, PartialEq)]
@@ -171,6 +182,98 @@ pub fn view(event: &Event, item: &str, subject: &str, at: &str) -> Value {
 mod tests {
     use super::*;
 
+    fn event_claim(observer: &str, at: &str) -> crate::model::ClaimInput {
+        let subject = "github-event/github/acme/garden/issue/7/closed";
+        let fields = BTreeMap::from([
+            ("version".into(), json!(1)),
+            ("item".into(), json!("resource/github/acme/garden/issue/7")),
+            ("event".into(), json!("closed")),
+            ("observer".into(), json!(observer)),
+            ("at".into(), json!(at)),
+            ("detail".into(), json!({"state": "closed"})),
+        ]);
+        let key = observation_key(subject, &fields).unwrap();
+        crate::model::ClaimInput {
+            subject: subject.into(), kind: fractal_schema::github_plugin::EVENT.into(),
+            actor: None, fields, evidence: Vec::new(), expected_subject: None,
+            idempotency_key: Some(key),
+        }
+    }
+
+    #[test]
+    fn observation_keys_cover_every_field_and_canonicalize_nested_objects() {
+        let input = event_claim("observer/a", "2026-11-11T10:00:00Z");
+        let original = observation_key(&input.subject, &input.fields).unwrap();
+        for (name, value) in [
+            ("observer", json!("observer/b")),
+            ("at", json!("2026-11-11T10:00:01Z")),
+            ("head_sha", json!("abc")),
+            ("detail", json!({"state": "open"})),
+            ("event", json!("reopened")),
+            ("version", json!(2)),
+            ("item", json!("resource/github/acme/garden/issue/8")),
+        ] {
+            let mut fields = input.fields.clone();
+            fields.insert(name.into(), value);
+            assert_ne!(observation_key(&input.subject, &fields).unwrap(), original, "{name}");
+        }
+        assert_ne!(observation_key("github-event/github/acme/garden/issue/8/closed", &input.fields).unwrap(), original);
+        let mut fields = input.fields;
+        fields.insert("detail".into(), serde_json::from_str(r#"{"b":{"y":2,"x":1},"a":0}"#).unwrap());
+        let first = observation_key(&input.subject, &fields).unwrap();
+        fields.insert("detail".into(), serde_json::from_str(r#"{"a":0,"b":{"x":1,"y":2}}"#).unwrap());
+        assert_eq!(observation_key(&input.subject, &fields).unwrap(), first);
+    }
+
+    /// Independent writers race before either sees the other's event. Old global keys and
+    /// new observation keys must fold by subject, regardless of replication arrival order.
+    #[test]
+    fn concurrent_event_observations_and_legacy_claims_fold_once_without_conflicts() {
+        use crate::store::Store;
+        const FLEET: &str = "018f6f0d-4a5d-7b8c-9d0e-123456789abc";
+        for (other_observer, legacy) in [("observer/b", false), ("observer/a", false), ("observer/b", true)] {
+            let a = Store::open_memory("alder").unwrap();
+            let b = Store::open_memory("birch").unwrap();
+            let forward = Store::open_memory("cedar").unwrap();
+            let reverse = Store::open_memory("dogwood").unwrap();
+            for store in [&a, &b, &forward, &reverse] {
+                store.bind_fleet(FLEET).unwrap();
+            }
+            let input = event_claim("observer/a", "2026-11-11T10:00:00Z");
+            let first = a.append_claim(&input).unwrap();
+            assert_eq!(a.append_claim(&input).unwrap().id, first.id, "exact retry");
+            let mut other = event_claim(other_observer, "2026-11-11T10:00:01Z");
+            if legacy {
+                other.idempotency_key = Some(format!("github-event:{}", other.subject));
+            }
+            let second = b.append_claim(&other).unwrap();
+            assert_ne!(first.body["_operation"]["id"], second.body["_operation"]["id"]);
+            for (target, sources) in [(&forward, [&a, &b]), (&reverse, [&b, &a])] {
+                for source in sources {
+                    let exchange = source.export_replication_exchange_answering(
+                        FLEET, &target.replication_inventory().unwrap(),
+                        &target.replication_signature_requests().unwrap(),
+                    ).unwrap();
+                    target.receive_replication_exchange(&source.origin, FLEET, &exchange).unwrap();
+                    target.validate_replication_backlog().unwrap();
+                    target.project_replication_backlog().unwrap();
+                }
+                assert_eq!(target.idempotency_conflicts(10).unwrap().0, 0);
+                assert_eq!(target.claims_for(&input.subject, None).unwrap().len(), 2, "provenance survives");
+                let events = target.github_events("resource/github/acme/garden", Some(7), 10).unwrap();
+                assert_eq!(events.len(), 1, "one event, not one row per observation");
+                assert_eq!(events, target.github_events("resource/github/acme/garden", None, 1).unwrap());
+                target.replay_replication_graph().unwrap();
+                assert_eq!(events, target.github_events("resource/github/acme/garden", Some(7), 10).unwrap());
+            }
+            assert_eq!(
+                forward.github_events("resource/github/acme/garden", None, 10).unwrap(),
+                reverse.github_events("resource/github/acme/garden", None, 10).unwrap(),
+                "canonical fold is independent of arrival",
+            );
+        }
+    }
+
     fn facts(value: Value) -> Map<String, Value> {
         value.as_object().unwrap().clone()
     }

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

@@ -541,10 +541,15 @@ impl Store {
         };
         let upper = prefix_successor(&prefix).unwrap_or_default();
         let connection = self.readers.get();
-        let mut statement = connection.prepare_cached(
+        // Old global operations and new observation operations share the event subject.
+        // Select its canonical first observation before applying the listing limit.
+        let mut statement = connection.prepare_cached(/* sqlite-large-value: allow-for-now lv-h05417f7bc8ab owner=fractal_store::store::resources TODO=trace-statement-and-value-representation */ &canonical_sql(
             "SELECT subject, json_extract(body, '$.fields'), accepted_at_unix_ms FROM claims
-             WHERE subject>=?1 AND subject<?2 AND kind='github.event'",
-        )?;
+             WHERE subject>=?1 AND subject<?2 AND kind='github.event'
+               AND id=(SELECT request.id FROM claims request
+                       WHERE request.subject=claims.subject AND request.kind='github.event'
+                       ORDER BY CANONICAL_ASC(request) LIMIT 1)",
+        ))?;
         let mut events = statement
             .query_map(params![prefix, upper], |row| {
                 Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?, row.get::<_, String>(2)?))
@@ -570,7 +575,9 @@ impl Store {
                 })))
             })
             .collect::<Vec<_>>();
-        events.sort_by_key(|(at, _)| std::cmp::Reverse(*at));
+        events.sort_by(|(left_at, left), (right_at, right)| {
+            right_at.cmp(left_at).then_with(|| left["subject"].as_str().cmp(&right["subject"].as_str()))
+        });
         events.truncate(limit);
         Ok(events.into_iter().map(|(_, event)| event).collect())
     }