fix(cortex): allocate DAG sequence numbers per author, seeded from the store

`DagAction::seq` is documented as a per-author sequence, and the DAG store keys
its author index on `(author, seq)`. The node allocated it from one
process-global counter that always started at 1, which broke that contract in
two ways — the second one loses history.

Gaps: interleaved writes by different authors (the node itself, the tool
surface) drew from the same counter, so each author's sequence came out full of
holes. Nothing in-tree depends on contiguity, but the number no longer means
what the type says it means.

Eviction: because the counter restarted at 1 on every process start, a node
reopening a persistent DAG re-issued sequence numbers the store already held.
`put` inserts into the author index unconditionally, so each new action silently
REPLACED the pre-restart action holding that key. Measured on a sled-backed
store: four writes across a restart leave four actions on disk and a two-entry
author chain. The evicted actions are still stored, still signed and still
linked by hash — but they disappear from every view built on the author chain:
`/api/v1/dag/chain`, the git-provenance list, the approval list. History that is
present and unreadable is not much better than history that is missing.

`AppState::next_dag_seq` now allocates per author and seeds an author's counter
from its highest recorded `seq` the first time it is asked, so numbering
continues the chain instead of colliding with it. All seven allocation sites —
the shared triple write path, delete, the two Raft-routed handlers, the GraphQL
mutations, custom actions, review approvals and git provenance — go through it.

Regression tests cover both symptoms: a restart against a persistent store must
leave all writes visible with a contiguous 1..4 sequence, and two authors
writing through the same node must each own a gapless sequence.

Author: ApiliumDevTeam

crates/aingle_cortex/src/graphql/resolvers.rs

@@ -176,9 +176,7 @@ impl MutationRoot {
                     .dag_author
                     .clone()
                     .unwrap_or_else(|| aingle_graph::NodeId::named("node:local"));
-                let dag_seq = state
-                    .dag_seq_counter
-                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                let dag_seq = state.next_dag_seq(&dag_author, Some(dag_store));
                 let parents = dag_store.tips().unwrap_or_default();
 
                 let mut action = aingle_graph::dag::DagAction {
@@ -238,9 +236,7 @@ impl MutationRoot {
                         .dag_author
                         .clone()
                         .unwrap_or_else(|| aingle_graph::NodeId::named("node:local"));
-                    let dag_seq = state
-                        .dag_seq_counter
-                        .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                    let dag_seq = state.next_dag_seq(&dag_author, Some(dag_store));
                     let parents = dag_store.tips().unwrap_or_default();
 
                     let mut action = aingle_graph::dag::DagAction {

crates/aingle_cortex/src/rest/dag.rs

@@ -640,15 +640,15 @@ pub async fn post_create_dag_action(
             .unwrap_or_else(|| aingle_graph::NodeId::named("node:local"))
     };
 
-    let dag_seq = state
-        .dag_seq_counter
-        .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-
     let graph = state.graph.read().await;
     let dag_store = graph
         .dag_store()
         .ok_or_else(|| Error::Internal("DAG not enabled".into()))?;
 
+    // Allocated after the store is in hand: the number continues this author's
+    // recorded chain instead of restarting at 1 and evicting its own history.
+    let dag_seq = state.next_dag_seq(&dag_author, Some(dag_store));
+
     let parents = dag_store
         .tips()
         .map_err(|e| Error::Internal(e.to_string()))?;

crates/aingle_cortex/src/rest/triples.rs

@@ -180,14 +180,14 @@ pub async fn create_triple(
         let dag_author = state.dag_author.clone().unwrap_or_else(|| {
             aingle_graph::NodeId::named(&format!("node:{}", state.cluster_node_id.unwrap_or(0)))
         });
-        let dag_seq = state
-            .dag_seq_counter
-            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-
-        // Get current tips
-        let parents = {
+        // Get current tips, and a sequence number that continues this author's
+        // recorded chain rather than restarting at 1 and evicting its own history
+        // from the (author, seq) index.
+        let (parents, dag_seq) = {
             let graph = state.graph.read().await;
-            graph.dag_tips().unwrap_or_default()
+            let parents = graph.dag_tips().unwrap_or_default();
+            let seq = state.next_dag_seq(&dag_author, graph.dag_store());
+            (parents, seq)
         };
 
         let mut action = aingle_graph::dag::DagAction {
@@ -459,19 +459,17 @@ pub async fn delete_triple(
         let dag_author = state.dag_author.clone().unwrap_or_else(|| {
             aingle_graph::NodeId::named(&format!("node:{}", state.cluster_node_id.unwrap_or(0)))
         });
-        let dag_seq = state
-            .dag_seq_counter
-            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-
-        let (parents, subject_for_dag) = {
+        let (parents, subject_for_dag, dag_seq) = {
             let graph = state.graph.read().await;
             let tips = graph.dag_tips().unwrap_or_default();
             let subj = graph
                 .get(&triple_id)
                 .ok()
                 .flatten()
                 .map(|t| crate::service::triples::dag_subject_name(&t.subject));
-            (tips, subj)
+            // Continues this author's recorded chain instead of restarting at 1.
+            let seq = state.next_dag_seq(&dag_author, graph.dag_store());
+            (tips, subj, seq)
         };
 
         let mut action = aingle_graph::dag::DagAction {

crates/aingle_cortex/src/service/git_provenance.rs

@@ -117,9 +117,7 @@ pub async fn record_git_provenance(
         .dag_author
         .clone()
         .unwrap_or_else(|| aingle_graph::NodeId::named("node:local"));
-    let seq = state
-        .dag_seq_counter
-        .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    let seq = state.next_dag_seq(&author, Some(dag_store));
     let parents = dag_store.tips().unwrap_or_default();
     let at = chrono::Utc::now();
     let summary = match &git_ref.branch {

crates/aingle_cortex/src/service/review.rs

@@ -52,9 +52,7 @@ pub async fn record_approval(
         .dag_author
         .clone()
         .unwrap_or_else(|| aingle_graph::NodeId::named("node:local"));
-    let seq = state
-        .dag_seq_counter
-        .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    let seq = state.next_dag_seq(&author, Some(dag_store));
     let parents = dag_store.tips().unwrap_or_default();
     let approved_at = chrono::Utc::now();
 

crates/aingle_cortex/src/service/triples.rs

@@ -70,9 +70,7 @@ pub(crate) fn record_insert_action(
     origin: Option<&str>,
 ) -> Result<()> {
     let dag_author = dag_action_author(state, origin);
-    let dag_seq = state
-        .dag_seq_counter
-        .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    let dag_seq = state.next_dag_seq(&dag_author, Some(dag_store));
     let parents = dag_store.tips().unwrap_or_default();
 
     let mut action = aingle_graph::dag::DagAction {
@@ -399,9 +397,7 @@ pub async fn delete_triple(
         if deleted {
             if let Some(dag_store) = graph.dag_store() {
                 let dag_author = dag_action_author(state, origin);
-                let dag_seq = state
-                    .dag_seq_counter
-                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                let dag_seq = state.next_dag_seq(&dag_author, Some(dag_store));
                 let parents = dag_store.tips().unwrap_or_default();
 
                 let mut action = aingle_graph::dag::DagAction {
@@ -781,6 +777,119 @@ mod tests {
         }
     }
 
+    /// A restart must continue each author's sequence, not restart it. The DAG
+    /// store keys its author index on `(author, seq)`, so re-issuing a number the
+    /// store already holds evicts the older action from that index: it survives
+    /// on disk and in the parent chain, but vanishes from every author-chain view
+    /// — history that is present and unreadable.
+    #[cfg(feature = "dag")]
+    #[tokio::test]
+    async fn a_restart_continues_the_author_sequence_instead_of_evicting_history() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("g.sled");
+        let p = path.to_str().unwrap().to_string();
+
+        let author = NodeId::named("node:local");
+
+        // First run: two writes.
+        {
+            let state = AppState::with_db_path(&p, None).unwrap();
+            {
+                let mut graph = state.graph.write().await;
+                graph.enable_dag_persistent(&p).unwrap();
+            }
+            create_triple(&state, req("ex:a", "ex:p", "ex:1"), None, None)
+                .await
+                .unwrap();
+            create_triple(&state, req("ex:b", "ex:p", "ex:2"), None, None)
+                .await
+                .unwrap();
+            state
+                .graph
+                .read()
+                .await
+                .dag_store()
+                .unwrap()
+                .flush()
+                .unwrap();
+        }
+
+        // Second run against the same store: a fresh in-memory allocator.
+        {
+            let state = AppState::with_db_path(&p, None).unwrap();
+            {
+                let mut graph = state.graph.write().await;
+                graph.enable_dag_persistent(&p).unwrap();
+            }
+            create_triple(&state, req("ex:c", "ex:p", "ex:3"), None, None)
+                .await
+                .unwrap();
+            create_triple(&state, req("ex:d", "ex:p", "ex:4"), None, None)
+                .await
+                .unwrap();
+
+            let graph = state.graph.read().await;
+            let store = graph.dag_store().unwrap();
+            let chain = store.chain(&author, 100).unwrap();
+            assert_eq!(
+                chain.len(),
+                4,
+                "all four writes must remain visible in the author chain after a restart"
+            );
+            let mut seqs: Vec<u64> = chain.iter().map(|a| a.seq).collect();
+            seqs.sort_unstable();
+            assert_eq!(
+                seqs,
+                vec![1, 2, 3, 4],
+                "an author's sequence must be contiguous, not restarted or gapped"
+            );
+        }
+    }
+
+    /// Two authors writing through the same node must not eat each other's
+    /// numbers: `seq` is documented and indexed per author.
+    #[cfg(feature = "dag")]
+    #[tokio::test]
+    async fn interleaved_authors_get_independent_sequences() {
+        let state = AppState::with_db_path(":memory:", None).unwrap();
+        {
+            let mut graph = state.graph.write().await;
+            graph.enable_dag();
+        }
+
+        for i in 0..3 {
+            create_triple(
+                &state,
+                req("ex:alice", "ex:knows", &format!("ex:local{i}")),
+                None,
+                None,
+            )
+            .await
+            .unwrap();
+            create_triple(
+                &state,
+                req("ex:alice", "ex:knows", &format!("ex:tool{i}")),
+                None,
+                Some("mcp"),
+            )
+            .await
+            .unwrap();
+        }
+
+        let graph = state.graph.read().await;
+        let store = graph.dag_store().unwrap();
+        for author in ["node:local", "mcp"] {
+            let mut seqs: Vec<u64> = store
+                .chain(&NodeId::named(author), 100)
+                .unwrap()
+                .iter()
+                .map(|a| a.seq)
+                .collect();
+            seqs.sort_unstable();
+            assert_eq!(seqs, vec![1, 2, 3], "{author} must own a gapless sequence");
+        }
+    }
+
     #[tokio::test]
     async fn list_triples_returns_inserted() {
         let state = AppState::with_db_path(":memory:", None).unwrap();

crates/aingle_cortex/src/state.rs

@@ -110,9 +110,22 @@ pub struct AppState {
     /// This node's author identity for DAG actions.
     #[cfg(feature = "dag")]
     pub dag_author: Option<aingle_graph::NodeId>,
-    /// Per-author monotonic sequence counter for DAG actions.
+    /// Next DAG sequence number to hand out, per author.
+    ///
+    /// A `DagAction`'s `seq` is documented — and indexed — as a *per-author*
+    /// number: the DAG store keys its author index on `(author, seq)`. A single
+    /// process-wide counter therefore broke in two ways. Interleaved writes by
+    /// different authors (the node itself, the tool surface) tore holes in each
+    /// author's sequence; and because the counter always restarted at 1, a
+    /// restart against a persistent DAG re-issued sequences the store had already
+    /// used, so each new action silently EVICTED the pre-restart action holding
+    /// that key — the actions survived on disk and in the chain, but dropped out
+    /// of every author-chain view (`/dag/chain`, git provenance, approvals).
+    ///
+    /// Allocate through [`AppState::next_dag_seq`], which seeds an author's
+    /// counter from the store the first time it is asked.
     #[cfg(feature = "dag")]
-    pub dag_seq_counter: std::sync::Arc<std::sync::atomic::AtomicU64>,
+    pub dag_seq: std::sync::Arc<std::sync::Mutex<std::collections::HashMap<String, u64>>>,
     /// Ed25519 signing key for DAG actions (mandatory in production).
     #[cfg(feature = "dag")]
     pub dag_signing_key: Option<std::sync::Arc<aingle_graph::dag::DagSigningKey>>,
@@ -189,7 +202,7 @@ impl AppState {
             #[cfg(feature = "dag")]
             dag_author: None,
             #[cfg(feature = "dag")]
-            dag_seq_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
+            dag_seq: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             #[cfg(feature = "dag")]
             dag_signing_key: None,
             #[cfg(feature = "mcp")]
@@ -248,7 +261,7 @@ impl AppState {
             #[cfg(feature = "dag")]
             dag_author: None,
             #[cfg(feature = "dag")]
-            dag_seq_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
+            dag_seq: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             #[cfg(feature = "dag")]
             dag_signing_key: None,
             #[cfg(feature = "mcp")]
@@ -307,7 +320,7 @@ impl AppState {
             #[cfg(feature = "dag")]
             dag_author: None,
             #[cfg(feature = "dag")]
-            dag_seq_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
+            dag_seq: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             #[cfg(feature = "dag")]
             dag_signing_key: None,
             #[cfg(feature = "mcp")]
@@ -483,7 +496,7 @@ impl AppState {
             #[cfg(feature = "dag")]
             dag_author: None,
             #[cfg(feature = "dag")]
-            dag_seq_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
+            dag_seq: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             #[cfg(feature = "dag")]
             dag_signing_key: None,
             #[cfg(feature = "mcp")]
@@ -668,6 +681,44 @@ impl AppState {
             .unwrap_or_default()
     }
 
+    /// Allocate the next DAG sequence number for `author`.
+    ///
+    /// The DAG store keys its author index on `(author, seq)`, so a sequence
+    /// number must be unique *within its author* and must not collide with one
+    /// already on disk. The first request for an author therefore seeds from the
+    /// store — the author's highest recorded `seq` plus one — instead of assuming
+    /// an empty history; every later request increments in memory. Reusing a
+    /// number a persistent DAG already holds silently evicts the older action
+    /// from the author index, hiding it from every chain view even though it is
+    /// still stored and still linked.
+    ///
+    /// `dag_store` is `None` only where the DAG is not enabled, in which case
+    /// numbering starts at 1 because there is no history to continue.
+    ///
+    /// A poisoned lock is recovered rather than panicked on: a mutation must not
+    /// be lost because some unrelated writer panicked while holding this map.
+    #[cfg(feature = "dag")]
+    pub fn next_dag_seq(
+        &self,
+        author: &aingle_graph::NodeId,
+        dag_store: Option<&aingle_graph::dag::DagStore>,
+    ) -> u64 {
+        let key = format!("{author}");
+        let mut map = self
+            .dag_seq
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let seq = match map.get(&key) {
+            Some(next) => *next,
+            None => dag_store
+                .and_then(|s| s.chain(author, 1).ok())
+                .and_then(|c| c.first().map(|a| a.seq + 1))
+                .unwrap_or(1),
+        };
+        map.insert(key, seq + 1);
+        seq
+    }
+
     /// Sets the workspace working-copy root that the note-write tools resolve
     /// paths against and that [`crate::service::ingest`] confines every ingest
     /// to.