Implement RFC-022 unified graph write protocol (#343)

* Implement unified graph write protocol

* Preserve recovery error wire compatibility

crates/omnigraph-server/src/handlers.rs

@@ -788,8 +788,9 @@ pub(crate) async fn run_query(
         (status = 400, description = "Bad request", body = ErrorOutput),
         (status = 401, description = "Unauthorized", body = ErrorOutput),
         (status = 403, description = "Forbidden", body = ErrorOutput),
-        (status = 409, description = "Merge conflict", body = ErrorOutput),
+        (status = 409, description = "Write-authority conflict", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -837,8 +838,9 @@ pub(crate) async fn server_change(
         (status = 400, description = "Bad request", body = ErrorOutput),
         (status = 401, description = "Unauthorized", body = ErrorOutput),
         (status = 403, description = "Forbidden", body = ErrorOutput),
-        (status = 409, description = "Merge conflict", body = ErrorOutput),
+        (status = 409, description = "Write-authority conflict", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -847,7 +849,8 @@ pub(crate) async fn server_change(
 /// Writes to the named `branch` (defaults to `main`). Mutations are atomic
 /// per call and produce a new commit. Returns counts of nodes and edges
 /// affected. **Destructive**: on success the branch is updated; rejected
-/// mutations may still acquire locks briefly. Returns 409 on merge conflict.
+/// mutations may still acquire locks briefly. Returns 409 when the prepared
+/// write authority changes before effects.
 ///
 /// Pairs with `POST /query` (read-only). The legacy `POST /change` route
 /// has identical semantics and is kept as a deprecated alias.
@@ -904,9 +907,10 @@ pub(crate) fn parse_optional_invoke_body(
         (status = 401, description = "Unauthorized", body = ErrorOutput),
         (status = 403, description = "Forbidden (the inner `change` gate for a stored mutation)", body = ErrorOutput),
         (status = 404, description = "Unknown stored query, or `invoke_query` denied — indistinguishable to a caller without the grant", body = ErrorOutput),
-        (status = 409, description = "Merge conflict", body = ErrorOutput),
+        (status = 409, description = "Stored mutation write-authority conflict", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
         (status = 500, description = "Policy evaluation error (a denial is reported as 404, not 500)", body = ErrorOutput),
+        (status = 503, description = "A stored mutation is blocked by a durable recovery intent", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -1203,25 +1207,11 @@ pub(crate) async fn server_schema_apply(
         .await
         .map_err(ApiError::from_omni)?
     };
-    // Prompt index convergence (iss-848): schema apply records `@index` intent
-    // but defers the physical build. On a long-lived server, materialize it
-    // promptly rather than waiting for the next `optimize` cron — spawned
-    // detached so it never blocks or fails the apply response. Best-effort: a
-    // failure is logged and the index still converges on the next optimize.
-    // The CLI is one-shot, so it has no equivalent; its convergence path is the
-    // operator's optimize cadence.
-    if result.applied {
-        let engine = Arc::clone(&handle.engine);
-        tokio::spawn(async move {
-            if let Err(err) = engine.ensure_indices().await {
-                tracing::warn!(
-                    target: "omnigraph::server",
-                    error = %err,
-                    "post-apply ensure_indices failed; indexes will converge on the next optimize",
-                );
-            }
-        });
-    }
+    // Physical indexes are derived state. Schema apply records intent only;
+    // explicit `ensure_indices` / `optimize` maintenance owns convergence on
+    // every surface, including a long-lived server. Keeping the handler free
+    // of detached physical writes also makes a successful response describe
+    // the complete effect envelope of this request.
     Ok(Json(schema_apply_output(handle.uri.as_str(), result)))
 }
 
@@ -1313,7 +1303,9 @@ async fn run_ingest(
         (status = 400, description = "Bad request", body = ErrorOutput),
         (status = 401, description = "Unauthorized", body = ErrorOutput),
         (status = 403, description = "Forbidden", body = ErrorOutput),
+        (status = 409, description = "Prepared load authority changed before effects", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -1357,7 +1349,9 @@ pub(crate) async fn server_load(
         (status = 400, description = "Bad request", body = ErrorOutput),
         (status = 401, description = "Unauthorized", body = ErrorOutput),
         (status = 403, description = "Forbidden", body = ErrorOutput),
+        (status = 409, description = "Prepared load authority changed before effects", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -1437,6 +1431,7 @@ pub(crate) async fn server_branch_list(
         (status = 403, description = "Forbidden", body = ErrorOutput),
         (status = 409, description = "Branch already exists", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -1518,6 +1513,7 @@ pub(crate) struct BranchPath {
         (status = 403, description = "Forbidden", body = ErrorOutput),
         (status = 404, description = "Branch not found", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]
@@ -1579,6 +1575,7 @@ pub(crate) async fn server_branch_delete(
         (status = 403, description = "Forbidden", body = ErrorOutput),
         (status = 409, description = "Merge conflict", body = ErrorOutput),
         (status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+        (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
     ),
     security(("bearer_token" = [])),
 )]

crates/omnigraph-server/src/lib.rs

@@ -286,10 +286,12 @@ impl Write for ExportStreamWriter {
 #[derive(Debug)]
 pub struct ApiError {
     status: StatusCode,
-    code: ErrorCode,
+    code: Option<ErrorCode>,
     message: String,
     merge_conflicts: Vec<api::MergeConflictOutput>,
     manifest_conflict: Option<api::ManifestConflictOutput>,
+    read_set_conflict: Option<api::ReadSetConflictOutput>,
+    recovery_required: Option<api::RecoveryRequiredOutput>,
 }
 
 impl AppState {
@@ -616,40 +618,48 @@ impl ApiError {
     pub fn unauthorized(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::UNAUTHORIZED,
-            code: ErrorCode::Unauthorized,
+            code: Some(ErrorCode::Unauthorized),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
     pub fn forbidden(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::FORBIDDEN,
-            code: ErrorCode::Forbidden,
+            code: Some(ErrorCode::Forbidden),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
     pub fn bad_request(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::BAD_REQUEST,
-            code: ErrorCode::BadRequest,
+            code: Some(ErrorCode::BadRequest),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
     pub fn not_found(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::NOT_FOUND,
-            code: ErrorCode::NotFound,
+            code: Some(ErrorCode::NotFound),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
@@ -660,30 +670,36 @@ impl ApiError {
     pub fn method_not_allowed(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::METHOD_NOT_ALLOWED,
-            code: ErrorCode::MethodNotAllowed,
+            code: Some(ErrorCode::MethodNotAllowed),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
     pub fn conflict(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::CONFLICT,
-            code: ErrorCode::Conflict,
+            code: Some(ErrorCode::Conflict),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
     pub fn internal(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::INTERNAL_SERVER_ERROR,
-            code: ErrorCode::Internal,
+            code: Some(ErrorCode::Internal),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
@@ -694,10 +710,12 @@ impl ApiError {
     pub fn too_many_requests(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::TOO_MANY_REQUESTS,
-            code: ErrorCode::TooManyRequests,
+            code: Some(ErrorCode::TooManyRequests),
             message: message.into(),
             merge_conflicts: Vec::new(),
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
@@ -715,20 +733,51 @@ impl ApiError {
     fn merge_conflict(conflicts: Vec<api::MergeConflictOutput>) -> Self {
         Self {
             status: StatusCode::CONFLICT,
-            code: ErrorCode::Conflict,
+            code: Some(ErrorCode::Conflict),
             message: summarize_merge_conflicts(&conflicts),
             merge_conflicts: conflicts,
             manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: None,
         }
     }
 
     fn manifest_version_conflict(message: String, details: api::ManifestConflictOutput) -> Self {
         Self {
             status: StatusCode::CONFLICT,
-            code: ErrorCode::Conflict,
+            code: Some(ErrorCode::Conflict),
             message,
             merge_conflicts: Vec::new(),
             manifest_conflict: Some(details),
+            read_set_conflict: None,
+            recovery_required: None,
+        }
+    }
+
+    fn read_set_conflict(message: String, details: api::ReadSetConflictOutput) -> Self {
+        Self {
+            status: StatusCode::CONFLICT,
+            code: Some(ErrorCode::Conflict),
+            message,
+            merge_conflicts: Vec::new(),
+            manifest_conflict: None,
+            read_set_conflict: Some(details),
+            recovery_required: None,
+        }
+    }
+
+    fn recovery_required(message: String, operation_id: String) -> Self {
+        Self {
+            status: StatusCode::SERVICE_UNAVAILABLE,
+            // `ErrorCode` is a closed rolling wire contract. The additive
+            // `recovery_required` field carries the new meaning while older
+            // clients continue to deserialize the otherwise familiar body.
+            code: None,
+            message,
+            merge_conflicts: Vec::new(),
+            manifest_conflict: None,
+            read_set_conflict: None,
+            recovery_required: Some(api::RecoveryRequiredOutput { operation_id }),
         }
     }
 
@@ -752,6 +801,18 @@ impl ApiError {
                             actual,
                         },
                     ),
+                    Some(ManifestConflictDetails::ReadSetChanged {
+                        member,
+                        expected,
+                        actual,
+                    }) => Self::read_set_conflict(
+                        err.message,
+                        api::ReadSetConflictOutput {
+                            member,
+                            expected,
+                            actual,
+                        },
+                    ),
                     _ => Self::conflict(err.message),
                 },
                 ManifestErrorKind::Internal => Self::internal(err.message),
@@ -762,6 +823,13 @@ impl ApiError {
                     .map(api::MergeConflictOutput::from)
                     .collect(),
             ),
+            OmniError::RecoveryRequired {
+                operation_id,
+                reason,
+            } => Self::recovery_required(
+                format!("recovery required for operation {operation_id}: {reason}"),
+                operation_id,
+            ),
             OmniError::Lance(message) => Self::internal(format!("storage: {message}")),
             OmniError::Io(err) => Self::internal(format!("io: {err}")),
             // Engine-layer policy enforcement (MR-722). All denials and
@@ -815,7 +883,7 @@ const RETRY_AFTER_SECONDS: &str = "60";
 impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         let mut headers = axum::http::HeaderMap::new();
-        if matches!(self.code, ErrorCode::TooManyRequests) {
+        if matches!(self.code, Some(ErrorCode::TooManyRequests)) {
             headers.insert(
                 axum::http::header::RETRY_AFTER,
                 axum::http::HeaderValue::from_static(RETRY_AFTER_SECONDS),
@@ -826,15 +894,42 @@ impl IntoResponse for ApiError {
             headers,
             Json(ErrorOutput {
                 error: self.message,
-                code: Some(self.code),
+                code: self.code,
                 merge_conflicts: self.merge_conflicts,
                 manifest_conflict: self.manifest_conflict,
+                read_set_conflict: self.read_set_conflict,
+                recovery_required: self.recovery_required,
             }),
         )
             .into_response()
     }
 }
 
+#[cfg(test)]
+mod api_error_tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn recovery_required_503_omits_closed_error_code() {
+        let response = ApiError::from_omni(OmniError::RecoveryRequired {
+            operation_id: "01JTESTRECOVERY".to_string(),
+            reason: "pending recovery intent".to_string(),
+        })
+        .into_response();
+
+        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let body = axum::body::to_bytes(response.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let error: ErrorOutput = serde_json::from_slice(&body).unwrap();
+        assert_eq!(error.code, None);
+        assert_eq!(
+            error.recovery_required.unwrap().operation_id,
+            "01JTESTRECOVERY"
+        );
+    }
+}
+
 pub fn init_tracing() {
     let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
     let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();