Merge pull request #843 from fabro-sh/remove/run-metadata-branches
Some checks are pending
TypeScript / Build (push) Waiting to run
Rust / Format (push) Waiting to run
Rust / Clippy (push) Waiting to run
Rust / Generated Docs (push) Waiting to run
Rust / Test (Linux) (push) Waiting to run
Rust / Sandbox plugins (stdio) (push) Waiting to run
Rust / Test (macOS) (push) Waiting to run
TypeScript / Typecheck (push) Waiting to run
TypeScript / Test (push) Waiting to run

Remove Git run metadata branches
This commit is contained in:
Bryan Helmkamp 2026-09-12 16:41:24 -06:00 committed by GitHub
commit 697d8622e9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
49 changed files with 302 additions and 4130 deletions

View file

@ -131,7 +131,7 @@ Fabro is an AI-powered workflow orchestration platform. Workflows are defined as
- **fabro-github** — GitHub App auth (JWT signing, installation tokens, PR creation)
- **fabro-mcp** — Model Context Protocol client/server
- **fabro-slack** — Slack integration (socket mode, blocks API)
- **fabro-checkpoint** — Git-based checkpoint storage with branch store and metadata branches
- **fabro-checkpoint** — Git checkpoint author identity and commit trailers
- **fabro-telemetry** — CLI analytics (Segment) and crash reporting (Sentry), with anonymous IDs, command sanitization, and detached subprocess delivery
- **fabro-util** — Shared utilities (redaction, terminal formatting)

8
Cargo.lock generated
View file

@ -2342,16 +2342,8 @@ version = "0.354.0-nightly.0"
name = "fabro-checkpoint"
version = "0.354.0-nightly.0"
dependencies = [
"chrono",
"fabro-config",
"fabro-store",
"fabro-types",
"git2",
"serde",
"serde_json",
"tempfile",
"thiserror 2.0.18",
"tracing",
]
[[package]]

View file

@ -250,7 +250,7 @@ the top-level `actor` envelope field.
### `metadata.snapshot.started`
Emitted when Fabro begins a durable metadata snapshot operation. These are product events for Fabro metadata snapshots, not tracing spans for the underlying git or filesystem work.
Historical event, no longer emitted. Recorded when Fabro began a Git metadata snapshot operation. Retained for reading older run streams.
Init and finalize metadata snapshots are unscoped. Checkpoint metadata snapshots use the checkpoint stage scope, so they include the checkpoint `node_id`, `node_label`, and `stage_id`.
@ -272,7 +272,7 @@ Init and finalize metadata snapshots are unscoped. Checkpoint metadata snapshots
### `metadata.snapshot.completed`
Emitted when Fabro commits and pushes a metadata snapshot successfully.
Historical event, no longer emitted. Recorded when Fabro committed and pushed a metadata snapshot successfully. Retained for reading older run streams.
```json
{
@ -300,7 +300,7 @@ Emitted when Fabro commits and pushes a metadata snapshot successfully.
### `metadata.snapshot.failed`
Emitted when a real metadata snapshot attempt fails. It is emitted before the matching compatibility `run.notice`, allowing human-facing consumers to suppress duplicate warning text. Compatibility notices with codes `checkpoint_metadata_write_failed` and `checkpoint_metadata_push_failed` may still appear in raw event streams. The `checkpoint_metadata_degraded` notice is a separate summary signal and should not be treated as a duplicate of this event.
Historical event, no longer emitted. Recorded when a metadata snapshot attempt failed, before the matching compatibility `run.notice`, allowing human-facing consumers to suppress duplicate warning text. Compatibility notices with codes `checkpoint_metadata_write_failed` and `checkpoint_metadata_push_failed` may still appear in raw event streams. The `checkpoint_metadata_degraded` notice is a separate summary signal and should not be treated as a duplicate of this event.
```json
{

View file

@ -1,6 +1,6 @@
# Run Scratch Files
This document maps the files that still live under a run scratch directory. Durable run state lives in the run store and metadata branch; scratch is mostly local runtime state and caches.
This document maps the files that still live under a run scratch directory. Durable run state lives in the durable event stream, event-derived projections, and CAS; scratch is mostly local runtime state and caches.
Scope:
- Scratch root: `~/.fabro/scratch/YYYYMMDD-{run_id}/`

View file

@ -227,11 +227,11 @@ When Fabro builds a [preamble](/execution/context#preamble-construction) for a d
This keeps preambles concise while still giving agents a path to read the full output if needed.
## Git storage
## Durable storage
Large offloaded context values are not stored on the Git [metadata branch](/execution/checkpoints#metadata-branch). The metadata branch keeps checkpoint JSON and stage metadata; blob payloads live in the durable blob store and are referenced by `blob://sha256/...`.
Checkpoint events and projections reference large offloaded context values with `blob://sha256/...` references. The payloads live in the durable content-addressed store (CAS).
Captured stage artifacts such as screenshots, videos, reports, and traces still use the artifact store and metadata export paths described below.
Captured stage artifacts such as screenshots, videos, reports, and traces still use the artifact store and export paths described below.
## Remote sandbox syncing

View file

@ -14737,7 +14737,6 @@ components:
- checkpoint
- clone
- run_branch
- meta_branch
- environment
- notifications
- interviews
@ -14776,8 +14775,6 @@ components:
$ref: "#/components/schemas/RunCloneSettings"
run_branch:
$ref: "#/components/schemas/RunBranchSettings"
meta_branch:
$ref: "#/components/schemas/RunMetaBranchSettings"
environment:
$ref: "#/components/schemas/RunEnvironmentSettings"
notifications:
@ -15032,15 +15029,6 @@ components:
push:
type: boolean
RunMetaBranchSettings:
type: object
required: [enabled, push]
properties:
enabled:
type: boolean
push:
type: boolean
RunEnvironmentSettings:
type: object
required: [id, provider, image, resources, network, lifecycle, labels, env]

View file

@ -3,18 +3,19 @@ title: "Checkpoints"
description: "How Fabro uses Git to checkpoint and resume workflow runs"
---
Fabro checkpoints every workflow run using Git plus the durable run store. After each node completes, Fabro commits the file changes and execution state so that interrupted runs can be resumed exactly where they left off. This happens automatically — no configuration required beyond running inside a Git repository.
Fabro checkpoints every workflow run using Git plus the durable run store. After each node completes, Fabro commits file changes to the run branch and records execution state in the durable event stream so that interrupted runs can be resumed exactly where they left off. This happens automatically — no configuration required beyond running inside a Git repository.
## Two branches, two purposes
## Code and execution history
Each run creates two Git branches that work in tandem:
Fabro stores code and execution state separately:
| Branch | Ref format | Contains |
|---|---|---|
| **Run branch** | `fabro/run/{run_id}` | File changes made by agents and commands — the actual work product |
| **Metadata branch** | `fabro/meta/{run_id}` | Checkpoint JSON, the workflow graph, run and start records, and offloaded artifacts |
| Storage | Contains |
|---|---|
| **Run branch** (`fabro/run/{run_id}`) | File changes made by agents and commands |
| **Durable run store** | Events and event-derived projections for checkpoints, stages, configuration, and conclusions |
| **Content-addressed store (CAS)** | Artifact and offloaded context payloads referenced by events and checkpoints |
The run branch is a regular Git branch that grows one commit per completed node. The metadata branch is an orphan branch (no shared history with your code) that stores structured data using Git's object database directly — no working tree needed.
The run branch is a regular Git branch. Checkpoint events record its commit SHAs, which link execution state to the corresponding code.
### Run branch commits
@ -25,7 +26,6 @@ fabro(01JKXYZ...): plan (succeeded)
Fabro-Run: 01JKXYZ...
Fabro-Completed: 2
Fabro-Checkpoint: a1b2c3d4...
```
The commit message follows a structured format:
@ -35,29 +35,26 @@ The commit message follows a structured format:
| Subject line | `fabro({run_id}): {node_id} ({status})` |
| `Fabro-Run` trailer | The run ID |
| `Fabro-Completed` trailer | Number of completed nodes so far |
| `Fabro-Checkpoint` trailer | SHA of the corresponding commit on the metadata branch |
The `Fabro-Checkpoint` trailer links each run branch commit to its metadata branch commit, so you can navigate from file changes to the full execution state and back.
The `git_commit_sha` in a `checkpoint.completed` event identifies the run branch commit. New commits do not include a `Fabro-Checkpoint` trailer.
Fabro disables Git commit and tag signing for checkpoint commits created inside a sandbox. Your personal or repository-level signing settings can stay enabled, but sandbox bookkeeping does not need access to your signing key.
Checkpoint commits, metadata commits, and any commit a workflow command or agent creates all carry the run's one resolved Git identity. Fabro derives it from the run's GitHub App bot account or token user, or from the generic `Fabro <noreply@fabro.sh>` identity when the run has no GitHub credential, and `[run.git.author]` overrides it. See [`[run.git.author]`](/administration/server-configuration#rungitauthor-section).
Checkpoint commits and any commit a workflow command or agent creates all carry the run's one resolved Git identity. Fabro derives it from the run's GitHub App bot account or token user, or from the generic `Fabro <noreply@fabro.sh>` identity when the run has no GitHub credential, and `[run.git.author]` overrides it. See [`[run.git.author]`](/administration/server-configuration#rungitauthor-section).
### Durable execution state
Fabro derives run state from persisted events. The projection includes the run spec, start and status records, checkpoints, stage results, conclusion, and sandbox information. Artifact payloads live in CAS.
Use `fabro dump` to export a run as files, including `run.json`, `graph.fabro`, and per-stage artifacts.
### Metadata branch
The metadata branch (`fabro/meta/{run_id}`) is an orphan branch that stores structured run data using Git's object storage directly. Fabro writes it from inside the sandbox with Git plumbing commands, without checking out a metadata worktree. It is initialized at run start with:
- **`run.json`** — Current projection snapshot: run spec, start/status records, current checkpoint, conclusion, sandbox, and other run-level metadata
- **`graph.fabro`** — Workflow source for the run
After each node, the metadata branch is updated with:
- **`run.json`** — Refreshed projection snapshot with the new current checkpoint
- **`stages/{rank:03}-{node_id}@{visit}/...`** — Execution-order-prefixed per-stage trace files (prompts, responses, status, diffs, command output, and tool metadata)
Fabro no longer creates or pushes `fabro/meta/{run_id}` branches. Existing metadata branches remain untouched. Historical metadata snapshot events remain readable.
## What's in a checkpoint
The `run.json.checkpoint` snapshot captures everything needed to resume a run:
The checkpoint projection captures the execution state needed to resume a run:
| Field | Description |
|---|---|
@ -89,7 +86,7 @@ This means your original working directory stays untouched while the agent makes
If the working directory has uncommitted changes, the worktree starts from committed `HEAD` and those uncommitted changes are not included. Fabro logs a warning so you can commit, stash, or run explicitly in place when that is what you want.
</Note>
For Docker and Daytona sandboxes, the repository is cloned into the sandbox and checkpoint Git operations run there. Both the run branch and metadata branch are pushed to origin from the sandbox after each checkpoint when pushing is configured.
For Docker and Daytona sandboxes, the repository is cloned into the sandbox and checkpoint Git operations run there. The run branch is pushed to origin from the sandbox after each checkpoint when pushing is configured.
## Resuming a run
@ -119,14 +116,14 @@ The number after `@` is the stage execution ordinal. It is separate from the gra
## The checkpoint cycle
Here's the full sequence that runs after every node completes:
After a node completes, Fabro:
1. **Append checkpoint event** — Persist the new checkpoint into durable run state
2. **Write metadata branch** — Serialize the checkpoint and any new artifacts to the metadata branch (shadow commit)
3. **Commit to run branch** — Stage all file changes, commit with structured trailers linking to the shadow commit SHA
4. **Update durable checkpoint** — Persist the `git_commit_sha` associated with the run-branch commit
1. Stores offloaded context payloads in CAS.
2. Creates a code checkpoint commit when Git checkpointing is enabled.
3. Pushes the run branch when configured and collects the code diff.
4. Emits a checkpoint event with execution state and the code commit SHA. The run store persists this event and updates the projection.
Steps 2-4 are best-effort — if any Git operation fails, the run continues and emits a `RunNotice` warning event. Resume still uses the durable checkpoint in the run store.
A checkpoint commit failure stops execution. Intermediate push and diff failures emit warning notices. A required final publish failure marks the run as failed.
## Inspecting run history
@ -142,8 +139,12 @@ git show fabro/run/01JKXYZ...
# Diff the full run against the starting point
git diff main..fabro/run/01JKXYZ...
# Read checkpoint data from the metadata branch
git show fabro/meta/01JKXYZ...:run.json | jq .checkpoint.current_node
# Inspect execution state and events
fabro inspect 01JKXYZ
fabro events 01JKXYZ
# Export the run projection and artifacts
fabro dump 01JKXYZ --output ./run-dump
```
## Rewinding to an earlier checkpoint
@ -182,7 +183,7 @@ fabro resume <NEW_RUN_ID>
Use **rewind** when a terminal run should be abandoned and replaced from an earlier point. Use **fork** when you want to try a different approach while keeping the original run as a reference.
`fabro rewind --list`, `fabro fork --list`, `fabro rewind`, and `fabro fork` are server-backed. Timeline listing reads durable run-store checkpoints; it does not rebuild missing metadata branches.
`fabro rewind --list`, `fabro fork --list`, `fabro rewind`, and `fabro fork` are server-backed. Timeline listing reads checkpoints from the durable run store.
See [`fabro fork`](/reference/cli#fabro-fork) for the full command reference.

View file

@ -260,23 +260,12 @@ push = true
| Field | Description |
|---|---|
| `enabled` | When `false`, Fabro does not create the managed run branch or checkpoint commits. This also disables metadata branch writes. |
| `enabled` | When `false`, Fabro does not create the managed run branch or checkpoint commits. |
| `push` | When `false`, Fabro creates local checkpoint commits but does not push `fabro/run/<id>` to the remote. |
### `[run.meta_branch]`
Configure Fabro's managed `fabro/meta/<id>` metadata branch.
```toml title="run.toml"
[run.meta_branch]
enabled = true
push = true
```
| Field | Description |
|---|---|
| `enabled` | When `false`, Fabro skips metadata branch snapshots. |
| `push` | When `false`, Fabro writes metadata snapshots locally but does not push `fabro/meta/<id>` to the remote. |
Metadata branches have been retired. Existing `enabled` and `push` settings in this table are accepted but ignored. Resolved settings and API responses omit `meta_branch`. You can remove the table from your configuration. Existing Git metadata branches remain untouched.
### `[run.environment]` and server-managed environments

View file

@ -29,7 +29,7 @@ The rest of this page describes the `app` strategy, which is required for browse
|---|---|
| **OAuth login** | Users sign in to the web UI with their GitHub account |
| **Private repo cloning** | Daytona and Docker sandboxes clone private repositories using short-lived Installation Access Tokens |
| **Checkpoint pushing** | After each workflow stage, Fabro pushes the run branch and metadata branch back to origin from inside the sandbox |
| **Checkpoint pushing** | After each workflow stage, Fabro pushes the run branch back to origin from inside the sandbox |
| **Auto-PR** | When `[run.pull_request] enabled = true` in the [run config](/execution/run-configuration#runpull_request), Fabro opens a PR from the agent's working branch after a successful run |
| **Auto-merge** | When `[run.pull_request] auto_merge = true`, Fabro enables GitHub's auto-merge on created PRs so they merge automatically once required checks pass |
| **Sandbox GITHUB_TOKEN** | When `[run.integrations.github.permissions]` are declared at any layer (workflow, project, or user settings), Fabro mints a scoped Installation Access Token and injects it as `GITHUB_TOKEN` in the sandbox |
@ -319,7 +319,7 @@ Every commit a run creates is authored and committed by the run's GitHub credent
### Checkpoint pushing
After each workflow stage, Fabro [checkpoints](/execution/checkpoints) by pushing the run branch and metadata branch to origin. Before a successful run becomes terminal, the publish stage pushes the final commit again and treats failure as a run failure. Inside remote sandboxes, the git remote URL is configured with the Installation Access Token for authenticated pushing.
After each workflow stage, Fabro [checkpoints](/execution/checkpoints) by pushing the run branch to origin. Before a successful run becomes terminal, the publish stage pushes the final commit again and treats failure as a run failure. Inside remote sandboxes, the git remote URL is configured with the Installation Access Token for authenticated pushing.
When pull request creation is enabled, Fabro then checks that GitHub reports the run branch at the exact final commit before opening the PR. A failed final push, branch check, or PR creation marks the run as failed with `publish_failed`; the terminal run event is emitted only after this step finishes.

View file

@ -974,10 +974,6 @@ fn attach_json_errors_without_prompting_for_human_input() {
"provider": null,
"slack": null
},
"meta_branch": {
"enabled": true,
"push": true
},
"metadata": {},
"model": {
"controls": {

View file

@ -159,10 +159,6 @@ fn inspect_resolves_selector_via_server_endpoint() {
"enabled": true,
"push": true
},
"meta_branch": {
"enabled": true,
"push": true
},
"environment": {
"id": "default",
"provider": "local",

View file

@ -4,7 +4,7 @@ edition.workspace = true
version.workspace = true
publish = false
license.workspace = true
description = "Git-backed checkpoint storage for Fabro workflows"
description = "Git checkpoint commit helpers for Fabro workflows"
repository = "https://github.com/brynary/arc"
[lib]
@ -15,14 +15,4 @@ workspace = true
[dependencies]
fabro-config = { path = "../../foundation/fabro-config" }
fabro-store = { path = "../fabro-store" }
fabro-types = { path = "../../foundation/fabro-types" }
git2.workspace = true
serde.workspace = true
serde_json.workspace = true
thiserror.workspace = true
tracing.workspace = true
[dev-dependencies]
chrono.workspace = true
tempfile = "3"

View file

@ -1,465 +0,0 @@
use git2::{Oid, Signature};
use tracing::{debug, warn};
use crate::Result;
use crate::git::{FileMode, Store, TreeEntries};
/// Metadata about a commit, returned by `log`.
#[derive(Debug)]
pub struct CommitInfo {
pub oid: Oid,
pub message: String,
pub author_name: String,
pub author_email: String,
pub time: git2::Time,
}
/// Key-value storage on a single git branch. Each write creates one commit.
/// The branch's tree grows monotonically — each commit's tree is a superset of
/// the previous.
pub struct BranchStore<'a> {
objects: &'a Store,
branch: String,
author: Signature<'static>,
}
impl<'a> BranchStore<'a> {
pub fn new(objects: &'a Store, branch: impl Into<String>, author: &Signature<'_>) -> Self {
// Clone to 'static by using Signature::now (author name/email are copied into
// owned strings)
let author_static = Signature::now(
author.name().unwrap_or("unknown"),
author.email().unwrap_or(""),
)
.expect("creating signature should not fail");
Self {
objects,
branch: branch.into(),
author: author_static,
}
}
/// Create branch with empty root commit if it doesn't exist.
pub fn ensure_branch(&self) -> Result<()> {
if self.objects.resolve_ref(&self.branch)?.is_some() {
return Ok(());
}
let empty_tree = self.objects.write_empty_tree()?;
let commit_oid =
self.objects
.write_commit(empty_tree, &[], "initialize branch", &self.author)?;
self.objects.update_ref(&self.branch, commit_oid)?;
debug!(branch = %self.branch, "Created checkpoint branch");
Ok(())
}
/// Core read-modify-write: read current tree, let caller mutate, write new
/// commit.
pub fn write_with(
&self,
message: &str,
f: impl FnOnce(&mut TreeEntries) -> Result<()>,
) -> Result<Oid> {
let parent_oid = self.objects.resolve_ref(&self.branch)?.ok_or_else(|| {
warn!(branch = %self.branch, "Branch not found during write");
crate::Error::BranchNotFound {
branch: self.branch.clone(),
}
})?;
let parent_commit = self.objects.repo().find_commit(parent_oid)?;
let tree_oid = parent_commit.tree_id();
let mut entries = self.objects.read_tree(tree_oid)?;
f(&mut entries)?;
let new_tree = self.objects.write_tree(&entries)?;
let commit_oid =
self.objects
.write_commit(new_tree, &[parent_oid], message, &self.author)?;
self.objects.update_ref(&self.branch, commit_oid)?;
debug!(branch = %self.branch, commit = %commit_oid, "Wrote checkpoint commit");
Ok(commit_oid)
}
/// Store a single file. Creates one commit.
pub fn write_entry(&self, path: &str, content: &[u8], message: &str) -> Result<Oid> {
let blob_oid = self.objects.write_blob(content)?;
self.write_with(message, |entries| {
entries.set(path, blob_oid, FileMode::Blob);
Ok(())
})
}
/// Atomically store multiple files in a single commit.
pub fn write_entries(&self, file_entries: &[(&str, &[u8])], message: &str) -> Result<Oid> {
let blobs: Vec<(String, Oid)> = file_entries
.iter()
.map(|(path, content)| {
let oid = self.objects.write_blob(content)?;
Ok((path.to_string(), oid))
})
.collect::<Result<Vec<_>>>()?;
self.write_with(message, |entries| {
for (path, oid) in &blobs {
entries.set(path.clone(), *oid, FileMode::Blob);
}
Ok(())
})
}
/// Remove a file. Creates one commit.
pub fn delete_entry(&self, path: &str, message: &str) -> Result<Oid> {
self.write_with(message, |entries| {
entries.remove(path);
Ok(())
})
}
/// Read a single file from the latest tree. Returns `None` if branch or
/// path doesn't exist.
pub fn read_entry(&self, path: &str) -> Result<Option<Vec<u8>>> {
let Some(commit_oid) = self.objects.resolve_ref(&self.branch)? else {
return Ok(None);
};
let commit = self.objects.repo().find_commit(commit_oid)?;
let tree = commit.tree()?;
match tree.get_path(std::path::Path::new(path)) {
Ok(entry) => {
let blob = self.objects.repo().find_blob(entry.id())?;
Ok(Some(blob.content().to_vec()))
}
Err(e) if e.code() == git2::ErrorCode::NotFound => Ok(None),
Err(e) => Err(e.into()),
}
}
/// Read multiple paths. Missing paths are omitted from the result.
pub fn read_entries<'b>(&self, paths: &[&'b str]) -> Result<Vec<(&'b str, Vec<u8>)>> {
let Some(commit_oid) = self.objects.resolve_ref(&self.branch)? else {
return Ok(vec![]);
};
let commit = self.objects.repo().find_commit(commit_oid)?;
let tree = commit.tree()?;
let mut results = Vec::new();
for path in paths {
match tree.get_path(std::path::Path::new(path)) {
Ok(entry) => {
let blob = self.objects.repo().find_blob(entry.id())?;
results.push((*path, blob.content().to_vec()));
}
Err(e) if e.code() == git2::ErrorCode::NotFound => {}
Err(e) => return Err(e.into()),
}
}
Ok(results)
}
/// List all paths under a prefix in the latest tree.
pub fn list_entries(&self, prefix: &str) -> Result<Vec<String>> {
let Some(commit_oid) = self.objects.resolve_ref(&self.branch)? else {
return Ok(vec![]);
};
let commit = self.objects.repo().find_commit(commit_oid)?;
let tree_oid = commit.tree_id();
let entries = self.objects.read_tree(tree_oid)?;
let paths: Vec<String> = entries
.under_prefix(prefix)
.map(|(k, _)| k.to_string())
.collect();
Ok(paths)
}
/// Full tree from branch tip.
pub fn tip_tree(&self) -> Result<TreeEntries> {
let commit_oid = self.objects.resolve_ref(&self.branch)?.ok_or_else(|| {
crate::Error::BranchNotFound {
branch: self.branch.clone(),
}
})?;
let commit = self.objects.repo().find_commit(commit_oid)?;
self.objects.read_tree(commit.tree_id())
}
/// Walk commits on the branch, newest first.
pub fn log(&self, limit: usize) -> Result<Vec<CommitInfo>> {
let Some(commit_oid) = self.objects.resolve_ref(&self.branch)? else {
return Ok(vec![]);
};
let mut revwalk = self.objects.repo().revwalk()?;
revwalk.set_sorting(git2::Sort::TIME | git2::Sort::TOPOLOGICAL)?;
revwalk.push(commit_oid)?;
let mut results = Vec::new();
for oid_result in revwalk.take(limit) {
let oid = oid_result?;
let commit = self.objects.repo().find_commit(oid)?;
results.push(CommitInfo {
oid,
message: commit.message().unwrap_or("").to_string(),
author_name: commit.author().name().unwrap_or("").to_string(),
author_email: commit.author().email().unwrap_or("").to_string(),
time: commit.author().when(),
});
}
Ok(results)
}
}
/// Split a hex ID into a sharded path.
///
/// ```text
/// sharded_path("a3b2c4d5e6f7", 2) → "a3/b2c4d5e6f7"
/// ```
pub fn sharded_path(id: &str, prefix_len: usize) -> String {
if id.len() <= prefix_len {
return id.to_string();
}
format!("{}/{}", &id[..prefix_len], &id[prefix_len..])
}
#[cfg(test)]
mod tests {
use git2::Repository;
use super::*;
use crate::git::FileMode;
fn temp_repo() -> (tempfile::TempDir, Store) {
let dir = tempfile::TempDir::new().unwrap();
let repo = Repository::init(dir.path()).unwrap();
(dir, Store::new(repo))
}
fn test_sig() -> Signature<'static> {
Signature::now("Test", "test@example.com").unwrap()
}
// -- sharded_path (pure function) --
#[test]
fn sharded_path_basic() {
assert_eq!(sharded_path("a3b2c4d5e6f7", 2), "a3/b2c4d5e6f7");
}
#[test]
fn sharded_path_short_id() {
assert_eq!(sharded_path("ab", 2), "ab");
}
#[test]
fn sharded_path_prefix_3() {
assert_eq!(sharded_path("abcdef", 3), "abc/def");
}
// -- ensure_branch --
#[test]
fn ensure_branch_creates_branch() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/metadata", &sig);
bs.ensure_branch().unwrap();
assert!(store.resolve_ref("test/metadata").unwrap().is_some());
}
#[test]
fn ensure_branch_idempotent() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/metadata", &sig);
bs.ensure_branch().unwrap();
let first_oid = store.resolve_ref("test/metadata").unwrap().unwrap();
bs.ensure_branch().unwrap();
let second_oid = store.resolve_ref("test/metadata").unwrap().unwrap();
assert_eq!(first_oid, second_oid);
}
// -- write_entry + read_entry roundtrip --
#[test]
fn write_and_read_entry() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("hello.txt", b"world", "add hello").unwrap();
let content = bs.read_entry("hello.txt").unwrap().unwrap();
assert_eq!(content, b"world");
}
// -- write_entries atomic multi-file --
#[test]
fn write_entries_atomic() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entries(&[("a.txt", b"alpha"), ("b.txt", b"beta")], "add both")
.unwrap();
assert_eq!(bs.read_entry("a.txt").unwrap().unwrap(), b"alpha");
assert_eq!(bs.read_entry("b.txt").unwrap().unwrap(), b"beta");
// Verify it was a single commit (2 total: init + write)
let log = bs.log(10).unwrap();
assert_eq!(log.len(), 2);
}
// -- delete_entry --
#[test]
fn delete_entry() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("to-delete.txt", b"content", "add file")
.unwrap();
bs.delete_entry("to-delete.txt", "remove file").unwrap();
assert!(bs.read_entry("to-delete.txt").unwrap().is_none());
}
// -- write_with closure --
#[test]
fn write_with_closure() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
let blob_oid = store.write_blob(b"custom content").unwrap();
bs.write_with("custom write", |entries| {
entries.set("custom.txt", blob_oid, FileMode::Blob);
Ok(())
})
.unwrap();
let content = bs.read_entry("custom.txt").unwrap().unwrap();
assert_eq!(content, b"custom content");
}
// -- read_entry on nonexistent --
#[test]
fn read_entry_nonexistent_branch() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "nonexistent", &sig);
assert!(bs.read_entry("anything.txt").unwrap().is_none());
}
#[test]
fn read_entry_nonexistent_path() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
assert!(bs.read_entry("nonexistent.txt").unwrap().is_none());
}
// -- list_entries --
#[test]
fn list_entries_with_prefix() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entries(
&[
("ab/data.json", b"{}"),
("ab/meta.json", b"{}"),
("cd/other.json", b"{}"),
],
"add files",
)
.unwrap();
let ab_entries = bs.list_entries("ab/").unwrap();
assert_eq!(ab_entries, vec!["ab/data.json", "ab/meta.json"]);
}
// -- tip_tree --
#[test]
fn tip_tree() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("file.txt", b"content", "add file").unwrap();
let tree = bs.tip_tree().unwrap();
assert_eq!(tree.len(), 1);
assert!(tree.get("file.txt").is_some());
}
// -- log --
#[test]
fn log_returns_history() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("a.txt", b"a", "first write").unwrap();
bs.write_entry("b.txt", b"b", "second write").unwrap();
let log = bs.log(10).unwrap();
assert_eq!(log.len(), 3); // init + 2 writes
assert_eq!(log[0].message, "second write");
assert_eq!(log[1].message, "first write");
assert_eq!(log[2].message, "initialize branch");
}
#[test]
fn log_respects_limit() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("a.txt", b"a", "first").unwrap();
bs.write_entry("b.txt", b"b", "second").unwrap();
let log = bs.log(2).unwrap();
assert_eq!(log.len(), 2);
}
#[test]
fn log_empty_branch() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "nonexistent", &sig);
let log = bs.log(10).unwrap();
assert!(log.is_empty());
}
// -- read_entries --
#[test]
fn read_entries_multiple() {
let (_dir, store) = temp_repo();
let sig = test_sig();
let bs = BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entries(&[("a.txt", b"alpha"), ("b.txt", b"beta")], "add files")
.unwrap();
let results = bs.read_entries(&["a.txt", "b.txt", "missing.txt"]).unwrap();
assert_eq!(results.len(), 2);
assert_eq!(results[0], ("a.txt", b"alpha".to_vec()));
assert_eq!(results[1], ("b.txt", b"beta".to_vec()));
}
}

View file

@ -1,32 +0,0 @@
use std::path::PathBuf;
pub type Result<T> = std::result::Result<T, Error>;
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error(transparent)]
Git(#[from] git2::Error),
#[error("reading file {path}: {source}")]
ReadFile {
path: PathBuf,
source: std::io::Error,
},
#[error("branch {branch} not found")]
BranchNotFound { branch: String },
}
#[derive(Debug, thiserror::Error)]
pub enum MetadataError {
#[error(transparent)]
Storage(#[from] Error),
#[error("deserialize {entity} on branch {branch}: {source}")]
Deserialize {
entity: &'static str,
branch: String,
#[source]
source: serde_json::Error,
},
}

View file

@ -1,602 +0,0 @@
#![expect(
clippy::disallowed_methods,
reason = "sync git2 operations dominate this module; std::fs usage is part of the same blocking path and not on a Tokio hot path"
)]
use std::collections::BTreeMap;
use std::path::Path;
use git2::{Oid, Repository, Signature};
use crate::{Error, Result};
/// Git file modes for tree entries.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FileMode {
Blob,
BlobExecutable,
Tree,
}
impl FileMode {
fn as_i32(self) -> i32 {
match self {
Self::Blob => 0o100_644,
Self::BlobExecutable => 0o100_755,
Self::Tree => 0o040_000,
}
}
fn from_i32(mode: i32) -> Self {
match mode {
0o100_755 => Self::BlobExecutable,
0o040_000 => Self::Tree,
_ => Self::Blob,
}
}
}
/// A single entry in a flat tree map.
#[derive(Debug, Clone)]
pub struct TreeEntry {
pub oid: Oid,
pub filemode: FileMode,
}
/// A flat, sorted map of paths to tree entries.
///
/// Intermediate representation between reading an existing git tree and writing
/// a new one. Paths use forward slashes and are relative to the tree root (e.g.
/// `"src/main.rs"`).
#[derive(Debug, Clone, Default)]
pub struct TreeEntries(BTreeMap<String, TreeEntry>);
impl TreeEntries {
pub fn new() -> Self {
Self(BTreeMap::new())
}
pub fn set(&mut self, path: impl Into<String>, oid: Oid, filemode: FileMode) {
self.0.insert(path.into(), TreeEntry { oid, filemode });
}
pub fn remove(&mut self, path: &str) {
self.0.remove(path);
}
pub fn get(&self, path: &str) -> Option<&TreeEntry> {
self.0.get(path)
}
pub fn merge(&mut self, other: &Self) {
for (path, entry) in &other.0 {
self.0.insert(path.clone(), entry.clone());
}
}
pub fn iter(&self) -> impl Iterator<Item = (&str, &TreeEntry)> {
self.0.iter().map(|(k, v)| (k.as_str(), v))
}
/// Iterate entries whose paths start with the given prefix.
pub fn under_prefix<'a>(
&'a self,
prefix: &'a str,
) -> impl Iterator<Item = (&'a str, &'a TreeEntry)> {
self.0
.range(prefix.to_string()..)
.take_while(move |(k, _)| k.starts_with(prefix))
.map(|(k, v)| (k.as_str(), v))
}
pub fn len(&self) -> usize {
self.0.len()
}
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
}
/// Wraps a `git2::Repository` with operations for creating blobs, trees,
/// commits, and refs.
pub struct Store {
repo: Repository,
}
impl Store {
pub fn new(repo: Repository) -> Self {
Self { repo }
}
pub fn repo(&self) -> &Repository {
&self.repo
}
pub fn repo_dir(&self) -> &Path {
self.repo
.workdir()
.or_else(|| self.repo.path().parent())
.unwrap_or(self.repo.path())
}
/// Store bytes as a git blob.
pub fn write_blob(&self, content: &[u8]) -> Result<Oid> {
Ok(self.repo.blob(content)?)
}
/// Read a file from disk, store as a blob.
/// Returns `(oid, filemode)` where filemode detects the executable bit on
/// unix.
pub fn write_blob_from_file(&self, path: &Path) -> Result<(Oid, FileMode)> {
let content = std::fs::read(path).map_err(|e| Error::ReadFile {
path: path.to_path_buf(),
source: e,
})?;
let mode = detect_filemode(path);
let oid = self.repo.blob(&content)?;
Ok((oid, mode))
}
/// Recursively flatten a git tree into `TreeEntries`.
pub fn read_tree(&self, oid: Oid) -> Result<TreeEntries> {
let tree = self.repo.find_tree(oid)?;
let mut entries = TreeEntries::new();
read_tree_recursive(&self.repo, &tree, "", &mut entries)?;
Ok(entries)
}
/// Build nested git tree objects from flat `TreeEntries`.
pub fn write_tree(&self, entries: &TreeEntries) -> Result<Oid> {
let root = build_dir_node(entries);
write_dir_node(&self.repo, &root)
}
/// Write an empty tree (zero entries).
pub fn write_empty_tree(&self) -> Result<Oid> {
let builder = self.repo.treebuilder(None)?;
Ok(builder.write()?)
}
/// Create a commit. Does NOT update any ref — caller does that via
/// `update_ref`. `author` is used for both author and committer fields.
pub fn write_commit(
&self,
tree_oid: Oid,
parents: &[Oid],
message: &str,
author: &Signature<'_>,
) -> Result<Oid> {
let tree = self.repo.find_tree(tree_oid)?;
let parent_commits: Vec<git2::Commit<'_>> = parents
.iter()
.map(|oid| self.repo.find_commit(*oid))
.collect::<std::result::Result<Vec<_>, _>>()?;
let parent_refs: Vec<&git2::Commit<'_>> = parent_commits.iter().collect();
let oid = self
.repo
.commit(None, author, author, message, &tree, &parent_refs)?;
Ok(oid)
}
/// Set a branch ref to point at `commit_oid`. Creates branch if needed.
pub fn update_ref(&self, branch: &str, commit_oid: Oid) -> Result<()> {
let refname = format!("refs/heads/{branch}");
self.repo
.reference(&refname, commit_oid, true, "update ref")?;
Ok(())
}
/// Resolve branch to commit OID. Returns `None` if branch doesn't exist.
pub fn resolve_ref(&self, branch: &str) -> Result<Option<Oid>> {
let refname = format!("refs/heads/{branch}");
match self.repo.find_reference(&refname) {
Ok(reference) => Ok(Some(reference.peel_to_commit()?.id())),
Err(e) if e.code() == git2::ErrorCode::NotFound => Ok(None),
Err(e) => Err(e.into()),
}
}
/// Read a blob from the tree of a specific commit. Returns `None` if the
/// path doesn't exist.
pub fn read_blob_at(&self, commit_oid: Oid, path: &str) -> Result<Option<Vec<u8>>> {
let commit = self.repo.find_commit(commit_oid)?;
let tree = commit.tree()?;
match tree.get_path(std::path::Path::new(path)) {
Ok(entry) => {
let blob = self.repo.find_blob(entry.id())?;
Ok(Some(blob.content().to_vec()))
}
Err(e) if e.code() == git2::ErrorCode::NotFound => Ok(None),
Err(e) => Err(e.into()),
}
}
/// Delete a branch reference. No-op if branch doesn't exist.
pub fn delete_ref(&self, branch: &str) -> Result<()> {
let refname = format!("refs/heads/{branch}");
match self.repo.find_reference(&refname) {
Ok(mut reference) => {
reference.delete()?;
Ok(())
}
Err(e) if e.code() == git2::ErrorCode::NotFound => Ok(()),
Err(e) => Err(e.into()),
}
}
}
/// Recursively read a git tree into flat `TreeEntries`.
fn read_tree_recursive(
repo: &Repository,
tree: &git2::Tree<'_>,
prefix: &str,
entries: &mut TreeEntries,
) -> Result<()> {
for entry in tree {
let name = entry.name().unwrap_or("");
let path = if prefix.is_empty() {
name.to_string()
} else {
format!("{prefix}/{name}")
};
let mode = FileMode::from_i32(entry.filemode());
if mode == FileMode::Tree {
let subtree = repo.find_tree(entry.id())?;
read_tree_recursive(repo, &subtree, &path, entries)?;
} else {
entries.set(path, entry.id(), mode);
}
}
Ok(())
}
/// Intermediate structure for building nested git trees from flat paths.
struct DirNode {
files: BTreeMap<String, TreeEntry>,
dirs: BTreeMap<String, Self>,
}
impl DirNode {
fn new() -> Self {
Self {
files: BTreeMap::new(),
dirs: BTreeMap::new(),
}
}
}
/// Build a `DirNode` tree from flat `TreeEntries`.
fn build_dir_node(entries: &TreeEntries) -> DirNode {
let mut root = DirNode::new();
for (path, entry) in entries.iter() {
let parts: Vec<&str> = path.split('/').collect();
insert_into_dir_node(&mut root, &parts, entry);
}
root
}
fn insert_into_dir_node(node: &mut DirNode, parts: &[&str], entry: &TreeEntry) {
match parts {
[name] => {
node.files.insert(name.to_string(), entry.clone());
}
[dir, rest @ ..] => {
let child = node
.dirs
.entry(dir.to_string())
.or_insert_with(DirNode::new);
insert_into_dir_node(child, rest, entry);
}
[] => {}
}
}
/// Recursively write a `DirNode` as nested git trees, bottom-up.
fn write_dir_node(repo: &Repository, node: &DirNode) -> Result<Oid> {
let mut builder = repo.treebuilder(None)?;
for (name, entry) in &node.files {
builder.insert(name, entry.oid, entry.filemode.as_i32())?;
}
for (name, child) in &node.dirs {
let child_oid = write_dir_node(repo, child)?;
builder.insert(name, child_oid, FileMode::Tree.as_i32())?;
}
Ok(builder.write()?)
}
/// Detect file mode (executable or not) from filesystem metadata.
#[cfg(unix)]
fn detect_filemode(path: &Path) -> FileMode {
use std::os::unix::fs::PermissionsExt;
match std::fs::metadata(path) {
Ok(meta) => {
if meta.permissions().mode() & 0o111 != 0 {
FileMode::BlobExecutable
} else {
FileMode::Blob
}
}
Err(_) => FileMode::Blob,
}
}
#[cfg(not(unix))]
fn detect_filemode(_path: &Path) -> FileMode {
FileMode::Blob
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_repo() -> (tempfile::TempDir, Store) {
let dir = tempfile::TempDir::new().unwrap();
let repo = Repository::init(dir.path()).unwrap();
(dir, Store::new(repo))
}
// -- TreeEntries tests (pure data, no git) --
#[test]
fn tree_entries_set_and_get() {
let mut entries = TreeEntries::new();
let oid = Oid::zero();
entries.set("src/main.rs", oid, FileMode::Blob);
let entry = entries.get("src/main.rs").unwrap();
assert_eq!(entry.oid, oid);
assert_eq!(entry.filemode, FileMode::Blob);
}
#[test]
fn tree_entries_remove() {
let mut entries = TreeEntries::new();
entries.set("a.txt", Oid::zero(), FileMode::Blob);
entries.set("b.txt", Oid::zero(), FileMode::Blob);
entries.remove("a.txt");
assert!(entries.get("a.txt").is_none());
assert!(entries.get("b.txt").is_some());
assert_eq!(entries.len(), 1);
}
#[test]
fn tree_entries_merge() {
let mut a = TreeEntries::new();
a.set("file1.txt", Oid::zero(), FileMode::Blob);
let mut b = TreeEntries::new();
b.set("file2.txt", Oid::zero(), FileMode::Blob);
a.merge(&b);
assert_eq!(a.len(), 2);
assert!(a.get("file1.txt").is_some());
assert!(a.get("file2.txt").is_some());
}
#[test]
fn tree_entries_under_prefix() {
let mut entries = TreeEntries::new();
entries.set("src/a.rs", Oid::zero(), FileMode::Blob);
entries.set("src/b.rs", Oid::zero(), FileMode::Blob);
entries.set("test/c.rs", Oid::zero(), FileMode::Blob);
let src: Vec<&str> = entries.under_prefix("src/").map(|(k, _)| k).collect();
assert_eq!(src, vec!["src/a.rs", "src/b.rs"]);
}
#[test]
fn tree_entries_empty() {
let entries = TreeEntries::new();
assert!(entries.is_empty());
assert_eq!(entries.len(), 0);
}
#[test]
fn tree_entries_iter_sorted() {
let mut entries = TreeEntries::new();
entries.set("z.txt", Oid::zero(), FileMode::Blob);
entries.set("a.txt", Oid::zero(), FileMode::Blob);
entries.set("m.txt", Oid::zero(), FileMode::Blob);
let keys: Vec<&str> = entries.iter().map(|(k, _)| k).collect();
assert_eq!(keys, vec!["a.txt", "m.txt", "z.txt"]);
}
// -- write_blob / read back --
#[test]
fn write_blob_and_read_back() {
let (_dir, store) = temp_repo();
let content = b"hello world";
let oid = store.write_blob(content).unwrap();
let blob = store.repo().find_blob(oid).unwrap();
assert_eq!(blob.content(), content);
}
// -- write_blob_from_file --
#[test]
fn write_blob_from_file_regular() {
let (_dir, store) = temp_repo();
let tmp = tempfile::NamedTempFile::new().unwrap();
std::fs::write(tmp.path(), b"file content").unwrap();
let (oid, mode) = store.write_blob_from_file(tmp.path()).unwrap();
assert_eq!(mode, FileMode::Blob);
let blob = store.repo().find_blob(oid).unwrap();
assert_eq!(blob.content(), b"file content");
}
#[cfg(unix)]
#[test]
fn write_blob_from_file_executable() {
use std::os::unix::fs::PermissionsExt;
let (_dir, store) = temp_repo();
let tmp = tempfile::NamedTempFile::new().unwrap();
std::fs::write(tmp.path(), b"#!/bin/sh").unwrap();
std::fs::set_permissions(tmp.path(), std::fs::Permissions::from_mode(0o755)).unwrap();
let (_oid, mode) = store.write_blob_from_file(tmp.path()).unwrap();
assert_eq!(mode, FileMode::BlobExecutable);
}
// -- write_empty_tree --
#[test]
fn write_empty_tree() {
let (_dir, store) = temp_repo();
let oid = store.write_empty_tree().unwrap();
let tree = store.repo().find_tree(oid).unwrap();
assert_eq!(tree.len(), 0);
}
// -- write_tree + read_tree roundtrip --
#[test]
fn write_and_read_tree_flat() {
let (_dir, store) = temp_repo();
let oid1 = store.write_blob(b"content a").unwrap();
let oid2 = store.write_blob(b"content b").unwrap();
let mut entries = TreeEntries::new();
entries.set("a.txt", oid1, FileMode::Blob);
entries.set("b.txt", oid2, FileMode::Blob);
let tree_oid = store.write_tree(&entries).unwrap();
let read_back = store.read_tree(tree_oid).unwrap();
assert_eq!(read_back.len(), 2);
assert_eq!(read_back.get("a.txt").unwrap().oid, oid1);
assert_eq!(read_back.get("b.txt").unwrap().oid, oid2);
}
#[test]
fn write_and_read_tree_nested() {
let (_dir, store) = temp_repo();
let oid1 = store.write_blob(b"main").unwrap();
let oid2 = store.write_blob(b"lib").unwrap();
let oid3 = store.write_blob(b"readme").unwrap();
let mut entries = TreeEntries::new();
entries.set("src/main.rs", oid1, FileMode::Blob);
entries.set("src/lib/mod.rs", oid2, FileMode::Blob);
entries.set("README.md", oid3, FileMode::Blob);
let tree_oid = store.write_tree(&entries).unwrap();
let read_back = store.read_tree(tree_oid).unwrap();
assert_eq!(read_back.len(), 3);
assert_eq!(read_back.get("src/main.rs").unwrap().oid, oid1);
assert_eq!(read_back.get("src/lib/mod.rs").unwrap().oid, oid2);
assert_eq!(read_back.get("README.md").unwrap().oid, oid3);
}
// -- write_commit --
#[test]
fn write_commit_orphan() {
let (_dir, store) = temp_repo();
let tree_oid = store.write_empty_tree().unwrap();
let sig = Signature::now("Test", "test@example.com").unwrap();
let commit_oid = store
.write_commit(tree_oid, &[], "initial commit", &sig)
.unwrap();
let commit = store.repo().find_commit(commit_oid).unwrap();
assert_eq!(commit.parent_count(), 0);
assert_eq!(commit.message(), Some("initial commit"));
}
#[test]
fn write_commit_with_parent() {
let (_dir, store) = temp_repo();
let tree_oid = store.write_empty_tree().unwrap();
let sig = Signature::now("Test", "test@example.com").unwrap();
let parent_oid = store
.write_commit(tree_oid, &[], "first commit", &sig)
.unwrap();
let child_oid = store
.write_commit(tree_oid, &[parent_oid], "second commit", &sig)
.unwrap();
let child = store.repo().find_commit(child_oid).unwrap();
assert_eq!(child.parent_count(), 1);
assert_eq!(child.parent_id(0).unwrap(), parent_oid);
}
// -- ref operations --
#[test]
fn update_and_resolve_ref() {
let (_dir, store) = temp_repo();
let tree_oid = store.write_empty_tree().unwrap();
let sig = Signature::now("Test", "test@example.com").unwrap();
let commit_oid = store.write_commit(tree_oid, &[], "initial", &sig).unwrap();
store.update_ref("test-branch", commit_oid).unwrap();
let resolved = store.resolve_ref("test-branch").unwrap();
assert_eq!(resolved, Some(commit_oid));
}
#[test]
fn resolve_ref_nonexistent() {
let (_dir, store) = temp_repo();
let resolved = store.resolve_ref("nonexistent").unwrap();
assert_eq!(resolved, None);
}
#[test]
fn delete_ref_existing() {
let (_dir, store) = temp_repo();
let tree_oid = store.write_empty_tree().unwrap();
let sig = Signature::now("Test", "test@example.com").unwrap();
let commit_oid = store.write_commit(tree_oid, &[], "initial", &sig).unwrap();
store.update_ref("to-delete", commit_oid).unwrap();
store.delete_ref("to-delete").unwrap();
assert_eq!(store.resolve_ref("to-delete").unwrap(), None);
}
#[test]
fn delete_ref_nonexistent_is_noop() {
let (_dir, store) = temp_repo();
store.delete_ref("nonexistent").unwrap();
}
// -- read_blob_at --
#[test]
fn read_blob_at_returns_content() {
let (_dir, store) = temp_repo();
let sig = Signature::now("Test", "test@example.com").unwrap();
let bs = crate::branch::BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("hello.txt", b"world", "add hello").unwrap();
let log = bs.log(1).unwrap();
let commit_oid = log[0].oid;
let content = store.read_blob_at(commit_oid, "hello.txt").unwrap();
assert_eq!(content.unwrap(), b"world");
}
#[test]
fn read_blob_at_returns_none_for_missing_path() {
let (_dir, store) = temp_repo();
let sig = Signature::now("Test", "test@example.com").unwrap();
let bs = crate::branch::BranchStore::new(&store, "test/data", &sig);
bs.ensure_branch().unwrap();
bs.write_entry("hello.txt", b"world", "add hello").unwrap();
let log = bs.log(1).unwrap();
let commit_oid = log[0].oid;
let content = store.read_blob_at(commit_oid, "nonexistent.txt").unwrap();
assert!(content.is_none());
}
}

View file

@ -1,9 +1,4 @@
//! Author identity and commit trailers for workflow code checkpoints.
pub mod author;
pub mod branch;
pub mod error;
pub mod git;
pub mod trailer;
pub const META_BRANCH_PREFIX: &str = "fabro/meta/";
pub use error::{Error, MetadataError, Result};

View file

@ -69,10 +69,6 @@ impl RepoCredentials {
Self::new(None)
}
pub(crate) fn source(&self) -> Option<&Arc<InstallationTokenSource>> {
self.source.as_ref()
}
pub(crate) fn managed(&self) -> bool {
self.source.is_some()
}

View file

@ -17,7 +17,7 @@ use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use fabro_github::GitHubCredentials;
use fabro_github::token_source::{InstallationTokenSource, TokenSnapshot};
use fabro_github::token_source::TokenSnapshot;
use fabro_types::SandboxProviderKind;
use fabro_util::workspace_glob::WorkspaceGlob;
use sandbox_driver::{
@ -983,10 +983,6 @@ impl RunSandbox {
Ok(Some(token.snapshot))
}
pub fn push_token_source(&self) -> Option<Arc<InstallationTokenSource>> {
self.workspace.credentials.source().cloned()
}
/// The local command that opens a shell in the sandbox, from the
/// provider's access facet. `None` when the provider has no such
/// command (the local sandbox is the host).

View file

@ -4427,6 +4427,76 @@ mod tests {
);
}
#[test]
fn historical_metadata_events_and_settings_remain_replayable() {
let mut settings = serde_json::to_value(WorkflowSettings::default()).unwrap();
settings["run"]["meta_branch"] = json!({"enabled": true, "push": true});
let mut events = vec![test_raw_event(
1,
"run.created",
&json!({
"title": "Historical run",
"settings": settings,
"graph": { "name": "test", "nodes": {}, "edges": [], "attrs": {} },
"labels": {},
"provenance": test_support::test_run_provenance()
}),
None,
)];
for (name, properties) in [
(
"metadata.snapshot.started",
json!({
"phase": "init", "branch": "fabro/meta/historical"
}),
),
(
"metadata.snapshot.completed",
json!({
"phase": "init", "branch": "fabro/meta/historical",
"duration_ms": 10, "entry_count": 2, "bytes": 42, "commit_sha": "abc123"
}),
),
(
"metadata.snapshot.failed",
json!({
"phase": "checkpoint", "branch": "fabro/meta/historical",
"duration_ms": 20, "failure_kind": "push", "error": "remote unavailable"
}),
),
] {
let event = test_raw_event(
u32::try_from(events.len()).unwrap() + 1,
name,
&properties,
None,
);
assert!(matches!(
event.event.body,
EventBody::MetadataSnapshotStarted(_)
| EventBody::MetadataSnapshotCompleted(_)
| EventBody::MetadataSnapshotFailed(_)
));
assert_eq!(event.event.event_name(), name);
assert_eq!(event.event.properties().unwrap(), properties);
events.push(event);
}
events.push(test_raw_event(
5,
"run.title.updated",
&json!({ "title": "Replayed historical run" }),
None,
));
let state = RunProjection::apply_events(&events).unwrap();
assert_eq!(state.title, "Replayed historical run");
assert!(
serde_json::to_value(&state.spec.settings).unwrap()["run"]
.get("meta_branch")
.is_none()
);
}
#[test]
fn projection_serialization_includes_manifest_and_definition_blob_refs() {
let manifest_blob = BlobHash::new(br#"{"version":1}"#).to_string();
@ -4487,8 +4557,7 @@ mod tests {
#[test]
fn terminal_conclusion_replays_stage_summaries_without_metadata() {
for terminal_name in ["run.completed", "run.failed"] {
let mut settings = WorkflowSettings::default();
settings.run.meta_branch.enabled = false;
let settings = WorkflowSettings::default();
let mut events = vec![
test_raw_event(
1,

View file

@ -27,7 +27,6 @@ fabro-config = { path = "../../foundation/fabro-config" }
fabro-graphviz = { path = "../fabro-graphviz" }
fabro-hooks = { path = "../fabro-hooks" }
fabro-validate = { path = "../fabro-validate" }
fabro-dump = { path = "../fabro-dump" }
fabro-sandbox = { path = "../fabro-sandbox" }
sandbox-driver.workspace = true
fabro-mcp = { path = "../fabro-mcp" }
@ -77,6 +76,7 @@ tempfile = "3"
toml.workspace = true
fabro-vault = { path = "../../foundation/fabro-vault" }
[dev-dependencies]
fabro-dump = { path = "../fabro-dump" }
fabro-client = { path = "../../foundation/fabro-client" }
fabro-workflow-version = { path = "../fabro-workflow-version" }
fabro-llm = { path = "../fabro-llm", features = ["test-support"] }

View file

@ -724,26 +724,10 @@ impl From<RunEventPersistenceError> for Error {
}
}
impl From<fabro_checkpoint::MetadataError> for Error {
fn from(err: fabro_checkpoint::MetadataError) -> Self {
match err {
err @ fabro_checkpoint::MetadataError::Deserialize {
entity: "checkpoint",
..
} => Self::Checkpoint(err.to_string()),
err => {
let message = err.to_string();
Self::engine_with_source(message, err)
}
}
}
}
pub type Result<T> = std::result::Result<T, Error>;
#[cfg(test)]
mod tests {
use fabro_checkpoint::MetadataError;
use fabro_llm::RetryClassification;
use super::*;
@ -924,41 +908,6 @@ mod tests {
assert!(err.is_err());
}
#[test]
fn metadata_checkpoint_deserialize_error_preserves_source_detail() {
let source = serde_json::from_str::<serde_json::Value>("not json").unwrap_err();
let source_message = source.to_string();
let fabro_error = Error::from(MetadataError::Deserialize {
entity: "checkpoint",
branch: "fabro/meta/run-1".to_string(),
source,
});
assert!(matches!(fabro_error, Error::Checkpoint(_)));
let message = fabro_error.to_string();
assert!(message.contains("deserialize checkpoint on branch fabro/meta/run-1"));
assert!(message.contains(&source_message));
}
#[test]
fn metadata_non_checkpoint_deserialize_error_maps_to_engine_with_source_detail() {
let source = serde_json::from_str::<serde_json::Value>("not json").unwrap_err();
let source_message = source.to_string();
let fabro_error = Error::from(MetadataError::Deserialize {
entity: "run spec",
branch: "fabro/meta/run-1".to_string(),
source,
});
assert!(matches!(fabro_error, Error::Stage {
stage: ErrorStage::Engine,
..
}));
let message = fabro_error.to_string();
assert!(message.contains("deserialize run spec on branch fabro/meta/run-1"));
assert!(message.contains(&source_message));
}
#[test]
fn cancelled_error_display() {
let err = Error::Cancelled;

View file

@ -1,10 +1,7 @@
use std::path::Path;
use std::process::Command;
use anyhow::Context as _;
pub use fabro_checkpoint::META_BRANCH_PREFIX;
pub use fabro_checkpoint::author::GitAuthor;
use fabro_checkpoint::git::Store;
use fabro_redact::DisplaySafeUrl;
use fabro_types::{DirtyStatus, GitContext, WorkflowSettings};
use tokio::task::{JoinError, spawn_blocking};
@ -268,34 +265,6 @@ pub fn remote_branch_sha_noninteractive(
Ok(None)
}
/// Push run and metadata branches to origin if a remote tracking branch exists.
///
/// Callers supply pre-built refspecs so they control force-push (`+` prefix).
#[allow(
clippy::print_stderr,
reason = "Git push status is operator feedback and should stay off stdout."
)]
pub fn push_run_branches(
store: &Store,
probe_branch: &str,
run_refspec: Option<&str>,
meta_refspec: &str,
label: &str,
) -> anyhow::Result<()> {
let repo_path = store.repo_dir();
let remote_ref = format!("refs/remotes/origin/{probe_branch}");
if store.repo().find_reference(&remote_ref).is_err() {
return Ok(());
}
eprintln!("Pushing {label} branches to origin...");
if let Some(refspec) = run_refspec {
push_branch(repo_path, "origin", refspec).context("failed to push run branch")?;
}
push_branch(repo_path, "origin", meta_refspec).context("failed to push metadata branch")?;
eprintln!("Remote refs updated.");
Ok(())
}
/// Error from [`blocking_push_with_timeout`].
pub enum BlockingPushError {
/// The git push itself failed.
@ -394,7 +363,6 @@ pub fn sync_status(repo: &Path, remote: &str, branch: Option<&str>) -> GitSyncSt
}
}
/// Filenames allowed in per-node directories on the shadow branch.
#[cfg(test)]
#[expect(
clippy::disallowed_methods,
@ -713,36 +681,6 @@ mod tests {
assert!(result.is_err());
}
#[test]
fn push_run_branches_preserves_push_error_causes() {
let dir = tempfile::tempdir().unwrap();
init_repo(dir.path());
let repo = git2::Repository::open(dir.path()).unwrap();
let head = repo.head().unwrap().target().unwrap();
repo.reference("refs/remotes/origin/main", head, true, "test")
.unwrap();
let store = Store::new(repo);
let err = push_run_branches(&store, "main", Some("main"), "fabro/meta/test-run", "test")
.unwrap_err();
let chain = err.chain().map(ToString::to_string).collect::<Vec<_>>();
assert!(
chain
.iter()
.any(|cause| cause.contains("failed to push run branch")),
"expected push context in chain, got {chain:#?}"
);
assert!(
chain.len() >= 2,
"expected push source to be preserved, got {chain:#?}"
);
assert!(
chain.iter().any(|cause| cause.contains("git push failed")),
"expected git source in chain, got {chain:#?}"
);
}
#[test]
fn branch_needs_push_when_no_remote_ref() {
let dir = tempfile::tempdir().unwrap();

View file

@ -325,7 +325,6 @@ pub use error::{Error, FailureCategory, FailureSignature, FailureSignatureExt, R
pub use fabro_types::ManifestPath;
pub use steering_hub::{PairControlError, SteeringHub};
pub mod run_materialization;
pub(crate) mod run_metadata;
pub mod run_options;
pub mod run_status;
pub mod runtime_store;

File diff suppressed because it is too large Load diff

View file

@ -37,7 +37,6 @@ use crate::event::Emitter;
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::outcome::{BilledModelUsage, Outcome};
use crate::run_control::RunControlState;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::run_options::RunOptions;
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git_runtime::SandboxGitRuntime;
@ -93,8 +92,6 @@ impl WorkflowLifecycle {
locations: &RunLocations,
run_options: &Arc<RunOptions>,
sandbox_git: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
is_resume: bool,
on_node: crate::OnNodeCallback,
run_control: Option<Arc<RunControlState>>,
@ -159,11 +156,8 @@ impl WorkflowLifecycle {
sandbox: Arc::clone(sandbox),
emitter: Arc::clone(emitter),
run_id: run_options.run_id,
run_store: run_store.clone(),
run_options: Arc::clone(run_options),
sandbox_git,
metadata_runtime,
metadata_writer,
start_node_id,
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
last_git_sha,
@ -419,8 +413,8 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
state: &WfRunState,
) -> CoreResult<()> {
// A StageStart hook can skip before any attempt reserved an execution
// scope. Ensure one exists so Git metadata-snapshot events and the
// `checkpoint.completed` envelope attach to a concrete execution;
// scope. Ensure one exists so the `checkpoint.completed` envelope
// attaches to a concrete execution;
// an existing reservation from the attempt path is reused as-is.
let execution = self
.stage_executions

View file

@ -51,7 +51,6 @@ use crate::pipeline::{
use crate::records::Checkpoint;
use crate::run_control::RunControlState;
use crate::run_materialization::resolve_run_model;
use crate::run_metadata::metadata_branch_name;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions, SetupCommand};
use crate::run_status::{FailureReason, RunStatus};
use crate::runtime_store::RunStoreHandle;
@ -413,7 +412,7 @@ impl RunSession {
Some(RunTarget::Git(_) | RunTarget::None {})
);
let git = (!dry_run_clone_target)
.then(|| git_checkpoint_options_from_start(settings, &record.run_id, state.start))
.then(|| git_checkpoint_options_from_start(settings, state.start))
.flatten();
let definition_blob = state.spec.definition_blob;
let accepted_definition = match definition_blob {
@ -752,7 +751,6 @@ async fn configured_providers_for_start(
fn git_checkpoint_options_from_start(
settings: &fabro_types::WorkflowSettings,
run_id: &RunId,
start: Option<fabro_types::StartRecord>,
) -> Option<GitCheckpointOptions> {
if !settings.run.run_branch.enabled {
@ -761,13 +759,8 @@ fn git_checkpoint_options_from_start(
let start = start?;
start.run_branch.as_ref().map(|_| GitCheckpointOptions {
base_sha: start.base_sha.clone(),
run_branch: start.run_branch.clone(),
meta_branch: settings
.run
.meta_branch
.enabled
.then(|| metadata_branch_name(&run_id.to_string())),
base_sha: start.base_sha.clone(),
run_branch: start.run_branch.clone(),
})
}
@ -2369,27 +2362,7 @@ mod tests {
base_sha: Some("abc123".to_string()),
};
assert!(
git_checkpoint_options_from_start(&settings, &fixtures::RUN_1, Some(start)).is_none()
);
}
#[test]
fn start_record_git_options_honor_disabled_meta_branch() {
let mut settings = WorkflowSettings::default();
settings.run.meta_branch.enabled = false;
let start = fabro_types::StartRecord {
start_time: Utc::now(),
run_branch: Some("fabro/run/test".to_string()),
base_sha: Some("abc123".to_string()),
};
let git = git_checkpoint_options_from_start(&settings, &fixtures::RUN_1, Some(start))
.expect("run branch should remain enabled");
assert_eq!(git.run_branch.as_deref(), Some("fabro/run/test"));
assert_eq!(git.base_sha.as_deref(), Some("abc123"));
assert_eq!(git.meta_branch, None);
assert!(git_checkpoint_options_from_start(&settings, Some(start)).is_none());
}
async fn persisted_workflow_with_settings(

View file

@ -160,8 +160,6 @@ pub async fn execute(init: Initialized) -> Executed {
&engine.run.locations,
&settings_arc,
Arc::clone(&engine.run.sandbox_git),
Arc::clone(&engine.run.metadata_runtime),
engine.run.metadata_writer.clone(),
checkpoint.is_some(),
on_node,
run_control,

View file

@ -1154,9 +1154,8 @@ async fn execute_persists_start_record_and_node_status() {
let dir = tempfile::tempdir().unwrap();
let mut run_options = test_run_options(dir.path(), "test-run");
run_options.git = Some(GitCheckpointOptions {
base_sha: Some("abc123".into()),
run_branch: Some(format!("fabro/run/{}", test_run_id("test-run"))),
meta_branch: None,
base_sha: Some("abc123".into()),
run_branch: Some(format!("fabro/run/{}", test_run_id("test-run"))),
});
let executed = execute_test_run_with_options(run_options, simple_graph(), None).await;

View file

@ -1,12 +1,7 @@
use std::sync::Arc;
use std::time::Instant;
use fabro_dump::RunDump;
use fabro_hooks::{HookContext, HookEvent};
use fabro_types::run_event::{MetadataSnapshotFailureKind, MetadataSnapshotPhase};
use fabro_types::{BilledTokenCounts, DiffSummary, EventBody, RunFailure, RunProjection};
use fabro_util::error::collect_causes;
use fabro_util::time::elapsed_ms;
use super::types::{Concluded, Executed, FinalizeOptions, Finalized, PublishOutcome, Published};
use crate::billing_rollup;
@ -14,7 +9,6 @@ use crate::error::{Error, run_failure_from_error, run_failure_from_outcome_failu
use crate::event::{Event, RunNoticeCode, RunNoticeLevel};
use crate::outcome::{Outcome, StageOutcome};
use crate::records::Conclusion;
use crate::run_metadata::{MetadataSnapshot, metadata_push_failure_is_transient};
use crate::run_options::RunOptions;
use crate::run_status::{FailureReason, RunStatus, SuccessReason};
use crate::runtime_store::RunStoreHandle;
@ -99,205 +93,8 @@ fn build_conclusion_from_projection(
}
}
/// `conclusion` is injected because the terminal event hasn't been emitted
/// yet — the run store's `projection.conclusion` is still `None` at this point.
pub async fn write_finalize_commit(
run_options: &RunOptions,
services: &RunServices,
conclusion: &Conclusion,
) {
if services.metadata_runtime.metadata_suspended() {
return;
}
let Some(writer) = services.metadata_writer.as_ref() else {
return;
};
let Some(meta_branch) = run_options
.git
.as_ref()
.and_then(|git| git.meta_branch.as_deref())
else {
return;
};
let phase = MetadataSnapshotPhase::Finalize;
let started = Instant::now();
emit_metadata_snapshot_started(services, phase, meta_branch);
let mut projection = match services.run_store.state().await {
Ok(state) => state,
Err(err) => {
let message = format!("failed to load run state for final metadata snapshot: {err}");
emit_metadata_snapshot_failed(
services,
phase,
meta_branch,
started,
MetadataSnapshotFailureKind::LoadState,
message.clone(),
collect_causes(err.as_ref()),
None,
None,
None,
);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
false,
);
return;
}
};
projection.conclusion = Some(conclusion.clone());
let dump = match RunDump::from_projection(&projection) {
Ok(dump) => dump,
Err(err) => {
let message = format!("failed to build run dump for final metadata snapshot: {err}");
emit_metadata_snapshot_failed(
services,
phase,
meta_branch,
started,
MetadataSnapshotFailureKind::Write,
message.clone(),
collect_causes(err.as_ref()),
None,
None,
None,
);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
false,
);
return;
}
};
match writer.write_snapshot(&dump, "finalize run").await {
Ok(snapshot) => {
if let Some(detail) = snapshot.push_error.as_deref() {
let message =
format!("failed to push metadata ref refs/heads/{meta_branch}: {detail}");
emit_metadata_snapshot_failed(
services,
phase,
meta_branch,
started,
MetadataSnapshotFailureKind::Push,
message.clone(),
Vec::new(),
Some(snapshot.commit_sha.clone()),
Some(snapshot.entry_count),
Some(snapshot.bytes),
);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataPushFailed,
message,
metadata_push_failure_is_transient(detail, snapshot.token.as_ref()),
);
} else {
services.metadata_runtime.clear_metadata_degraded();
emit_metadata_snapshot_completed(services, phase, meta_branch, started, &snapshot);
}
}
Err(err) => {
let message = format!("failed to write final checkpoint metadata: {err}");
emit_metadata_snapshot_failed(
services,
phase,
meta_branch,
started,
MetadataSnapshotFailureKind::Write,
message.clone(),
collect_causes(&err),
None,
None,
None,
);
emit_metadata_warning(
services,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
false,
);
}
}
}
fn emit_metadata_snapshot_started(
services: &RunServices,
phase: MetadataSnapshotPhase,
branch: &str,
) {
services.emitter.emit(&Event::MetadataSnapshotStarted {
phase,
branch: branch.to_string(),
});
}
fn emit_metadata_snapshot_completed(
services: &RunServices,
phase: MetadataSnapshotPhase,
branch: &str,
started: Instant,
snapshot: &MetadataSnapshot,
) {
services.emitter.emit(&Event::MetadataSnapshotCompleted {
phase,
branch: branch.to_string(),
duration_ms: elapsed_ms(started),
entry_count: snapshot.entry_count,
bytes: snapshot.bytes,
commit_sha: snapshot.commit_sha.clone(),
});
}
#[allow(
clippy::too_many_arguments,
reason = "Metadata failure event carries the full event contract explicitly."
)]
fn emit_metadata_snapshot_failed(
services: &RunServices,
phase: MetadataSnapshotPhase,
branch: &str,
started: Instant,
failure_kind: MetadataSnapshotFailureKind,
error: String,
causes: Vec<String>,
commit_sha: Option<String>,
entry_count: Option<usize>,
bytes: Option<u64>,
) {
services.emitter.emit(&Event::MetadataSnapshotFailed {
phase,
branch: branch.to_string(),
duration_ms: elapsed_ms(started),
failure_kind,
error,
causes,
commit_sha,
entry_count,
bytes,
exec_output_tail: None,
});
}
fn emit_metadata_warning(
services: &RunServices,
code: RunNoticeCode,
message: String,
transient: bool,
) {
if services.metadata_runtime.mark_metadata_degraded(transient) {
services.emitter.notice(RunNoticeLevel::Warn, code, message);
}
}
/// Failed and cancelled runs use a shorter diff timeout so a corrupted
/// workspace can't stall downstream consumers waiting on the terminal event.
/// workspace cannot stall consumers waiting on the terminal event.
async fn compute_final_patch(
run_options: &RunOptions,
services: &RunServices,
@ -507,16 +304,6 @@ pub async fn finalize(published: Published, options: &FinalizeOptions) -> Result
conclusion.status = final_status;
conclusion.failure = failure;
write_finalize_commit(&run_options, &services, &conclusion).await;
if services.metadata_runtime.metadata_degraded() {
services.emitter.notice(
RunNoticeLevel::Warn,
RunNoticeCode::CheckpointMetadataDegraded,
"checkpoint metadata archive writes were degraded for this run".to_string(),
);
}
let terminal_event = build_terminal_event(
&outcome,
conclusion.timing,
@ -576,16 +363,13 @@ mod tests {
use std::time::Duration;
use anyhow::Result;
use async_trait::async_trait;
use bytes::Bytes;
use fabro_auth::test_support as auth_test_support;
use fabro_graphviz::graph::Graph;
use fabro_sandbox::test_support::MockSandbox;
use fabro_store::{Database, EventEnvelope, RunDatabase, RunProjection};
use fabro_types::run_event::{MetadataSnapshotFailureKind, MetadataSnapshotPhase};
use fabro_store::{Database, RunDatabase, RunProjection};
use fabro_types::{
BilledTokenCounts, BlobHash, EventBody, RunEvent, RunId, RunSpec, StageCompletion,
WorkflowSettings, first_event_seq, fixtures, test_support,
BilledTokenCounts, EventBody, RunEvent, RunId, RunSpec, StageCompletion, WorkflowSettings,
first_event_seq, fixtures, test_support,
};
use object_store::memory::InMemory;
@ -594,9 +378,8 @@ mod tests {
use crate::error::ErrorStage;
use crate::event::{Emitter, StoreProgressLogger, append_event};
use crate::records::Checkpoint;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::run_options::{GitCheckpointOptions, RunOptions};
use crate::runtime_store::{RunStoreBackend, RunStoreHandle};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::EngineServices;
@ -622,16 +405,6 @@ mod tests {
}
}
fn test_git_run_options(run_dir: &std::path::Path, meta_branch: &str) -> RunOptions {
let mut options = test_run_options(run_dir);
options.git = Some(GitCheckpointOptions {
base_sha: None,
run_branch: None,
meta_branch: Some(meta_branch.to_string()),
});
options
}
fn test_executed(
graph: Graph,
outcome: Result<Outcome, Error>,
@ -682,11 +455,12 @@ mod tests {
run_id: test_run_id(),
title: None,
settings: serde_json::to_value(WorkflowSettings::default()).unwrap(),
graph: serde_json::to_value(fabro_types::Graph::new("metadata")).unwrap(),
graph: serde_json::to_value(fabro_types::Graph::new("checkpoint"))
.unwrap(),
workflow_source: None,
labels: std::collections::BTreeMap::new(),
source_directory: Some("/tmp/project".to_string()),
workflow_slug: Some("metadata".to_string()),
workflow_slug: Some("checkpoint".to_string()),
workflow_version_id: None,
target: None,
automation: None,
@ -706,7 +480,7 @@ mod tests {
#[expect(
clippy::disallowed_methods,
reason = "metadata event tests use synchronous git commands to set up temporary repositories"
reason = "checkpoint tests use synchronous git commands to set up temporary repositories"
)]
fn init_git_repo(repo: &Path) {
let init = std::process::Command::new("git")
@ -733,7 +507,7 @@ mod tests {
#[expect(
clippy::disallowed_methods,
reason = "metadata event tests use synchronous git commands to set up temporary repositories"
reason = "checkpoint tests use synchronous git commands to set up temporary repositories"
)]
fn git_commit_all(repo: &Path, msg: &str) -> String {
let add = std::process::Command::new("git")
@ -982,8 +756,6 @@ mod tests {
run_store: RunStoreHandle,
emitter: Arc<Emitter>,
sandbox: Arc<fabro_sandbox::RunSandbox>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
) -> Arc<RunServices> {
let locations = crate::services::RunLocations::for_sandbox(
None,
@ -1002,8 +774,6 @@ mod tests {
auth_test_support::vault_only_credential_source(),
Arc::new(fabro_llm::test_support::test_catalog()),
Arc::new(SandboxGitRuntime::new()),
metadata_runtime,
metadata_writer,
crate::stage_execution::StageExecutionTracker::default(),
)
}
@ -1037,8 +807,6 @@ mod tests {
auth_test_support::vault_only_credential_source(),
Arc::new(fabro_llm::test_support::test_catalog()),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
crate::stage_execution::StageExecutionTracker::default(),
);
let executed = test_executed(
@ -1064,204 +832,6 @@ mod tests {
assert_eq!(concluded.conclusion.status, StageOutcome::Succeeded);
}
#[tokio::test]
async fn finalize_metadata_snapshot_success_emits_started_completed_unscoped() {
let repo_dir = tempfile::tempdir().unwrap();
init_git_repo(repo_dir.path());
let branch = "fabro/metadata/run";
let run_store = seeded_run_store().await;
let handle = RunStoreHandle::local(run_store.clone());
let conclusion = Conclusion {
timestamp: chrono::Utc::now(),
status: StageOutcome::Succeeded,
timing: fabro_types::RunTiming::wall_only(10),
failure: None,
final_git_commit_sha: None,
stages: Vec::new(),
billing: None,
total_retries: 0,
diff: fabro_types::RunDiff::default(),
};
let emitter = Arc::new(Emitter::new(test_run_id()));
let events = record_events(&emitter);
let services = test_services(
handle,
emitter,
Arc::new(
fabro_sandbox::local_sandbox(repo_dir.path().to_path_buf())
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
Some(RunMetadataWriterHandle::new_for_test_repo(
repo_dir.path(),
branch,
)),
);
let run_options = test_git_run_options(repo_dir.path(), branch);
write_finalize_commit(&run_options, &services, &conclusion).await;
let events = events.lock().unwrap();
assert_eq!(events.len(), 2);
assert_eq!(events[0].event_name(), "metadata.snapshot.started");
assert_eq!(events[1].event_name(), "metadata.snapshot.completed");
assert!(events[0].node_id.is_none());
match &events[1].body {
EventBody::MetadataSnapshotCompleted(props) => {
assert_eq!(props.phase, MetadataSnapshotPhase::Finalize);
assert_eq!(props.branch, branch);
assert!(!props.commit_sha.is_empty());
}
other => panic!("expected metadata completed event, got {other:?}"),
}
}
#[tokio::test]
async fn finalize_metadata_load_state_failure_emits_failed_before_notice() {
let repo_dir = tempfile::tempdir().unwrap();
init_git_repo(repo_dir.path());
let emitter = Arc::new(Emitter::new(test_run_id()));
let events = record_events(&emitter);
let services = test_services(
RunStoreHandle::new(Arc::new(FailingStateStore)),
emitter,
Arc::new(
fabro_sandbox::local_sandbox(repo_dir.path().to_path_buf())
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
Some(RunMetadataWriterHandle::new_for_test_repo(
repo_dir.path(),
"fabro/metadata/run",
)),
);
let run_options = test_git_run_options(repo_dir.path(), "fabro/metadata/run");
let conclusion = Conclusion {
timestamp: chrono::Utc::now(),
status: StageOutcome::Succeeded,
timing: fabro_types::RunTiming::wall_only(10),
failure: None,
final_git_commit_sha: None,
stages: Vec::new(),
billing: None,
total_retries: 0,
diff: fabro_types::RunDiff::default(),
};
write_finalize_commit(&run_options, &services, &conclusion).await;
let events = events.lock().unwrap();
let names = events.iter().map(RunEvent::event_name).collect::<Vec<_>>();
assert_eq!(names, vec![
"metadata.snapshot.started",
"metadata.snapshot.failed",
"run.notice",
]);
match &events[1].body {
EventBody::MetadataSnapshotFailed(props) => {
assert_eq!(props.phase, MetadataSnapshotPhase::Finalize);
assert_eq!(props.failure_kind, MetadataSnapshotFailureKind::LoadState);
}
other => panic!("expected metadata failed event, got {other:?}"),
}
}
#[tokio::test]
async fn degraded_metadata_runtime_skips_finalize_metadata_events() {
let repo_dir = tempfile::tempdir().unwrap();
init_git_repo(repo_dir.path());
let run_store = seeded_run_store().await;
let emitter = Arc::new(Emitter::new(test_run_id()));
let events = record_events(&emitter);
let runtime = Arc::new(RunMetadataRuntime::new());
runtime.mark_metadata_degraded(false);
let services = test_services(
RunStoreHandle::local(run_store),
emitter,
Arc::new(
fabro_sandbox::local_sandbox(repo_dir.path().to_path_buf())
.await
.unwrap(),
),
runtime,
Some(RunMetadataWriterHandle::new_for_test_repo(
repo_dir.path(),
"fabro/metadata/run",
)),
);
let run_options = test_git_run_options(repo_dir.path(), "fabro/metadata/run");
let conclusion = Conclusion {
timestamp: chrono::Utc::now(),
status: StageOutcome::Succeeded,
timing: fabro_types::RunTiming::wall_only(10),
failure: None,
final_git_commit_sha: None,
stages: Vec::new(),
billing: None,
total_retries: 0,
diff: fabro_types::RunDiff::default(),
};
write_finalize_commit(&run_options, &services, &conclusion).await;
assert!(events.lock().unwrap().is_empty());
}
#[tokio::test]
async fn finalize_emits_metadata_snapshot_before_run_completed() {
let repo_dir = tempfile::tempdir().unwrap();
init_git_repo(repo_dir.path());
let run_store = seeded_run_store().await;
let emitter = Arc::new(Emitter::new(test_run_id()));
let events = record_events(&emitter);
let services = test_services(
RunStoreHandle::local(run_store),
Arc::clone(&emitter),
Arc::new(
fabro_sandbox::local_sandbox(repo_dir.path().to_path_buf())
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
Some(RunMetadataWriterHandle::new_for_test_repo(
repo_dir.path(),
"fabro/metadata/run",
)),
);
let executed = test_executed(
Graph::new("test"),
Ok(Outcome::success()),
test_git_run_options(repo_dir.path(), "fabro/metadata/run"),
5,
services,
);
finalize_executed(executed, &FinalizeOptions {
run_dir: repo_dir.path().to_path_buf(),
run_id: test_run_id(),
workflow_name: "test".to_string(),
preserve_sandbox: false,
stop_on_terminal: true,
last_git_sha: None,
})
.await
.unwrap();
let names = events
.lock()
.unwrap()
.iter()
.map(|event| event.event_name().to_string())
.collect::<Vec<_>>();
assert_eq!(names, vec![
"metadata.snapshot.started",
"metadata.snapshot.completed",
"run.completed",
]);
}
#[tokio::test]
async fn configured_run_branch_without_remote_is_not_reported_as_pushed() {
let repo_dir = tempfile::tempdir().unwrap();
@ -1271,14 +841,11 @@ mod tests {
RunStoreHandle::local(seeded_run_store().await),
emitter,
MockSandbox::linux().sandbox(),
Arc::new(RunMetadataRuntime::new()),
None,
);
let mut run_options = test_run_options(repo_dir.path());
run_options.git = Some(GitCheckpointOptions {
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
meta_branch: None,
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
});
let executed = test_executed(
Graph::new("test"),
@ -1330,14 +897,11 @@ mod tests {
RunStoreHandle::local(seeded_run_store().await),
emitter,
sandbox,
Arc::new(RunMetadataRuntime::new()),
None,
);
let mut run_options = test_run_options(repo_dir.path());
run_options.git = Some(GitCheckpointOptions {
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
meta_branch: None,
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
});
let executed = test_executed(
Graph::new("test"),
@ -1425,15 +989,12 @@ mod tests {
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
None,
);
let mut run_options = test_run_options(repo_dir.path());
run_options.base_branch = Some("main".to_string());
run_options.git = Some(GitCheckpointOptions {
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
meta_branch: None,
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
});
let executed = test_executed(
Graph::new("test"),
@ -1492,15 +1053,12 @@ mod tests {
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
None,
);
let mut run_options = test_run_options(repo_dir.path());
run_options.base_branch = Some("main".to_string());
run_options.git = Some(GitCheckpointOptions {
base_sha: Some("base-sha".to_string()),
run_branch: Some("fabro/run/test".to_string()),
meta_branch: None,
base_sha: Some("base-sha".to_string()),
run_branch: Some("fabro/run/test".to_string()),
});
let executed = test_executed(
Graph::new("test"),
@ -1552,15 +1110,12 @@ mod tests {
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
None,
);
let mut run_options = test_run_options(repo_dir.path());
run_options.base_branch = Some("main".to_string());
run_options.git = Some(GitCheckpointOptions {
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
meta_branch: None,
base_sha: None,
run_branch: Some("fabro/run/test".to_string()),
});
let executed = test_executed(
Graph::new("test"),
@ -1623,8 +1178,6 @@ mod tests {
RunStoreHandle::local(seeded_run_store().await),
Arc::new(Emitter::new(test_run_id())),
sandbox.sandbox(),
Arc::new(RunMetadataRuntime::new()),
None,
);
let executed = test_executed(
Graph::new("test"),
@ -1657,8 +1210,6 @@ mod tests {
RunStoreHandle::local(seeded_run_store().await),
Arc::new(Emitter::new(test_run_id())),
sandbox.sandbox(),
Arc::new(RunMetadataRuntime::new()),
None,
);
let executed = test_executed(
Graph::new("test"),
@ -1708,14 +1259,11 @@ mod tests {
.await
.unwrap(),
),
Arc::new(RunMetadataRuntime::new()),
None,
);
let mut run_options = test_git_run_options(repo, "fabro/metadata/run");
let mut run_options = test_run_options(repo);
run_options.git = Some(GitCheckpointOptions {
base_sha: Some(base),
run_branch: None,
meta_branch: None,
base_sha: Some(base),
run_branch: None,
});
let executed = test_executed(
Graph::new("test"),
@ -1751,33 +1299,4 @@ mod tests {
})
);
}
struct FailingStateStore;
#[async_trait]
impl RunStoreBackend for FailingStateStore {
async fn load_state(&self) -> Result<RunProjection> {
Err(anyhow::anyhow!("state unavailable"))
}
async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
Ok(Vec::new())
}
async fn append_run_event(&self, _event: &RunEvent) -> Result<()> {
Ok(())
}
async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
Ok(BlobHash::new(data))
}
async fn read_blob(&self, _blob_hash: &BlobHash) -> Result<Option<Bytes>> {
Ok(None)
}
async fn read_run_log(&self) -> Result<Option<Vec<u8>>> {
Ok(None)
}
}
}

View file

@ -28,7 +28,6 @@ use crate::handler::llm::{AgentAcpBackend, BackendRouter, PebbleBackend, routing
use crate::handler::{HandlerRegistry, default_registry};
#[cfg(test)]
use crate::model_fallback::ModelFallbackPolicy;
use crate::run_metadata::{RunMetadataRuntime, build_metadata_writer, metadata_branch_name};
use crate::run_options::{GitCheckpointOptions, RunOptions};
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::{
@ -375,7 +374,6 @@ pub async fn initialize(
let search_secrets = search_secrets_from_configured_sources(&options.vault).await;
let catalog = Arc::clone(&options.catalog);
let sandbox_git = Arc::new(SandboxGitRuntime::new());
let metadata_runtime = Arc::new(RunMetadataRuntime::new());
let hook_runner = if options.hooks.hooks.is_empty() {
None
@ -635,13 +633,6 @@ pub async fn initialize(
options.run_options.git = Some(GitCheckpointOptions {
base_sha,
run_branch: Some(info.run_branch.clone()),
meta_branch: options
.run_options
.settings
.run
.meta_branch
.enabled
.then(|| metadata_branch_name(&options.run_options.run_id.to_string())),
});
if options.run_options.base_branch.is_none() {
options.run_options.base_branch = info.base_branch;
@ -720,22 +711,6 @@ pub async fn initialize(
});
}
let metadata_writer =
match build_metadata_writer(&options.run_options, sandbox.push_token_source()) {
Ok(writer) => writer,
Err(err) => {
let message = format!("failed to initialize checkpoint metadata writer: {err}");
if metadata_runtime.mark_metadata_degraded(false) {
options.emitter.notice(
RunNoticeLevel::Warn,
RunNoticeCode::CheckpointMetadataWriteFailed,
message,
);
}
None
}
};
let run_services = RunServices::new(
options.run_store.clone(),
Arc::clone(&options.emitter),
@ -748,8 +723,6 @@ pub async fn initialize(
Arc::clone(&llm_source),
catalog,
sandbox_git,
metadata_runtime,
metadata_writer,
StageExecutionTracker::seeded(stage_executions),
);
let engine = Arc::new(EngineServices {

View file

@ -13,7 +13,7 @@ pub use execute::execute;
pub(crate) use finalize::build_conclusion_from_store;
#[cfg(any(test, feature = "test-support"))]
pub(crate) use finalize::{billing_from_projection, build_terminal_event};
pub use finalize::{classify_engine_result, conclude, finalize, write_finalize_commit};
pub use finalize::{classify_engine_result, conclude, finalize};
pub use initialize::initialize;
pub use parse::parse;
pub(crate) use persist::persist;

File diff suppressed because it is too large Load diff

View file

@ -10,9 +10,8 @@ use crate::git::{GitAuthor, git_author_from_settings};
/// Git checkpoint options for a workflow run.
#[derive(Clone)]
pub struct GitCheckpointOptions {
pub base_sha: Option<String>,
pub run_branch: Option<String>,
pub meta_branch: Option<String>,
pub base_sha: Option<String>,
pub run_branch: Option<String>,
}
/// Options for a workflow run.
@ -30,7 +29,7 @@ pub struct RunOptions {
pub labels: HashMap<String, String>,
/// Workflow directory slug (e.g. "smoke" from `.fabro/workflows/smoke/`).
pub workflow_slug: Option<String>,
/// GitHub credentials for pushing metadata branches to origin.
/// GitHub credentials for sandbox repository access.
pub github_app: Option<fabro_github::GitHubCredentials>,
/// Submitter-side git context captured before the run was created.
pub pre_run_git: Option<GitContext>,

View file

@ -64,7 +64,6 @@ pub async fn git_checkpoint(
node_id: &str,
status: &str,
completed_count: usize,
shadow_sha: Option<String>,
checkpoint: &RunCheckpointSettings,
author: &GitAuthor,
) -> std::result::Result<String, GitCommandError> {
@ -89,7 +88,7 @@ pub async fn git_checkpoint(
let subject = format!("fabro({run_id}): {node_id} ({status})");
let completed_str = completed_count.to_string();
let mut trailers = vec![
let trailers = vec![
Trailer {
key: "Fabro-Run",
value: run_id,
@ -99,13 +98,6 @@ pub async fn git_checkpoint(
value: &completed_str,
},
];
let shadow_sha_ref = shadow_sha.as_deref().unwrap_or("");
if shadow_sha.is_some() {
trailers.push(Trailer {
key: "Fabro-Checkpoint",
value: shadow_sha_ref,
});
}
let mut message = trailerlink::format_message(&subject, "", &trailers);
author.append_footer(&mut message);
@ -129,7 +121,6 @@ pub(crate) async fn checked_git_checkpoint(
node_id: &str,
status: &str,
completed_count: usize,
shadow_sha: Option<String>,
checkpoint: &RunCheckpointSettings,
author: &GitAuthor,
) -> std::result::Result<String, SharedError> {
@ -142,7 +133,6 @@ pub(crate) async fn checked_git_checkpoint(
node_id,
status,
completed_count,
shadow_sha,
checkpoint,
author,
)
@ -560,7 +550,6 @@ mod tests {
"work",
"success",
1,
None,
&RunCheckpointSettings::default(),
&crate::git::GitAuthor::default(),
)
@ -592,7 +581,6 @@ mod tests {
"work",
"success",
1,
None,
&RunCheckpointSettings::default(),
&crate::git::GitAuthor::default(),
)
@ -622,7 +610,6 @@ mod tests {
"work",
"success",
1,
None,
&RunCheckpointSettings::default(),
&crate::git::GitAuthor::default(),
)
@ -642,7 +629,6 @@ mod tests {
"work",
"success",
1,
None,
&RunCheckpointSettings::default(),
&crate::git::GitAuthor::default(),
)
@ -672,7 +658,6 @@ mod tests {
"work",
"success",
1,
Some("feedface".to_owned()),
&checkpoint,
&author,
)
@ -698,7 +683,8 @@ mod tests {
assert!(commit.contains("'--allow-empty'"), "{commit}");
assert!(
commit.contains("fabro(run1): work (success)")
&& commit.contains("Fabro-Checkpoint: feedface"),
&& commit.contains("Fabro-Run: run1")
&& !commit.contains("Fabro-Checkpoint"),
"{commit}"
);
assert!(
@ -804,7 +790,6 @@ mod tests {
"work",
"success",
1,
None,
&RunCheckpointSettings::default(),
&author,
)

View file

@ -19,7 +19,6 @@ use crate::event::Emitter;
use crate::git_identity;
use crate::handler::HandlerRegistry;
use crate::interview_runtime::RunInterviewBlocker;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::stage_execution::StageExecutionTracker;
@ -103,8 +102,6 @@ pub struct RunServices {
pub llm_source: Arc<dyn CredentialProvider>,
pub catalog: Arc<Catalog>,
pub(crate) sandbox_git: Arc<SandboxGitRuntime>,
pub(crate) metadata_runtime: Arc<RunMetadataRuntime>,
pub(crate) metadata_writer: Option<RunMetadataWriterHandle>,
pub(crate) interview_blocker: Arc<RunInterviewBlocker>,
/// Run-scoped stage execution allocator, shared between the core
/// lifecycle and direct-dispatch handlers such as parallel branches.
@ -125,8 +122,6 @@ impl RunServices {
llm_source: Arc<dyn CredentialProvider>,
catalog: Arc<Catalog>,
sandbox_git: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
stage_executions: StageExecutionTracker,
) -> Arc<Self> {
Arc::new(Self {
@ -141,8 +136,6 @@ impl RunServices {
llm_source,
catalog,
sandbox_git,
metadata_runtime,
metadata_writer,
interview_blocker: Arc::new(RunInterviewBlocker::new()),
stage_executions,
})
@ -339,8 +332,6 @@ impl EngineServices {
Arc::new(StubCredentialSource),
Arc::new(fabro_llm::default_catalog()),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
StageExecutionTracker::default(),
),
registry: Arc::new(HandlerRegistry::new(Box::new(start::StartHandler))),

View file

@ -28,7 +28,6 @@ use crate::pipeline;
use crate::pipeline::types::{Executed, Initialized};
use crate::pipeline::{billing_from_projection, build_terminal_event};
use crate::records::Checkpoint;
use crate::run_metadata::RunMetadataRuntime;
use crate::run_options::RunOptions;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::{EngineServices, RunLocations, RunServices};
@ -278,8 +277,6 @@ async fn initialized(
.unwrap_or_else(auth_test_support::vault_only_credential_source),
Arc::new(test_catalog()),
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
StageExecutionTracker::default(),
),
registry: Arc::new(registry),

View file

@ -774,9 +774,8 @@ async fn daytona_git_checkpoint_remote_emits_events() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
run_branch: Some(branch_name),
meta_branch: None,
base_sha: Some(base_sha),
run_branch: Some(branch_name),
}),
};
let outcome = engine
@ -828,14 +827,13 @@ async fn daytona_git_checkpoint_remote_emits_events() {
}
// ---------------------------------------------------------------------------
// Daytona shadow commit E2E with sandbox-native metadata
// Daytona checkpoint E2E without metadata branches
// ---------------------------------------------------------------------------
/// End-to-end test: pipeline with git checkpointing enabled + `meta_branch`
/// writes shadow branch in the sandbox repo and includes `Fabro-Checkpoint`
/// End-to-end test: checkpoint code commits without a metadata branch or
/// trailer in sandbox commits.
#[fabro_macros::e2e_test(live("DAYTONA_API_KEY"), live("GITHUB_APP_PRIVATE_KEY"))]
async fn daytona_git_checkpoint_with_shadow_branch() {
async fn daytona_git_checkpoint_without_metadata_branch() {
let env = create_env().await;
env.initialize().await.unwrap();
let env: Arc<RunSandbox> = Arc::new(env);
@ -867,10 +865,10 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
let (run_id, base_sha, branch_name) = setup_daytona_git(&env).await;
// Pipeline: start -> work -> exit
let mut graph = Graph::new("DaytonaShadowBranch");
let mut graph = Graph::new("DaytonaCodeCheckpoint");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("Test Daytona shadow branch".to_string()),
AttrValue::String("Test Daytona code checkpoints".to_string()),
);
let mut start = Node::new("start");
@ -903,7 +901,6 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
let meta_branch = format!("fabro/meta/{run_id}");
let engine = WorkflowRunner::new(registry, Arc::new(Emitter::default()), env.clone());
let run_options = RunOptions {
settings: WorkflowSettings::default(),
@ -919,9 +916,8 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
run_branch: Some(branch_name),
meta_branch: Some(meta_branch.clone()),
base_sha: Some(base_sha),
run_branch: Some(branch_name),
}),
};
let outcome = engine
@ -930,34 +926,21 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageOutcome::Succeeded);
// Assert shadow branch in the sandbox has checkpoint data
let run_json = env
// Metadata refs are never created in the sandbox.
let refs = env
.exec_command(
&format!("git show refs/heads/{meta_branch}:run.json"),
"git for-each-ref refs/heads/fabro/meta/",
10_000,
None,
None,
None,
)
.await
.expect("git show should succeed");
assert_eq!(run_json.exit_code, Some(0), "{}", run_json.stderr_lossy());
let projection: fabro_store::RunProjection =
serde_json::from_slice(run_json.stdout_lossy().as_bytes()).expect("run.json should parse");
let checkpoint = projection
.current_checkpoint()
.cloned()
.expect("shadow branch should contain checkpoint data");
assert!(
!checkpoint.completed_nodes.is_empty(),
"checkpoint should have completed nodes"
);
assert!(
checkpoint.completed_nodes.contains(&"work".to_string()),
"checkpoint should contain the 'work' node"
);
.expect("git ref listing should succeed");
assert_eq!(refs.exit_code, Some(0), "{}", refs.stderr_lossy());
assert!(refs.stdout_lossy().trim().is_empty());
// Assert sandbox commit has Fabro-Checkpoint trailer
// Run identity remains in code commits, without a metadata SHA.
let log_result = env
.exec_command("git log --format=%B -1", 10_000, None, None, None)
.await
@ -965,8 +948,8 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
assert_eq!(log_result.exit_code, Some(0));
let commit_msg = log_result.stdout_lossy().trim().to_string();
assert!(
commit_msg.contains("Fabro-Checkpoint:"),
"sandbox commit should have Fabro-Checkpoint trailer, got:\n{commit_msg}"
!commit_msg.contains("Fabro-Checkpoint:"),
"sandbox commit should not have Fabro-Checkpoint trailer, got:\n{commit_msg}"
);
assert!(
commit_msg.contains("Fabro-Run:"),
@ -1351,9 +1334,8 @@ async fn daytona_git_push_run_branch_to_origin() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
run_branch: Some(branch_name.clone()),
meta_branch: None,
base_sha: Some(base_sha),
run_branch: Some(branch_name.clone()),
}),
};
let outcome = engine

View file

@ -306,9 +306,8 @@ async fn git_checkpoint_skips_start_node() {
let mut run_options = test_run_options(run_tmp.path());
run_options.git = Some(GitCheckpointOptions {
base_sha: Some(base_sha),
run_branch: None,
meta_branch: Some(format!("fabro/meta/{}", fixtures::RUN_2)),
base_sha: Some(base_sha),
run_branch: None,
});
Box::pin(run_graph(
@ -562,7 +561,6 @@ async fn remote_prompt_demotion_stays_outside_checkout_and_survives_checkpoint()
"work",
"succeeded",
1,
None,
&RunCheckpointSettings::default(),
&GitAuthor::default(),
)
@ -699,9 +697,8 @@ async fn run_identity_governs_engine_and_workflow_commits_everywhere() {
let mut run_options = test_run_options(run_tmp.path());
run_options.git_identity = Some(identity.clone());
run_options.git = Some(GitCheckpointOptions {
base_sha: Some(base_sha),
run_branch: None,
meta_branch: None,
base_sha: Some(base_sha),
run_branch: None,
});
// A `[run.environment]` entry and an inherited host variable both name a

View file

@ -11068,9 +11068,8 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha.clone()),
run_branch: Some(run_branch),
meta_branch: None,
base_sha: Some(base_sha.clone()),
run_branch: Some(run_branch),
}),
};
// 5. Run pipeline
@ -11128,11 +11127,9 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() {
.output();
}
/// End-to-end test: pipeline with git checkpointing enabled + `meta_branch`
/// but no worker-side GitHub credentials still writes run-branch checkpoint
/// commits and skips metadata-branch snapshots.
/// Git checkpointing writes code commits while execution state stays in events.
#[tokio::test]
async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
async fn git_checkpoint_retains_run_history_without_metadata_branch() {
// 1. Create a temporary git repo with an initial commit
let repo = tempfile::tempdir().unwrap();
std::process::Command::new("git")
@ -11156,7 +11153,7 @@ async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
.unwrap();
// 2. Create a branch and worktree
let run_id = test_run_id("test-shadow");
let run_id = test_run_id("test-code-history");
let base_sha = {
let out = std::process::Command::new("git")
.args(["rev-parse", "HEAD"])
@ -11179,14 +11176,26 @@ async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
.output()
.unwrap();
let historical_branch = "fabro/meta/historical";
let historical = std::process::Command::new("git")
.args(["branch", historical_branch, &base_sha])
.current_dir(repo.path())
.output()
.unwrap();
assert!(historical.status.success());
// Write a file in the worktree so there's something to commit
std::fs::write(worktree_path.join("shadow_test.txt"), "shadow branch test").unwrap();
std::fs::write(
worktree_path.join("checkpoint_test.txt"),
"code checkpoint test",
)
.unwrap();
// 3. Build a simple pipeline: start -> work -> exit
let mut graph = Graph::new("ShadowBranchTest");
let mut graph = Graph::new("CodeHistoryTest");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("Test shadow branch".to_string()),
AttrValue::String("Test code history".to_string()),
);
let mut start = Node::new("start");
start.attrs.insert(
@ -11207,11 +11216,10 @@ async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
graph.edges.push(Edge::new("start", "work"));
graph.edges.push(Edge::new("work", "exit"));
// 4. Set up engine with meta_branch
// 4. Set up the workflow engine
let run_dir = tempfile::tempdir().unwrap();
// Write graph.fabro so init_run can read it
std::fs::write(run_dir.path().join("graph.fabro"), "digraph {}").unwrap();
let emitter = Emitter::default();
let events = collect_events(&emitter);
let env: Arc<fabro_sandbox::RunSandbox> = Arc::new(
fabro_sandbox::local_sandbox(worktree_path.clone())
@ -11223,7 +11231,6 @@ async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
registry.register("exit", Box::new(ExitHandler));
let engine = WorkflowRunner::new(registry, Arc::new(emitter), env);
let meta_branch = format!("fabro/meta/{run_id}");
let run_options = RunOptions {
settings: WorkflowSettings::default(),
run_dir: run_dir.path().to_path_buf(),
@ -11238,29 +11245,60 @@ async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
run_branch: Some(format!("fabro/run/{run_id}")),
meta_branch: Some(meta_branch.clone()),
base_sha: Some(base_sha.clone()),
run_branch: Some(format!("fabro/run/{run_id}")),
}),
};
// 5. Run pipeline
let outcome = engine
.run(&graph, &run_options)
let (outcome, state) = engine
.run_with_state(&graph, &run_options)
.await
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageOutcome::Succeeded);
// 6. Without pre-run GitHub credentials, metadata snapshots are disabled.
let run_json = std::process::Command::new("git")
.args(["show", &format!("refs/heads/{meta_branch}:run.json")])
// Existing metadata refs stay unchanged; the run creates none.
let refs = std::process::Command::new("git")
.args([
"for-each-ref",
"--format=%(refname) %(objectname)",
"refs/heads/fabro/meta/",
])
.current_dir(repo.path())
.output()
.expect("git show should run");
assert!(
!run_json.status.success(),
"metadata run.json should not exist without writer prerequisites"
.unwrap();
assert!(refs.status.success());
assert_eq!(
String::from_utf8_lossy(&refs.stdout).trim(),
format!("refs/heads/{historical_branch} {base_sha}")
);
// Events retain the code link and context used by resume and history views.
let events = events.lock().unwrap();
assert!(
!events
.iter()
.any(|event| event.event_name().starts_with("metadata.snapshot."))
);
let checkpoint_event = events
.iter()
.rev()
.find(|event| event.event_name() == "checkpoint.completed")
.expect("checkpoint event");
let properties = checkpoint_event.properties().unwrap();
let sha = properties["git_commit_sha"].as_str().unwrap();
let head = std::process::Command::new("git")
.args(["rev-parse", "HEAD"])
.current_dir(&worktree_path)
.output()
.unwrap();
assert!(head.status.success());
assert_eq!(sha, String::from_utf8_lossy(&head.stdout).trim());
assert_eq!(properties["context_values"]["my_flag"], "set");
let checkpoint = state.current_checkpoint().unwrap();
assert_eq!(checkpoint.git_commit_sha.as_deref(), Some(sha));
assert_eq!(checkpoint.context_values["my_flag"], "set");
assert!(!state.conclusion.as_ref().unwrap().stages.is_empty());
// 7. Assert run-branch commit still has the run checkpoint trailers.
let output = std::process::Command::new("git")
.args(["log", "--format=%B", "-1"])
@ -11278,7 +11316,7 @@ async fn git_checkpoint_host_skips_metadata_branch_without_writer_prereqs() {
);
assert!(
!commit_msg.contains("Fabro-Checkpoint:"),
"run-branch commit should not have Fabro-Checkpoint trailer without metadata snapshot, got:\n{commit_msg}"
"run-branch commit should not have Fabro-Checkpoint trailer after metadata branch removal, got:\n{commit_msg}"
);
// Cleanup worktree
@ -11425,9 +11463,8 @@ async fn parallel_shared_checkout_host_e2e() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha.clone()),
run_branch: Some(run_branch.clone()),
meta_branch: None,
base_sha: Some(base_sha.clone()),
run_branch: Some(run_branch.clone()),
}),
};
// 5. Run pipeline
@ -11680,9 +11717,8 @@ async fn git_checkpoint_host_skips_empty_diff_patch() {
pre_run_git: None,
fork_source_ref: None,
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha.clone()),
run_branch: Some(run_branch),
meta_branch: None,
base_sha: Some(base_sha.clone()),
run_branch: Some(run_branch),
}),
};
let outcome = engine

View file

@ -19,10 +19,6 @@ enabled = true
enabled = true
push = true
[run.meta_branch]
enabled = true
push = true
[run.environment]
id = "default"

View file

@ -41,7 +41,8 @@ pub struct RunLayer {
pub clone: Option<RunCloneLayer>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub run_branch: Option<RunRunBranchLayer>,
#[serde(default, skip_serializing_if = "Option::is_none")]
/// Legacy input only. Metadata branches are no longer written.
#[serde(default, skip_serializing)]
pub meta_branch: Option<RunMetaBranchLayer>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub environment: Option<RunEnvironmentLayer>,
@ -336,7 +337,7 @@ pub struct RunRunBranchLayer {
pub push: Option<bool>,
}
/// `[run.meta_branch]` — Fabro-managed checkpoint metadata branch policy.
/// Legacy `[run.meta_branch]` input, accepted and ignored on config load.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, fabro_macros::Combine)]
#[serde(deny_unknown_fields)]
pub struct RunMetaBranchLayer {

View file

@ -9,8 +9,8 @@ use fabro_types::settings::run::{
NotificationRouteSettings, PreparedStep, PreparedStepRun, PullRequestSettings,
ResolvedMcpEntry, RunAgentSettings, RunBranchSettings, RunCheckpointSettings, RunCloneSettings,
RunExecutionSettings, RunGitSettings, RunGoal, RunIntegrationsGithubSettings,
RunIntegrationsSettings, RunInterviewsSettings, RunMetaBranchSettings, RunModelControls,
RunModelSettings, RunNamespace, RunPrepareSettings, RunScmSettings, ScmGitHubSettings, TlsMode,
RunIntegrationsSettings, RunInterviewsSettings, RunModelControls, RunModelSettings,
RunNamespace, RunPrepareSettings, RunScmSettings, ScmGitHubSettings, TlsMode,
};
use fabro_util::workspace_glob::WorkspaceGlob;
@ -20,8 +20,8 @@ use crate::{
InterviewsLayer, McpEntryLayer, MergeMap, ModelRefOrSplice, NotificationProviderLayer,
NotificationRouteLayer, RunAgentLayer, RunArtifactsLayer, RunCheckpointLayer, RunCloneLayer,
RunExecutionLayer, RunGitLayer, RunGoalLayer, RunIntegrationsGithubLayer, RunIntegrationsLayer,
RunLayer, RunMetaBranchLayer, RunModelLayer, RunPrepareLayer, RunPullRequestLayer,
RunRunBranchLayer, RunScmLayer, StickyMap, StringOrSplice,
RunLayer, RunModelLayer, RunPrepareLayer, RunPullRequestLayer, RunRunBranchLayer, RunScmLayer,
StickyMap, StringOrSplice,
};
pub fn resolve_run(
@ -32,19 +32,6 @@ pub fn resolve_run(
) -> RunNamespace {
let clone = resolve_clone(layer.clone.as_ref(), errors);
let run_branch = resolve_run_branch(layer.run_branch.as_ref());
let mut meta_branch = resolve_meta_branch(layer.meta_branch.as_ref());
if !run_branch.enabled {
if meta_branch.enabled || meta_branch.push {
tracing::debug!(
run_branch_enabled = run_branch.enabled,
"Disabling metadata branch because run branch is disabled"
);
}
meta_branch = RunMetaBranchSettings {
enabled: false,
push: false,
};
}
let pull_request = resolve_pull_request(layer.pull_request.as_ref());
if pull_request.is_some() && (!run_branch.enabled || !run_branch.push) {
errors.push(ResolveError::Invalid {
@ -69,7 +56,6 @@ pub fn resolve_run(
checkpoint: resolve_checkpoint(layer.checkpoint.as_ref()),
clone,
run_branch,
meta_branch,
environment: resolve_run_environment(layer.environment.as_ref(), environments, errors),
notifications: layer
.notifications
@ -395,17 +381,6 @@ fn resolve_run_branch(run_branch: Option<&RunRunBranchLayer>) -> RunBranchSettin
}
}
fn resolve_meta_branch(meta_branch: Option<&RunMetaBranchLayer>) -> RunMetaBranchSettings {
RunMetaBranchSettings {
enabled: meta_branch
.and_then(|meta_branch| meta_branch.enabled)
.unwrap_or(true),
push: meta_branch
.and_then(|meta_branch| meta_branch.push)
.unwrap_or(true),
}
}
fn resolve_notification_route(route: &NotificationRouteLayer) -> NotificationRouteSettings {
NotificationRouteSettings {
enabled: route.enabled.unwrap_or(false),

View file

@ -128,8 +128,6 @@ fn resolves_run_defaults_from_empty_settings() {
assert_eq!(settings.clone.depth, 100);
assert!(settings.run_branch.enabled);
assert!(settings.run_branch.push);
assert!(settings.meta_branch.enabled);
assert!(settings.meta_branch.push);
assert!(settings.pull_request.is_none());
}
@ -314,8 +312,6 @@ push = false
assert_eq!(settings.clone.depth, 1);
assert!(settings.run_branch.enabled);
assert!(!settings.run_branch.push);
assert!(settings.meta_branch.enabled);
assert!(!settings.meta_branch.push);
}
#[test]
@ -358,7 +354,7 @@ depth = -1
}
#[test]
fn disabling_run_branch_forces_meta_branch_off() {
fn legacy_meta_branch_settings_are_ignored() {
let settings = super::workflow_settings_from_toml(
r"
_version = 1
@ -375,8 +371,20 @@ push = true
.run;
assert!(!settings.run_branch.enabled);
assert!(!settings.meta_branch.enabled);
assert!(!settings.meta_branch.push);
assert!(
serde_json::to_value(&settings)
.unwrap()
.get("meta_branch")
.is_none()
);
let layer: SettingsLayer = "[run.meta_branch]\nenabled = true\npush = true\n"
.parse()
.unwrap();
assert!(
serde_json::to_value(&layer).unwrap()["run"]
.get("meta_branch")
.is_none()
);
}
#[test]

View file

@ -144,10 +144,16 @@ pub enum EventBody {
RunFailed(RunFailedProps),
#[serde(rename = "run.notice")]
RunNotice(RunNoticeProps),
/// Historical metadata snapshot event. Retained for replay; no longer
/// emitted.
#[serde(rename = "metadata.snapshot.started")]
MetadataSnapshotStarted(MetadataSnapshotStartedProps),
/// Historical metadata snapshot event. Retained for replay; no longer
/// emitted.
#[serde(rename = "metadata.snapshot.completed")]
MetadataSnapshotCompleted(MetadataSnapshotCompletedProps),
/// Historical metadata snapshot event. Retained for replay; no longer
/// emitted.
#[serde(rename = "metadata.snapshot.failed")]
MetadataSnapshotFailed(MetadataSnapshotFailedProps),
#[serde(rename = "stage.started")]

View file

@ -36,7 +36,6 @@ pub struct RunNamespace {
pub checkpoint: RunCheckpointSettings,
pub clone: RunCloneSettings,
pub run_branch: RunBranchSettings,
pub meta_branch: RunMetaBranchSettings,
pub environment: RunEnvironmentSettings,
pub notifications: HashMap<String, NotificationRouteSettings>,
pub interviews: RunInterviewsSettings,
@ -66,7 +65,6 @@ impl Default for RunNamespace {
checkpoint: RunCheckpointSettings::default(),
clone: RunCloneSettings::default(),
run_branch: RunBranchSettings::default(),
meta_branch: RunMetaBranchSettings::default(),
environment: RunEnvironmentSettings::default(),
notifications: HashMap::new(),
interviews: RunInterviewsSettings::default(),
@ -1115,21 +1113,6 @@ impl Default for RunBranchSettings {
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunMetaBranchSettings {
pub enabled: bool,
pub push: bool,
}
impl Default for RunMetaBranchSettings {
fn default() -> Self {
Self {
enabled: true,
push: true,
}
}
}
#[derive(
Debug,
Clone,

View file

@ -387,7 +387,6 @@ models/run-interviews-settings.ts
models/run-lifecycle.ts
models/run-links.ts
models/run-manifest.ts
models/run-meta-branch-settings.ts
models/run-mode.ts
models/run-model-controls.ts
models/run-model-settings.ts

View file

@ -358,7 +358,6 @@ export * from './run-interviews-settings';
export * from './run-lifecycle';
export * from './run-links';
export * from './run-manifest';
export * from './run-meta-branch-settings';
export * from './run-mode';
export * from './run-model';
export * from './run-model-controls';

View file

@ -1,20 +0,0 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
export interface RunMetaBranchSettings {
'enabled': boolean;
'push': boolean;
}

View file

@ -57,9 +57,6 @@ import type { RunIntegrationsSettings } from './run-integrations-settings';
import type { RunInterviewsSettings } from './run-interviews-settings';
// May contain unused imports in some cases
// @ts-ignore
import type { RunMetaBranchSettings } from './run-meta-branch-settings';
// May contain unused imports in some cases
// @ts-ignore
import type { RunModelSettings } from './run-model-settings';
// May contain unused imports in some cases
// @ts-ignore
@ -83,7 +80,6 @@ export interface RunNamespace {
'checkpoint': RunCheckpointSettings;
'clone': RunCloneSettings;
'run_branch': RunBranchSettings;
'meta_branch': RunMetaBranchSettings;
'environment': RunEnvironmentSettings;
'notifications': { [key: string]: NotificationRouteSettings; };
'interviews': RunInterviewsSettings;