[Parked] feat(snapshot): E2B alias rebuild semantics on template publish

src/api/proxy.rs

@@ -224,7 +224,18 @@ where
     I: AsRef<ApiImpl> + Send + Sync,
 {
     if !has_routing_header(request.headers()) {
-        return StatusCode::NOT_FOUND.into_response();
+        // Unmatched control-plane route: return the API error envelope so
+        // JSON clients surface "route not found" instead of failing to parse
+        // an empty 404 body.
+        let error = agentenv_http_server::models::Error::new(
+            404,
+            format!(
+                "route not found: {} {}",
+                request.method(),
+                request.uri().path()
+            ),
+        );
+        return (StatusCode::NOT_FOUND, axum::Json(error)).into_response();
     }
     let forward_path = request.uri().path().to_owned();
     with_route_source(
@@ -2120,6 +2131,24 @@ mod tests {
             .unwrap();
 
         assert_eq!(response.status(), StatusCode::NOT_FOUND);
+        let content_type = response
+            .headers()
+            .get(axum::http::header::CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or_default()
+            .to_owned();
+        assert!(
+            content_type.starts_with("application/json"),
+            "unexpected content-type: {content_type}"
+        );
+        let body = response.into_body().collect().await.unwrap().to_bytes();
+        let payload: Value = serde_json::from_slice(&body).unwrap();
+        assert_eq!(payload["code"], 404);
+        let message = payload["message"].as_str().unwrap();
+        assert!(
+            message.contains("route not found: GET /nonexistent/path"),
+            "unexpected message: {message}"
+        );
     }
 
     #[tokio::test]

src/snapshot/repository/backends/oss/repository.rs

@@ -164,25 +164,28 @@ impl SnapshotRepository for OssSnapshotRepository {
                 reason: format!("snapshot '{}' already exists", record.id),
             });
         }
+        // When the alias already points at a live snapshot, leave the binding
+        // untouched so the existing template keeps resolving while the new
+        // build runs; a successful publish moves the alias to the new snapshot
+        // (E2B rebuild semantics).
+        let mut bind_on_create = true;
         if let Some(alias) = record.alias.as_ref() {
             if let Some(existing) = self.load_alias_target(alias.as_ref()).await? {
                 if existing != record.id && self.snapshot_exists(&existing).await? {
-                    return Err(RepositoryError::AliasConflict {
-                        alias: alias.to_string(),
-                        existing,
-                        new_id: record.id.clone(),
-                    });
+                    bind_on_create = false;
                 }
             }
         }
         self.write_record(&record).await?;
-        if let Some(alias) = record.alias.as_ref() {
-            if let Err(error) = self.bind_alias(alias.as_ref(), &record.id).await {
-                let _ = self
-                    .client
-                    .delete(&OssSnapshotArtifactLayout::record_key(&record.id))
-                    .await;
-                return Err(error);
+        if bind_on_create {
+            if let Some(alias) = record.alias.as_ref() {
+                if let Err(error) = self.bind_alias(alias.as_ref(), &record.id, false).await {
+                    let _ = self
+                        .client
+                        .delete(&OssSnapshotArtifactLayout::record_key(&record.id))
+                        .await;
+                    return Err(error);
+                }
             }
         }
         Ok(record)
@@ -287,26 +290,63 @@ impl SnapshotRepository for OssSnapshotRepository {
                 disk_publications: disk_publications.clone(),
             };
 
-            // 5. Bind alias (if present) with conflict detection.
+            // 5. Commit the record before moving the alias. This prevents an
+            //    alias from ever resolving to a snapshot whose catalog record
+            //    has not been published yet. The tradeoff is a crash window:
+            //    dying after the record write but before `bind_alias` leaves a
+            //    committed record whose `alias` field names an alias that still
+            //    resolves to the previous snapshot, so readers of `record.alias`
+            //    (listings) may observe the stale claim until the next rebind.
+            //    Nothing reconciles that state automatically.
+            let previous_record = self.read_record(id).await?;
+            let previous_alias_target = if let Some(alias) = metadata.alias.as_ref() {
+                match self.load_alias_target(alias.as_ref()).await? {
+                    Some(existing) if self.snapshot_exists(&existing).await? => Some(existing),
+                    _ => None,
+                }
+            } else {
+                None
+            };
+            let record = self
+                .write_committed_record(
+                    metadata.id.clone(),
+                    metadata.alias.clone(),
+                    metadata.resources,
+                    committed,
+                    metadata.source.clone(),
+                )
+                .await?;
+
+            // 6. Move the alias only after the new record is readable. If the
+            //    bind fails, restore the pending record and old alias.
             if let Some(ref alias) = metadata.alias {
-                if let Err(e) = self.bind_alias(alias.as_ref(), id).await {
-                    // Best-effort rollback. Content-addressed managed layers are intentionally left
-                    // in place; they are shared across snapshots and require separate GC.
-                    if let Err(error) = self.client.delete_prefix(&layout.artifact_prefix()).await {
-                        warn!(snapshot_id = %id, error = %error, "failed to roll back snapshot artifacts after alias bind failure");
+                if let Err(error) = self.bind_alias(alias.as_ref(), id, true).await {
+                    self.restore_alias_after_failed_bind(
+                        alias.as_ref(),
+                        id,
+                        previous_alias_target.as_ref(),
+                    )
+                    .await;
+                    self.restore_record_after_failed_publish(id, previous_record.as_ref())
+                        .await;
+                    return Err(error);
+                }
+                if let Some(previous_id) = previous_alias_target
+                    .as_ref()
+                    .filter(|previous_id| *previous_id != id)
+                {
+                    if let Err(error) = self.clear_record_alias(previous_id, alias.as_ref()).await {
+                        warn!(
+                            alias = %alias,
+                            previous_snapshot_id = %previous_id,
+                            error = %error,
+                            "failed to clear previous snapshot alias metadata"
+                        );
                     }
-                    return Err(e);
                 }
             }
 
-            self.write_committed_record(
-                metadata.id.clone(),
-                metadata.alias.clone(),
-                metadata.resources,
-                committed,
-                metadata.source.clone(),
-            )
-            .await
+            Ok(record)
         }
         .await;
 
@@ -584,7 +624,8 @@ impl OssSnapshotRepository {
     /// Instead the algorithm is:
     ///   1. Read the current alias target.
     ///   2. If it already points to `id`, return success (idempotent).
-    ///   3. If it points to a live snapshot, return `AliasConflict`.
+    ///   3. If it points to a live snapshot: with `rebind` move the alias to
+    ///      `id` (E2B rebuild semantics), otherwise return `AliasConflict`.
     ///   4. If it points to a deleted snapshot, remove the stale alias.
     ///   5. Write our binding unconditionally.
     ///   6. Read back and verify we won the race.  If someone else wrote a
@@ -595,7 +636,7 @@ impl OssSnapshotRepository {
     /// interval between our write and the subsequent read.  This is weaker
     /// than a true CAS but sufficient for the current deployment model
     /// where concurrent publishes for the *same alias* are rare.
-    async fn bind_alias(&self, alias: &str, id: &SnapshotId) -> RepositoryResult<()> {
+    async fn bind_alias(&self, alias: &str, id: &SnapshotId, rebind: bool) -> RepositoryResult<()> {
         let key = validated_alias_key(alias)?;
         let payload = serde_json::to_vec(id)
             .map_err(|e| RepositoryError::backend("serialize alias binding", e))?;
@@ -608,18 +649,19 @@ impl OssSnapshotRepository {
                 }
 
                 let still_exists = self.snapshot_exists(&existing_id).await?;
-                if still_exists {
+                if still_exists && !rebind {
                     return Err(RepositoryError::AliasConflict {
                         alias: alias.to_string(),
                         existing: existing_id,
                         new_id: id.clone(),
                     });
                 }
-
-                self.client
-                    .delete(&key)
-                    .await
-                    .map_err(|e| RepositoryError::backend("delete stale alias", e))?;
+                if !still_exists {
+                    self.client
+                        .delete(&key)
+                        .await
+                        .map_err(|e| RepositoryError::backend("delete stale alias", e))?;
+                }
             }
 
             // Step 5: write our binding (unconditional — OSS does not
@@ -669,6 +711,95 @@ impl OssSnapshotRepository {
         })
     }
 
+    async fn restore_record_after_failed_publish(
+        &self,
+        id: &SnapshotId,
+        previous_record: Option<&SnapshotRecord>,
+    ) {
+        let result = match previous_record {
+            Some(record) => self.write_record(record).await,
+            None => self
+                .client
+                .delete(&OssSnapshotArtifactLayout::record_key(id))
+                .await
+                .map_err(|error| RepositoryError::backend("remove failed snapshot record", error)),
+        };
+        if let Err(error) = result {
+            warn!(snapshot_id = %id, error = %error, "failed to restore snapshot record after publish failure");
+        }
+    }
+
+    async fn restore_alias_after_failed_bind(
+        &self,
+        alias: &str,
+        id: &SnapshotId,
+        previous_id: Option<&SnapshotId>,
+    ) {
+        let current = match self.load_alias_target(alias).await {
+            Ok(current) => current,
+            Err(error) => {
+                warn!(alias, snapshot_id = %id, error = %error, "failed to inspect alias during publish rollback");
+                return;
+            }
+        };
+        // Skip when a concurrent publisher already moved the alias elsewhere.
+        // This only narrows the lost-update window: like `bind_alias`, the
+        // rollback cannot be atomic on a store without conditional writes, so a
+        // publisher that rebinds between this read and the write below is still
+        // clobbered.
+        if current.as_ref() != Some(id) {
+            return;
+        }
+
+        let key = match validated_alias_key(alias) {
+            Ok(key) => key,
+            Err(error) => {
+                warn!(alias, snapshot_id = %id, error = %error, "failed to validate alias during publish rollback");
+                return;
+            }
+        };
+        let result =
+            match previous_id {
+                Some(previous_id) => match serde_json::to_vec(previous_id) {
+                    Ok(payload) => {
+                        self.client.put_bytes(&key, payload).await.map_err(|error| {
+                            RepositoryError::backend("restore alias binding", error)
+                        })
+                    }
+                    Err(error) => Err(RepositoryError::backend(
+                        "serialize restored alias binding",
+                        error,
+                    )),
+                },
+                None => self.client.delete(&key).await.map_err(|error| {
+                    RepositoryError::backend("remove failed alias binding", error)
+                }),
+            };
+        if let Err(error) = result {
+            warn!(alias, snapshot_id = %id, error = %error, "failed to restore alias after publish failure");
+        }
+    }
+
+    /// Clears the alias field on the record that previously owned a rebound
+    /// alias so template listings do not report the moved name twice.
+    ///
+    /// Only `moved_alias` is cleared; a previous owner that already claims a
+    /// different name keeps it.
+    async fn clear_record_alias(&self, id: &SnapshotId, moved_alias: &str) -> RepositoryResult<()> {
+        if let Some(mut previous) = self.read_record(id).await? {
+            let claims_moved_alias = previous
+                .alias
+                .as_ref()
+                .is_some_and(|alias| alias.as_ref() == moved_alias);
+            if claims_moved_alias {
+                previous.alias = None;
+                previous.updated_at_unix_ms = now_unix_ms();
+                self.write_record(&previous).await?;
+            }
+        }
+        Ok(())
+    }
+
     async fn export_managed_disk_image(
         &self,
         image_config_path: &Path,