Checkpoint source 7 workflows and in-progress diagram authoring redesign

Preserve current core, source, REST, desktop, recovery, documentation and validation work, including AGENTS.md and the development summary. This is a work-in-progress checkpoint: the new compact palette has outstanding compilation errors and native acceptance; earlier passing gates retain their documented scope.

crates/oa-server/src/lib.rs

@@ -3,6 +3,9 @@ pub mod error;
 mod routes;
 pub mod store;
 
+#[cfg(test)]
+mod preparation_http_tests;
+
 use axum::{
     Router,
     extract::{DefaultBodyLimit, Request, State},

crates/oa-server/src/preparation_http_tests.rs

@@ -0,0 +1,151 @@
+use crate::{
+    app,
+    store::{Store, StoredProject},
+};
+use oa_core::{Id, Project};
+use reqwest::{Client, StatusCode};
+use serde_json::{Value, json};
+use std::{sync::Arc, time::Duration};
+use tokio::{sync::oneshot, time::timeout};
+
+#[tokio::test]
+async fn prepared_source_http_rechecks_revision_without_holding_catalog() {
+    let directory = tempfile::tempdir().unwrap();
+    let path = directory.path().to_owned();
+    let store = Arc::new(
+        tokio::task::spawn_blocking(move || Store::open(path))
+            .await
+            .unwrap()
+            .unwrap(),
+    );
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base = format!("http://{}/api/v1", listener.local_addr().unwrap());
+    let (stop, done) = oneshot::channel();
+    let router = app(store.clone(), None);
+    let task = tokio::spawn(async move {
+        axum::serve(listener, router)
+            .with_graceful_shutdown(async {
+                let _ = done.await;
+            })
+            .await
+            .unwrap();
+    });
+    let client = Client::new();
+    let created = client
+        .post(format!("{base}/projects"))
+        .json(&json!({"name":"Initial review"}))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(created.status(), StatusCode::CREATED);
+    let etag = created.headers()["etag"].to_str().unwrap().to_owned();
+    let summary: Value = created.json().await.unwrap();
+    let id: Id = summary["resource_id"].as_str().unwrap().parse().unwrap();
+    let project_url = format!("{base}/projects/{id}");
+    let initial: StoredProject = client
+        .get(&project_url)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let (entered, release) = store.pause_next_replacement(id, etag.clone());
+    // Real parsing, legacy admission and source preparation all run. Runtime
+    // cost is irrelevant: the gate proves the exact pre-commit phase instead.
+    let source = "arcadia 2; project prepared \"Prepared stale head\" { oa_activity root \"Root\" {} oa_activity child \"Child\" { parent @root; } }";
+    let proposed_client = client.clone();
+    let proposed_url = format!("{project_url}/source");
+    let proposed_etag = etag.clone();
+    let proposed = tokio::spawn(async move {
+        proposed_client
+            .put(proposed_url)
+            .header("if-match", proposed_etag)
+            .header("content-type", "text/plain")
+            .body(source)
+            .send()
+            .await
+            .unwrap()
+    });
+    let evidence = timeout(Duration::from_secs(10), entered)
+        .await
+        .expect("source did not reach validated pre-commit phase")
+        .unwrap();
+    assert_eq!(evidence.model.name, "Prepared stale head");
+    assert_eq!(evidence.model.elements.len(), 2);
+    let compiled =
+        oa_core::source_project::compile_bundle(&evidence.bundle, crate::store::SOURCE_LIMITS)
+            .unwrap();
+    assert_eq!(compiled.decoded.project, evidence.model);
+    assert!(
+        !proposed.is_finished(),
+        "prepared request must still overlap the winner"
+    );
+    let catalog = timeout(
+        Duration::from_secs(2),
+        client.get(format!("{base}/projects")).send(),
+    )
+    .await
+    .expect("catalog locked during admitted candidate lifetime")
+    .unwrap();
+    assert_eq!(catalog.status(), StatusCode::OK);
+    let winner = timeout(
+        Duration::from_secs(2),
+        client
+            .patch(&project_url)
+            .header("if-match", &etag)
+            .json(&json!({"name":"Concurrent accepted review"}))
+            .send(),
+    )
+    .await
+    .expect("prepared worker held the commit lock")
+    .unwrap();
+    assert_eq!(winner.status(), StatusCode::OK);
+    let winner_etag = winner.headers()["etag"].to_str().unwrap().to_owned();
+    assert!(
+        !proposed.is_finished(),
+        "stale candidate must await explicit release"
+    );
+    release.release();
+    let loser = timeout(Duration::from_secs(10), proposed)
+        .await
+        .expect("released candidate did not finish")
+        .unwrap();
+    assert_eq!(loser.status(), StatusCode::PRECONDITION_FAILED);
+    let failure: Value = loser.json().await.unwrap();
+    assert_eq!(failure["code"], "revision_conflict");
+    assert_eq!(failure["current_revision"], 2);
+    let committed_response = client.get(&project_url).send().await.unwrap();
+    assert_eq!(
+        committed_response.headers()["etag"].to_str().unwrap(),
+        winner_etag
+    );
+    let committed: StoredProject = committed_response
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let mut expected_model: Project = initial.model;
+    expected_model.name = "Concurrent accepted review".into();
+    assert_eq!(committed.model, expected_model);
+    assert_eq!(committed.revision, 2);
+    assert!(committed.model.elements.is_empty());
+    let installed: StoredProject = serde_json::from_slice(
+        &std::fs::read(directory.path().join(format!("{id}.json"))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(installed.model, committed.model);
+    assert_eq!(installed.source_project, committed.source_project);
+    assert_eq!(installed.revision, 2);
+    assert_eq!(installed.resource_id, committed.resource_id);
+    assert_eq!(installed.migration, committed.migration);
+    let _ = stop.send(());
+    drop(client);
+    timeout(Duration::from_secs(5), task)
+        .await
+        .unwrap()
+        .unwrap();
+}

crates/oa-server/src/routes.rs

@@ -1069,10 +1069,11 @@ fn check_transition_request(request: &oa_core::transitions::TransitionRequest) -
     if request.sources.len() > 100
         || request.mappings.len() > 200
         || request.communication_endpoints.len() > 200
+        || request.operation_omissions.len() > 200
         || request.resolutions.len() > 400
     {
         return Err(ApiError::bad_request(
-            "Transition requests allow at most 100 sources, 200 mappings or communication endpoint decisions and 400 field resolutions.",
+            "Transition requests allow at most 100 sources, 200 mappings, communication endpoint decisions or operation omissions and 400 field resolutions.",
         ));
     }
     Ok(())
@@ -1195,6 +1196,7 @@ fn check_semantic_request(request: &oa_core::semantic::SemanticTransitionRequest
     if transition.sources.len() > 100
         || transition.mappings.len() > 200
         || transition.communication_endpoints.len() > 200
+        || transition.operation_omissions.len() > 200
         || transition.resolutions.len() > 400
         || request.traces.len() > 100
         || request.references.len() > 5_000
@@ -1204,7 +1206,7 @@ fn check_semantic_request(request: &oa_core::semantic::SemanticTransitionRequest
         || request.accepted_impacts.len() > 200
     {
         return Err(ApiError::bad_request(
-            "Semantic transitions allow at most 100 sources/traces/reviews, 200 mappings/communication endpoint decisions/candidates/accepted impacts, 400 prose resolutions and 5,000 reference mappings or combined facet resolutions/resets.",
+            "Semantic transitions allow at most 100 sources/traces/reviews, 200 mappings/communication endpoint decisions/operation omissions/candidates/accepted impacts, 400 prose resolutions and 5,000 reference mappings or combined facet resolutions/resets.",
         ));
     }
     Ok(())

crates/oa-server/src/store.rs

@@ -13,6 +13,10 @@ use std::{
     sync::{Mutex, MutexGuard},
 };
 
+#[cfg(test)]
+#[path = "store_preparation_gate.rs"]
+pub(crate) mod preparation_gate;
+
 pub const MAX_MODEL_BYTES: usize = 8 * 1024 * 1024;
 pub const MAX_SOURCE_PROJECT_BYTES: usize = 8 * 1024 * 1024;
 pub const MAX_OBJECTS: usize = 5_000;
@@ -89,6 +93,8 @@ pub enum StoreError {
 
 pub struct Store {
     inner: Mutex<StoreInner>,
+    #[cfg(test)]
+    preparation_gate: Mutex<Option<preparation_gate::Armed>>,
     // Keep the OS lock alive for the store's entire lifetime, including requests.
     _directory_lock: File,
 }
@@ -242,6 +248,8 @@ impl Store {
     ) -> ApiResult<(StoredProject, bool)> {
         let current = self.checked(id, expected)?;
         let prepared = PreparedProject::replacement(&current, decoded)?;
+        #[cfg(test)]
+        self.wait_after_preparation(id, expected, &prepared)?;
         self.lock()?.replace(id, expected, prepared)
     }
 
@@ -381,6 +389,8 @@ impl Store {
             }
         }
         Ok(Self {
+            #[cfg(test)]
+            preparation_gate: Mutex::new(None),
             inner: Mutex::new(StoreInner {
                 directory,
                 projects,

crates/oa-server/src/store_preparation_gate.rs

@@ -0,0 +1,141 @@
+//! Private synchronization for real-HTTP tests; absent from production builds.
+use super::{ApiError, ApiResult, Id, PreparedProject, Project, SourceBundle, Store};
+use std::{sync::mpsc, time::Duration};
+use tokio::sync::oneshot;
+
+pub(crate) struct PreparedEvidence {
+    pub model: Project,
+    pub bundle: SourceBundle,
+}
+
+pub(crate) struct Armed {
+    id: Id,
+    expected: String,
+    entered: oneshot::Sender<PreparedEvidence>,
+    resume: mpsc::Receiver<bool>,
+}
+
+/// Dropping an unfinished test cancels the worker instead of stranding it.
+pub(crate) struct ReleaseOnDrop(Option<mpsc::Sender<bool>>);
+
+impl ReleaseOnDrop {
+    pub(crate) fn release(mut self) {
+        self.0.take().unwrap().send(true).unwrap();
+    }
+}
+
+impl Drop for ReleaseOnDrop {
+    fn drop(&mut self) {
+        if let Some(sender) = self.0.take() {
+            let _ = sender.send(false);
+        }
+    }
+}
+
+impl Store {
+    pub(crate) fn pause_next_replacement(
+        &self,
+        id: Id,
+        expected: String,
+    ) -> (oneshot::Receiver<PreparedEvidence>, ReleaseOnDrop) {
+        let (entered, receiver) = oneshot::channel();
+        let (resume, wait) = mpsc::channel();
+        let mut slot = self.preparation_gate.lock().unwrap();
+        assert!(slot.is_none(), "one gate per store instance");
+        *slot = Some(Armed {
+            id,
+            expected,
+            entered,
+            resume: wait,
+        });
+        (receiver, ReleaseOnDrop(Some(resume)))
+    }
+
+    pub(super) fn wait_after_preparation(
+        &self,
+        id: Id,
+        expected: &str,
+        prepared: &PreparedProject,
+    ) -> ApiResult<()> {
+        let armed = {
+            let mut slot = self.preparation_gate.lock().unwrap();
+            if slot
+                .as_ref()
+                .is_some_and(|gate| gate.id == id && gate.expected == expected)
+            {
+                slot.take()
+            } else {
+                None
+            }
+        };
+        let Some(armed) = armed else {
+            return Ok(());
+        };
+        // Neither the catalog nor the gate mutex is held while the real model
+        // worker waits. Evidence comes from the already validated candidate.
+        armed
+            .entered
+            .send(PreparedEvidence {
+                model: prepared.model().clone(),
+                bundle: prepared.bundle().clone(),
+            })
+            .map_err(|_| ApiError::internal("Preparation test observer disappeared."))?;
+        match armed.resume.recv_timeout(Duration::from_secs(15)) {
+            Ok(true) => Ok(()),
+            _ => Err(ApiError::internal(
+                "Preparation test cancelled or timed out.",
+            )),
+        }
+    }
+}
+
+#[tokio::test]
+async fn preparation_gate_drop_cancels_without_installing_candidate() {
+    let directory = tempfile::tempdir().unwrap();
+    let path = directory.path().to_owned();
+    let (store, initial) = tokio::task::spawn_blocking(move || {
+        let store = std::sync::Arc::new(Store::open(path).unwrap());
+        let initial = store.create(Project::empty()).unwrap();
+        (store, initial)
+    })
+    .await
+    .unwrap();
+    let original_bytes = std::fs::read(
+        directory
+            .path()
+            .join(format!("{}.json", initial.resource_id)),
+    )
+    .unwrap();
+    let (entered, release) = store.pause_next_replacement(initial.resource_id, initial.etag());
+    let mut model = initial.model.clone();
+    model.name = "Must not be installed".into();
+    let worker_store = store.clone();
+    let id = initial.resource_id;
+    let etag = initial.etag();
+    let worker = tokio::task::spawn_blocking(move || {
+        worker_store.replace_decoded(
+            id,
+            &etag,
+            oa_core::persistence::DecodeResult {
+                project: model,
+                migration: None,
+            },
+        )
+    });
+    tokio::time::timeout(Duration::from_secs(10), entered)
+        .await
+        .unwrap()
+        .unwrap();
+    drop(release);
+    let error = tokio::time::timeout(Duration::from_secs(2), worker)
+        .await
+        .expect("dropping the test guard must release its worker")
+        .unwrap()
+        .unwrap_err();
+    assert_eq!(error.detail, "Preparation test cancelled or timed out.");
+    assert_eq!(store.get(id).unwrap().model, initial.model);
+    assert_eq!(
+        std::fs::read(directory.path().join(format!("{id}.json"))).unwrap(),
+        original_bytes
+    );
+}