graflo.cli.commit

Git-shaped verbs over a manifest's commit history.

Distinct from migrate-schema, which plans and executes changes against a database. These record and replay changes to the manifest -- the contract plane. Nothing here touches a backend.

The verbs are deliberately the ones a git user already knows: commit, log, verify, checkout, merge, revert. Where the semantics differ from git they differ visibly -- checkout replays from a base manifest rather than restoring a snapshot, because a manifest history stores change sets, not trees.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

checkout_cmd(commit_id, base_path, store, output_path)

Reconstruct the manifest as of a commit (default: the head).

Source code in graflo/cli/commit.py

@click.command("checkout")
@click.argument("commit_id", required=False)
@click.option(
    "--base",
    "base_path",
    type=click.Path(exists=True, path_type=Path),
    required=True,
    help="Manifest the root commit starts from; replay is exact from here.",
)
@_store_option
@click.option("--output-path", type=click.Path(path_type=Path), default=None)
def checkout_cmd(
    commit_id: str | None, base_path: Path, store: Path, output_path: Path | None
) -> None:
    """Reconstruct the manifest as of a commit (default: the head)."""
    base = _load(base_path)
    history = FileCommitStore(store).load()
    try:
        result = checkout(base, history, commit_id)
    except CommitError as exc:
        raise click.ClickException(str(exc)) from exc

    click.echo(f"manifest hash: {manifest_hash(result)[:12]}")
    _write(result, output_path)

commit_cmd(from_path, to_path, label, onto, hints_path, store, as_root, dry)

Record the change between two manifests as a commit.

Source code in graflo/cli/commit.py

@click.command("commit")
@click.option(
    "--from-manifest",
    "from_path",
    type=click.Path(exists=True, path_type=Path),
    required=True,
    help="Base manifest (the state before the change).",
)
@click.option(
    "--to-manifest",
    "to_path",
    type=click.Path(exists=True, path_type=Path),
    required=True,
    help="Target manifest (the state after the change).",
)
@click.option("-m", "--label", default=None, help="Short human-readable name.")
@click.option(
    "--onto",
    default=None,
    help="Parent commit id. Defaults to the single head; required when forked.",
)
@click.option(
    "--hints",
    "hints_path",
    type=click.Path(exists=True, path_type=Path),
    default=None,
    help="YAML RenameHints; a drop plus an add is otherwise not a rename.",
)
@_store_option
@click.option(
    "--root",
    "as_root",
    is_flag=True,
    default=False,
    help=(
        "Start a new lineage instead of extending the head. A store can hold "
        "several unrelated lineages, which is what merge later joins."
    ),
)
@click.option("--dry", is_flag=True, default=False, help="Print without storing.")
def commit_cmd(
    from_path: Path,
    to_path: Path,
    label: str | None,
    onto: str | None,
    hints_path: Path | None,
    store: Path,
    as_root: bool,
    dry: bool,
) -> None:
    """Record the change between two manifests as a commit."""
    base, target = _load(from_path), _load(to_path)
    ops, warnings = diff_manifests_verified(base, target, hints=_hints(hints_path))

    for warning in warnings:
        click.echo(f"warning: {warning}", err=True)
    if not ops:
        click.echo("No operations derived; nothing to record.")
        return

    history = FileCommitStore(store).load()
    parents = _resolve_parents(history, onto, as_root=as_root)

    try:
        entry = build_commit(
            base,
            ops,
            parents=parents,
            label=label,
            created_at=datetime.now(UTC).isoformat(),
        )
    except CommitError as exc:
        raise click.ClickException(str(exc)) from exc

    click.echo(f"commit    : {entry.id}")
    click.echo(f"parents   : {', '.join(entry.parents) or '-'}")
    click.echo(f"label     : {entry.label or '-'}")
    click.echo(f"reversible: {entry.reversible}")
    click.echo(f"tree      : {(entry.tree_before or '-')[:12]} -> {entry.tree[:12]}")
    click.echo("\noperations:")
    click.echo(ops_to_yaml_str(list(entry.ops)))

    if warnings:
        click.echo(
            "Refusing to store: the derived change set does not fully reproduce "
            "the target manifest (see warnings above).",
            err=True,
        )
        raise click.ClickException("incomplete change set")
    if dry:
        click.echo("(dry run -- not stored)")
        return

    click.echo(f"stored: {_append(store, entry)}")

commit_group()

The verbs this module contributes to the graflo group.

Source code in graflo/cli/commit.py

def commit_group() -> dict[str, click.Command]:
    """The verbs this module contributes to the ``graflo`` group."""
    return {
        "commit": commit_cmd,
        "log": log_cmd,
        "verify": verify_cmd,
        "checkout": checkout_cmd,
        "merge3": merge3_cmd,
        "revert": revert_cmd,
        "stamp": stamp_cmd,
        "rehash": rehash_cmd,
    }

log_cmd(store, graph)

List the commit history, oldest first.

Source code in graflo/cli/commit.py

@click.command("log")
@_store_option
@click.option("--graph", is_flag=True, default=False, help="Show parent edges.")
def log_cmd(store: Path, graph: bool) -> None:
    """List the commit history, oldest first."""
    history = FileCommitStore(store).load()
    if not history.commits:
        click.echo("No commits stored.")
        return

    heads = {head.id for head in history.heads()}
    click.echo(f"{'commit':<10} {'kind':<8} {'rev?':<5} {'ops':<4} label")
    for entry in history.topological():
        marker = " (head)" if entry.id in heads else ""
        click.echo(
            f"{entry.short():<10} {entry.kind:<8} {entry.reversible!s:<5} "
            f"{len(entry.ops):<4} {entry.label or '-'}{marker}"
        )
        if graph and entry.parents:
            click.echo(
                f"           └─ parents: {', '.join(p[:8] for p in entry.parents)}"
            )

    if len(heads) > 1:
        click.echo(
            f"\nhistory has {len(heads)} heads -- it has forked. "
            "Use `graflo merge3` to reconcile them."
        )

merge3_cmd(left, right, base_path, store, take, label, output_path, dry, plot_path, history_plot_path)

Merge two commits, reconciling against their common ancestor.

Source code in graflo/cli/commit.py

@click.command("merge3")
@click.argument("left")
@click.argument("right")
@click.option(
    "--base",
    "base_path",
    type=click.Path(exists=True, path_type=Path),
    required=True,
    help="Manifest the root commit starts from.",
)
@_store_option
@click.option(
    "--take",
    type=click.Choice(["left", "right"]),
    default=None,
    help="Resolve every conflict by taking one side. Omit to see them first.",
)
@click.option("-m", "--label", default=None, help="Short human-readable name.")
@click.option("--output-path", type=click.Path(path_type=Path), default=None)
@click.option("--dry", is_flag=True, default=False, help="Report without storing.")
@click.option(
    "--plot",
    "plot_path",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help=(
        "Draw the slot tree here -- where the two branches met, and what each "
        "did. Written even when conflicts stop the merge."
    ),
)
@click.option(
    "--plot-history",
    "history_plot_path",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help="Draw the commit DAG here, with the merge base marked.",
)
def merge3_cmd(
    left: str,
    right: str,
    base_path: Path,
    store: Path,
    take: str | None,
    label: str | None,
    output_path: Path | None,
    dry: bool,
    plot_path: Path | None,
    history_plot_path: Path | None,
) -> None:
    """Merge two commits, reconciling against their common ancestor."""
    base_manifest = _load(base_path)
    history = FileCommitStore(store).load()

    left_commit = history.require(left)
    right_commit = history.require(right)

    merge_base_id = find_merge_base(history, left_commit.id, right_commit.id)
    if merge_base_id is None:
        raise click.ClickException(
            f"{left_commit.short()} and {right_commit.short()} share no ancestor. "
            "Unrelated lineages are joined by `graflo merge` (declared "
            "equivalence), not by a three-way merge3."
        )

    ancestor = checkout(base_manifest, history, merge_base_id)
    left_state = checkout(base_manifest, history, left_commit.id)
    right_state = checkout(base_manifest, history, right_commit.id)

    click.echo(f"merge base: {merge_base_id[:8]}")
    if history_plot_path is not None:
        _plot_history(
            history,
            history_plot_path,
            heads=[left_commit.id, right_commit.id],
            merge_base=merge_base_id,
        )

    merged, result = merge_three_way(ancestor, left_state, right_state)

    # Drawn before the refusal below, since an unresolved merge is exactly the
    # case the picture is for.
    if plot_path is not None:
        _plot_slots(result, plot_path)

    if result.conflicts and take is None:
        click.echo(f"\n{len(result.conflicts)} conflict(s):")
        for conflict in result.conflicts:
            click.echo(f"  {describe_slot(conflict.slot_key)}  -- {conflict.reason}")
            click.echo(f"    left : {[op.op for op in conflict.left_ops]}")
            click.echo(f"    right: {[op.op for op in conflict.right_ops]}")
        raise click.ClickException(
            "unresolved conflicts; re-run with --take left/right, or resolve "
            "them through the server API"
        )

    resolutions = []
    if result.conflicts:
        chooser = take_left if take == "left" else take_right
        resolutions = [chooser(conflict) for conflict in result.conflicts]
        merged, result = merge_three_way(
            ancestor, left_state, right_state, resolutions=resolutions
        )

    if merged is None:
        raise click.ClickException("merge did not resolve; nothing to record")

    for warning in result.warnings:
        click.echo(f"warning: {warning}", err=True)
    click.echo(f"merged hash: {manifest_hash(merged)[:12]}")

    recipe = build_recipe(ancestor, left_state, right_state, resolutions=resolutions)
    stamp_provenance(
        merged,
        content_hash=manifest_hash(merged),
        canon=CANON_VERSION,
        parents=[left_commit.id, right_commit.id],
        merge_recipe=recipe.content_hash(),
    )
    _write(merged, output_path)

    if dry:
        click.echo("(dry run -- not stored)")
        return

    from graflo.architecture.evolution.commit import MergeRecipeRef

    try:
        entry = build_multi_parent_commit(
            left_state,
            merged,
            parents=[left_commit.id, right_commit.id],
            label=label or f"merge {right_commit.short()} into {left_commit.short()}",
            created_at=datetime.now(UTC).isoformat(),
            merge_recipe=MergeRecipeRef(
                hash=recipe.content_hash(), kind=recipe.kind, payload=recipe.to_dict()
            ),
        )
    except CommitError as exc:
        raise click.ClickException(str(exc)) from exc

    click.echo(f"commit: {entry.id}")
    click.echo(f"stored: {_append(store, entry)}")

rehash_cmd(store, dry)

Recompute every commit id from the current op serialization.

A commit id is derived from its ops as serialized, so a release that renames an op field changes the id of every commit carrying that op -- and, through the parent chain, of every commit after it. This walks the history in topological order, recomputes each id against its already-rehashed parents, and rewrites the store. Trees are untouched: the ops still replay to the same manifests, only their names on disk move.

Source code in graflo/cli/commit.py

@click.command("rehash")
@_store_option
@click.option("--dry", is_flag=True, default=False, help="Report without rewriting.")
def rehash_cmd(store: Path, dry: bool) -> None:
    """Recompute every commit id from the current op serialization.

    A commit id is derived from its ops as serialized, so a release that renames
    an op field changes the id of every commit carrying that op -- and, through
    the parent chain, of every commit after it. This walks the history in
    topological order, recomputes each id against its already-rehashed parents,
    and rewrites the store. Trees are untouched: the ops still replay to the same
    manifests, only their names on disk move.
    """
    history = FileCommitStore(store).load()
    mapping: dict[str, str] = {}
    rebuilt: list[Commit] = []
    for commit in history.topological():
        parents = [mapping.get(parent, parent) for parent in commit.parents]
        new_id = compute_commit_id(list(commit.ops), parents)
        mapping[commit.id] = new_id
        rebuilt.append(commit.model_copy(update={"id": new_id, "parents": parents}))

    changed = {old: new for old, new in mapping.items() if old != new}
    if not changed:
        click.echo(f"every commit id is current ({len(history.commits)} commit(s))")
        return
    for old, new in changed.items():
        click.echo(f"{old} -> {new}")
    if dry:
        click.echo(f"{len(changed)} commit id(s) would change (dry run)")
        return
    FileCommitStore(store).save(History(commits=rebuilt))
    click.echo(f"rewrote {len(changed)} commit id(s) in {store}")

revert_cmd(commit_id, base_path, store, label, output_path, dry)

Record a new commit undoing an earlier one.

History is append-only: undoing a change moves forward, it never edits what was recorded.

Source code in graflo/cli/commit.py

@click.command("revert")
@click.argument("commit_id")
@click.option(
    "--base",
    "base_path",
    type=click.Path(exists=True, path_type=Path),
    required=True,
    help="Manifest the root commit starts from.",
)
@_store_option
@click.option("-m", "--label", default=None)
@click.option("--output-path", type=click.Path(path_type=Path), default=None)
@click.option("--dry", is_flag=True, default=False)
def revert_cmd(
    commit_id: str,
    base_path: Path,
    store: Path,
    label: str | None,
    output_path: Path | None,
    dry: bool,
) -> None:
    """Record a new commit undoing an earlier one.

    History is append-only: undoing a change moves forward, it never edits what
    was recorded.
    """
    base_manifest = _load(base_path)
    history = FileCommitStore(store).load()
    target = history.require(commit_id)

    heads = history.heads()
    if len(heads) != 1:
        raise click.ClickException(
            f"expected a single head, found {len(heads)}; reconcile them first"
        )
    head = heads[0]
    current = checkout(base_manifest, history, head.id)

    try:
        entry = build_revert_commit(
            current,
            target,
            parents=[head.id],
            label=label,
            created_at=datetime.now(UTC).isoformat(),
        )
    except CommitError as exc:
        raise click.ClickException(str(exc)) from exc

    click.echo(f"commit: {entry.id}")
    click.echo(f"tree  : {(entry.tree_before or '-')[:12]} -> {entry.tree[:12]}")
    click.echo("\noperations:")
    click.echo(ops_to_yaml_str(list(entry.ops)))

    if output_path is not None:
        from graflo.architecture.evolution.apply import apply_evolution

        _write(
            apply_evolution(
                current, list(entry.ops), bump_version=False, finish_init=False
            ),
            output_path,
        )
    if dry:
        click.echo("(dry run -- not stored)")
        return
    click.echo(f"stored: {_append(store, entry)}")

stamp_cmd(manifest_path, commit_id, store, output_path)

Write the content address and lineage into a manifest's metadata.

Stamping is explicit rather than automatic because a manifest that stamps itself on every apply would disagree with itself about its own lineage.

Source code in graflo/cli/commit.py

@click.command("stamp")
@click.argument("manifest_path", type=click.Path(exists=True, path_type=Path))
@click.option("--commit", "commit_id", default=None, help="Commit that produced it.")
@_store_option
@click.option("--output-path", type=click.Path(path_type=Path), default=None)
def stamp_cmd(
    manifest_path: Path, commit_id: str | None, store: Path, output_path: Path | None
) -> None:
    """Write the content address and lineage into a manifest's metadata.

    Stamping is explicit rather than automatic because a manifest that stamps
    itself on every apply would disagree with itself about its own lineage.
    """
    manifest = _load(manifest_path)
    history = FileCommitStore(store).load()

    parents: list[str] = []
    if commit_id is not None:
        commit = history.require(commit_id)
        parents = list(commit.parents)
        commit_id = commit.id

    provenance = stamp_provenance(
        manifest,
        content_hash=manifest_hash(manifest),
        canon=CANON_VERSION,
        commit=commit_id,
        parents=parents,
    )
    click.echo(f"content_hash: {provenance.content_hash}")
    click.echo(f"canon       : {provenance.canon}")
    click.echo(f"commit      : {provenance.commit or '-'}")
    click.echo(f"parents     : {', '.join(provenance.parents) or '-'}")
    _write(manifest, output_path or manifest_path)

verify_cmd(base_path, against_path, store)

Replay every head, checking each recorded tree hash.

Source code in graflo/cli/commit.py

@click.command("verify")
@click.option(
    "--base",
    "base_path",
    type=click.Path(exists=True, path_type=Path),
    required=True,
    help="Manifest the root commit starts from.",
)
@click.option(
    "--against",
    "against_path",
    type=click.Path(exists=True, path_type=Path),
    default=None,
    help="Manifest the fully replayed history should equal.",
)
@_store_option
def verify_cmd(base_path: Path, against_path: Path | None, store: Path) -> None:
    """Replay every head, checking each recorded tree hash."""
    base = _load(base_path)
    history = FileCommitStore(store).load()

    problems = verify_history(base, history)
    if problems:
        for problem in problems:
            click.echo(f"error: {problem}", err=True)
        raise click.ClickException(f"{len(problems)} head(s) failed to replay")
    click.echo(
        f"history replays cleanly ({len(history.commits)} commit(s), "
        f"{len(history.heads())} head(s))"
    )

    if against_path is not None:
        expected = manifest_hash(_load(against_path))
        actual = manifest_hash(checkout(base, history))
        if actual != expected:
            raise click.ClickException(
                f"replayed manifest hashes {actual[:12]}, expected {expected[:12]}"
            )
        click.echo(f"matches {against_path}")