feat(branch): start a branch at a graph commit

changelog.d/branch-create-at-a-commit.added.md

@@ -0,0 +1,14 @@
+- `branch create` starts a branch at a past graph commit: `--from-commit` on
+  the CLI, `from commit "<id>"` in the `branch create` statement (GQ language
+  version 2.3) and `from_commit` on `POST /branches`. The new branch holds that
+  commit's rows and history, so a fix or revert can be tried where a change was
+  made and merged back with the commit as its base. A commit `cleanup` no
+  longer retains (`410`), one written under an earlier schema or before the
+  storage upgrade, one whose branch was deleted, an unpublished commit and an
+  unattached snapshot (`409`) start no branch; the new `branch_start_refusal` error detail names the
+  reason. Every branch creation receipt now carries `from_commit_id`, the
+  commit the new branch starts at, and the CLI prints it:
+  `created branch <name> from <branch> at <commit-id>`. See
+  [Start a branch at a commit][branch-at-commit].
+
+[branch-at-commit]: ../docs/user/branching/index.md#start-a-branch-at-a-commit

crates/omnigraph-api-types/src/lib.rs

@@ -390,8 +390,13 @@ pub struct SnapshotOutput {
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct BranchCreateRequest {
-    /// Parent branch to fork from. Defaults to `main`.
+    /// Parent branch to fork from; the new branch starts at its head.
+    /// Mutually exclusive with `from_commit`. Defaults to `main`.
     pub from: Option<String>,
+    /// A published graph commit id to start the new branch at, with that
+    /// commit's data and history. Mutually exclusive with `from`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub from_commit: Option<String>,
     /// Name of the new branch. Must not already exist.
     pub name: String,
 }
@@ -402,6 +407,10 @@ pub struct BranchCreateOutput {
     pub from: String,
     pub name: String,
     pub actor_id: Option<String>,
+    /// The graph commit the new branch starts at: `from`'s head when the
+    /// branch was created, or the commit `from_commit` named. For a commit
+    /// start, `from` is the branch that wrote the commit.
+    pub from_commit_id: String,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
@@ -612,8 +621,14 @@ pub struct ReadPageOutput {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum BranchOutcomeOutput {
-    /// `name` was forked off `from`.
-    Created { from: String, name: String },
+    /// `name` was forked off `from` at the commit `from_commit_id`: the
+    /// head of `from`, or the commit a `from commit "<id>"` start named,
+    /// which `from` wrote.
+    Created {
+        from: String,
+        name: String,
+        from_commit_id: String,
+    },
     /// `name` was removed.
     Deleted { name: String },
     /// `source` was merged into `target`; `merge` is the three-way result,
@@ -2106,6 +2121,42 @@ pub struct ChangeDiffRefusalOutput {
     pub type_name: Option<String>,
 }
 
+/// Why a branch could not start at the graph commit `from_commit` named.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum BranchStartRefusalReason {
+    /// HTTP 410: cleanup reclaimed data the commit's snapshot names.
+    NotRetained,
+    /// HTTP 409: the commit was written under an earlier schema.
+    SchemaChanged,
+    /// HTTP 409: the commit was written before the storage upgrade.
+    EarlierStorageFormat,
+    /// HTTP 409: the branch that wrote the commit was deleted, and may have
+    /// been created again under the same name since.
+    BranchDeleted,
+    /// HTTP 409: the commit is unpublished; publish it first.
+    UnpublishedCommit,
+    /// HTTP 409: the commit is an unattached snapshot, never branch history.
+    UnattachedSnapshot,
+    /// A reason added by a newer server; the branch was not created.
+    #[serde(other)]
+    Unknown,
+}
+
+/// A branch create refused at the graph commit `from_commit` named; nothing
+/// was created.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct BranchStartRefusalOutput {
+    pub reason: BranchStartRefusalReason,
+    pub graph_commit_id: String,
+    /// For `not_retained`, the graph type whose stored version cleanup reclaimed.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub type_name: Option<String>,
+    /// For `branch_deleted`, the branch that wrote the commit.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub branch: Option<String>,
+}
+
 /// A selected full-text index cannot safely serve the current analyzer (HTTP 409).
 /// This is not a retryable write conflict: an operator must rebuild the live
 /// branch's indexes. Historical snapshots stay unchanged; branch old content
@@ -2282,6 +2333,10 @@ pub struct ErrorOutput {
     /// commit or an unprovable schema boundary).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub change_diff_refusal: Option<ChangeDiffRefusalOutput>,
+    /// Set with HTTP 409, or 410 for `not_retained`, when `branch create`
+    /// cannot start a branch at the commit `from_commit` named.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub branch_start_refusal: Option<BranchStartRefusalOutput>,
     /// Set with HTTP 409 when a selected full-text index requires an explicit
     /// rebuild before search can succeed. Unlike a write-authority conflict,
     /// this condition is not cleared by retrying. This additive discriminator
@@ -2318,6 +2373,7 @@ impl ErrorOutput {
             blob_precondition_failure: None,
             change_feed_gap: None,
             change_diff_refusal: None,
+            branch_start_refusal: None,
             full_text_index_rebuild_required: None,
             full_text_index_required: None,
             diagnostic: None,

crates/omnigraph-catalog/src/legacy.rs

@@ -479,7 +479,7 @@ pub(crate) enum RetiredHead {
 /// `Some` when `dataset`'s version is read from the legacy history of a root
 /// that has one: it predates native commit records, or is a supported pending
 /// conversion. The value says whether it carries the pending key.
-fn legacy_version(dataset: &Dataset) -> Option<bool> {
+pub(crate) fn legacy_version(dataset: &Dataset) -> Option<bool> {
     let metadata = &dataset.schema().metadata;
     let fenced = metadata.contains_key(UPGRADE_PENDING_KEY);
     let legacy = match metadata

crates/omnigraph-catalog/src/lib.rs

@@ -2385,15 +2385,65 @@ impl ManifestCoordinator {
     #[doc(hidden)]
     pub async fn create_branch(&mut self, name: &str) -> Result<()> {
         crate::branch_names::ensure_logical_branch_name(name)?;
-        let mut ds = self.dataset.clone();
+        let version = self.version();
+        Self::create_branch_from(self.dataset.clone(), name, version).await
+    }
+
+    /// Create the logical branch `name` as a fresh native incarnation whose
+    /// first `__manifest` version is a clone of this branch's version
+    /// `version`, the version whose publish made `graph_commit_id` the head:
+    /// the new branch starts with that commit's table pins, lineage and
+    /// schema contract, exactly as a branch created while the commit was the
+    /// head. The version must hold `graph_commit_id` as its head in the
+    /// format this build writes; a version only the legacy history serves is
+    /// refused. Every refusal here precedes any effect.
+    #[doc(hidden)]
+    pub async fn create_branch_at(
+        &mut self,
+        name: &str,
+        version: u64,
+        graph_commit_id: &str,
+    ) -> Result<()> {
+        crate::branch_names::ensure_logical_branch_name(name)?;
+        let dataset = if version == self.version() {
+            self.dataset.clone()
+        } else {
+            self.dataset
+                .checkout_version(version)
+                .await
+                .map_err(|error| OmniError::storage(error).before_effect())?
+        };
+        if crate::legacy::legacy_version(&dataset).is_some() {
+            return Err(OmniError::branch_start_refused(
+                graph_commit_id,
+                crate::error::BranchStartRefusal::EarlierStorageFormat,
+            )
+            .before_effect());
+        }
+        let head = state::read_manifest_rows_projected(&dataset)
+            .await
+            .map_err(OmniError::before_effect)?
+            .head;
+        if head.graph_commit_id != graph_commit_id {
+            return Err(OmniError::manifest_internal(format!(
+                "`__manifest` version {version} of {} holds head commit '{}', not '{graph_commit_id}' \
+                 its record names; no branch was created",
+                self.native_branch.as_deref().unwrap_or("main"),
+                head.graph_commit_id
+            ))
+            .before_effect());
+        }
+        Self::create_branch_from(dataset, name, version).await
+    }
+
+    /// Create `name` as a clone of `ds`, checked out at `version`.
+    async fn create_branch_from(mut ds: Dataset, name: &str, version: u64) -> Result<()> {
         crate::branch_control::ensure_manifest_branch_create_namespace(&ds, name)
             .await
             .map_err(OmniError::before_effect)?;
         let native =
             crate::branch_names::native_branch_name(name, &crate::branch_names::mint_incarnation());
-        match crate::branch_control::create_branch_recoverably(&mut ds, &native, self.version())
-            .await?
-        {
+        match crate::branch_control::create_branch_recoverably(&mut ds, &native, version).await? {
             crate::branch_control::BranchCreateOutcome::Created => Ok(()),
             crate::branch_control::BranchCreateOutcome::RefAlreadyExists => Err(
                 OmniError::manifest_conflict(format!("branch '{}' already exists", name)),

crates/omnigraph-cli/src/app_tests.rs

@@ -1082,7 +1082,11 @@ async fn maintenance_delivery_failure_preserves_one_rebuild_and_its_actor() {
                 .await
                 .unwrap();
             client
-                .branch_create_from("main", "search", &mut Vec::new())
+                .branch_create(
+                    omnigraph::db::ReadTarget::branch("main"),
+                    "search",
+                    &mut Vec::new(),
+                )
                 .await
                 .unwrap();
             let main_before = client

crates/omnigraph-cli/src/client.rs

@@ -26,7 +26,7 @@ use std::sync::Arc;
 
 use color_eyre::Result;
 use color_eyre::eyre::{bail, eyre};
-use omnigraph::db::{Omnigraph, ReadTarget};
+use omnigraph::db::{Omnigraph, ReadTarget, SnapshotId};
 use omnigraph::settings::{SessionSettings, SettingId, SettingValue, Source};
 use omnigraph::{BLOB_READ_RANGE_MAX_BYTES, BlobContent, Session};
 use omnigraph_api_types::{
@@ -43,7 +43,7 @@ use omnigraph_api_types::{
     schema_apply_output, show_read_output, snapshot_payload,
 };
 use omnigraph_compiler::catalog::Catalog;
-use omnigraph_compiler::query::ast::{BranchRead, BranchWrite, CommitWrite};
+use omnigraph_compiler::query::ast::{BranchCreateFrom, BranchRead, BranchWrite, CommitWrite};
 use omnigraph_compiler::query::parser::parse_query;
 use reqwest::header::{CONTENT_RANGE, RANGE};
 use reqwest::{Method, StatusCode};
@@ -1544,12 +1544,22 @@ impl GraphClient {
                 let query_name = write.statement_name().to_string();
                 let (branch, commit, outcome) = match write {
                     BranchWrite::Create { name, from } => {
-                        let from = from.unwrap_or_else(|| "main".to_string());
-                        self.branch_create_from(&from, &name, stderr).await?;
+                        let start = match from {
+                            Some(BranchCreateFrom::Commit(commit)) => {
+                                ReadTarget::snapshot(SnapshotId::new(commit))
+                            }
+                            Some(BranchCreateFrom::Branch(branch)) => ReadTarget::branch(branch),
+                            None => ReadTarget::branch("main"),
+                        };
+                        let created = self.branch_create(start, &name, stderr).await?;
                         (
                             name.clone(),
                             None,
-                            BranchOutcomeOutput::Created { from, name },
+                            BranchOutcomeOutput::Created {
+                                from: created.from,
+                                name,
+                                from_commit_id: created.from_commit_id,
+                            },
                         )
                     }
                     BranchWrite::Delete { name } => {
@@ -2116,9 +2126,10 @@ impl GraphClient {
         }
     }
 
-    pub(crate) async fn branch_create_from(
+    /// Create `name` at `start`: a branch's head, or a published graph commit.
+    pub(crate) async fn branch_create(
         &self,
-        from: &str,
+        start: ReadTarget,
         name: &str,
         stderr: &mut dyn Write,
     ) -> Result<BranchCreateOutput> {
@@ -2133,9 +2144,17 @@ impl GraphClient {
                     http,
                     Method::POST,
                     remote_url(base_url, &["branches"], &[])?,
-                    Some(serde_json::to_value(BranchCreateRequest {
-                        from: Some(from.to_string()),
-                        name: name.to_string(),
+                    Some(serde_json::to_value(match start {
+                        ReadTarget::Branch(branch) => BranchCreateRequest {
+                            from: Some(branch),
+                            from_commit: None,
+                            name: name.to_string(),
+                        },
+                        ReadTarget::Snapshot(commit) => BranchCreateRequest {
+                            from: None,
+                            from_commit: Some(commit.as_str().to_string()),
+                            name: name.to_string(),
+                        },
                     })?),
                     token.as_deref(),
                 )
@@ -2144,13 +2163,13 @@ impl GraphClient {
             GraphClient::Embedded { uri, actor } => {
                 let db = Self::open_write(uri, stderr).await?;
                 let actor = actor.as_deref();
-                db.branch_create_from_as(ReadTarget::branch(from), name, actor)
-                    .await?;
+                let created = db.branch_create_from_as(start, name, actor).await?;
                 Ok(BranchCreateOutput {
                     uri: uri.clone(),
-                    from: from.to_string(),
+                    from: created.from,
                     name: name.to_string(),
                     actor_id: actor.map(String::from),
+                    from_commit_id: created.from_commit_id,
                 })
             }
             GraphClient::Catalogue => Err(catalogue_scope_refusal("branch create from")),

crates/omnigraph-cli/src/command_outcome.rs

@@ -231,6 +231,7 @@ impl Failure {
             || output.external_blob_source.is_some()
             || !output.merge_conflicts.is_empty()
             || output.diagnostic.is_some()
+            || output.branch_start_refusal.is_some()
         {
             Action::Refresh
         } else {
@@ -293,6 +294,7 @@ pub(crate) fn is_plain_refusal(output: &ErrorOutput) -> bool {
         && output.blob_precondition_failure.is_none()
         && output.change_feed_gap.is_none()
         && output.change_diff_refusal.is_none()
+        && output.branch_start_refusal.is_none()
         && output.full_text_index_rebuild_required.is_none()
         && output.full_text_index_required.is_none()
         && output.diagnostic.is_none()

crates/omnigraph-cli/src/commands/history/args.rs

@@ -50,8 +50,14 @@ pub(crate) enum BranchCommand {
         /// Graph URI
         #[arg(long)]
         uri: Option<String>,
+        /// Branch to fork; the new branch starts at its head. Defaults to
+        /// `main`.
         #[arg(long)]
         from: Option<String>,
+        /// A published graph commit id to start the branch at, with that
+        /// commit's data and history, instead of a branch head.
+        #[arg(long, conflicts_with = "from")]
+        from_commit: Option<String>,
         name: String,
         #[arg(long)]
         json: bool,

crates/omnigraph-cli/src/commands/history/branch.rs

@@ -5,6 +5,7 @@ use crate::client::{self, GraphClient};
 use crate::helpers::{confirm_destructive, resolve_branch, write_target};
 use crate::output::{write_json, write_merge_base_human};
 use color_eyre::eyre::Result;
+use omnigraph::db::{ReadTarget, SnapshotId};
 use std::io::Write;
 
 pub(crate) async fn run_branch(
@@ -18,6 +19,7 @@ pub(crate) async fn run_branch(
         BranchCommand::Create {
             uri,
             from,
+            from_commit,
             name,
             json,
         } => {
@@ -33,22 +35,25 @@ pub(crate) async fn run_branch(
                 stderr,
             )
             .await?;
-            let from = resolve_branch(from, None, "main");
+            let start = match from_commit {
+                Some(commit) => ReadTarget::snapshot(SnapshotId::new(commit)),
+                None => ReadTarget::branch(resolve_branch(from, None, "main")),
+            };
             write_target(
                 scope.quiet,
                 "branch create",
                 client.uri(),
                 client.is_remote(),
                 stderr,
             )?;
-            let payload = client.branch_create_from(&from, &name, stderr).await?;
+            let payload = client.branch_create(start, &name, stderr).await?;
             if json {
                 write_json(&payload, stdout)?;
             } else {
                 writeln!(
                     stdout,
-                    "created branch {} from {}",
-                    payload.name, payload.from
+                    "created branch {} from {} at {}",
+                    payload.name, payload.from, payload.from_commit_id
                 )?;
             }
         }