fix: treat a graph copy as finalized only when it carries committee pre-signatures

node/src/aggregates/graph.rs

@@ -19,6 +19,7 @@ use crate::engine_bridge::record_event_in;
 use crate::messages::{GOATMessage, GOATMessageContent, verify_init_graph_signature};
 use crate::metrics_service::node_metrics_state;
 use crate::middleware::get_topic_name;
+use crate::utils::stored_definition_committee_pre_signed;
 use anyhow::anyhow;
 use async_trait::async_trait;
 use bitvm_lib::actors::Actor;
@@ -1169,7 +1170,17 @@ impl AggregateHandler for GraphAggregate {
                         )));
                     }
                 };
-                if row.is_none() {
+                // A finalized lookup also runs when only the operator's copy is
+                // stored, whatever the status.
+                let wanted = match (&row, &pull.lookup) {
+                    (None, _) => true,
+                    (Some(_), GraphLookup::Id { finalized: true, .. }) => {
+                        stored_definition_committee_pre_signed(ctx.storage, &graph_id).await?
+                            != Some(true)
+                    }
+                    (Some(_), _) => false,
+                };
+                if wanted {
                     let params = bincode::serialize(&pull.lookup)
                         .map_err(|error| anyhow::anyhow!("encode graph lookup: {error}"))?;
                     self.enqueue_setup_job(ctx, JOB_FETCH_GRAPH, &params);

node/src/aggregates/graph_fetch.rs

@@ -7,8 +7,9 @@
 use crate::middleware::get_graph::{GetGraphRequest, GetGraphResponse};
 use crate::middleware::publisher::NetworkPublisher;
 use crate::utils::{
-    load_validated_graph_definition, parse_graph_raw_data, store_finalized_graph_if_needed,
-    store_operator_presigned_graph, validate_graph_instance_parameters,
+    load_validated_graph_definition, parse_graph_raw_data, raw_definition_committee_pre_signed,
+    store_finalized_graph_if_needed, store_operator_presigned_graph,
+    validate_graph_instance_parameters,
 };
 use anyhow::{Context, bail};
 use bitvm_lib::operator::verify_graph_operator_pre_signatures;
@@ -18,7 +19,6 @@ use client::goat_chain::GOATClient;
 use libp2p::PeerId;
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
-use store::GraphStatus;
 use store::localdb::LocalDB;
 use tracing::{info, warn};
 use uuid::Uuid;
@@ -73,13 +73,15 @@ pub async fn serve_graph_request(
     {
         return Ok(GetGraphResponse::not_found());
     }
-    // Only the operator's copy is stored while the graph is OperatorPresigned.
-    if request.finalized && row.status == GraphStatus::OperatorPresigned.to_string() {
-        return Ok(GetGraphResponse::not_found());
-    }
     let Some(raw) = storage.find_graph_raw_data(&row.graph_id).await? else {
         return Ok(GetGraphResponse::not_found());
     };
+    // A finalized copy is one with committee pre-signatures. The status says
+    // nothing about it: a Goat event moves an operator-only copy past
+    // OperatorPresigned too.
+    if request.finalized && !raw_definition_committee_pre_signed(&raw.raw_data)? {
+        return Ok(GetGraphResponse::not_found());
+    }
     Ok(GetGraphResponse::found(row.instance_id, row.graph_id, raw.raw_data))
 }
 
@@ -346,3 +348,65 @@ mod admission_tests {
         assert!(db.acquire().await.unwrap().find_graph(&graph_id).await.unwrap().is_none());
     }
 }
+
+#[cfg(test)]
+mod serve_tests {
+    use super::*;
+    use store::{Graph, GraphRawData, GraphStatus, GraphStatusSource};
+
+    #[tokio::test]
+    async fn a_finalized_request_is_served_only_a_copy_with_committee_pre_signatures() {
+        let dir = tempfile::tempdir().unwrap();
+        let db = LocalDB::new_with_busy_timeout(
+            &format!("sqlite:{}", dir.path().join("serve.db").display()),
+            false,
+            store::DEFAULT_BUSY_TIMEOUT,
+        )
+        .await;
+        db.migrate().await;
+        for committee_pre_signed in [false, true] {
+            let (instance_id, graph_id) = (Uuid::new_v4(), Uuid::new_v4());
+            let mut storage = db.acquire().await.unwrap();
+            storage
+                .upsert_graph_definition(&Graph {
+                    graph_id,
+                    instance_id,
+                    status: GraphStatus::OperatorPresigned.to_string(),
+                    definition_hash: "definition".to_string(),
+                    kickoff_index: i64::from(committee_pre_signed),
+                    ..Default::default()
+                })
+                .await
+                .unwrap();
+            // Past OperatorPresigned either way: a Goat event moves the status
+            // of an operator-only copy too.
+            storage
+                .transition_graph_status(
+                    instance_id,
+                    graph_id,
+                    GraphStatus::OperatorDataPushed,
+                    GraphStatusSource::ChainReconcile,
+                    None,
+                )
+                .await
+                .unwrap();
+            let raw =
+                serde_json::json!({ "committee_pre_signed": committee_pre_signed }).to_string();
+            storage
+                .upsert_graph_raw_data(
+                    instance_id,
+                    GraphRawData { graph_id, raw_data: raw, created_at: 0, updated_at: 0 },
+                    "definition",
+                )
+                .await
+                .unwrap();
+            drop(storage);
+
+            let finalized = GetGraphRequest::by_id(instance_id, graph_id, true);
+            let served = serve_graph_request(&db, &finalized).await.unwrap();
+            assert_eq!(served.graph.is_some(), committee_pre_signed);
+            let any = GetGraphRequest::by_id(instance_id, graph_id, false);
+            assert!(serve_graph_request(&db, &any).await.unwrap().graph.is_some());
+        }
+    }
+}

node/src/aggregates/graph_tests.rs

@@ -526,6 +526,37 @@ async fn a_definition_waits_for_its_predecessor_and_pulls_it() {
     t.stop().await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_finalized_pull_runs_while_only_the_operator_copy_is_stored() {
+    let t = start(Actor::Verifier, false).await;
+    for (nonce, committee_pre_signed) in [(0, false), (1, true)] {
+        let instance_id = Uuid::new_v4();
+        let graph_id = t.insert_graph(instance_id, nonce, GraphStatus::OperatorDataPushed).await;
+        let raw = serde_json::json!({ "committee_pre_signed": committee_pre_signed }).to_string();
+        let mut storage = t.db.acquire().await.unwrap();
+        storage
+            .upsert_graph_raw_data(
+                instance_id,
+                store::GraphRawData { graph_id, raw_data: raw, created_at: 0, updated_at: 0 },
+                "definition",
+            )
+            .await
+            .unwrap();
+        drop(storage);
+        let lookup = GraphLookup::Id { instance_id, graph_id, finalized: true };
+        let event = graph_pull_event(graph_id, lookup.clone()).unwrap();
+        let event_id = t.submit(event).await;
+        t.wait_inbox(&event_id, EngineInboxState::Applied).await;
+        if committee_pre_signed {
+            settle().await;
+            assert!(t.jobs_of(graph_id).await.iter().all(|job| job.action != JOB_FETCH_GRAPH));
+        } else {
+            t.wait_job(graph_id, JOB_FETCH_GRAPH, EngineJobState::Done).await;
+        }
+    }
+    t.stop().await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn a_value_from_an_unregistered_author_waits_without_a_pull() {
     let t = start_with(Actor::Committee, false, Arc::new(Unacquainted)).await;

node/src/scheduled_tasks/event_watch/gateway.rs

@@ -10,7 +10,7 @@ use crate::clock::current_time_secs;
 use crate::engine_bridge::record_event_in;
 use crate::utils::{
     generate_instance_from_bridge_in_request_event, obsolete_instance_graphs_except,
-    reflect_goat_address, strip_hex_prefix_owned,
+    reflect_goat_address, stored_definition_committee_pre_signed, strip_hex_prefix_owned,
 };
 use alloy::primitives::U256;
 use bitvm_lib::actors::Actor;
@@ -665,10 +665,16 @@ async fn handle_post_graph_data_events<'a>(
                 )
                 .await?;
                 // Watching roles fetch a finalized definition when the posted graph is absent
-                // locally.
-                if matches!(outcome, GraphStatusTransitionOutcome::NotFound)
-                    && matches!(actor, Actor::Verifier | Actor::Watchtower)
-                {
+                // locally. Any role holding only the operator's copy fetches it too: the
+                // status just moved past OperatorPresigned, but the copy still lacks the
+                // committee pre-signatures.
+                let pull = if matches!(outcome, GraphStatusTransitionOutcome::NotFound) {
+                    matches!(actor, Actor::Verifier | Actor::Watchtower)
+                } else {
+                    stored_definition_committee_pre_signed(storage_processor, &graph_id).await?
+                        == Some(false)
+                };
+                if pull {
                     record_event_in(
                         storage_processor,
                         graph_pull_event(
@@ -763,4 +769,56 @@ mod tests {
             );
         }
     }
+
+    /// A graph row with a stored definition; only its committee flag is read.
+    async fn stored(db: &LocalDB, slot: i64, committee_pre_signed: bool) -> (Uuid, Uuid) {
+        let (instance_id, graph_id) = (Uuid::new_v4(), Uuid::new_v4());
+        let mut storage = db.acquire().await.unwrap();
+        storage
+            .upsert_graph_definition(&store::Graph {
+                graph_id,
+                instance_id,
+                status: GraphStatus::OperatorPresigned.to_string(),
+                definition_hash: "definition".to_string(),
+                kickoff_index: slot,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        let raw = serde_json::json!({ "committee_pre_signed": committee_pre_signed }).to_string();
+        storage
+            .upsert_graph_raw_data(
+                instance_id,
+                store::GraphRawData { graph_id, raw_data: raw, created_at: 0, updated_at: 0 },
+                "definition",
+            )
+            .await
+            .unwrap();
+        (instance_id, graph_id)
+    }
+
+    #[tokio::test]
+    async fn a_posted_graph_held_only_as_the_operator_copy_is_pulled_finalized() {
+        let (db, _dir) = db().await;
+        let mut slot = 0;
+        for actor in [Actor::Verifier, Actor::Watchtower, Actor::Operator, Actor::Committee] {
+            for committee_pre_signed in [false, true] {
+                slot += 1;
+                let (instance_id, graph_id) = stored(&db, slot, committee_pre_signed).await;
+                let mut tx = db.start_transaction().await.unwrap();
+                handle_post_graph_data_events(&actor, &mut tx, vec![posted(instance_id, graph_id)])
+                    .await
+                    .unwrap();
+                tx.commit().await.unwrap();
+                let mut storage = db.acquire().await.unwrap();
+                let row = storage.find_graph(&graph_id).await.unwrap().unwrap();
+                assert_eq!(row.status, GraphStatus::OperatorDataPushed.to_string());
+                let pull = storage
+                    .find_engine_inbox_event(&format!("graph:{graph_id}:pull:finalized"))
+                    .await
+                    .unwrap();
+                assert_eq!(pull.is_some(), !committee_pre_signed, "{actor}");
+            }
+        }
+    }
 }

node/src/utils/graph_store.rs

@@ -427,6 +427,28 @@ async fn confirm_finalized_graph_definition(
     }
 }
 
+/// Whether a raw definition carries committee pre-signatures, read without
+/// parsing the graph. The store paths verify the signatures before they write
+/// a definition with the flag set, so the flag is what "finalized" means.
+pub fn raw_definition_committee_pre_signed(raw_data: &str) -> Result<bool> {
+    #[derive(serde::Deserialize)]
+    struct Flags {
+        committee_pre_signed: bool,
+    }
+    Ok(serde_json::from_str::<Flags>(raw_data)?.committee_pre_signed)
+}
+
+/// Whether the stored definition is finalized; `None` when none is stored.
+pub async fn stored_definition_committee_pre_signed(
+    storage_processor: &mut StorageProcessor<'_>,
+    graph_id: &Uuid,
+) -> Result<Option<bool>> {
+    let Some(raw) = storage_processor.find_graph_raw_data(graph_id).await? else {
+        return Ok(None);
+    };
+    raw_definition_committee_pre_signed(&raw.raw_data).map(Some)
+}
+
 /// Parse raw graph data JSON string to SimplifiedBitvmGcGraph using spawn_blocking
 /// to handle large data and potential stack overflow issues
 pub async fn parse_graph_raw_data(