//! `stack create`, `stack resubmit` and `stack merge`, driven as sequences.
//!
//! These are the commands with the most moving parts and the least
//! observable failure modes. What a stack *is* — a chain of pull records
//! each `dependentOn` the one beneath, correlated to commits by a change-id
//! that lives in a patch header and nowhere else — is a relationship between
//! several records and several commits, and no single function holds it. A
//! unit test of the reconcile planner proves the plan; only a sequence
//! proves the plan was executed against the right records, in the right
//! order, by the right account.
//!
//! So the tests that matter here are sequences. A `create` on its own cannot
//! be wrong about a change-id, because nothing has read one back yet; a
//! `resubmit`'s correctness is defined entirely by what the `create` before
//! it wrote. Most of what follows is a `create` and one command after it,
//! which is the shortest sequence that can be wrong about anything — but the
//! shortest is not the only length, and a bug has already been found living
//! at the third command. `every_edit_leaves_the_two_readers_agreeing` is the
//! long one, and the section it sits in says why length is worth paying for.
//!
//! See `tests/support/mod.rs` for the environment and the argument for it.

mod support;

use support::world::KNOT_PATCH;
use support::{ALICE, BOB, CAROL, PULL_NSID, PULL_STATUS_NSID, REPO_DID, Scenario};

const BOTTOM: &str = "Ibottom00000000000000000000000000000000a";
const MIDDLE: &str = "Imiddle00000000000000000000000000000000a";
const TOP: &str = "Itop00000000000000000000000000000000000a";

/// Three commits with change-ids, on a branch, ready to stack.
///
/// The mock knot is taught to agree with them. `stack create` pushes the
/// branch and then asks `sh.tangled.repo.compare` what the knot holds — not
/// for the patch, which is formatted per commit here, but as the proof the
/// push landed — and the default world answers about one commit and a
/// mailbox with no change-ids in it. Both are true of *some* branch and
/// neither is true of this one, so the ordinary tests would run under two
/// notes about a disagreement they are not testing.
fn three_commit_branch(world: &Scenario) {
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", Some(MIDDLE));
    world
        .checkout
        .commit("three.txt", "three\n", "feat: top", Some(TOP));
    world.with(|w| w.compare = Ok((3, knot_mailbox(&[BOTTOM, MIDDLE, TOP]))));
}

/// A knot's format-patch for `ids`, in shape only: what `stack create` reads
/// out of it is whether the `Change-Id:` headers are there at all.
fn knot_mailbox(ids: &[&str]) -> String {
    ids.iter()
        .map(|id| format!("{KNOT_PATCH}Change-Id: {id}\n"))
        .collect()
}

/// Four commits with change-ids: one more than an unmarked stack may open
/// without being asked about.
fn four_commit_branch(world: &Scenario) {
    three_commit_branch(world);
    world.checkout.commit(
        "four.txt",
        "four\n",
        "feat: fourth",
        Some("Ifourth0000000000000000000000000000000a"),
    );
}

/// A stacked branch that has already been published.
fn published_stack(label: &str) -> Scenario {
    let world = Scenario::new(label);
    three_commit_branch(&world);
    world.run(&["stack", "create"]).success();
    world.clear_journal();
    world
}

/// The record keys of Alice's pulls, bottom first.
fn keys(world: &Scenario) -> Vec<String> {
    world.pulls(ALICE).into_iter().map(|(k, _)| k).collect()
}

/// The `dependentOn` chain as record keys, bottom first. `None` is the
/// bottom, which depends on nothing.
fn chain(world: &Scenario, did: &str) -> Vec<(String, Option<String>)> {
    world
        .pulls(did)
        .into_iter()
        .map(|(rkey, value)| {
            let parent = value["dependentOn"]
                .as_str()
                .map(|uri| uri.rsplit('/').next().unwrap_or_default().to_string());
            (rkey, parent)
        })
        .collect()
}

/// Assert the chain is exactly `keys` in order, each depending on the one
/// before it and the first on `anchor`.
fn assert_chained(world: &Scenario, did: &str, anchor: Option<&str>) {
    let chain = chain(world, did);
    assert_eq!(
        chain[0].1.as_deref(),
        anchor,
        "the bottom of the chain points at the wrong thing: {chain:?}"
    );
    for pair in chain.windows(2) {
        assert_eq!(
            pair[1].1.as_deref(),
            Some(pair[0].0.as_str()),
            "the chain is broken between {} and {}: {chain:?}",
            pair[0].0,
            pair[1].0,
        );
    }
}

/// Every member's title in *chain* order, walked down `dependentOn`.
///
/// [`titles`] reads them in record-key order, which is minting order, and the
/// two agree only while every record was minted in one pass. A re-cut mints a
/// new member in the middle of an existing stack, so it is the case that
/// tells the two apart — and chain order is the one that matters, since it is
/// the order Tangled reviews and merges in.
fn chain_titles(world: &Scenario, did: &str) -> Vec<String> {
    let pulls = world.pulls(did);
    let parent_of = |v: &serde_json::Value| {
        v["dependentOn"]
            .as_str()
            .map(|uri| uri.rsplit('/').next().unwrap_or_default().to_string())
    };
    let mut next = pulls
        .iter()
        .find(|(_, v)| parent_of(v).is_none())
        .map(|(rkey, _)| rkey.clone());
    let mut out = Vec::new();
    while let Some(rkey) = next {
        let (_, value) = pulls
            .iter()
            .find(|(k, _)| *k == rkey)
            .expect("a chain names records that exist");
        out.push(value["title"].as_str().unwrap_or_default().to_string());
        next = pulls
            .iter()
            .find(|(_, v)| parent_of(v).as_deref() == Some(rkey.as_str()))
            .map(|(k, _)| k.clone());
    }
    out
}

fn titles(world: &Scenario, did: &str) -> Vec<String> {
    world
        .pulls(did)
        .into_iter()
        .map(|(_, v)| v["title"].as_str().unwrap_or_default().to_string())
        .collect()
}

// ---------------------------------------------------------------------------
// create
// ---------------------------------------------------------------------------

/// The whole point of `stack create`: one record per commit, chained bottom
/// to top, aimed at the repo's own DID, written as one batch.
///
/// Four claims in one test because they are one write — splitting them would
/// mean four `stack create` runs asserting four quarters of the same
/// `applyWrites` body.
#[test]
fn create_writes_one_chained_record_per_commit_in_a_single_batch() {
    let world = Scenario::new("create-chain");
    three_commit_branch(&world);

    world.run(&["stack", "create"]).success();

    // One record per commit, bottom first: TIDs are monotonic, so record-key
    // order is minting order, which is stack order.
    assert_eq!(
        titles(&world, ALICE),
        ["feat: bottom", "feat: middle", "feat: top"],
    );
    assert_chained(&world, ALICE, None);

    // One applyWrites, not three writes. A chain written record by record
    // passes through states the appview rejects at ingest — two pulls
    // depending on the same target — even though the end state is linear.
    let batches = world.with(|w| w.calls_to("com.atproto.repo.applyWrites").len());
    assert_eq!(batches, 1, "a stack is one atomic batch");

    // Every record aims at the repo's own DID, which on Tangled is never its
    // owner's. Sending the owner's is silent: both are well-formed DIDs.
    for (rkey, value) in world.pulls(ALICE) {
        assert_eq!(
            value["target"]["repoDid"].as_str(),
            Some(REPO_DID),
            "{rkey} aims at the wrong repo: {value:#}"
        );
        assert_eq!(value["target"]["branch"].as_str(), Some("main"));
        assert_eq!(value["source"]["branch"].as_str(), Some("feature"));
    }
}

/// A member's change-id lives in the *patch*, as a mail header, and nowhere
/// else in the record. Every later reconcile matches on it, so a create that
/// dropped it would produce a stack no resubmit could ever touch.
#[test]
fn create_puts_each_commits_change_id_in_its_patch_header() {
    let world = Scenario::new("create-change-ids");
    three_commit_branch(&world);

    world.run(&["stack", "create"]).success();

    let ids: Vec<String> = keys(&world)
        .iter()
        .map(|rkey| world.change_id(ALICE, rkey))
        .collect();
    assert_eq!(ids, [BOTTOM, MIDDLE, TOP]);
}

/// A branch pointing into the range ends a pull request there.
///
/// The shape the protocol always allowed and these commands did not: a
/// member is a *run* of commits, and the run's patch is a mailbox carrying
/// one message — and one `Change-Id:` header — per commit. Everything the
/// later reconcile needs to find this member again lives in those headers,
/// so they are what this asserts rather than the byte count.
#[test]
fn a_branch_in_the_range_ends_a_pull_there() {
    let world = Scenario::new("create-grouped");
    three_commit_branch(&world);
    // `part1` on the middle commit: the bottom two are one pull, the top
    // one is its own.
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();

    world.run(&["stack", "create"]).success();

    let rkeys = keys(&world);
    assert_eq!(rkeys.len(), 2, "--group 2,1 is two pull requests");
    // The bottom member is titled by its bottom commit, and carries both.
    assert_eq!(titles(&world, ALICE), ["feat: bottom", "feat: top"]);
    assert_chained(&world, ALICE, None);

    let bottom = world.latest_patch(ALICE, &rkeys[0]);
    assert_eq!(
        bottom.matches("\nSubject: ").count(),
        2,
        "the bottom member should carry two commits:\n{bottom}"
    );
    for id in [BOTTOM, MIDDLE] {
        assert!(
            bottom.contains(&format!("Change-Id: {id}")),
            "{id} is missing from the grouped member's patch:\n{bottom}"
        );
    }
    assert!(
        bottom.contains("one.txt") && bottom.contains("two.txt"),
        "both commits' diffs belong in the one round:\n{bottom}"
    );
    let top = world.latest_patch(ALICE, &rkeys[1]);
    assert!(
        top.contains(&format!("Change-Id: {TOP}")) && !top.contains(MIDDLE),
        "the top member must carry its commit alone:\n{top}"
    );
}

/// Marks that cut the range into a single member are refused: that is a
/// pull request, not a stack.
///
/// The commit-count check cannot catch this — it runs before the cut is
/// decided — so a mark on the top commit used to produce a "stack" of one
/// pull with no `dependentOn`, which every later stack command then refused
/// to touch.
#[test]
fn a_cut_that_leaves_one_member_is_refused() {
    let world = Scenario::new("create-one-member");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "whole", "HEAD"]).success();

    world
        .run(&["stack", "create"])
        .refused("one pull request of 3 commit(s)");
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// `--per-commit` ignores the marks: the old default, still reachable.
#[test]
fn per_commit_ignores_the_marks() {
    let world = Scenario::new("create-per-commit");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();

    world.run(&["stack", "create", "--per-commit"]).success();

    assert_eq!(
        titles(&world, ALICE),
        ["feat: bottom", "feat: middle", "feat: top"],
        "--per-commit must cut at every commit"
    );
}

/// A branch that merely *points* into the range is not a mark.
///
/// The bug this exists to keep out: `git branch backup` before a rebase, an
/// abandoned worktree's branch, a colleague's branch checked out last week —
/// each of them sits on a commit in the range, and inferring cuts from that
/// re-shapes somebody's stack with nothing said. Only a recorded mark cuts.
#[test]
fn an_unrecorded_branch_does_not_cut_the_stack() {
    let world = Scenario::new("create-stray-branch");
    three_commit_branch(&world);
    world.checkout.mark("backup", "HEAD~1");

    world.run(&["stack", "create"]).success();

    assert_eq!(
        titles(&world, ALICE),
        ["feat: bottom", "feat: middle", "feat: top"],
        "a branch nobody marked with re-cut the stack"
    );
}

/// A branch nothing points into is one pull per commit, exactly as before
/// branch marks existed. The stacks written by every earlier version look
/// like this, and a `create` that quietly grouped them would be a different
/// tool wearing the same name.
#[test]
fn an_unmarked_branch_is_still_one_pull_per_commit() {
    let world = Scenario::new("create-unmarked");
    three_commit_branch(&world);

    world.run(&["stack", "create"]).success();

    assert_eq!(
        titles(&world, ALICE),
        ["feat: bottom", "feat: middle", "feat: top"],
    );
}

/// The identity question, asked the only way it can be answered: two
/// accounts are logged in, one is named, and every request that leaves the
/// machine must have been made as that one.
///
/// This is the shape of the bug class this suite exists for. Both accounts
/// can write a well-formed stack, and nothing in the resulting records says
/// which credentials were spent. Only the journal does.
#[test]
fn every_write_in_a_stack_is_made_by_the_selected_account() {
    let world = Scenario::new("create-identity");
    three_commit_branch(&world);

    world.run_as(BOB, &["stack", "create"]).success();

    // The records are Bob's, even though Alice is the active account and
    // owns the repo — a pull lives in its author's PDS whatever it targets.
    assert!(
        world.pulls(ALICE).is_empty(),
        "nothing may land in Alice's repository"
    );
    assert_eq!(world.pulls(BOB).len(), 3);

    let actors = world.with(|w| {
        w.journal
            .iter()
            .filter(|c| c.actor.is_some())
            .map(|c| (c.label(), c.actor.clone().unwrap()))
            .collect::<Vec<_>>()
    });
    assert!(!actors.is_empty(), "no authenticated call was made at all");
    for (label, actor) in &actors {
        assert_eq!(actor, BOB, "{label} went out as the wrong account");
    }
}

/// `--dry-run` reaches the plan and stops: nothing uploaded, nothing
/// written, and the identifiers in its JSON are null rather than invented.
#[test]
fn a_dry_run_creates_nothing_and_touches_no_blob() {
    let world = Scenario::new("create-dry-run");
    three_commit_branch(&world);

    let plan = world
        .run(&["stack", "create", "--dry-run", "--json"])
        .success()
        .json();
    assert_eq!(plan["dry_run"], true);
    assert_eq!(plan["total"], 3);
    assert!(
        plan["members"]
            .as_array()
            .expect("members")
            .iter()
            .all(|m| m["uri"].is_null()),
        "a dry run's identifiers are null: {plan:#}"
    );

    world.with(|w| {
        assert!(w.collection(ALICE, PULL_NSID).is_empty());
        assert!(w.blobs.is_empty(), "a dry run uploaded a blob");
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "a dry run wrote records"
        );
        // The push is now the first thing that would leave the machine, so
        // "nothing sent" has to cover it and the compare that confirms it.
        assert!(
            w.calls_to("sh.tangled.repo.compare").is_empty(),
            "a dry run asked the knot to compare"
        );
    });
    assert_eq!(plan["pushed"], false);
    assert_eq!(world.checkout.pushed_head("feature"), None);
}

/// A reconcile republishes the branch, which is the half `stack resubmit`
/// did not do at all: every member records `source: {branch}`, and rounds
/// written against a branch left behind describe commits the knot does not
/// have. The rewrite a reconcile follows means the push has to be leased,
/// not fast-forward.
#[test]
fn a_reconcile_republishes_the_rewritten_branch() {
    let world = published_stack("resubmit-publishes");
    let published = world.checkout.pushed_head("feature");
    assert_eq!(published, Some(world.checkout.head()));

    world.checkout.amend_below(1, "two.txt", "two, revised\n");
    let report = world.run(&["stack", "resubmit", "--json"]).success().json();

    assert_eq!(report["pushed"], true);
    assert_eq!(
        world.checkout.pushed_head("feature"),
        Some(world.checkout.head()),
        "the records describe commits the knot never received"
    );
    assert_ne!(world.checkout.pushed_head("feature"), published);
}

/// A branch somebody else moved stops the reconcile before any of it: no
/// push, and no records either. The stack's whole chain is rewritten in one
/// batch, so this is the one refusal that has to happen first.
#[test]
fn a_reconcile_refuses_over_a_branch_something_else_moved() {
    let world = published_stack("resubmit-moved-branch");
    let before = keys(&world);

    world.checkout.branch("theirs");
    let theirs = world
        .checkout
        .commit("theirs.txt", "theirs\n", "feat: not mine", None);
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.publish_elsewhere("feature", &theirs);

    world.checkout.amend_below(1, "two.txt", "two, revised\n");
    world
        .run(&["stack", "resubmit"])
        .refused("something else moved the branch");

    assert_eq!(keys(&world), before, "records changed under a refusal");
    assert_eq!(
        world.checkout.pushed_head("feature"),
        Some(theirs),
        "their commit was overwritten"
    );
}

/// The branch is on the knot before any record says it is.
///
/// `source: {branch}` is the field the appview tells a branch-based pull
/// from a patch-based one by, and it never checks it — so for a stack it
/// decides the tree link, the should-resubmit indicator, the web's resubmit
/// route, and (closer to home) whether `stack view`/`resubmit`/`merge` can
/// find the chain from this checkout at all. The push is what makes it true,
/// and the compare after it is the proof it landed.
#[test]
fn create_pushes_the_branch_before_recording_it_as_the_source() {
    let world = Scenario::new("create-pushes");
    three_commit_branch(&world);

    let created = world.run(&["stack", "create", "--json"]).success().json();
    assert_eq!(created["pushed"], true);

    assert_eq!(
        world.checkout.pushed_head("feature"),
        Some(world.checkout.head()),
        "every member records the branch as its source, and nothing published it"
    );
    assert!(
        world.with(|w| !w.calls_to("sh.tangled.repo.compare").is_empty()),
        "the push was never confirmed against the knot"
    );
    for (rkey, value) in world.pulls(ALICE) {
        assert_eq!(
            value["source"]["branch"].as_str(),
            Some("feature"),
            "{rkey} lost its source: {value:#}"
        );
    }
}

/// Unlike `pr create`, the patches stay local: a member's patch is one
/// commit's, and the knot answers about a range. So the compare is read and
/// not stored — a member carrying the knot's mailbox would be carrying the
/// whole stack.
#[test]
fn a_members_patch_is_its_own_commit_and_not_the_knots_mailbox() {
    let world = Scenario::new("create-local-patches");
    three_commit_branch(&world);

    world.run(&["stack", "create"]).success();

    for (rkey, _) in world.pulls(ALICE) {
        let patch = world.round_patch(ALICE, &rkey, 0);
        assert!(
            !patch.contains("as the knot formatted it"),
            "{rkey} stored the knot's mailbox: {patch}"
        );
    }
}

/// A target that cannot be pushed to refuses the whole stack, and says what
/// there is instead. There is deliberately no `--patch-only` for a stack:
/// the source is how every other stack command finds the chain, so a
/// sourceless stack would be records nothing could read back.
#[test]
fn a_branch_that_cannot_be_pushed_refuses_before_any_record() {
    let world = Scenario::new("create-push-refused");
    three_commit_branch(&world);
    // The bare repo `url.<bare>.pushInsteadOf` rewrites the push target to.
    // Without it there is nowhere for the branch to land, which is what a
    // knot refusing the push looks like from here.
    std::fs::remove_dir_all(&world.checkout.bare).expect("remove the push target");

    world
        .run(&["stack", "create"])
        .refused("pr create --patch-only");

    assert!(
        world.pulls(ALICE).is_empty(),
        "a stack was recorded for a branch that was never published"
    );
}

/// The appview's own refusal, in its own words: a knot with nothing between
/// the target and the branch cannot be describing this stack. Reachable
/// after a successful push, because the knot answers about what it has.
#[test]
fn a_knot_with_nothing_between_the_revisions_stops_the_create() {
    let world = Scenario::new("create-empty-compare");
    three_commit_branch(&world);
    world.with(|w| w.compare = Ok((0, String::new())));

    world.run(&["stack", "create"]).refused("finds no commits");
    assert!(world.pulls(ALICE).is_empty(), "records were written anyway");
}

/// A knot that formats these commits without `Change-Id:` headers is a knot
/// whose repo the *website* cannot resubmit this stack from — it reads that
/// header from a jj change-id in the commit object and never from a
/// `Change-Id:` trailer, so a `--add-change-ids` branch has none as far as
/// the knot is concerned. Said, not refused: the stack itself is fine and
/// `stack resubmit` injects the headers itself.
#[test]
fn a_knot_that_cannot_see_the_change_ids_says_the_web_resubmit_will_refuse() {
    let world = Scenario::new("create-knot-blind-to-ids");
    three_commit_branch(&world);
    world.with(|w| w.compare = Ok((3, KNOT_PATCH.repeat(3))));

    let run = world.run(&["stack", "create"]).success();
    assert!(
        run.stderr.contains("from the web will refuse"),
        "nothing said the web resubmit cannot work\n--- stderr ---\n{}",
        run.stderr
    );
    assert_eq!(world.pulls(ALICE).len(), 3, "the stack was written anyway");
}

/// Creating twice is refused, and the refusal names the command that does
/// what the second run meant. A sequence necessarily: the state that makes
/// the second run wrong is what the first run wrote.
#[test]
fn a_second_create_on_the_same_branch_is_refused_and_points_at_resubmit() {
    let world = published_stack("create-twice");

    world
        .run(&["stack", "create"])
        .refused("already has a stack");

    assert_eq!(
        world.pulls(ALICE).len(),
        3,
        "the refused run must not have added records"
    );
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "the refusal came after a write"
        )
    });
}

/// A commit with no change-id refuses rather than guessing, and says how to
/// get one. Nothing is written.
#[test]
fn a_commit_with_no_change_id_refuses_before_anything_is_sent() {
    let world = Scenario::new("create-no-change-id");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world
        .checkout
        .commit("two.txt", "two\n", "feat: no id", None);

    world
        .run(&["stack", "create"])
        .refused("carry no change-id");

    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// `--add-change-ids` on a branch whose base has moved is refused, because
/// the rewrite it performs would turn the stack into a revert.
///
/// Reported against a real branch: the rewrite is `git commit-tree`, which
/// reuses each commit's tree and only changes its parent, so reparenting
/// onto a base the tree has never seen makes every patch carry a deletion of
/// whatever that base added. Nothing said so at the time — the only visible
/// sign was a patch coming out five times the size its own dry run had
/// printed a minute before.
///
/// The assertion is on the *patches*, not on the refusal's wording, because
/// what must never happen is a `deleted file` for a file no commit on the
/// branch touched. If the guard is ever removed, this fails on the thing
/// that matters rather than on a string.
#[test]
fn adding_change_ids_on_a_stale_base_never_produces_a_revert() {
    let world = Scenario::new("create-stale-base");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    world.checkout.commit("two.txt", "two\n", "feat: top", None);

    // Somebody else lands a file on main, and this branch has not caught up.
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("theirs.txt", "landed elsewhere\n", "feat: theirs", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);

    let run = world
        .command(&["stack", "create", "--add-change-ids"])
        .env("ATGC_ACCOUNT", ALICE)
        .finish();

    // Whatever it decided, no patch may propose removing a file this branch
    // never touched.
    for (rkey, _) in world.pulls(ALICE) {
        let patch = world.latest_patch(ALICE, &rkey);
        assert!(
            !patch.contains("theirs.txt"),
            "the stack proposes reverting a file it never touched:\n{patch}"
        );
    }
    // And it must not have quietly succeeded by writing nothing either: the
    // branch is stale, so this is a refusal that names the rebase.
    run.refused("rebase");
}

/// A branch with no change-ids on it, which is what `--add-change-ids` is
/// for and the only state in which the rewrite runs at all.
fn unstacked_branch(world: &Scenario) {
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    world.checkout.commit("two.txt", "two\n", "feat: top", None);
}

/// A session that cannot be resumed leaves the branch exactly where it was.
///
/// `--add-change-ids` moves `refs/heads/<branch>` to a tip whose shas are all
/// new, and until this test the session was resumed a hundred lines below
/// that. So the ordinary case — a grant that expired an hour after login,
/// which `auth::client_metadata` documents as having happened to every
/// session there was — rewrote the commits and *then* failed, leaving shas
/// nobody asked for and no pull requests to show for them. Recoverable from
/// the reflog, but the run that needed to hear about the reflog was the one
/// path that never mentioned it.
///
/// Asserted on the ref rather than on the wording: what must never happen is
/// a rewrite this run cannot use.
#[test]
fn a_refused_session_never_rewrites_the_branch() {
    let world = Scenario::new("create-no-session");
    unstacked_branch(&world);
    let tip = world.checkout.head();

    world.forget_sessions();
    world
        .run(&["stack", "create", "--add-change-ids"])
        .refused("no OAuth session");

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch was rewritten by a run that could not have written anything"
    );
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// The same, for the reconcile — where the consequence is worse. A rewritten
/// branch whose ids match no member is a stack the next `stack resubmit`
/// offers to `--prune`, so a failed session here costs the pull records and
/// their review comments rather than a `git reset`.
#[test]
fn a_refused_session_never_rewrites_the_branch_under_resubmit() {
    let world = published_stack("resubmit-no-session");
    // One more commit, without a trailer, so a rewrite has something to do.
    world
        .checkout
        .commit("four.txt", "four\n", "feat: fourth", None);
    let tip = world.checkout.head();

    world.forget_sessions();
    world
        .run(&["stack", "resubmit", "--add-change-ids"])
        .refused("no OAuth session");

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch was rewritten anyway"
    );
    assert_eq!(world.pulls(ALICE).len(), 3, "the stack was disturbed");
}

/// A branch rebuilt without its `Change-Id:` trailers is refused *before*
/// the rewrite, not after it.
///
/// The ids `--add-change-ids` mints come from the commits' own shas, so they
/// can match no record that exists: every open member is orphaned the moment
/// the rewrite runs, and the only remedy the reconcile can then name is
/// `--prune`, which deletes those pulls and every review comment on them.
/// Asked in the other order the answer is identical and nothing has moved,
/// which is the difference between a refusal and a dilemma.
#[test]
fn a_rewrite_that_would_orphan_the_stack_is_refused_before_it_runs() {
    let world = published_stack("resubmit-orphan");
    let before = keys(&world);

    // The branch, rebuilt by hand: the same three changes, no trailers.
    world
        .checkout
        .git(&["reset", "-q", "--hard", "origin/main"]);
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", None);
    world
        .checkout
        .commit("three.txt", "three\n", "feat: top", None);
    let tip = world.checkout.head();

    world
        .run(&["stack", "resubmit", "--add-change-ids"])
        .refused("--prune");

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch was rewritten before the refusal that made it pointless"
    );
    assert_eq!(keys(&world), before, "the refusal deleted something anyway");
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "the refusal came after a write"
        )
    });
}

/// A dry run describes the rewrite and does not perform it.
///
/// Checked rather than assumed. `atgc agent` tells people every mutating
/// command takes `--dry-run` and to prefer it first, so a dry run that moved
/// `refs/heads/<branch>` would be a worse defect than the ordering this file
/// is otherwise about — it would be the one command nobody expects to change
/// anything doing the single destructive thing these two commands can do.
#[test]
fn a_dry_run_that_would_add_change_ids_rewrites_nothing() {
    let world = Scenario::new("create-dry-run-rewrite");
    unstacked_branch(&world);
    let tip = world.checkout.head();

    let plan = world
        .run(&["stack", "create", "--add-change-ids", "--dry-run", "--json"])
        .success()
        .json();
    assert_eq!(plan["rewrite_pending"], true, "{plan:#}");

    assert_eq!(world.checkout.head(), tip, "a dry run rewrote the branch");
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// The summary marks an abbreviated change-id as abbreviated.
///
/// Nine characters of a forty-one character value, in a fixed column, reads
/// as the whole value. Somebody rebuilding a branch by hand retyped what
/// they saw; the resulting commits matched no pull, and `stack resubmit`
/// then correctly offered `--prune`, which would have deleted four pull
/// records and every review comment on them. The ellipsis is the difference
/// between recognizing a value and believing you have copied it.
#[test]
fn an_abbreviated_change_id_says_that_it_is_abbreviated() {
    let world = Scenario::new("create-id-column");
    three_commit_branch(&world);

    let run = world.run(&["stack", "create", "--dry-run"]).success();
    assert!(
        run.stdout.contains(&format!("{}…", &BOTTOM[..9])),
        "the change-id column is cut with nothing saying so:\n{}",
        run.stdout,
    );
    // And `--json` still carries it whole, which is where a caller that
    // needs the value rather than a glance is meant to read it.
    let plan = world
        .run(&["stack", "create", "--dry-run", "--json"])
        .success()
        .json();
    assert_eq!(plan["members"][0]["change_id"].as_str(), Some(BOTTOM));
}

// ---------------------------------------------------------------------------
// resubmit
// ---------------------------------------------------------------------------

/// Rerunning a reconcile against an unchanged branch writes nothing at all.
///
/// This is the documented recovery story for a partial failure — "just run
/// it again" — and it only works if identical bytes append no round. It is
/// also the property that a naive `resubmit` breaks silently: every member
/// gains a round per run, and nothing complains.
#[test]
fn resubmitting_an_unchanged_branch_is_a_no_op() {
    let world = published_stack("resubmit-no-op");
    let before = keys(&world);

    let report = world.run(&["stack", "resubmit", "--json"]).success().json();
    assert_eq!(report["changed"], false, "{report:#}");

    assert_eq!(keys(&world), before, "the record keys moved");
    for rkey in &before {
        assert_eq!(world.rounds(ALICE, rkey), 1, "{rkey} gained a round");
    }
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "a no-op reconcile sent a batch"
        )
    });
}

/// Amending one commit appends the new round to the record carrying that
/// commit's change-id — not to a neighbour, and not to a new record.
///
/// The bug this is aimed at is a round landing on the wrong member. Every
/// outcome leaves three pulls and one more round than before; only the
/// record keys and the change-ids tell the right one from the wrong one.
///
/// The commit *below* the amend keeps its sha and gains nothing. The one
/// above it does gain a round, and that is worth stating plainly rather than
/// working around: `git format-patch` puts the commit sha in a patch's first
/// line, a rebase gives every commit above the rewritten one a new sha, and
/// the reconcile compares patch bytes. So a member above an amend is
/// "changed" even though its diff is identical. Defensible — its patch does
/// now apply to a different base — but not obvious, and this is where it is
/// written down.
#[test]
fn amending_one_commit_appends_a_round_to_the_record_carrying_its_change_id() {
    let world = published_stack("resubmit-amend");
    let before = keys(&world);

    world.checkout.amend_below(1, "two.txt", "two, revised\n");
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(keys(&world), before, "a reconcile must reuse its records");
    assert_eq!(
        world.rounds(ALICE, &before[0]),
        1,
        "the member below the amend gained a round"
    );
    assert_eq!(
        world.rounds(ALICE, &before[1]),
        2,
        "the amend appended none"
    );
    assert_eq!(
        world.rounds(ALICE, &before[2]),
        2,
        "the member above the amend was rebased, so its patch bytes moved"
    );

    // The new content landed on the member whose change-id owns it, and on
    // no other. Every record still answers to the id it started with.
    assert_eq!(
        [
            world.change_id(ALICE, &before[0]),
            world.change_id(ALICE, &before[1]),
            world.change_id(ALICE, &before[2]),
        ],
        [BOTTOM, MIDDLE, TOP],
        "a record changed which commit it answers to"
    );
    assert!(
        world
            .latest_patch(ALICE, &before[1])
            .contains("two, revised"),
        "the amended content is not in the middle member's new round"
    );
    assert!(
        !world
            .latest_patch(ALICE, &before[2])
            .contains("two, revised"),
        "the amended content leaked into the member above it"
    );
    // And one batch again, not one write per member.
    world.with(|w| assert_eq!(w.calls_to("com.atproto.repo.applyWrites").len(), 1));
}

/// The description a stacked pull holds, if it holds one.
fn body(world: &Scenario, rkey: &str) -> Option<String> {
    world
        .pulls(ALICE)
        .into_iter()
        .find(|(k, _)| k == rkey)
        .and_then(|(_, value)| value["body"].as_str().map(str::to_string))
}

/// A description written by hand is not reverted by the next round.
///
/// The bug this holds shut: a stacked pull's body is generated from its
/// commit message, and every resubmit regenerated it — so a description
/// edited afterwards (screenshots, context, everything a reviewer actually
/// reads) was silently replaced by `%b` the next time any commit changed.
/// Two stacks, thirteen pulls, rewritten by hand.
#[test]
fn an_edited_body_survives_the_rounds_that_follow_it() {
    let world = published_stack("resubmit-edited-body");
    let before = keys(&world);
    let written = "Why this exists, with a screenshot.";
    world
        .run(&["pr", "edit", &before[1], "--body", written])
        .success();

    world.checkout.amend_below(1, "two.txt", "two, revised\n");
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(
        world.rounds(ALICE, &before[1]),
        2,
        "the round itself must still land"
    );
    assert_eq!(
        body(&world, &before[1]).as_deref(),
        Some(written),
        "the edited body was regenerated from the commit message"
    );
}

/// And an amended commit message does not overrule it either.
///
/// The comparison is against the body the *stored* round generated, not
/// against the incoming commit, so once somebody has written over a
/// description the commit message stops driving it. The alternative — let
/// a message amend win — is a silent overwrite of the same text again, in
/// the one case where it is least expected.
#[test]
fn an_amended_message_does_not_overrule_an_edited_body() {
    let world = published_stack("resubmit-edited-body-amend");
    let before = keys(&world);
    let written = "Hand-written, and it stays.";
    world
        .run(&["pr", "edit", &before[2], "--body", written])
        .success();

    world.checkout.amend_message(&format!(
        "feat: top\n\nA generated description.\n\nChange-Id: {TOP}\n"
    ));
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(world.rounds(ALICE, &before[2]), 2, "the round still landed");
    assert_eq!(body(&world, &before[2]).as_deref(), Some(written));
}

/// A body nobody has touched still follows the commit message it came from.
///
/// The other half of the same rule, and the reason it is a comparison
/// rather than a blanket "never rewrite a body": the commit message is
/// still the source of a stacked pull's description until somebody says
/// otherwise.
#[test]
fn an_untouched_body_follows_an_amended_commit_message() {
    let world = published_stack("resubmit-body-follows");
    let before = keys(&world);
    assert_eq!(body(&world, &before[2]), None, "these commits have no body");

    world.checkout.amend_message(&format!(
        "feat: top\n\nA fuller explanation.\n\nChange-Id: {TOP}\n"
    ));
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(
        body(&world, &before[2]).as_deref(),
        Some("A fuller explanation."),
        "an untouched body must still track its commit"
    );
}

/// Reordering the branch re-points the chain onto the new order, reusing
/// every record.
///
/// The claim under test is that a reorder is a *relink*, not a rebuild: no
/// record is created or destroyed, and each keeps the change-id it started
/// with. The member that did not move gains no round; the two that were
/// replayed do, for the same first-line-of-the-patch reason as an amend.
#[test]
fn reordering_the_branch_relinks_the_chain_onto_the_new_order() {
    let world = published_stack("resubmit-reorder");
    let before = keys(&world);

    world.checkout.swap_top_two();
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(keys(&world), before, "reordering must not mint records");
    assert_eq!(
        world.rounds(ALICE, &before[0]),
        1,
        "the member that did not move gained a round"
    );

    // The chain now runs bottom → top → middle, which is the branch's new
    // order and not the record-key order. Checking it by change-id rather
    // than by key is the point: the keys did not move, the links did.
    let order: Vec<String> = {
        let mut by_key: std::collections::BTreeMap<String, Option<String>> =
            chain(&world, ALICE).into_iter().collect();
        let mut order = Vec::new();
        let mut parent: Option<String> = None;
        while let Some((key, _)) = by_key
            .iter()
            .find(|(_, p)| p.as_deref() == parent.as_deref())
        {
            let key = key.clone();
            by_key.remove(&key);
            order.push(world.change_id(ALICE, &key));
            parent = Some(key);
        }
        order
    };
    assert_eq!(order, [BOTTOM, TOP, MIDDLE], "the chain was not relinked");
}

/// A commit added on top gets a new record, chained onto the existing top.
#[test]
fn a_commit_added_on_top_joins_the_existing_chain() {
    let world = published_stack("resubmit-add");
    let before = keys(&world);

    world.checkout.commit(
        "four.txt",
        "four\n",
        "feat: fourth",
        Some("Ifourth0000000000000000000000000000000a"),
    );
    world.run(&["stack", "resubmit"]).success();

    let after = keys(&world);
    assert_eq!(after.len(), 4, "the new commit got no record");
    assert_eq!(&after[..3], &before[..], "the existing records moved");
    assert_chained(&world, ALICE, None);
    assert_eq!(
        titles(&world, ALICE).last().map(String::as_str),
        Some("feat: fourth")
    );
    for rkey in &before {
        assert_eq!(world.rounds(ALICE, rkey), 1, "{rkey} gained a round");
    }
}

/// A commit that left the branch is not deleted on a guess: the reconcile
/// refuses, names what it would remove, and says which flag authorizes it.
/// Then `--prune` does it, and the chain closes over the gap.
#[test]
fn a_vanished_commit_needs_prune_and_then_its_record_goes() {
    let world = published_stack("resubmit-prune");
    let before = keys(&world);

    world.checkout.drop_top();

    world.run(&["stack", "resubmit"]).refused("--prune");
    assert_eq!(
        world.pulls(ALICE).len(),
        3,
        "the refusal deleted something anyway"
    );

    world.run(&["stack", "resubmit", "--prune"]).success();
    let after = keys(&world);
    assert_eq!(after, before[..2], "the wrong record was pruned");
    assert_chained(&world, ALICE, None);
}

/// A reconcile run as an account that holds no stack for this branch does
/// not touch the account that does.
///
/// The failure it guards is the mirror of the create-time one: a command
/// that announced one account and then reconciled another's records would
/// rewrite a stack the person never named.
#[test]
fn a_reconcile_by_another_account_leaves_the_owners_stack_alone() {
    let world = published_stack("resubmit-identity");
    let before: Vec<(String, serde_json::Value)> = world.pulls(ALICE);

    world.checkout.amend_file("three.txt", "three, revised\n");
    world.run_as(BOB, &["stack", "resubmit"]).refused("stack");

    assert_eq!(
        world.pulls(ALICE),
        before,
        "Bob's reconcile rewrote Alice's records"
    );
    assert!(world.pulls(BOB).is_empty());
}

// ---------------------------------------------------------------------------
// merge
// ---------------------------------------------------------------------------

/// A merge is a knot call *and* a batch of status records, in that order.
/// The knot moves the branch; the records are the only thing that makes
/// every listing stop calling these pulls open.
#[test]
fn merging_checks_with_the_knot_then_records_a_merged_status_per_pull() {
    let world = published_stack("merge-all");
    let pulls = keys(&world);

    world.run(&["stack", "merge"]).success();

    let labels = world.with(|w| w.labels());
    let knot_calls: Vec<&String> = labels.iter().filter(|l| l.starts_with("knot ")).collect();
    assert_eq!(
        knot_calls,
        [
            "knot sh.tangled.repo.mergeCheck",
            "knot sh.tangled.repo.merge"
        ],
        "the merge must be checked before it is made: {labels:?}"
    );

    // One status record per pull, all merged, all naming a pull of this
    // stack — written as one batch, so a crash cannot leave half the stack
    // reading open with nothing to say which half landed.
    let statuses = world.with(|w| w.collection(ALICE, PULL_STATUS_NSID));
    assert_eq!(statuses.len(), 3, "one merged status per pull");
    for (rkey, record) in &statuses {
        // The lexicon's token, not the bare word: a status record's `status`
        // is a `sh.tangled.repo.pull.status.*` knownValue, and writing the
        // short form is a record every indexer would ignore.
        assert_eq!(
            record.value["status"].as_str(),
            Some("sh.tangled.repo.pull.status.merged"),
            "{rkey}"
        );
        let names = record.value["pull"].as_str().unwrap_or_default();
        assert!(
            pulls.iter().any(|p| names.ends_with(p)),
            "{rkey} points at {names}, which is not in this stack"
        );
    }
    world.with(|w| {
        assert_eq!(
            w.calls_to("com.atproto.repo.applyWrites").len(),
            1,
            "the statuses are one batch"
        )
    });
}

/// `--through N` lands the bottom N and leaves the rest open. Merging a
/// stack from the top down is meaningless — every member's patch assumes the
/// ones beneath it — so the subset can only ever be a prefix.
#[test]
fn merging_through_a_position_lands_only_the_bottom_of_the_stack() {
    let world = published_stack("merge-through");
    let pulls = keys(&world);

    world.run(&["stack", "merge", "--through", "2"]).success();

    let statuses = world.with(|w| w.collection(ALICE, PULL_STATUS_NSID));
    assert_eq!(statuses.len(), 2, "--through 2 landed the wrong count");
    let landed: Vec<String> = statuses
        .iter()
        .map(|(_, r)| {
            r.value["pull"]
                .as_str()
                .unwrap_or_default()
                .rsplit('/')
                .next()
                .unwrap_or_default()
                .to_string()
        })
        .collect();
    assert!(
        landed.contains(&pulls[0]) && landed.contains(&pulls[1]),
        "--through 2 landed {landed:?}, not the bottom two of {pulls:?}"
    );
}

/// A knot that reports a conflict stops the merge, and stops it *before* any
/// status record is written. A stack marked merged that never landed is the
/// worst outcome here: every listing would then hide work that is still
/// undone.
#[test]
fn a_conflicting_merge_check_writes_no_status_records() {
    let world = published_stack("merge-conflict");
    world.with(|w| w.merge_check = Err("would not apply".to_string()));

    world.run(&["stack", "merge"]).refused("conflict");

    world.with(|w| {
        assert!(
            w.collection(ALICE, PULL_STATUS_NSID).is_empty(),
            "a refused merge recorded a status"
        );
        assert!(
            w.calls_to("sh.tangled.repo.merge").is_empty(),
            "the merge was attempted after a failed check"
        );
    });
}

/// A dry-run merge reaches the knot's check and stops there: no merge, no
/// records.
#[test]
fn a_dry_run_merge_checks_and_stops() {
    let world = published_stack("merge-dry-run");

    let report = world
        .run(&["stack", "merge", "--dry-run", "--json"])
        .success()
        .json();
    assert_eq!(report["dry_run"], true);
    assert_eq!(report["merged"], false, "{report:#}");

    world.with(|w| {
        assert_eq!(w.calls_to("sh.tangled.repo.mergeCheck").len(), 1);
        assert!(w.calls_to("sh.tangled.repo.merge").is_empty());
        assert!(w.collection(ALICE, PULL_STATUS_NSID).is_empty());
    });
}

/// The full lifecycle, in one run: create, land the bottom, rebase past it,
/// reconcile. The merged member is never touched and becomes the anchor the
/// shortened chain hangs from.
///
/// This is the sequence that cannot be assembled from unit tests at all. The
/// reconcile's `anchor` branch only fires when a *previous* merge left a
/// record whose commits are gone from the branch, and "gone from the branch"
/// is a fact about git, not about any value a planner could be handed.
#[test]
fn a_merged_bottom_becomes_the_anchor_of_the_next_reconcile() {
    let world = published_stack("merge-then-resubmit");
    let before = keys(&world);

    world.run(&["stack", "merge", "--through", "1"]).success();

    // The branch is rebased past what landed: the bottom commit is now part
    // of main, and `feature` carries only the two above it.
    world.checkout.git(&["checkout", "-q", "main"]);
    world.checkout.git(&["cherry-pick", "feature~2"]);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.git(&["rebase", "-q", "origin/main"]);

    world.clear_journal();
    let report = world.run(&["stack", "resubmit", "--json"]).success().json();

    // The merged member is named as the anchor and left out of the chain,
    // which is now the two commits still on the branch.
    assert_eq!(
        report["anchor"]["rkey"].as_str(),
        Some(before[0].as_str()),
        "the merged member is not the anchor: {report:#}"
    );
    assert_eq!(report["total"], 2, "{report:#}");

    // Three records still: the merged one is kept, never rewritten.
    assert_eq!(keys(&world), before, "the merged record was disturbed");
    assert_eq!(
        world.rounds(ALICE, &before[0]),
        1,
        "a merged member must never gain a round"
    );
    // And the chain still hangs off it, so the stack stays connected to the
    // history it landed into rather than restarting from nothing.
    assert_chained(&world, ALICE, None);
}

/// Closing a member of a stack says which pulls are left depending on it.
///
/// Not a refusal: closing a member is a documented way to take work out of a
/// stack, and `stack resubmit` relinks the chain around it. What it is is a
/// fact about pulls the command was not asked about and does not otherwise
/// mention, which is exactly the kind that goes unnoticed.
#[test]
fn closing_a_stack_member_names_the_pulls_left_depending_on_it() {
    let world = published_stack("close-warns-dependents");
    let keys = keys(&world);
    assert_eq!(keys.len(), 3, "{keys:?}");

    // The bottom: both members above it depend on it, one directly and one
    // through the other, and the warning has to reach both.
    let run = world.run(&["pr", "close", &keys[0]]).success();
    assert!(
        run.stderr.contains("2 open pull request(s)"),
        "{}",
        run.stderr
    );
    assert!(run.stderr.contains("feat: middle"), "{}", run.stderr);
    assert!(
        run.stderr.contains("feat: top"),
        "the transitive dependent was missed: {}",
        run.stderr
    );

    // The top: nothing depends on it, so there is nothing to say.
    let run = world.run(&["pr", "close", &keys[2]]).success();
    assert!(
        !run.stderr.contains("depend on it"),
        "warned about nothing: {}",
        run.stderr
    );
}

/// Reopening never warns. Restoring a member's open state repairs the
/// dependency a close broke, which is the opposite of a thing to flag.
#[test]
fn reopening_a_stack_member_says_nothing_about_dependents() {
    let world = published_stack("reopen-warns-nothing");
    let keys = keys(&world);
    world.run(&["pr", "close", &keys[0]]).success();
    let run = world.run(&["pr", "reopen", &keys[0]]).success();
    assert!(!run.stderr.contains("depend on it"), "{}", run.stderr);
}

// ---------------------------------------------------------------------------
// a retired member
// ---------------------------------------------------------------------------
//
// Closing a member and taking its commit off the branch is a documented way
// out of a stack: `stack resubmit` *retires* the pull — keeps the record, so
// the close and its review comments survive — and routes the chain past it.
//
// The record it keeps still says `dependentOn: <the member below>`, and the
// member above now says the same thing. Two pulls on one parent is a fork,
// and every command that orders a chain refuses one. So the sanctioned way
// out of a stack ends with a stack no stack command will read — including
// the resubmit that would have been the way back.
//
// These drive that end to end. Each one is a sequence because the fork is
// not visible in any single command: the resubmit that creates it reports
// success, and it is the *next* command that cannot run.

/// A stack whose middle member was closed and retired, with the branch
/// rebased past it. Two members left, `feat: bottom` and `feat: top`.
fn retired_middle(label: &str) -> Scenario {
    let world = published_stack(label);
    let keys = keys(&world);
    world.run(&["pr", "close", &keys[1]]).success();
    world.checkout.drop_below(1);
    let report = world.run(&["stack", "resubmit", "--json"]).success().json();
    assert_eq!(
        report["retired"][0]["rkey"].as_str(),
        Some(keys[1].as_str()),
        "the closed member was not retired, so this scenario is not set up: {report:#}"
    );
    world.clear_journal();
    world
}

/// The stack is still readable after a member of it has been retired.
///
/// The plainest statement of the bug: `stack resubmit` reports the retire
/// and exits 0, and then the command anyone would run next cannot order the
/// records it just wrote.
#[test]
fn a_retired_member_leaves_the_chain_readable() {
    let world = retired_middle("retired-view");

    let json = world.run(&["stack", "view", "--json"]).success().json();
    let titles: Vec<&str> = json["members"]
        .as_array()
        .expect("members")
        .iter()
        .map(|m| m["title"].as_str().unwrap_or_default())
        .collect();
    // Top first, as everywhere else, and the retired member is not in it.
    assert_eq!(titles, ["feat: top", "feat: bottom"], "{json:#}");
}

/// And the next reconcile runs.
///
/// This is the half that makes the bug a brick rather than a blemish.
/// Rerunning `stack resubmit` is the documented recovery for anything that
/// went wrong in the last one, so a state it cannot read is a state with no
/// way back out through atgc at all.
#[test]
fn a_retired_member_does_not_block_the_next_reconcile() {
    let world = retired_middle("retired-resubmit");
    let keys = keys(&world);

    world.checkout.amend_file("three.txt", "three, revised\n");
    world.run(&["stack", "resubmit"]).success();

    // The top member took the round; nothing else was touched, the retired
    // record least of all.
    assert_eq!(world.rounds(ALICE, &keys[2]), 3, "the top took no round");
    assert_eq!(
        world.rounds(ALICE, &keys[1]),
        1,
        "the retired member was written to"
    );
}

/// And so does a merge.
///
/// `stack merge` orders the chain for its own reason — it lands a member
/// plus everything unmerged below it — so it refuses on the same walk, and
/// a stack that cannot be merged is the most expensive shape of this bug.
#[test]
fn a_retired_member_does_not_block_a_merge() {
    let world = retired_middle("retired-merge");
    let keys = keys(&world);

    world.run(&["stack", "merge", "--through", "1"]).success();

    // The bottom landed. The retired member is not below it in the chain any
    // more, so nothing about it may be dragged into the merge — and `merge`
    // does land everything unmerged beneath what it is pointed at, which is
    // exactly how a stale link turns into a pull nobody asked to merge.
    let merged: Vec<String> = world
        .records(ALICE, PULL_STATUS_NSID)
        .into_iter()
        .filter(|(_, v)| v["status"].as_str() == Some("sh.tangled.repo.pull.status.merged"))
        .map(|(_, v)| v["pull"].as_str().unwrap_or_default().to_string())
        .collect();
    assert!(
        merged.iter().any(|p| p.ends_with(&keys[0])),
        "the bottom did not land: {merged:?}"
    );
    assert!(
        !merged.iter().any(|p| p.ends_with(&keys[1])),
        "the retired member was merged: {merged:?}"
    );
}

/// `pr view` degrades rather than refuses, so it never broke — but it did
/// start warning that the branch's pulls are not an orderable stack, on a
/// stack atgc itself had just written. Nothing is wrong with these records.
#[test]
fn pr_view_of_a_retired_stack_says_nothing_about_a_damaged_chain() {
    let world = retired_middle("retired-pr-view");

    let run = world.run(&["pr", "view"]).success();
    assert!(
        !run.stderr.contains("orderable stack"),
        "warned about a chain it wrote itself:\n{}",
        run.stderr
    );
}

/// The record itself: retiring unlinks it, and takes nothing else away.
///
/// The write the other tests here only see the effect of. A retired pull
/// keeps its rounds and its title — the close and the comments explaining it
/// are the reason the record is kept at all — and loses the one field that
/// has stopped being true.
#[test]
fn a_retired_member_keeps_everything_but_its_link() {
    let world = published_stack("retired-record");
    let keys = keys(&world);
    let before = world
        .pulls(ALICE)
        .into_iter()
        .find(|(k, _)| *k == keys[1])
        .expect("the middle member")
        .1;

    world.run(&["pr", "close", &keys[1]]).success();
    world.checkout.drop_below(1);
    world.run(&["stack", "resubmit"]).success();

    let after = world
        .pulls(ALICE)
        .into_iter()
        .find(|(k, _)| *k == keys[1])
        .expect("the retired member is kept")
        .1;
    assert!(
        after["dependentOn"].is_null(),
        "the retired member is still in the chain: {after:#}"
    );
    assert_eq!(after["title"], before["title"], "{after:#}");
    assert_eq!(
        world.rounds(ALICE, &keys[1]),
        1,
        "the retired member was given a round"
    );
}

/// Retiring the *top* relinks nothing, and still has a write to make.
///
/// The case with no fork in it: nothing is relinked past a closed top, so no
/// parent gains a second dependent. What it has instead is a chain that
/// walks *up* into a closed pull — `stack view` listing a member that was
/// closed and taken off the branch — and a reconcile with no slot to
/// rewrite, which is how the one op owed here came to be skipped as "the
/// stack already matches the branch".
#[test]
fn retiring_the_top_unlinks_it_even_with_nothing_else_to_write() {
    let world = published_stack("retired-top");
    let keys = keys(&world);
    world.run(&["pr", "close", &keys[2]]).success();
    world.checkout.drop_top();
    world.run(&["stack", "resubmit"]).success();

    let top = world
        .pulls(ALICE)
        .into_iter()
        .find(|(k, _)| *k == keys[2])
        .expect("the retired member is kept")
        .1;
    assert!(
        top["dependentOn"].is_null(),
        "the retired top is still in the chain: {top:#}"
    );

    let json = world.run(&["stack", "view", "--json"]).success().json();
    let titles: Vec<&str> = json["members"]
        .as_array()
        .expect("members")
        .iter()
        .map(|m| m["title"].as_str().unwrap_or_default())
        .collect();
    assert_eq!(titles, ["feat: middle", "feat: bottom"], "{json:#}");
}

/// Retiring the *bottom* never forked anything, and must not start.
///
/// The contrast that keeps the fix honest: a closed bottom has no
/// `dependentOn` of its own, so the member above it relinks to nothing and
/// no parent ever gains a second dependent. Whatever teaches the walk about
/// closed records has to leave this case exactly as it is.
#[test]
fn retiring_the_bottom_leaves_the_stack_readable() {
    let world = published_stack("retired-bottom");
    let keys = keys(&world);
    world.run(&["pr", "close", &keys[0]]).success();
    world.checkout.drop_below(2);
    world.run(&["stack", "resubmit"]).success();

    let json = world.run(&["stack", "view", "--json"]).success().json();
    let titles: Vec<&str> = json["members"]
        .as_array()
        .expect("members")
        .iter()
        .map(|m| m["title"].as_str().unwrap_or_default())
        .collect();
    assert_eq!(titles, ["feat: top", "feat: middle"], "{json:#}");
}

/// `pr resubmit` still refuses a member with live members beside it.
#[test]
fn pr_resubmit_refuses_a_member_of_a_live_stack() {
    let world = published_stack("pr-resubmit-live-stack");
    let keys = keys(&world);
    world
        .run(&["pr", "resubmit", &keys[1]])
        .refused("stack resubmit");
}

/// But a member whose whole chain below it has merged takes a round.
///
/// The objection to a whole-branch round on a stack member is in the
/// message: the round would carry every other member's commits. A merged
/// member's commits are in the target branch already, so they are ancestors
/// of the base the round is cut against and it cannot carry them — counting
/// them left the last member of a landed stack with no verb that would take
/// a round at all. `stack resubmit` is not the answer either: it reconciles
/// by change-id and cannot adopt a pull whose patches carry none, which is
/// every pull `pr create` writes.
///
/// This is not hypothetical. A chain in this repo reached exactly that state
/// and needed a hand-written `com.atproto.repo.putRecord` to escape it.
#[test]
fn pr_resubmit_takes_a_round_once_the_rest_of_the_stack_has_merged() {
    let world = published_stack("pr-resubmit-merged-below");
    let keys = keys(&world);

    // Land everything below the top, and take its commits off the branch the
    // way a rebase past a merge does.
    world.run(&["stack", "merge", "--through", "2"]).success();
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .git(&["cherry-pick", "feature~2", "feature~1"]);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.git(&["rebase", "-q", "origin/main"]);

    let before = world.rounds(ALICE, &keys[2]);
    world.run(&["pr", "resubmit", &keys[2]]).success();
    assert_eq!(
        world.rounds(ALICE, &keys[2]),
        before + 1,
        "the last member of a landed stack should take a round"
    );
}

/// A grouped stack reconciles without being told how it was grouped.
///
/// The grouping is not a flag anyone has to remember: each member's round
/// carries a `Change-Id:` header per commit it holds, so the records
/// themselves say where the cuts are. Amending a commit in the *middle* of a
/// two-commit member is the case that proves it — a reconcile that had
/// forgotten the grouping would split that member into two pulls.
#[test]
fn a_grouped_stack_keeps_its_grouping_across_a_reconcile() {
    let world = Scenario::new("resubmit-grouped");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    world.run(&["stack", "create"]).success();
    world.clear_journal();
    let before = keys(&world);

    // Rewrite the lower of the bottom member's two commits.
    world.checkout.amend_below(2, "one.txt", "one, revised\n");
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(
        keys(&world),
        before,
        "the grouping splintered into new pulls"
    );
    assert_eq!(
        world.rounds(ALICE, &before[0]),
        2,
        "the amend appended no round"
    );
    let bottom = world.latest_patch(ALICE, &before[0]);
    assert_eq!(
        bottom.matches("\nSubject: ").count(),
        2,
        "the member lost a commit in the reconcile:\n{bottom}"
    );
    assert!(
        bottom.contains("one, revised"),
        "the amended content is not in the grouped member's new round:\n{bottom}"
    );
}

/// A commit written into the middle of a grouped member joins it.
///
/// It sits inside that member's span on the branch, and a member is a run of
/// commits: the alternative — a new pull wedged between a member's own two
/// commits — is not a shape a stack can even hold.
#[test]
fn a_commit_added_inside_a_member_joins_it() {
    let world = Scenario::new("resubmit-grouped-insert");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    world.run(&["stack", "create"]).success();
    world.clear_journal();
    let before = keys(&world);

    // Above the bottom commit, so it lands between the grouped member's own
    // two commits — inside its span, not between the two members.
    world.checkout.insert_below(
        2,
        "extra.txt",
        "extra\n",
        "feat: extra",
        Some("Iextra000000000000000000000000000000000a"),
    );
    world.run(&["stack", "resubmit"]).success();

    assert_eq!(keys(&world), before, "a member's own commit minted a pull");
    let bottom = world.latest_patch(ALICE, &before[0]);
    assert_eq!(
        bottom.matches("\nSubject: ").count(),
        3,
        "the inserted commit did not join the member it sits inside:\n{bottom}"
    );
}

/// Moving a branch mark re-cuts a stack that already exists: two members
/// become one, and the record that lost its commits is a drop like any
/// other. This is the whole ergonomic claim of branch marks — you re-cut a
/// stack with `git branch -f`, not by restating a spec.
#[test]
fn moving_a_branch_mark_recuts_an_existing_stack() {
    let world = published_stack("resubmit-regroup");
    let before = keys(&world);

    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    let report = world
        .run(&["stack", "resubmit", "--prune", "--json"])
        .success()
        .json();
    assert_eq!(report["total"], 2, "{report:#}");

    let after = keys(&world);
    assert_eq!(after.len(), 2, "the re-cut left {} pulls", after.len());
    assert_eq!(after[0], before[0], "the bottom member lost its record");
    let bottom = world.latest_patch(ALICE, &after[0]);
    assert_eq!(
        bottom.matches("\nSubject: ").count(),
        2,
        "the bottom member did not absorb the commit above it:\n{bottom}"
    );
}

// ---------------------------------------------------------------------------
// rebase
// ---------------------------------------------------------------------------

/// `stack rebase` replays the branch onto its target and takes the marks
/// with it.
///
/// The reason this command exists rather than a line of advice: a plain `git
/// rebase` leaves every mark on a commit the branch no longer has, so the
/// cut silently disappears and the next reconcile sees an uncut branch. The
/// assertion that matters is the last one — the mark still ends the same
/// member afterwards, on a sha that did not exist when it was set.
#[test]
fn rebase_replays_the_branch_and_carries_its_marks() {
    let world = Scenario::new("rebase-marks");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    let mark_before = world.checkout.git(&["rev-parse", "part1"]);
    let tip_before = world.checkout.head();

    // main moves under the branch, which is the whole situation.
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("unrelated.txt", "theirs\n", "feat: landed meanwhile", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);

    world.run(&["stack", "rebase"]).success();

    // Replayed: a new tip, the target's commit now an ancestor.
    assert_ne!(world.checkout.head(), tip_before, "nothing was replayed");
    let base_in = world
        .checkout
        .git(&["merge-base", "--is-ancestor", "origin/main", "HEAD"]);
    assert_eq!(
        base_in.trim(),
        "",
        "the branch is not on top of origin/main"
    );

    // And the mark travelled: a different sha, still one below the tip, so
    // the cut it describes is the one it described before.
    let mark_after = world.checkout.git(&["rev-parse", "part1"]);
    assert_ne!(
        mark_after, mark_before,
        "the mark was left on a commit the branch no longer has"
    );
    assert_eq!(
        mark_after.trim(),
        world.checkout.git(&["rev-parse", "HEAD~1"]).trim(),
        "the mark no longer ends the member it ended before"
    );

    // Which the reconcile then agrees with: still two members, not three.
    world.run(&["stack", "create"]).success();
    assert_eq!(titles(&world, ALICE), ["feat: bottom", "feat: top"]);
}

/// A dirty tree is refused before anything is fetched or replayed.
#[test]
fn rebase_refuses_a_dirty_working_tree() {
    let world = Scenario::new("rebase-dirty");
    three_commit_branch(&world);
    world.checkout.write("one.txt", b"edited, uncommitted\n");

    world
        .run(&["stack", "rebase"])
        .refused("uncommitted changes");
    // And says the same thing whether or not a stack exists yet.
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// `stack sync` is the review cycle in one word: catch up with the target,
/// then reconcile the records with what the branch became.
///
/// The claim under test is that the two halves happen in the right order and
/// both take effect — a reconcile planned before the rebase would compare
/// the old shas, append rounds for them, and leave the stack describing a
/// branch that no longer exists.
#[test]
fn sync_rebases_and_then_reconciles_in_one_command() {
    let world = published_stack("sync-both");
    let before = keys(&world);

    // main moves, and the top commit is amended: one half of the work for
    // each half of the command.
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("unrelated.txt", "theirs\n", "feat: landed meanwhile", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.amend_file("three.txt", "three, revised\n");

    world.run(&["stack", "sync"]).success();

    // Rebased: the target's commit is an ancestor now.
    assert_eq!(
        world
            .checkout
            .git(&["merge-base", "--is-ancestor", "origin/main", "HEAD"])
            .trim(),
        "",
        "sync did not rebase onto the target"
    );
    // And reconciled: same records, the amended member holding the new text.
    assert_eq!(keys(&world), before, "a reconcile must reuse its records");
    assert!(
        world
            .latest_patch(ALICE, &before[2])
            .contains("three, revised"),
        "the amend never reached the records"
    );
    assert_eq!(
        world.change_id(ALICE, &before[0]),
        BOTTOM,
        "a record changed which commit it answers to"
    );
}

/// `--dry-run` reaches both plans and moves neither the branch nor a record.
#[test]
fn a_dry_run_sync_moves_nothing() {
    let world = published_stack("sync-dry-run");
    let tip = world.checkout.head();
    let before = keys(&world);

    world.run(&["stack", "sync", "--dry-run"]).success();

    assert_eq!(world.checkout.head(), tip, "a dry run rebased the branch");
    assert_eq!(keys(&world), before);
    for rkey in &before {
        assert_eq!(world.rounds(ALICE, rkey), 1, "{rkey} gained a round");
    }
}

// ---------------------------------------------------------------------------
// the failure paths of the commands that move the branch
// ---------------------------------------------------------------------------

/// Put `origin/main` and the branch in conflict over the same file.
///
/// The bottom commit of `three_commit_branch` writes `one.txt`; main writes
/// it differently, so replaying the branch onto main cannot apply cleanly.
fn conflicting_main(world: &Scenario) {
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("one.txt", "theirs, not yours\n", "feat: same file", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
}

/// A rebase that conflicts stops where git stops, says so, and leaves the
/// rebase in progress for `git rebase --continue` or `--abort`.
///
/// The failure path of the one command here that rewrites the branch. Left
/// untested it is the path most likely to be wrong, because it is the one
/// nobody runs on purpose — and a second `stack rebase` on top of a
/// half-finished one is how somebody loses the first one's conflict work,
/// which is the second half of this test.
#[test]
fn a_conflicting_rebase_stops_and_says_where() {
    let world = published_stack("rebase-conflict");
    conflicting_main(&world);

    world
        .run(&["stack", "rebase"])
        .refused("the rebase stopped");

    // git is mid-rebase, and the advice it was given is the advice that
    // works from here.
    let dir = world
        .checkout
        .git(&["rev-parse", "--git-path", "rebase-merge"]);
    assert!(
        world.checkout.path.join(dir.trim()).exists(),
        "the rebase was rolled back, so `git rebase --continue` cannot work"
    );

    // And a second run refuses rather than starting another one on top.
    world
        .run(&["stack", "rebase"])
        .refused("already in progress");

    world.checkout.git(&["rebase", "--abort"]);
}

/// `stack sync` stops after a failed rebase and writes nothing.
///
/// The whole promise of the combined verb: the reconcile is planned against
/// the branch the rebase produced, so a rebase that did not finish must not
/// be followed by one. A reconcile here would compare the pre-rebase shas,
/// append a round to every member, and leave the records describing a branch
/// that is mid-rebase.
#[test]
fn sync_stops_when_the_rebase_stops_and_writes_nothing() {
    let world = published_stack("sync-conflict");
    let before = keys(&world);
    conflicting_main(&world);

    world.run(&["stack", "sync"]).refused("the rebase stopped");

    assert_eq!(
        keys(&world),
        before,
        "records were written after a failed rebase"
    );
    for rkey in &before {
        assert_eq!(world.rounds(ALICE, rkey), 1, "{rkey} gained a round");
    }
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "a batch went out after the rebase stopped"
        )
    });

    world.checkout.git(&["rebase", "--abort"]);
}

/// Merging a stack whose bottom member holds two commits lands both of them.
///
/// Every other merge test drives single-commit members, so nothing said what
/// the knot is handed once a member is a run of commits. The patch it merges
/// has to carry every commit of every member being merged, in order: a merge
/// that quietly dropped the second commit of a member would land a change
/// that no longer builds, and the records would say it went perfectly.
#[test]
fn merging_a_multi_commit_member_lands_every_commit_in_it() {
    let world = Scenario::new("merge-grouped");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    world.run(&["stack", "create"]).success();
    world.clear_journal();

    world.run(&["stack", "merge"]).success();

    let merged = world.with(|w| {
        w.calls_to("sh.tangled.repo.merge")
            .first()
            .map(|c| c.body.clone())
            .expect("the knot was asked to merge")
    });
    let patch = merged["patch"].as_str().unwrap_or_default().to_string();
    assert_eq!(
        patch.matches("\nSubject: ").count(),
        3,
        "every commit of every member must be in the merged patch:\n{patch}"
    );
    for file in ["one.txt", "two.txt", "three.txt"] {
        assert!(
            patch.contains(file),
            "{file} is missing from the merge:\n{patch}"
        );
    }
    // Two members, so two statuses — the grouping survives all the way to
    // what is recorded about the merge.
    let statuses = world.with(|w| w.collection(ALICE, PULL_STATUS_NSID));
    assert_eq!(
        statuses.len(),
        2,
        "one merged status per pull, not per commit"
    );
}

/// One argument is a revision, and the mark is named after the commit it
/// lands on. Naming a cut is a chore, and the name somebody would have typed
/// is sitting in the commit subject.
#[test]
fn a_mark_names_itself_after_the_commit_it_ends() {
    let world = Scenario::new("mark-autoname");
    three_commit_branch(&world);

    let placed = world.run(&["stack", "mark", "HEAD~1"]).success();
    assert!(placed.stdout.contains("middle"), "{}", placed.stdout);

    // It is a real branch, at the commit named, and recorded as a cut.
    assert_eq!(
        world.checkout.git(&["rev-parse", "middle"]).trim(),
        world.checkout.git(&["rev-parse", "HEAD~1"]).trim(),
    );
    world.run(&["stack", "create"]).success();
    assert_eq!(titles(&world, ALICE), ["feat: bottom", "feat: top"]);
}

/// Marking with the branch you are on is refused with the reason, not with
/// git's sentence about worktrees.
#[test]
fn marking_with_the_branch_you_are_on_is_refused() {
    let world = Scenario::new("mark-self");
    three_commit_branch(&world);

    world
        .run(&["stack", "mark", "feature", "HEAD~1"])
        .refused("already ends the top pull request");
}

/// A mark can be forgotten, and one left outside the range is listed as such.
///
/// Forgetting is how a cut is undone without losing the sha, and the
/// out-of-range line is the only thing that makes a stranded mark visible —
/// the state a plain `git rebase` leaves behind, where the cut silently
/// stopped existing.
#[test]
fn a_mark_can_be_forgotten_and_a_stranded_one_is_named() {
    let world = published_stack("mark-forget");
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();

    // Stranded: the branch is rewritten out from under the mark, the way a
    // rebase with no --update-refs would.
    world.checkout.amend_below(2, "one.txt", "one, rewritten\n");
    let listed = world.run(&["stack", "mark"]).success();
    assert!(
        listed.stdout.contains("outside the range"),
        "a mark left behind by a rewrite must be visible:\n{}",
        listed.stdout
    );

    let forgotten = world.run(&["stack", "mark", "--forget", "part1"]).success();
    assert!(
        forgotten.stdout.contains("forgot mark part1"),
        "{}",
        forgotten.stdout
    );
    // The branch is left alone: forgetting a cut is not deleting work.
    assert!(
        !world
            .checkout
            .git(&["rev-parse", "--verify", "--quiet", "refs/heads/part1"])
            .trim()
            .is_empty(),
        "forgetting a mark deleted its branch"
    );
    let after = world.run(&["stack", "mark"]).success();
    assert!(
        after.stdout.contains("no marks on feature"),
        "{}",
        after.stdout
    );
}

/// `resubmit --per-commit` re-cuts a marked stack back to one pull per
/// commit, keeping every record that still holds a commit.
///
/// Worth pinning because the obvious guess is wrong: splitting a two-commit
/// member does not *drop* anything, so it needs no `--prune`. The lower half
/// keeps the record — it still carries the change-id the record answers to —
/// and the upper half becomes a new pull. A re-cut that deleted the record
/// and minted two would throw away the review comments on it.
#[test]
fn per_commit_recuts_a_marked_stack_without_dropping_records() {
    let world = Scenario::new("resubmit-per-commit");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    world.run(&["stack", "create"]).success();
    world.clear_journal();
    let before = keys(&world);
    assert_eq!(before.len(), 2);

    world.run(&["stack", "resubmit", "--per-commit"]).success();

    // Chain order, not record-key order: the new middle member is minted
    // last, so the two disagree here — and the chain is what Tangled reads.
    assert_eq!(
        chain_titles(&world, ALICE),
        ["feat: bottom", "feat: middle", "feat: top"],
        "the re-cut did not reach one pull per commit, in order"
    );
    let after = keys(&world);
    assert_eq!(after.len(), 3);
    assert!(
        before.iter().all(|k| after.contains(k)),
        "a re-cut destroyed a record instead of splitting around it: {before:?} -> {after:?}"
    );
}

/// An unmarked branch wide enough to be an accident is asked about rather
/// than opened.
///
/// The failure this prevents is the one that reads as success: eight commits
/// become eight pull requests of one commit each, nobody can review them,
/// and closing them is eight more acts. Two remedies, both named, and a
/// config for anybody who really does want a pull per commit every time.
#[test]
fn a_wide_unmarked_branch_is_asked_about_before_it_opens_a_pull_per_commit() {
    let world = Scenario::new("create-wide-unmarked");
    four_commit_branch(&world);

    world
        .run(&["stack", "create"])
        .refused("4 pull requests of one commit each");
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));

    // Saying it out loud goes through.
    world.run(&["stack", "create", "--per-commit"]).success();
    assert_eq!(world.pulls(ALICE).len(), 4);
}

/// Marking the branch answers the question, without --per-commit.
#[test]
fn marking_a_wide_branch_is_the_other_way_past_the_question() {
    let world = Scenario::new("create-wide-marked");
    four_commit_branch(&world);
    world.run(&["stack", "mark", "HEAD~2"]).success();

    world.run(&["stack", "create"]).success();
    assert_eq!(world.pulls(ALICE).len(), 2, "the marks decide the count");
}

/// `stack.askWhenUnmarked false` turns the question off for a checkout that
/// has heard it and meant it. Named for what it does: marks still decide the
/// cut, so a config called `perCommit` would promise more than it delivers.
#[test]
fn a_checkout_can_turn_the_unmarked_question_off() {
    let world = Scenario::new("create-wide-configured");
    four_commit_branch(&world);
    world
        .checkout
        .git(&["config", "--local", "stack.askWhenUnmarked", "false"]);

    world.run(&["stack", "create"]).success();
    assert_eq!(world.pulls(ALICE).len(), 4);
}

/// `pr create` on a marked branch says the marks are there.
///
/// Marks mean somebody meant several pull requests; this command files one.
/// Not a refusal — marks left over from a landed stack are ordinary, and a
/// stack is not always wanted — but silence here costs a pull request that
/// has to be closed by hand.
#[test]
fn pr_create_says_when_the_branch_has_marks_on_it() {
    let world = Scenario::new("pr-create-marked");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "HEAD~1"]).success();

    let run = world
        .run(&["pr", "create", "--title", "one pull"])
        .success();

    assert!(
        run.stderr.contains("mark(s) on it") && run.stderr.contains("middle"),
        "the marks went unmentioned:\n{}",
        run.stderr
    );
    // And it really did file one pull request, not a stack.
    assert_eq!(world.pulls(ALICE).len(), 1);
}

/// A reconcile against a target that has moved says so, and names the one
/// command that fixes it — the rounds it is about to write carry patches
/// against a base nothing has any more.
#[test]
fn resubmitting_behind_the_target_points_at_sync() {
    let world = published_stack("resubmit-behind");
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("unrelated.txt", "theirs\n", "feat: landed meanwhile", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.amend_file("three.txt", "three, revised\n");

    let run = world.run(&["stack", "resubmit"]).success();

    assert!(
        run.stderr.contains("atgc stack sync"),
        "a stack behind its target should be told the one-command fix:\n{}",
        run.stderr
    );
}

/// `stack view` says how many commits each member holds.
///
/// The one fact a stack of runs has that a stack of commits did not, and the
/// one a reviewer decides from: a member of four commits and a member of one
/// look identical in a listing that only counts rounds. It is read off the
/// patch, because a pull record does not say.
#[test]
fn view_says_how_many_commits_each_member_holds() {
    let world = Scenario::new("view-commits");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "HEAD~1"]).success();
    world.run(&["stack", "create"]).success();

    let run = world.run(&["stack", "view"]).success();
    assert!(
        run.stdout.contains("2 commits, 1 round"),
        "the grouped member should say what it holds:\n{}",
        run.stdout
    );

    let json = world.run(&["stack", "view", "--json"]).success().json();
    let members = json["members"].as_array().expect("members");
    // Top first in the array, as everywhere else.
    assert_eq!(members[0]["commits"], 1, "{json:#}");
    assert_eq!(members[1]["commits"], 2, "{json:#}");
}

/// `stack sync --json` is *one* object, carrying both halves.
///
/// Rule 1 of `--json` is one value on stdout, and this is the command most
/// able to break it: it runs a rebase and a reconcile, each of which used to
/// emit its own object. The two were refactored to return their reports for
/// exactly this reason, and nothing checked it until now — `.json()` parses
/// the whole of stdout, so a second object fails here rather than in
/// somebody's `jq`.
#[test]
fn sync_json_is_one_object_describing_both_halves() {
    let world = published_stack("sync-json");
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("unrelated.txt", "theirs\n", "feat: landed meanwhile", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.amend_file("three.txt", "three, revised\n");

    let report = world.run(&["stack", "sync", "--json"]).success().json();

    assert_eq!(report["rebase"]["rebased"], true, "{report:#}");
    assert_ne!(
        report["rebase"]["was"], report["rebase"]["now"],
        "a rebase that moved nothing is not a rebase: {report:#}"
    );
    assert_eq!(report["reconcile"]["changed"], true, "{report:#}");
    assert_eq!(report["reconcile"]["total"], 3, "{report:#}");
}

/// `stack rebase --json` on a branch already on top of its target says so
/// and moves nothing.
#[test]
fn rebase_json_reports_an_up_to_date_branch() {
    let world = published_stack("rebase-json-noop");
    let tip = world.checkout.head();

    let report = world.run(&["stack", "rebase", "--json"]).success().json();

    assert_eq!(report["rebased"], false, "{report:#}");
    assert_eq!(report["was"], report["now"], "{report:#}");
    assert_eq!(world.checkout.head(), tip, "the branch moved anyway");
}

/// `stack mark --json` lists the marks, their positions, and the subjects
/// they end.
#[test]
fn mark_json_lists_positions_and_subjects() {
    let world = Scenario::new("mark-json");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();

    let report = world.run(&["stack", "mark", "--json"]).success().json();

    assert_eq!(report["branch"], "feature", "{report:#}");
    assert_eq!(report["commits"], 3, "{report:#}");
    let marks = report["marks"].as_array().expect("marks array");
    assert_eq!(marks.len(), 1, "{report:#}");
    assert_eq!(marks[0]["name"], "part1", "{report:#}");
    assert_eq!(marks[0]["position"], 2, "{report:#}");
    assert_eq!(marks[0]["subject"], "feat: middle", "{report:#}");
}

/// A plan that cannot be published refuses before `--add-change-ids` moves
/// the branch, not after it.
///
/// Both of these were decidable from the commits as they stood, and both
/// used to arrive with the rewrite already run: new shas, no pull requests,
/// and a reader who now has to fix the original problem on commits that are
/// no longer the ones they wrote. Asserted on the ref, like the refused
/// session above — the wording is not the point, an unusable rewrite is.
#[test]
fn a_duplicate_change_id_refuses_before_the_rewrite_runs() {
    let world = Scenario::new("create-duplicate-before-rewrite");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some("Iduplicated"));
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", Some("Iduplicated"));
    // A third commit with no trailer, so the rewrite has something to do and
    // would really run if the refusal came after it.
    world
        .checkout
        .commit("three.txt", "three\n", "feat: top", None);
    let tip = world.checkout.head();

    world
        .run(&["stack", "create", "--add-change-ids"])
        .refused("all carry the change-id Iduplicated");

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch was rewritten by a run that refused anyway"
    );
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// The same, for a commit that changes nothing: the knot will not merge an
/// empty patch, so the stack is refused — with the branch where it was.
#[test]
fn an_empty_commit_refuses_before_the_rewrite_runs() {
    let world = Scenario::new("create-empty-before-rewrite");
    unstacked_branch(&world);
    world
        .checkout
        .git(&["commit", "-q", "--allow-empty", "-m", "chore: nothing"]);
    let tip = world.checkout.head();

    world
        .run(&["stack", "create", "--add-change-ids"])
        .refused("change nothing");

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch was rewritten by a run that refused anyway"
    );
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

// ---------------------------------------------------------------------------
// link
// ---------------------------------------------------------------------------

/// Two pulls opened separately, `upper`'s branch on top of `lower`'s, with
/// nothing chaining them. Hands back their record keys, bottom first.
///
/// This is the shape `stack link` exists for: pulls that were opened one at a
/// time, each already carrying its number, its rounds and whatever review it
/// has been through. Everything else in this file starts from a branch and
/// opens the whole chain at once.
fn two_flat_pulls(world: &Scenario) -> (String, String) {
    world.checkout.branch("lower");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    let bottom = open_pull(world, "the lower half");
    world.checkout.branch("upper");
    world.checkout.commit("two.txt", "two\n", "feat: top", None);
    let top = open_pull(world, "the upper half");
    world.clear_journal();
    (bottom, top)
}

/// Open one pull from the branch that is checked out, and hand back its key.
fn open_pull(world: &Scenario, title: &str) -> String {
    let created = world
        .run(&["pr", "create", "--title", title, "--json"])
        .success()
        .json();
    created["uri"]
        .as_str()
        .expect("pr create --json carries the uri it wrote")
        .rsplit('/')
        .next()
        .expect("an at-uri ends in a record key")
        .to_string()
}

/// The `dependentOn` of one pull, as a record key.
fn parent_of(world: &Scenario, rkey: &str) -> Option<String> {
    world
        .pulls(ALICE)
        .into_iter()
        .find(|(k, _)| k == rkey)
        .expect("the pull")
        .1["dependentOn"]
        .as_str()
        .map(|uri| uri.rsplit('/').next().unwrap_or_default().to_string())
}

/// Chaining two pulls that already exist writes both records in one go and
/// leaves everything else about them alone.
///
/// One `applyWrites` is the requirement, not an optimisation: written a
/// record at a time, the chain passes through a state the appview refuses at
/// ingest.
#[test]
fn link_chains_open_pulls_in_one_write() {
    let world = Scenario::new("link-two");
    let (bottom, top) = two_flat_pulls(&world);

    let run = world.run(&["stack", "link", &bottom, &top]).success();

    assert_eq!(parent_of(&world, &bottom), None, "{}", run.stdout);
    assert_eq!(
        parent_of(&world, &top),
        Some(bottom.clone()),
        "the upper half should depend on the lower:\n{}",
        run.stdout
    );
    let writes = world.with(|w| w.calls_to("com.atproto.repo.applyWrites").len());
    assert_eq!(writes, 1, "a chain must be written atomically");
    assert_eq!(
        world.rounds(ALICE, &top),
        1,
        "linking must not append a round"
    );
}

/// One mailbox message per sha, in the shape `git format-patch` writes.
fn mailbox(shas: &[&str]) -> String {
    shas.iter()
        .map(|sha| {
            format!(
                "From {sha} Mon Sep 17 00:00:00 2001\n\
                 From: alice <alice@alice.test>\n\
                 Subject: [PATCH] a commit\n\n---\n"
            )
        })
        .collect()
}

/// Two pulls whose patches share a commit cannot be a stack, and `link`
/// refuses before it writes.
///
/// **The shape this command used to accept, and the one it required.** A
/// merge takes a member plus everything unmerged below it and applies them
/// as one series, so two members carrying the same commit apply it twice.
/// The knot answers that with "patch doesn't apply" and "file already
/// exists" — which names a file rather than the cause, and sends the reader
/// looking for a conflict that is not there.
///
/// It is not a hypothetical: a chain linked this way was opened against this
/// repo and the knot refused to merge it, naming six files that were nothing
/// to do with the problem. The only checks here were about *order*, and the
/// ancestry they require is precisely what makes one branch's patch contain
/// the other's commits.
#[test]
fn link_refuses_members_whose_patches_share_a_commit() {
    let world = Scenario::new("link-overlap");
    let lower_sha = "1111111111111111111111111111111111111111";
    let upper_sha = "2222222222222222222222222222222222222222";

    world.checkout.branch("lower");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    world.with(|w| w.compare = Ok((1, mailbox(&[lower_sha]))));
    let bottom = open_pull(&world, "the lower half");

    world.checkout.branch("upper");
    world.checkout.commit("two.txt", "two\n", "feat: top", None);
    // What `pr create` records for a branch opened on top of another: its
    // own commit *and* the one below it, because the patch spans the target
    // branch to this branch's head.
    world.with(|w| w.compare = Ok((2, mailbox(&[lower_sha, upper_sha]))));
    let top = open_pull(&world, "the upper half");
    world.clear_journal();

    let run = world
        .run(&["stack", "link", &bottom, &top])
        .refused("apply it twice");
    assert!(
        run.stderr.contains(&lower_sha[..12]),
        "the refusal has to name the commit they share:\n{}",
        run.stderr
    );
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "a refused link must write nothing"
        )
    });
}

/// Patches that share nothing still link.
///
/// The rule above is about overlap and not about branches, so the case
/// `link` exists for — pulls whose patches are already separate ranges —
/// goes through untouched.
#[test]
fn link_accepts_members_whose_patches_are_separate() {
    let world = Scenario::new("link-disjoint");

    world.checkout.branch("lower");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    world.with(|w| w.compare = Ok((1, mailbox(&["3333333333333333333333333333333333333333"]))));
    let bottom = open_pull(&world, "the lower half");

    world.checkout.branch("upper");
    world.checkout.commit("two.txt", "two\n", "feat: top", None);
    world.with(|w| w.compare = Ok((1, mailbox(&["4444444444444444444444444444444444444444"]))));
    let top = open_pull(&world, "the upper half");
    world.clear_journal();

    world.run(&["stack", "link", &bottom, &top]).success();
    assert_eq!(parent_of(&world, &top), Some(bottom));
}

/// Taking a member out of a chain closes the gap behind it.
///
/// The inverse of `link`, and the half that was missing: `link` wrote
/// `dependentOn` and nothing removed it. `stack resubmit` clears one only as
/// a side effect of retiring a *closed* member, and it cannot touch a chain
/// whose patches carry no change-id — which is every pull `pr create` writes.
/// A chain in this repo reached that state and its only exit was a
/// hand-written `com.atproto.repo.putRecord`.
#[test]
fn unlink_takes_a_member_out_and_closes_the_gap() {
    let world = published_stack("unlink-middle");
    let keys = keys(&world);
    world.clear_journal();

    world.run(&["stack", "unlink", &keys[1]]).success();

    assert_eq!(
        parent_of(&world, &keys[1]),
        None,
        "it is still in the chain"
    );
    assert_eq!(
        parent_of(&world, &keys[2]),
        Some(keys[0].clone()),
        "the member above should have inherited what the one below it had"
    );
    assert_eq!(parent_of(&world, &keys[0]), None, "the bottom moved");
    world.with(|w| {
        assert_eq!(
            w.calls_to("com.atproto.repo.applyWrites").len(),
            1,
            "a chain must be rewritten atomically"
        )
    });
}

/// Naming every member dissolves the chain, and needs no separate verb.
#[test]
fn unlink_dissolves_a_whole_chain_when_every_member_is_named() {
    let world = published_stack("unlink-all");
    let keys = keys(&world);

    world
        .run(&["stack", "unlink", &keys[0], &keys[1], &keys[2]])
        .success();

    for key in &keys {
        assert_eq!(parent_of(&world, key), None, "{key} is still chained");
    }
}

/// Unlinking the bottom leaves the one above it depending on nothing.
#[test]
fn unlinking_the_bottom_makes_the_next_one_the_bottom() {
    let world = published_stack("unlink-bottom");
    let keys = keys(&world);

    world.run(&["stack", "unlink", &keys[0]]).success();

    assert_eq!(parent_of(&world, &keys[1]), None, "it should be the bottom");
    assert_eq!(parent_of(&world, &keys[2]), Some(keys[1].clone()));
}

/// A pull that is in no chain is not an error, and writes nothing.
#[test]
fn unlinking_something_unchained_writes_nothing() {
    let world = Scenario::new("unlink-flat");
    unstacked_branch(&world);
    let created = world
        .run(&["pr", "create", "--title", "a flat pull", "--json"])
        .success()
        .json();
    let rkey = created["uri"]
        .as_str()
        .expect("pr create --json carries the uri it wrote")
        .rsplit('/')
        .next()
        .expect("an at-uri ends in a record key")
        .to_string();
    world.clear_journal();

    let run = world.run(&["stack", "unlink", &rkey]).success();

    assert!(run.stdout.contains("nothing to write"), "{}", run.stdout);
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "nothing should have been written"
        )
    });
}

/// A dry run says what it would do and sends nothing.
#[test]
fn a_dry_run_unlink_writes_nothing() {
    let world = published_stack("unlink-dry-run");
    let keys = keys(&world);
    world.clear_journal();

    let run = world
        .run(&["stack", "unlink", &keys[1], "--dry-run"])
        .success();

    assert!(run.stdout.contains("dry run"), "{}", run.stdout);
    assert_eq!(
        parent_of(&world, &keys[1]),
        Some(keys[0].clone()),
        "a dry run rewrote the chain"
    );
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "a dry run must send nothing"
        )
    });
}

/// The order is checked against git where git can check it, and a chain that
/// does not stack is refused before anything is written.
#[test]
fn link_refuses_an_order_git_says_does_not_stack() {
    let world = Scenario::new("link-backwards");
    let (bottom, top) = two_flat_pulls(&world);

    let run = world.run(&["stack", "link", &top, &bottom]);

    assert_ne!(run.code, Some(0), "{}", run.stdout);
    assert!(
        run.stderr.contains("not an ancestor"),
        "the refusal should name the ancestry it checked:\n{}",
        run.stderr
    );
    assert_eq!(parent_of(&world, &bottom), None, "{}", run.stderr);
    assert_eq!(parent_of(&world, &top), None, "{}", run.stderr);
}

/// Naming a pull twice is refused: a pull holds one place in a chain.
#[test]
fn link_refuses_the_same_pull_twice() {
    let world = Scenario::new("link-duplicate");
    let (bottom, _top) = two_flat_pulls(&world);

    let run = world.run(&["stack", "link", &bottom, &bottom]);

    assert_ne!(run.code, Some(0), "{}", run.stdout);
    assert!(run.stderr.contains("named twice"), "{}", run.stderr);
}

/// A chain that is already in that order writes nothing at all, so `link` is
/// safe to re-run — the way an agent that cannot remember whether it ran will
/// re-run it.
#[test]
fn linking_an_already_chained_stack_writes_nothing() {
    let world = Scenario::new("link-idempotent");
    let (bottom, top) = two_flat_pulls(&world);
    world.run(&["stack", "link", &bottom, &top]).success();
    world.clear_journal();

    let run = world.run(&["stack", "link", &bottom, &top]).success();

    assert!(run.stdout.contains("already chained"), "{}", run.stdout);
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "nothing should have been written"
        )
    });
}

/// A dry run says what it would chain and sends nothing.
#[test]
fn a_dry_run_link_writes_nothing() {
    let world = Scenario::new("link-dry-run");
    let (bottom, top) = two_flat_pulls(&world);

    let run = world
        .run(&["stack", "link", &bottom, &top, "--dry-run"])
        .success();

    assert!(run.stdout.contains("dry run"), "{}", run.stdout);
    assert_eq!(parent_of(&world, &top), None, "{}", run.stdout);
}

/// Only the account that authored a pull can chain it: the `dependentOn` is
/// a field on the record, and the record lives in its author's PDS.
#[test]
fn link_refuses_a_pull_that_is_not_yours() {
    let world = Scenario::new("link-not-mine");
    let (bottom, _top) = two_flat_pulls(&world);
    world.checkout.git(&["checkout", "-q", "-b", "bobs"]);
    world
        .checkout
        .commit("bob.txt", "bob\n", "feat: bob's work", None);
    let bobs = world
        .run_as(BOB, &["pr", "create", "--title", "bob's pull", "--json"])
        .success()
        .json()["uri"]
        .as_str()
        .expect("bob's uri")
        .to_string();

    let run = world.run(&["stack", "link", &bottom, &bobs]);

    assert_ne!(run.code, Some(0), "{}", run.stdout);
    assert!(
        run.stderr.contains("only the account that authored"),
        "{}",
        run.stderr
    );
}

/// `stack link --json` is one object, bottom first, saying what each member
/// was made to depend on and whether anything was written for it.
#[test]
fn link_json_describes_the_chain_bottom_first() {
    let world = Scenario::new("link-json");
    let (bottom, top) = two_flat_pulls(&world);

    let report = world
        .run(&["stack", "link", &bottom, &top, "--json"])
        .success()
        .json();

    assert_eq!(report["changed"], true, "{report:#}");
    let members = report["members"].as_array().expect("members");
    assert_eq!(members.len(), 2, "{report:#}");
    assert_eq!(members[0]["position"], 1, "{report:#}");
    assert_eq!(members[0]["rkey"], bottom.as_str(), "{report:#}");
    assert_eq!(
        members[0]["dependent_on"],
        serde_json::Value::Null,
        "{report:#}"
    );
    assert_eq!(members[1]["rkey"], top.as_str(), "{report:#}");
    assert!(
        members[1]["dependent_on"]
            .as_str()
            .expect("a parent")
            .ends_with(&bottom),
        "{report:#}"
    );
    // One write, not two: the bottom already depended on nothing, so only
    // the member whose parent changed was sent.
    assert_eq!(
        report["wrote"].as_array().expect("wrote").len(),
        1,
        "{report:#}"
    );
    assert_eq!(members[0]["changed"], false, "{report:#}");
    assert_eq!(members[1]["changed"], true, "{report:#}");
}

// ---------------------------------------------------------------------------
// navigation
// ---------------------------------------------------------------------------

/// A three-commit branch cut into three members: marks at the bottom two
/// commits, the branch itself ending the top one.
fn marked_three(label: &str) -> Scenario {
    let world = Scenario::new(label);
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~2"]).success();
    world.run(&["stack", "mark", "part2", "HEAD~1"]).success();
    world
}

/// The branch a scenario's checkout is standing on.
fn on(world: &Scenario) -> String {
    world
        .checkout
        .git(&["rev-parse", "--abbrev-ref", "HEAD"])
        .trim()
        .to_string()
}

/// `up` and `down` walk the members, and each step is an ordinary checkout.
#[test]
fn up_and_down_walk_the_members() {
    let world = marked_three("nav-walk");

    world.run(&["stack", "bottom"]).success();
    assert_eq!(on(&world), "part1");
    world.run(&["stack", "up"]).success();
    assert_eq!(on(&world), "part2");
    world.run(&["stack", "up"]).success();
    assert_eq!(on(&world), "feature");
    world.run(&["stack", "down", "2"]).success();
    assert_eq!(on(&world), "part1");
    world.run(&["stack", "top"]).success();
    assert_eq!(on(&world), "feature");
}

/// Walking past the end stops there and says so, rather than failing: a
/// script that runs `up` until it stops needs the answer, not an error.
#[test]
fn walking_past_the_top_stops_and_says_so() {
    let world = marked_three("nav-past-the-end");

    let run = world.run(&["stack", "up", "9"]).success();

    assert_eq!(on(&world), "feature");
    assert!(run.stdout.contains("already on"), "{}", run.stdout);
}

/// Navigation works from a lower member too, where the branch underfoot is a
/// mark and the stack it belongs to has to be found from the config.
#[test]
fn navigating_from_a_lower_member_finds_the_stack_it_belongs_to() {
    let world = marked_three("nav-from-below");
    world.checkout.git(&["checkout", "-q", "part1"]);

    let report = world.run(&["stack", "up", "--json"]).success().json();

    assert_eq!(report["from"], "part1", "{report:#}");
    assert_eq!(report["to"], "part2", "{report:#}");
    assert_eq!(report["position"], 2, "{report:#}");
    assert_eq!(report["total"], 3, "{report:#}");
    assert_eq!(report["moved"], true, "{report:#}");
    let layers = report["layers"].as_array().expect("layers");
    assert_eq!(layers[0], "part1", "bottom first: {report:#}");
    assert_eq!(layers[2], "feature", "{report:#}");
}

/// `checkout` takes a position or a mark name, and refuses a position the
/// stack does not have.
#[test]
fn checkout_takes_a_position_or_a_name() {
    let world = marked_three("nav-checkout");

    world.run(&["stack", "checkout", "2"]).success();
    assert_eq!(on(&world), "part2");
    world.run(&["stack", "checkout", "part1"]).success();
    assert_eq!(on(&world), "part1");

    let run = world.run(&["stack", "checkout", "9"]);
    assert_ne!(run.code, Some(0), "{}", run.stdout);
    assert!(run.stderr.contains("this stack has 3"), "{}", run.stderr);
    assert_eq!(on(&world), "part1", "a refusal must not move HEAD");
}

/// A branch that is not part of a stack is told so, and pointed at the
/// commands that are for it.
#[test]
fn navigating_an_unstacked_branch_is_refused_with_a_pointer() {
    let world = Scenario::new("nav-unstacked");
    three_commit_branch(&world);

    let run = world.run(&["stack", "up"]);

    assert_ne!(run.code, Some(0), "{}", run.stdout);
    assert!(run.stderr.contains("not part of a stack"), "{}", run.stderr);
    assert!(run.stderr.contains("stack mark"), "{}", run.stderr);
}

// ---------------------------------------------------------------------------
// seams
// ---------------------------------------------------------------------------

/// `stack view` says `?` for a member whose patch it could not read, rather
/// than failing the whole listing.
///
/// How many commits a member holds is written in exactly one place — its
/// latest round's patch — so the count is one blob read per member, and a
/// blob that has gone is the ordinary consequence of that. The listing is
/// most wanted when something is wrong with the records, so it degrades
/// instead of refusing.
#[test]
fn view_says_question_mark_for_a_member_whose_patch_is_gone() {
    let world = published_stack("view-unreadable-patch");
    world.with(|w| w.blobs.clear());

    let run = world.run(&["stack", "view"]).success();

    assert!(
        run.stdout.contains("? commits"),
        "an unreadable patch should show as ?:\n{}",
        run.stdout
    );
    let json = world.run(&["stack", "view", "--json"]).success().json();
    assert_eq!(
        json["members"][0]["commits"],
        serde_json::Value::Null,
        "{json:#}"
    );
}

/// A mark whose branch somebody deleted is named as gone, and the stack it
/// cut falls back to the cuts that are left.
#[test]
fn a_mark_whose_branch_was_deleted_is_named_and_stops_cutting() {
    let world = Scenario::new("mark-branch-deleted");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    world.checkout.git(&["branch", "-D", "part1"]);

    let listed = world.run(&["stack", "mark"]).success();
    assert!(
        listed.stdout.contains("branch is gone"),
        "{}",
        listed.stdout
    );

    // With no cut left in the range, the branch is one pull per commit —
    // the unmarked shape — rather than a stack built around a mark that no
    // longer exists.
    world.run(&["stack", "create"]).success();
    assert_eq!(keys(&world).len(), 3, "{:#?}", world.pulls(ALICE));
}

/// `--add-change-ids` rewrites every commit in the range, which moves the
/// branches the marks are, and the cut has to survive it.
///
/// This is the seam the rewrite is most able to break silently: the marks
/// are remapped inside the rewrite, before the stacked branch's own ref is
/// moved, and getting the order wrong leaves them pointing at commits that
/// are no longer in the range — a stack that quietly re-cuts itself into one
/// pull per commit.
#[test]
fn add_change_ids_carries_the_marks_through_the_rewrite() {
    let world = Scenario::new("add-change-ids-marks");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", None);
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", None);
    world
        .checkout
        .commit("three.txt", "three\n", "feat: top", None);
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    let before = world.checkout.git(&["rev-parse", "part1"]);

    world
        .run(&["stack", "create", "--add-change-ids"])
        .success();

    let after = world.checkout.git(&["rev-parse", "part1"]);
    assert_ne!(before, after, "the rewrite should have moved the mark");
    let listed = world.run(&["stack", "mark"]).success();
    assert!(
        listed.stdout.contains("2/3"),
        "part1 should still cut after the middle commit:\n{}",
        listed.stdout
    );
    assert_eq!(
        keys(&world).len(),
        2,
        "the cut should have held: {:#?}",
        world.pulls(ALICE)
    );
}

/// `pr diff` on a member that carries several commits prints the whole
/// mailbox, not just the first message.
#[test]
fn pr_diff_on_a_multi_commit_member_prints_every_commit() {
    let world = Scenario::new("diff-multi-commit");
    three_commit_branch(&world);
    world.run(&["stack", "mark", "HEAD~1"]).success();
    world.run(&["stack", "create"]).success();

    let bottom = keys(&world).remove(0);
    let run = world.run(&["pr", "diff", &bottom]).success();

    assert!(run.stdout.contains("feat: bottom"), "{}", run.stdout);
    assert!(
        run.stdout.contains("feat: middle"),
        "a two-commit member's patch holds both:\n{}",
        run.stdout
    );
}

// ---------------------------------------------------------------------------
// what the other reader sees
// ---------------------------------------------------------------------------
//
// Every test above asks whether atgc sent what it meant to send. These ask
// something the rest of the suite cannot: whether the records it sent say the
// same thing to the service that renders them.
//
// The two are different questions. A stack can satisfy every rule a PDS
// enforces and still be one tangled.org draws wrongly, and the retire bug was
// exactly that — legal bytes, a green suite, and a live member that could
// silently vanish from the stack view. `support::appview` is a model of the
// appview's own traversal; see its module doc for what it covers and what
// that is worth.

/// A stack atgc has just written reads the same to both.
#[test]
fn a_created_stack_reads_the_same_to_both() {
    let world = published_stack("agree-create");
    world.assert_stack_reads_alike(ALICE);
    assert_eq!(
        world.appview_stack(ALICE, &keys(&world)[0]).members(),
        keys(&world),
        "the appview orders the stack bottom first, as atgc does"
    );
}

/// And still does after every kind of edit a stack takes.
///
/// One test and one long sequence on purpose. The disagreements worth finding
/// are not in any single write — each of these operations was already proved
/// correct on its own above — but in what a stack accumulates: a reorder over
/// an amend over an insert, with a member retired in the middle of it and the
/// bottom merged out from under the rest. That history is where a stale link
/// survives, and it is not reachable by any pair of commands.
#[test]
fn every_edit_leaves_the_two_readers_agreeing() {
    let world = published_stack("agree-sequence");
    world.assert_stack_reads_alike(ALICE);

    // An amend: a round on one member, nothing relinked.
    world.checkout.amend_file("three.txt", "three, revised\n");
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);

    // A reorder: no rounds, every link rewritten.
    world.checkout.swap_top_two();
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);

    // An insert in the middle: a new record minted between two that exist.
    world.checkout.insert_below(
        1,
        "inserted.txt",
        "inserted\n",
        "feat: inserted",
        Some("Iinserted000000000000000000000000000000a"),
    );
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);

    // A retire: the record stays, and the chain has to close over it in a
    // way both readers agree about. This is the step that used to fork.
    //
    // Found by title, not by position: after a reorder and an insert, record
    // order is minting order and no longer chain order, and picking the
    // wrong record here would close a pull whose commit is still on the
    // branch.
    let inserted = world
        .pulls(ALICE)
        .into_iter()
        .find(|(_, v)| v["title"].as_str() == Some("feat: inserted"))
        .expect("the inserted member")
        .0;
    world.run(&["pr", "close", &inserted]).success();
    world.checkout.drop_below(1);
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);

    // A merge, and the rebase past it: the bottom leaves the branch and the
    // rest stays chained to it.
    world.run(&["stack", "merge", "--through", "1"]).success();
    world.checkout.git(&["checkout", "-q", "main"]);
    world.checkout.git(&["cherry-pick", "feature~2"]);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world.checkout.git(&["rebase", "-q", "origin/main"]);
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);
}

/// The model has teeth: the shape atgc used to leave behind is a coin toss.
///
/// An oracle that cannot fail proves nothing, so this plants the pre-fix
/// records directly — a retired member still naming the pull below it, and
/// the member above relinked to the same place — and asserts the appview
/// would have to choose between them. `GetStack` asks for *the* dependent,
/// so the stack it returns depends on which row the database hands back:
/// the live top can be the one dropped.
///
/// Planted rather than driven, because atgc cannot be made to write this any
/// more. That is the point of the fix, and the reason the shape needs a
/// fixture to survive as a test at all.
#[test]
fn the_shape_that_used_to_fork_a_stack_is_a_coin_toss_to_the_appview() {
    let world = published_stack("agree-fork");
    let keys = keys(&world);

    // Relink the top onto the bottom, and leave the middle pointing there
    // too: a retire as it was written before the record was unlinked.
    world.with(|w| {
        let mut top = w
            .get(ALICE, PULL_NSID, &keys[2])
            .expect("the top member")
            .value
            .clone();
        top["dependentOn"] = serde_json::json!(format!("at://{ALICE}/{PULL_NSID}/{}", keys[0]));
        w.plant(ALICE, PULL_NSID, &keys[2], top);
    });

    assert_eq!(
        world.appview_stack(ALICE, &keys[0]),
        support::appview::Stack::Ambiguous {
            parent: keys[0].clone(),
            dependents: {
                let mut d = vec![keys[1].clone(), keys[2].clone()];
                d.sort();
                d
            },
        },
    );
}

/// A link naming a pull the index does not hold is malformed, not a stack
/// that stops there.
///
/// What a deep listing or a deleted record looks like from the appview's
/// side. atgc has its own name for this — `missing_below`, which the write
/// commands refuse on — and the two readers agreeing that the records are
/// unreadable is as much an agreement as any other.
#[test]
fn a_link_naming_nothing_is_malformed_to_the_appview() {
    let world = published_stack("agree-malformed");
    let keys = keys(&world);
    let gone = format!("at://{ALICE}/{PULL_NSID}/nosuchrecord");

    world.with(|w| {
        let mut bottom = w
            .get(ALICE, PULL_NSID, &keys[0])
            .expect("the bottom member")
            .value
            .clone();
        bottom["dependentOn"] = serde_json::json!(gone);
        w.plant(ALICE, PULL_NSID, &keys[0], bottom);
    });

    assert_eq!(
        world.appview_stack(ALICE, &keys[2]),
        support::appview::Stack::Malformed(gone)
    );
}

/// And a cycle is a cycle to both.
#[test]
fn a_cycle_is_a_cycle_to_the_appview() {
    let world = published_stack("agree-cycle");
    let keys = keys(&world);

    world.with(|w| {
        let mut bottom = w
            .get(ALICE, PULL_NSID, &keys[0])
            .expect("the bottom member")
            .value
            .clone();
        bottom["dependentOn"] = serde_json::json!(format!("at://{ALICE}/{PULL_NSID}/{}", keys[2]));
        w.plant(ALICE, PULL_NSID, &keys[0], bottom);
    });

    assert_eq!(
        world.appview_stack(ALICE, &keys[1]),
        support::appview::Stack::Cyclic
    );
    world.run(&["stack", "view"]).refused("loops");
}

// ---------------------------------------------------------------------------
// what the index says
// ---------------------------------------------------------------------------
//
// Every test above reads one source. The stack write commands read three:
// `Source::EVERY` is the PDS, Bobbin and the web scrape, unconditionally and
// whatever `ATGC_USE_BOBBIN` says, because a merge has to see a stack member
// somebody else opened. So an index row is an input to a reconcile, not a
// convenience for a listing — and Bobbin's state is *preferred* over the one
// the account's own status records give.
//
// That is a deliberate trade, argued where `EVERY` is defined: a stale row
// can cost a refusal that turns out to be unnecessary, and cannot cost a
// merge that should not have happened. These drive it, because the trade is
// only safe in the direction it was argued in.

/// Bobbin's listing of `world`'s own pulls, with a state per record key.
///
/// The index as it would answer if it agreed with the PDS about everything
/// except what the caller overrides — which is how a *disagreement* is
/// staged without also staging an index that has never heard of the repo.
fn bobbin_agrees_except(world: &Scenario, states: &[(&str, &str)]) {
    let rows: Vec<serde_json::Value> = world
        .pulls(ALICE)
        .into_iter()
        .map(|(rkey, value)| {
            let state = states
                .iter()
                .find(|(k, _)| *k == rkey)
                .map(|(_, s)| *s)
                .unwrap_or("open");
            serde_json::json!({
                "uri": format!("at://{ALICE}/{PULL_NSID}/{rkey}"),
                "state": state,
                "commentCount": 0,
                "value": value,
            })
        })
        .collect();
    world.with(|w| w.bobbin_pulls = rows);
}

/// The default reads no index at all.
///
/// A deliberate decision with a cost attached — your own records are the
/// whole answer on your own repo, and the index lags — so it is worth a
/// regression test rather than a comment. Nothing about `stack view` may
/// start asking Bobbin without somebody deciding to.
#[test]
fn stack_view_asks_no_index_unless_asked_to() {
    let world = published_stack("index-default");
    world.clear_journal();
    world.run(&["stack", "view"]).success();
    let asked = world.with(|w| {
        w.journal
            .iter()
            .filter(|c| c.service == "bobbin")
            .map(|c| c.label())
            .collect::<Vec<_>>()
    });
    assert!(asked.is_empty(), "the default read the index: {asked:?}");
}

/// A member missing from the index is still a member.
///
/// Index lag is the ordinary state of a pull opened a moment ago, and the
/// stack writes read the index by design. A chain assembled from the union
/// of the sources survives that; one assembled from the index alone would
/// silently lose its newest member, which for a reconcile means re-minting a
/// record that already exists.
#[test]
fn a_member_the_index_has_not_caught_up_with_is_not_lost() {
    let world = published_stack("index-lag");
    let keys = keys(&world);
    // Bobbin has the bottom two and has never heard of the top.
    bobbin_agrees_except(&world, &[]);
    world.with(|w| {
        w.bobbin_pulls
            .retain(|p| !p["uri"].as_str().unwrap_or_default().ends_with(&keys[2]))
    });

    let json = world
        .command(&["stack", "view", "--source", "pds,bobbin", "--json"])
        .finish()
        .success()
        .json();
    let members: Vec<&str> = json["members"]
        .as_array()
        .expect("members")
        .iter()
        .map(|m| m["rkey"].as_str().unwrap_or_default())
        .collect();
    assert_eq!(members.len(), 3, "the lagging member was dropped: {json:#}");
    assert_eq!(members[0], keys[2], "{json:#}");
}

/// A stale index row does not turn a retire into a deletion.
///
/// The direction the `EVERY` trade would not be safe in, and the reason the
/// state rule turns around on your own repo. Bobbin's state normally wins,
/// because most pulls are written by people whose PDSes are not being read —
/// but Bobbin accepts a status record from only the pull's author and the
/// repo's owner, so when the account being read is both, there is nobody left
/// to be better informed and a disagreement just means the index is behind.
///
/// What it would cost here if the rule did not turn around: an index that has
/// not caught up with a close reports the member open, an open member whose
/// commit has left the branch is a *drop*, and a drop under `--prune` is a
/// deleted record — the one holding the close and the comments explaining it.
/// A retire would become a deletion on the word of a row that is merely late.
#[test]
fn a_stale_index_row_does_not_turn_a_retire_into_a_deletion() {
    let world = published_stack("index-stale-close");
    let keys = keys(&world);
    world.run(&["pr", "close", &keys[1]]).success();
    world.checkout.drop_below(1);
    // The close is a record on Alice's own PDS; the index still says open.
    bobbin_agrees_except(&world, &[]);

    let report = world.run(&["stack", "resubmit", "--json"]).success().json();

    // Retired on the strength of the account's own record, with no --prune
    // asked for and nothing deleted.
    assert_eq!(
        report["retired"][0]["rkey"].as_str(),
        Some(keys[1].as_str()),
        "a stale index row overruled the account's own close: {report:#}"
    );
    assert!(
        report["drops"].as_array().is_some_and(|d| d.is_empty()),
        "{report:#}"
    );
    assert!(
        world.pulls(ALICE).iter().any(|(k, _)| *k == keys[1]),
        "the closed member's record went"
    );
}

/// A stack on a repo somebody else owns: Alice is a contributor here.
///
/// The `sh.tangled.repo` record moves to Bob's PDS, which is the whole of
/// what ownership is — `ownership::owns_repo` looks for the record in the
/// acting account's own collection. The repo is still findable and the knot
/// is still named; what changes is that Alice is no longer one of the two
/// accounts Bobbin accepts a status record from.
fn contributors_stack(label: &str) -> Scenario {
    let world = published_stack(label);
    world.with(|w| {
        let mut record = w
            .get(ALICE, support::REPO_NSID, "demo")
            .expect("the repo record")
            .value
            .clone();
        record["owner"] = serde_json::json!(BOB);
        w.repo(ALICE)
            .records
            .remove(&(support::REPO_NSID.to_string(), "demo".to_string()));
        w.plant(BOB, support::REPO_NSID, "demo", record);
    });
    world
}

/// On somebody else's repo, an unindexed stack has no states at all.
///
/// Not a defect in the reading: a pull's state lives in status records that
/// the author *and the repo's owner* may both write, so on a repo you do not
/// own, your own PDS is no longer the whole answer — a maintainer's close or
/// merge is a record in their PDS, which is not being read. `?` is the honest
/// answer, and the distance between "nobody acted" and "we cannot see who
/// acted" is the distance the `?` exists to hold.
///
/// It is worth pinning because it is invisible from the owner's side: every
/// other stack test in this file runs on Alice's own repo, where the same
/// records settle to `open`, and nothing would have caught this reading
/// changing.
#[test]
fn a_contributors_stack_has_no_states_without_the_index() {
    let world = contributors_stack("contrib-unsettled");

    // The column a person reads.
    let run = world.run(&["stack", "view"]).success();
    assert_eq!(
        run.stdout.matches(" ?  ").count(),
        3,
        "every member should read unsettled:\n{}",
        run.stdout
    );

    // And `null` rather than a guess in `--json`, which is the shape a script
    // can tell apart from "open".
    let json = world.run(&["stack", "view", "--json"]).success().json();
    for member in json["members"].as_array().expect("members") {
        assert!(member["state"].is_null(), "{json:#}");
    }
}

/// And the write commands refuse it rather than guess.
///
/// Both refusals are the safe half of that `?`. A merge lands the member
/// named *and everything unmerged below it*, so a row it cannot settle is one
/// it cannot know it should skip; a reconcile that cannot settle a member
/// whose commit has left the branch cannot tell a merge from an abandonment,
/// and one of those two answers deletes a record.
#[test]
fn an_unsettled_stack_is_refused_rather_than_guessed_at() {
    let world = contributors_stack("contrib-refusals");

    world
        .run(&["stack", "merge", "--dry-run"])
        .refused("open pulls only");

    world.checkout.drop_top();
    world
        .run(&["stack", "resubmit"])
        .refused("state cannot be settled");
}

/// The index is what settles it, and asking is enough.
///
/// The other half of the same trade: on a repo you do not own, Bobbin *is*
/// better informed, and the states it carries are the ones the refusals above
/// are waiting for. This is the whole recovery story for a contributor's
/// stack, and it is worth a test because it is the only one there is.
#[test]
fn the_index_settles_a_contributors_stack() {
    let world = contributors_stack("contrib-indexed");
    bobbin_agrees_except(&world, &[]);

    let json = world
        .command(&["stack", "view", "--source", "pds,bobbin", "--json"])
        .finish()
        .success()
        .json();
    let states: Vec<&str> = json["members"]
        .as_array()
        .expect("members")
        .iter()
        .map(|m| m["state"].as_str().unwrap_or_default())
        .collect();
    assert_eq!(states, ["open", "open", "open"], "{json:#}");

    // And the refusal lifts: the reconcile can tell a closed member from one
    // whose state it never learned.
    world.checkout.drop_top();
    world.run(&["stack", "resubmit"]).refused("--prune");
}

/// **A stack almost never starts as one.** It starts as `atgc pr create`,
/// and the second feature arrives on top of it later — and until this, that
/// was a shape no command could produce. `stack create` refused ("already
/// has a pull request"), `stack resubmit` refused ("not stacked"), and
/// `stack link` refuses two pulls whose patches share a commit, which is
/// exactly what `pr create` records for a branch sitting on top of another.
/// Three refusals around the ordinary case.
///
/// So the branch's existing pull becomes the stack's bottom member. It keeps
/// its record key — and with it its number, its comments and its rounds —
/// and gains a round, because the patch `pr create` recorded spans the whole
/// branch and a member's spans only its own cut.
#[test]
fn a_branch_whose_pull_predates_the_stack_is_adopted_as_its_bottom() {
    let world = Scenario::new("create-adopts-the-flat-pull");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world.with(|w| w.compare = Ok((1, knot_mailbox(&[BOTTOM]))));
    world
        .run(&["pr", "create", "--title", "the first feature"])
        .success();

    let flat = keys(&world);
    assert_eq!(flat.len(), 1, "the fixture wanted one flat pull");
    world.clear_journal();

    // The second feature, on top of the first.
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", Some(MIDDLE));
    world.with(|w| w.compare = Ok((2, knot_mailbox(&[BOTTOM, MIDDLE]))));

    let run = world.run(&["stack", "create"]).success();
    assert!(
        run.stdout.contains("adopt:"),
        "nothing said the existing pull was adopted\n--- stdout ---\n{}",
        run.stdout
    );

    // Two members, and the bottom is the *same record* as before: a new
    // record here would be a second pull request for a commit that already
    // had one, and the comments on it would be stranded.
    let after = keys(&world);
    assert_eq!(after.len(), 2, "a stack of two was not written");
    assert!(
        after.contains(&flat[0]),
        "the existing pull was re-minted rather than adopted: {flat:?} -> {after:?}"
    );
    assert_chained(&world, ALICE, None);

    // The adopted member holds a second round: its first patch was the whole
    // branch, and its cut is one commit of it.
    let bottom = world
        .pulls(ALICE)
        .into_iter()
        .find(|(k, _)| k == &flat[0])
        .expect("the adopted record");
    assert_eq!(
        bottom.1["rounds"].as_array().map(Vec::len),
        Some(2),
        "the adopted member did not get the round its new patch needs"
    );
}

/// Adoption is for an *open* pull. A merged or closed one is history, and a
/// stack hung off it is a stack whose bottom is never going to be reviewed —
/// so this refuses rather than building on it, and says which state it found.
#[test]
fn a_closed_pull_on_the_branch_is_not_adopted() {
    let world = Scenario::new("create-will-not-adopt-a-closed-pull");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world.with(|w| w.compare = Ok((1, knot_mailbox(&[BOTTOM]))));
    world
        .run(&["pr", "create", "--title", "the first feature"])
        .success();
    let flat = keys(&world);
    world.run(&["pr", "close", &flat[0]]).success();
    world.clear_journal();

    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", Some(MIDDLE));
    world.with(|w| w.compare = Ok((2, knot_mailbox(&[BOTTOM, MIDDLE]))));

    world.run(&["stack", "create"]).refused("is closed");
    assert_eq!(
        keys(&world).len(),
        1,
        "the refused run must not have added records"
    );
}

/// A chain that is already forked stops every write over it, and stops it
/// before anything is sent.
///
/// **This pins the read-time refusal, not the write-time guard.** Worth
/// saying, because the two are easy to confuse and this test was written
/// expecting the second: the refusal it gets comes from `own_chain`, which
/// walks the recorded chain before a reconcile plans anything, and it would
/// fire without `refuse_new_damage` existing at all. What is genuinely new
/// here is the second assertion — that nothing reached `applyWrites` — since
/// a partial write into a forked stack is the state none of these commands
/// can repair.
///
/// The write-time guard has no flow test because no CLI path reaches it
/// today: every verb that can currently damage a chain already refuses by
/// hand, which is exactly the arrangement `refuse_new_damage` is a backstop
/// for. Its proof is the unit tests over `refuse_new_damage` in
/// `cmd::stack::tests`, which hand it the ops a careless verb would build.
#[test]
fn a_forked_chain_stops_a_reconcile_before_anything_is_sent() {
    let world = published_stack("guard-refuses-a-fork");
    let keys = keys(&world);

    // A second live pull hanging off the bottom member: a fork, on record,
    // that no atgc command created and none would.
    world.with(|w| {
        let bottom = format!("at://{ALICE}/{PULL_NSID}/{}", keys[0]);
        w.plant(
            ALICE,
            PULL_NSID,
            "3interloper000",
            serde_json::json!({
                "title": "a second branch off the bottom",
                "dependentOn": bottom,
                "source": {"branch": "claude/stack"},
                "target": {"repo": REPO_DID, "branch": "main"},
                "rounds": [],
                "createdAt": "2026-08-28T00:00:00Z",
            }),
        );
    });
    world.clear_journal();

    // Any stack write now has to answer for the shape it leaves behind. The
    // reconcile cannot: the chain it would write is one the reader refuses.
    let run = world.run(&["stack", "resubmit"]);
    assert_ne!(
        run.code,
        Some(0),
        "a reconcile over a forked chain was allowed\n--- stderr ---\n{}",
        run.stderr
    );

    // And it refused before sending, which is the half that matters: a
    // partial write here is a stack nobody can repair with these commands.
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "records were written despite the refusal"
        );
    });
}

/// **A rewrite is undone when the command around it fails.**
///
/// The two tests above pin the refusals that were *hoisted* above the
/// rewrite, one at a time, as each was found. That approach has now failed
/// twice: the rule was written down, applied to `stack create` on
/// 2026-08-15, and broken again in `stack resubmit` ten days later, because
/// "every step that can refuse" is a list nobody keeps current.
///
/// So the rewrite is undone instead of being carefully sequenced around. This
/// takes the one failure that genuinely cannot be hoisted — the PDS refusing
/// the batch, which is only knowable by sending it — and asserts the branch
/// comes back anyway.
#[test]
fn a_refused_batch_puts_the_rewritten_branch_back() {
    let world = Scenario::new("create-batch-refused");
    unstacked_branch(&world);
    let tip = world.checkout.head();
    world.with(|w| w.fail_next_batch_swap = true);

    let run = world.run(&["stack", "create", "--add-change-ids"]);
    assert_ne!(
        run.code,
        Some(0),
        "the PDS refused the batch and the command did not\n{}",
        run.stderr
    );

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch was left carrying shas from a run that wrote nothing"
    );
    assert!(
        run.stderr.contains("has been put back"),
        "the failure did not say the branch was restored\n--- stderr ---\n{}",
        run.stderr
    );
    world.with(|w| assert!(w.collection(ALICE, PULL_NSID).is_empty()));
}

/// **The two newest verbs, in a sequence, with both readers checked after
/// every step.**
///
/// `every_edit_leaves_the_two_readers_agreeing` covers amend, reorder,
/// insert, retire and merge, and predates both `stack create`'s adoption of
/// an existing pull and `stack unlink`. Those are the least battle-tested
/// paths in this epic and the two most recent chances to leave a chain the
/// appview reads differently, which is exactly the failure the oracle here
/// exists to catch and no single-command test can.
///
/// The sequence: a plain `pr create`, grown into a stack by adopting it, a
/// member added on top, the middle taken out with `unlink`, and a reconcile
/// over what is left.
#[test]
fn adopting_and_unlinking_leave_a_chain_both_readers_agree_on() {
    let world = Scenario::new("adopt-then-unlink");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world.with(|w| w.compare = Ok((1, knot_mailbox(&[BOTTOM]))));
    world
        .run(&["pr", "create", "--title", "the first feature"])
        .success();
    let flat = keys(&world);
    assert_eq!(flat.len(), 1, "the fixture wanted one flat pull");

    // Grown into a stack: the existing pull becomes the bottom member.
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", Some(MIDDLE));
    world.with(|w| w.compare = Ok((2, knot_mailbox(&[BOTTOM, MIDDLE]))));
    world.run(&["stack", "create"]).success();
    world.assert_stack_reads_alike(ALICE);
    assert!(
        keys(&world).contains(&flat[0]),
        "the adopted pull was re-minted rather than kept"
    );

    // A third member on top.
    world
        .checkout
        .commit("three.txt", "three\n", "feat: top", Some(TOP));
    world.with(|w| w.compare = Ok((3, knot_mailbox(&[BOTTOM, MIDDLE, TOP]))));
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);

    // The middle taken out of the chain. The member above it has to inherit
    // the one below, or the top is left depending on a pull that is no
    // longer in the stack — the gap `unlink` exists to close.
    let before_unlink = chain(&world, ALICE);
    assert_eq!(before_unlink.len(), 3, "the fixture wanted three members");
    let (middle, top) = (before_unlink[1].0.clone(), before_unlink[2].0.clone());
    world.run(&["stack", "unlink", &middle]).success();
    world.assert_stack_reads_alike(ALICE);

    // **Agreement is not correctness**, and this is the assertion that says
    // so. `assert_stack_reads_alike` checks that atgc and the appview read
    // the same chain off the same records — two readers can agree perfectly
    // on a chain that is wrong. An unlink that dropped the relink leaves the
    // top depending on the member just removed, which is a shape both
    // readers walk quite happily; it just is not the stack anybody asked
    // for. So the shape itself is named.
    let after = chain(&world, ALICE);
    let parent_of = |rkey: &str| {
        after
            .iter()
            .find(|(k, _)| k == rkey)
            .map(|(_, p)| p.clone())
            .expect("the record is still there")
    };
    assert_eq!(
        parent_of(&top),
        Some(flat[0].clone()),
        "the top did not inherit the bottom when the middle was unlinked"
    );
    // The unlinked member keeps its record — `unlink` takes a pull *out of a
    // chain*, it does not delete it — and what makes it out is having no
    // parent and nothing depending on it.
    assert_eq!(
        parent_of(&middle),
        None,
        "the unlinked member kept its link"
    );
    assert!(
        !after
            .iter()
            .any(|(_, p)| p.as_deref() == Some(middle.as_str())),
        "something is still depending on the unlinked member"
    );

    // And a reconcile over what unlink left, which is the step that finds
    // out whether the chain it wrote is one the reconcile can still plan
    // against.
    world.run(&["stack", "resubmit"]).success();
    world.assert_stack_reads_alike(ALICE);
}

/// **The repo owner merging a contributor's stack**, which is the direction
/// every other contributor-stack test here leaves out.
///
/// `a_contributors_stack_has_no_states_without_the_index` and its two
/// neighbours all have Alice — the contributor — acting on her own stack in
/// Bob's repo. The maintainer's side is the one that actually lands work,
/// and it moves records between two PDSes in a way nothing else does: the
/// pulls are Alice's and stay Alice's, while the `merged` status Bob writes
/// is a record in *Bob's* repository. A merge that wrote into the author's
/// PDS would need push access nobody has.
#[test]
fn the_owner_merges_a_contributors_stack_from_their_own_pds() {
    let world = contributors_stack("owner-merges-contributors-stack");
    let keys = keys(&world);
    // The maintainer needs an index to see a stack that is not theirs at
    // all: a contributor's pull records live in the contributor's PDS, and
    // `stack merge` reads `Source::EVERY` precisely so that it can.
    bobbin_agrees_except(&world, &[]);
    world.clear_journal();

    // Bob owns the repo and the knot lets him push; the stack is Alice's.
    world
        .run_as(BOB, &["stack", "merge", "--through", "1"])
        .success();

    // The status record is the owner's, in the owner's repository.
    let merged: Vec<String> = world
        .records(BOB, PULL_STATUS_NSID)
        .into_iter()
        .filter(|(_, v)| v["status"].as_str() == Some("sh.tangled.repo.pull.status.merged"))
        .map(|(_, v)| v["pull"].as_str().unwrap_or_default().to_string())
        .collect();
    assert!(
        merged.iter().any(|p| p.ends_with(&keys[0])),
        "the owner's merge left no status record of its own: {merged:?}"
    );
    assert!(
        world
            .records(ALICE, PULL_STATUS_NSID)
            .iter()
            .all(|(_, v)| v["status"].as_str() != Some("sh.tangled.repo.pull.status.merged")),
        "a merged status was written into the author's PDS"
    );

    // And the author's pull records are untouched: a merge records an
    // outcome, it does not edit the pull it is about.
    world.with(|w| {
        assert_eq!(
            w.collection(ALICE, PULL_NSID).len(),
            3,
            "the owner's merge added or removed one of the author's pulls"
        );
    });

    // **The merge commit belongs to the person whose patch it is.** git
    // attribution is permanent and shows in every log; a maintainer landing
    // a contributor's work under their own name is a wrong that no record
    // here would show and no reconcile could undo.
    world.with(|w| {
        let merges = w.calls_to("sh.tangled.repo.merge");
        assert_eq!(merges.len(), 1, "the knot was not asked to merge");
        assert_eq!(
            merges[0].body["authorEmail"].as_str(),
            Some(ALICE),
            "the merge was attributed to the maintainer, not the author: {}",
            merges[0].body
        );
    });

    // The chain the author wrote is still the chain both readers see.
    world.assert_stack_reads_alike(ALICE);
}

/// The knot is what decides whether a merge may happen at all, and an owner
/// who has lost push access is refused there rather than here — the same
/// answer a contributor gets, from the same place. Pinned because the
/// ownership check above and the push check are different questions, and
/// passing the first is not passing the second.
#[test]
fn an_owner_without_push_access_cannot_merge_a_contributors_stack() {
    let world = contributors_stack("owner-without-push");
    bobbin_agrees_except(&world, &[]);
    // A knot that lets only Alice push, so Bob owns the repo and still
    // cannot land anything on it.
    world.with(|w| w.knot_push_allowed = Some(vec![ALICE.to_string()]));

    world
        .run_as(BOB, &["stack", "merge", "--through", "1"])
        .refused("push");

    assert!(
        world
            .records(BOB, PULL_STATUS_NSID)
            .iter()
            .all(|(_, v)| v["status"].as_str() != Some("sh.tangled.repo.pull.status.merged")),
        "a refused merge wrote a merged status anyway"
    );
}

/// Two *other* accounts with a pull on the same branch name is a question a
/// merge cannot answer, so it names them instead of guessing.
///
/// Own pulls still win the entry, which is what keeps this from being a
/// regression in the ordinary case: a branch name is not unique across
/// accounts, and landing a stranger's stack because it shares a name with
/// yours would merge the wrong work.
#[test]
fn a_branch_two_strangers_both_have_a_pull_on_is_refused_rather_than_guessed() {
    let world = contributors_stack("merge-ambiguous-branch");
    bobbin_agrees_except(&world, &[]);

    // A third account's pull, on the same branch, in its own repository —
    // and in the index, which is the only way a maintainer sees either of
    // them. Planting the record alone proves nothing: `stack merge` reads
    // the listing, and a record no source returns is not in it.
    const CAROL: &str = "did:plc:cccccccccccccccccccccccd";
    let carols = serde_json::json!({
        "title": "carol's unrelated work",
        "source": {"branch": "feature"},
        "target": {"repo": REPO_DID, "branch": "main"},
        "rounds": [],
        "createdAt": "2026-08-28T00:00:00Z",
    });
    world.with(|w| {
        w.plant(CAROL, PULL_NSID, "3carols000000", carols.clone());
        w.bobbin_pulls.push(serde_json::json!({
            "uri": format!("at://{CAROL}/{PULL_NSID}/3carols000000"),
            "state": "open",
            "commentCount": 0,
            "value": carols,
        }));
    });

    let run = world.run_as(BOB, &["stack", "merge", "--through", "1"]);
    assert_ne!(run.code, Some(0), "a merge picked one of two stacks");
    assert!(
        run.stderr
            .contains("accounts have a pull request on branch"),
        "the refusal did not name the ambiguity\n--- stderr ---\n{}",
        run.stderr
    );
    assert!(
        world
            .records(BOB, PULL_STATUS_NSID)
            .iter()
            .all(|(_, v)| v["status"].as_str() != Some("sh.tangled.repo.pull.status.merged")),
        "a refused merge wrote a merged status anyway"
    );
}

/// A chain whose members belong to two accounts: Alice's pull at the bottom,
/// Bob's sitting on it. No atgc command can build this — `stack create` cuts
/// one branch and `stack link` refuses a pull that is not yours — but
/// Tangled's web UI can, and `chain_containing` reads it, so what every verb
/// does with one is a real question.
fn mixed_author_chain(label: &str) -> Scenario {
    let world = Scenario::new(label);
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world.with(|w| w.compare = Ok((1, knot_mailbox(&[BOTTOM]))));
    world
        .run(&["pr", "create", "--title", "alice's bottom"])
        .success();
    let bottom = keys(&world).remove(0);

    world.with(|w| {
        w.plant(
            BOB,
            PULL_NSID,
            "3bobstop00000",
            serde_json::json!({
                "title": "bob's member on top",
                "dependentOn": format!("at://{ALICE}/{PULL_NSID}/{bottom}"),
                "source": {"branch": "feature"},
                "target": {"repo": REPO_DID, "branch": "main"},
                "rounds": [],
                "createdAt": "2026-08-28T00:00:00Z",
            }),
        );
    });
    world
}
/// **A pull somebody else has stacked on top of reads as "not stacked", and
/// the refusal now says what it did not look at.**
///
/// A pull is stacked when something depends on it, and that something is a
/// record in the other account's repository — so "not stacked" is a claim
/// about records the acting account does not hold. Off an index it is not a
/// claim atgc is in a position to make, and it was making it anyway: the
/// author of the bottom member was told their pull stands alone while
/// somebody else's work sat on top of it.
///
/// The same shape as most of this session's bugs — a partial view asserted
/// as a fact — and the same fix: say what was read.
#[test]
fn a_pull_someone_else_stacked_on_says_what_it_could_not_see() {
    let world = mixed_author_chain("mixed-caveat-unindexed");

    // `stack view` reads the PDS alone by default, so the caveat names the
    // index rather than blaming it.
    let run = world.run(&["stack", "view"]);
    assert_eq!(run.code, Some(2), "{}", run.stderr);
    assert!(
        run.stderr.contains("only your own records were read"),
        "the refusal claimed more than it read\n--- stderr ---\n{}",
        run.stderr
    );

    // The write paths read `Source::EVERY`, so theirs is the weaker caveat:
    // the index was asked, and its ingest stalls.
    let run = world.run(&["stack", "resubmit", "--dry-run"]);
    assert_eq!(run.code, Some(2), "{}", run.stderr);
    assert!(
        run.stderr.contains("the index was read too"),
        "the refusal did not admit the index can lag\n--- stderr ---\n{}",
        run.stderr
    );
}

/// The member on top *can* see the chain it sits in, and says the part it
/// cannot hold rather than reporting a stack of one.
///
/// Already right, and pinned because the asymmetry is worth keeping on
/// purpose: a `dependentOn` points down, so the member above names the one
/// below and can report it missing, while the member below has nothing
/// pointing at whatever depends on it.
#[test]
fn the_member_above_reports_the_part_of_the_chain_it_cannot_hold() {
    let world = mixed_author_chain("mixed-above");
    let run = world.run_as(BOB, &["stack", "view"]).success();
    assert!(
        run.stdout.contains("continues below"),
        "a truncated chain was reported as the whole of it\n--- stdout ---\n{}",
        run.stdout
    );
}

/// **Neither account can reconcile a chain they only half own.** `resubmit`
/// writes every member's record, so a chain spanning two repositories is one
/// no single account can reconcile — a boundary of the design rather than a
/// gap in it, and untested until now. Both sides are refused by name, and
/// neither writes anything.
#[test]
fn a_two_author_chain_cannot_be_reconciled_by_either_account() {
    let world = mixed_author_chain("mixed-reconcile");
    bobbin_agrees_except(&world, &[]);
    world.with(|w| {
        let bobs = w
            .get(BOB, PULL_NSID, "3bobstop00000")
            .expect("bob's member")
            .value
            .clone();
        w.bobbin_pulls.push(serde_json::json!({
            "uri": format!("at://{BOB}/{PULL_NSID}/3bobstop00000"),
            "state": "open",
            "commentCount": 0,
            "value": bobs,
        }));
    });

    for who in [ALICE, BOB] {
        let run = world.run_as(who, &["stack", "resubmit", "--dry-run"]);
        assert_ne!(run.code, Some(0), "a half-owned chain was reconciled");
        assert!(
            run.stderr.contains("only a stack's author can write it"),
            "the refusal did not name the ownership problem\n--- stderr ---\n{}",
            run.stderr
        );
    }
}
/// **The stack write paths say out loud when the index they rely on is
/// down**, and carry on with what they can read.
///
/// `Source::EVERY` is the deliberate exception to "indexes are opt-in", and
/// its argument is that these commands read the listing to find a reason to
/// *stop* — so seeing less costs a refusal that turns out to be unnecessary.
/// A dead index inverts that: fewer rows means fewer reasons to stop.
///
/// Two of the three ways that could hurt are structurally guarded. A member
/// below one that is visible is named by its `dependentOn`, so an invisible
/// one shows up as `missing_below` and both verbs refuse outright. The third
/// is not guardable at all: nothing points *upward*, so a contributor's
/// member stacked on top is invisible with no trace it ever existed. The
/// warning is the whole of the mitigation available, which is why it is
/// pinned rather than left to chance.
#[test]
fn a_write_path_says_when_the_index_it_relies_on_is_down() {
    for args in [
        &["stack", "resubmit", "--dry-run"][..],
        &["stack", "merge", "--dry-run"],
    ] {
        let world = published_stack(&format!("index-down-{}", args[1]));
        world.with(|w| w.bobbin_fails = Some("InternalServerError"));

        let run = world.run(args);
        assert!(
            run.stderr.contains("could not reach Bobbin"),
            "`atgc {}` did not say the index was unreachable\n--- stderr ---\n{}",
            args.join(" "),
            run.stderr
        );
    }
}

/// `stack view` reads no index unless asked, so a dead one is not its
/// business and it says nothing about it. Pinned so that a future change
/// which starts consulting Bobbin here has to notice it is also inheriting
/// the warning.
#[test]
fn stack_view_is_untroubled_by_an_index_it_never_asked() {
    let world = published_stack("index-down-view");
    world.with(|w| w.bobbin_fails = Some("InternalServerError"));

    let run = world.run(&["stack", "view"]).success();
    assert!(
        !run.stderr.contains("Bobbin"),
        "the default read reached for an index\n--- stderr ---\n{}",
        run.stderr
    );
}

/// **The knot merged and the records did not**, which is the one half-state
/// this tool cannot avoid and had never exercised.
///
/// A merge is two writes to two services with no transaction over them: the
/// knot moves the branch, then the PDS records a merged status per pull. The
/// second failing leaves work that is genuinely landed and pulls that every
/// listing still calls open — recoverable, but only by somebody who knows
/// exactly which pulls are in that state.
///
/// So the refusal has to name them, and nothing checked that it did. This is
/// the message a person reads at the worst moment they will have with this
/// command.
#[test]
fn a_merge_whose_statuses_fail_names_the_pulls_left_open() {
    let world = published_stack("merge-statuses-fail");
    let keys = keys(&world);
    world.with(|w| w.fail_next_batch_swap = true);

    let run = world.run(&["stack", "merge", "--through", "1"]);
    assert_ne!(run.code, Some(0), "a failed status write reported success");
    assert!(
        run.stderr.contains("the knot merged the patch"),
        "the failure did not say the branch had already moved\n--- stderr ---\n{}",
        run.stderr
    );
    assert!(
        run.stderr.contains(&keys[0]),
        "the pull left reading open was not named\n--- stderr ---\n{}",
        run.stderr
    );

    // The knot really was asked to merge: this is a half-state, not a
    // refusal before anything happened. Getting that backwards would make
    // the message a lie in the more alarming direction.
    world.with(|w| {
        assert_eq!(
            w.calls_to("sh.tangled.repo.merge").len(),
            1,
            "the merge never reached the knot, so nothing is half-done"
        );
    });
}

/// A dead appview costs a pull its *number* and nothing else: the record key
/// is the identifier that always exists, and every command takes one.
///
/// Worth separating from the refusals above. Losing the appview stops
/// `pr close`, because the owner it resolves decides whether a write is
/// safe; it must not stop a read that was only going to make the output
/// prettier. A number is a convenience the appview mints and atgc never
/// holds, so its absence is a cosmetic degradation and the command has to
/// carry on.
#[test]
fn a_dead_appview_costs_a_number_and_not_the_listing() {
    let world = published_stack("stack-view-appview-down");
    let keys = keys(&world);
    world.with(|w| w.web_fails = true);

    let run = world.run(&["stack", "view"]).success();
    for rkey in &keys {
        assert!(
            run.stdout.contains(rkey.as_str()),
            "a member vanished when the appview went down\n--- stdout ---\n{}",
            run.stdout
        );
    }
}
/// **A knot that refuses has done nothing; a knot that dies may have merged
/// anyway**, and the two must not read alike.
///
/// `Exit::Unreachable` means "retry later; a host did not answer", which is
/// right for a check that never ran and wrong for a merge whose outcome is
/// unknown: retrying could be retrying something already sitting on the
/// target branch. The distinction is whether the knot composed a refusal —
/// a tagged 4xx means it got far enough to decide — or simply stopped.
#[test]
fn a_merge_call_that_dies_says_the_branch_may_have_moved() {
    let world = published_stack("knot-dies-mid-merge");
    world.with(|w| w.knot_fails = Some(("sh.tangled.repo.merge", "BadGateway")));

    let run = world.run(&["stack", "merge", "--through", "1"]);
    assert_eq!(run.code, Some(6), "{}", run.stderr);
    assert!(
        run.stderr.contains("may have merged anyway"),
        "an unknown outcome was reported as a plain failure\n--- stderr ---\n{}",
        run.stderr
    );
    // Nothing was recorded, which is what makes the warning necessary rather
    // than merely informative: the records and the branch may now disagree.
    assert!(
        world
            .records(ALICE, PULL_STATUS_NSID)
            .iter()
            .all(|(_, v)| v["status"].as_str() != Some("sh.tangled.repo.pull.status.merged")),
        "a merge of unknown outcome recorded itself as done"
    );
}

/// The check dying is unambiguous: nothing has run, so "retry later" is the
/// whole of the advice and the branch is not mentioned. Pinned beside the
/// case above, because a warning that fires on every knot hiccup would be
/// noise and would stop being read.
#[test]
fn a_merge_check_that_dies_does_not_claim_anything_may_have_landed() {
    let world = published_stack("knot-dies-at-check");
    world.with(|w| w.knot_fails = Some(("sh.tangled.repo.mergeCheck", "BadGateway")));

    let run = world.run(&["stack", "merge", "--through", "1"]);
    assert_eq!(run.code, Some(6), "{}", run.stderr);
    assert!(
        !run.stderr.contains("may have merged anyway"),
        "a check that never ran was reported as possibly landed\n--- stderr ---\n{}",
        run.stderr
    );
}

/// A knot that refuses on access control keeps its own explanation and gains
/// no ambiguity warning: it decided, and what it decided is actionable.
#[test]
fn a_refused_merge_is_still_reported_as_a_refusal() {
    let world = published_stack("knot-refuses-merge");
    world.with(|w| w.knot_push_allowed = Some(vec![BOB.to_string()]));

    let run = world.run(&["stack", "merge", "--through", "1"]);
    assert!(
        !run.stderr.contains("may have merged anyway"),
        "a decided refusal was reported as an unknown outcome\n--- stderr ---\n{}",
        run.stderr
    );
    assert!(
        run.stderr.contains("push access"),
        "the refusal lost its explanation\n--- stderr ---\n{}",
        run.stderr
    );
}

/// **A refused push leaves the branch where it was**, which is the last of
/// the three knot failures and the only one that happens before any record
/// exists.
///
/// `stack create` pushes the branch before it writes anything, because the
/// `source: {branch}` every member records is a claim the push is what makes
/// true. With `--add-change-ids` the branch has just been rewritten to carry
/// the trailers, so a refusal here is the case the undo was built for —
/// reached through a completely different road than the batch failure that
/// tests it elsewhere.
#[test]
fn a_knot_that_refuses_the_push_leaves_no_rewrite_behind() {
    let world = Scenario::new("push-refused");
    unstacked_branch(&world);
    let tip = world.checkout.head();
    world.checkout.refuse_pushes();

    let run = world.run(&["stack", "create", "--add-change-ids"]);
    assert_ne!(run.code, Some(0), "a refused push reported success");

    assert_eq!(
        world.checkout.head(),
        tip,
        "the branch kept shas from a run that published nothing"
    );
    assert!(
        run.stderr.contains("has been put back"),
        "the failure did not say the branch was restored\n--- stderr ---\n{}",
        run.stderr
    );
    world.with(|w| {
        assert!(
            w.collection(ALICE, PULL_NSID).is_empty(),
            "records were written for a branch the knot never took"
        );
    });
}

/// The same refusal on a reconcile, where the stack already exists: the
/// records must be left exactly as they were, because a half-updated chain
/// is the state no command can repair.
#[test]
fn a_refused_push_on_a_reconcile_writes_no_records() {
    let world = published_stack("push-refused-resubmit");
    let before = chain(&world, ALICE);
    world.checkout.refuse_pushes();
    world.checkout.commit(
        "four.txt",
        "four\n",
        "feat: fourth",
        Some("Ifourth0000000000000000000000000000000a"),
    );
    world.clear_journal();

    let run = world.run(&["stack", "resubmit"]);
    assert_ne!(run.code, Some(0), "a refused push reported success");
    assert_eq!(
        chain(&world, ALICE),
        before,
        "the chain moved for a push the knot refused"
    );
    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.repo.applyWrites").is_empty(),
            "records were written after the push was refused"
        );
    });
}

/// **Your own PDS being down stops a merge before the knot is asked**, which
/// is the ordering that keeps the worst half-state unreachable.
///
/// A merge is two writes to two services with no transaction: the knot moves
/// the branch, then the PDS records it. If the PDS is already known to be
/// unreachable, asking the knot first would land the patch and then
/// certainly fail to record it — manufacturing by hand the exact half-state
/// the code elsewhere apologises for. Reading the listing first is what makes
/// that impossible, and it is worth pinning as an ordering rather than
/// leaving it to the order the lines happen to sit in.
#[test]
fn a_merge_never_reaches_the_knot_when_the_pds_is_already_down() {
    let world = published_stack("merge-own-pds-down");
    world.clear_journal();
    world.with(|w| {
        w.pds_down.insert(ALICE.to_string());
    });

    let run = world.run(&["stack", "merge", "--through", "1"]);
    assert_ne!(run.code, Some(0), "a merge ran with no way to record it");
    world.with(|w| {
        assert!(
            w.calls_to("sh.tangled.repo.merge").is_empty(),
            "the knot was asked to merge with no way to record the outcome"
        );
    });
}

/// **The PDS dying between the knot and the records**, which is the same
/// half-state a refused batch produces and arrives by a different road.
///
/// A refusal is the PDS deciding; unreachability is it not answering. Both
/// leave the branch moved and the pulls reading open, so both have to reach
/// the message that names them — and a classification that treated an
/// unreachable PDS as something else would send somebody looking for a
/// conflict that is not there.
#[test]
fn a_pds_that_dies_after_the_merge_still_names_the_pulls_left_open() {
    let world = published_stack("merge-pds-dies-after");
    let keys = keys(&world);
    world.with(|w| w.pds_method_fails = Some("com.atproto.repo.applyWrites"));

    let run = world.run(&["stack", "merge", "--through", "1"]);
    assert_ne!(run.code, Some(0), "a failed status write reported success");
    assert!(
        run.stderr.contains("the knot merged the patch"),
        "the failure did not say the branch had already moved\n--- stderr ---\n{}",
        run.stderr
    );
    assert!(
        run.stderr.contains(&keys[0]),
        "the pull left reading open was not named\n--- stderr ---\n{}",
        run.stderr
    );
    world.with(|w| {
        assert_eq!(
            w.calls_to("sh.tangled.repo.merge").len(),
            1,
            "the branch did not actually move, so this is not the half-state"
        );
    });
}

/// **Adopting an existing pull leases the batch; building a stack from
/// nothing does not.**
///
/// A stack built from nothing writes only creates — no record can be
/// clobbered, and an unleased batch is right. Adoption changed that: the
/// bottom member becomes an `Op::Update` against a record that already
/// exists, and without a `swapCommit` it lands whatever happened to that
/// pull since it was read. Another session appending a round in between
/// would be overwritten with no sign of it.
///
/// The asymmetry is the point, so both halves are asserted: a lease that
/// appeared on every create would refuse stacks for no reason.
#[test]
fn a_create_leases_its_batch_only_when_it_adopts() {
    // Adopting: the batch carries a lease.
    let world = Scenario::new("create-adopt-leases");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("one.txt", "one\n", "feat: bottom", Some(BOTTOM));
    world.with(|w| w.compare = Ok((1, knot_mailbox(&[BOTTOM]))));
    world
        .run(&["pr", "create", "--title", "the first feature"])
        .success();
    world
        .checkout
        .commit("two.txt", "two\n", "feat: middle", Some(MIDDLE));
    world.with(|w| w.compare = Ok((2, knot_mailbox(&[BOTTOM, MIDDLE]))));
    world.clear_journal();
    world.run(&["stack", "create"]).success();

    world.with(|w| {
        let writes = w.calls_to("com.atproto.repo.applyWrites");
        assert_eq!(writes.len(), 1, "the stack was not written in one batch");
        assert!(
            writes[0].body["swapCommit"].is_string(),
            "an adopting create overwrote a record with no lease: {}",
            writes[0].body
        );
    });

    // Building from nothing: no lease, because nothing can be lost.
    let fresh = Scenario::new("create-fresh-unleased");
    three_commit_branch(&fresh);
    fresh.clear_journal();
    fresh.run(&["stack", "create"]).success();
    fresh.with(|w| {
        let writes = w.calls_to("com.atproto.repo.applyWrites");
        assert_eq!(writes.len(), 1);
        assert!(
            writes[0].body["swapCommit"].is_null(),
            "a create of nothing but new records leased against a repo it is not editing: {}",
            writes[0].body
        );
    });
}

/// **A mark cuts the range, so it has to be in it**, and two marks on one
/// commit would end two pull requests at the same place.
///
/// Neither was refused. `stack mark x origin/main` recorded a branch that
/// cut nothing — the cut is decided by where a mark's commit sits among
/// `base..HEAD`, and one outside them has no position, so the stack that
/// came out was the unmarked one with a branch left behind claiming
/// otherwise. Two marks on one commit leave a member with no commits in it.
#[test]
fn a_mark_that_would_cut_nothing_is_refused() {
    let world = published_stack("mark-outside-range");
    world
        .run(&["stack", "mark", "outside", "origin/main"])
        .refused("would cut nothing");

    let world = published_stack("mark-same-commit");
    world.run(&["stack", "mark", "one", "HEAD~1"]).success();
    world
        .run(&["stack", "mark", "two", "HEAD~1"])
        .refused("no commits in it");
}

/// One condition, one status. `stack view` answered an unstacked branch with
/// `Usage` and the navigation verbs answered it with the unclassified `1`,
/// which is the shape `exit_status.rs` exists to stop: a caller testing for
/// a status got a different answer depending on which verb it ran.
///
/// `--through` out of range joins them: a number the command line got wrong
/// is `Usage`, not "something went wrong, try again".
#[test]
fn one_status_for_a_branch_that_is_not_a_stack() {
    let world = Scenario::new("nav-unstacked-status");
    world.checkout.branch("feature");
    world
        .checkout
        .commit("a.txt", "a\n", "feat: one", Some(BOTTOM));

    for verb in ["up", "down", "top", "bottom"] {
        let run = world.run(&["stack", verb]);
        assert_eq!(
            run.code,
            Some(2),
            "`atgc stack {verb}` answered a branch that is not a stack with {:?}\n{}",
            run.code,
            run.stderr
        );
    }

    let stacked = published_stack("through-out-of-range");
    let run = stacked.run(&["stack", "merge", "--through", "9"]);
    assert_eq!(run.code, Some(2), "{}", run.stderr);
}

/// **A record key alone does not say whose record it is**, and the refusal
/// now says which account it was looked up in and how to name another.
///
/// `author_of_rkey` answers the acting account when nothing else names one,
/// so a key belonging to somebody else is looked up in the wrong repository
/// and comes back a flat 404 naming a DID the reader never mentioned. Both
/// ways out — an at-uri, or `--author` — are named.
#[test]
fn a_record_key_that_is_not_yours_says_where_it_looked() {
    let world = published_stack("bare-key-not-yours");
    let rkey = keys(&world).remove(0);

    let run = world.run_as(BOB, &["stack", "unlink", &rkey]);
    assert_ne!(run.code, Some(0), "somebody else's pull was unlinked");
    assert!(
        run.stderr.contains("the account this command is acting as"),
        "the refusal did not say whose repository it searched\n--- stderr ---\n{}",
        run.stderr
    );
    assert!(
        run.stderr.contains("--author"),
        "the refusal named no way to reach the right account\n--- stderr ---\n{}",
        run.stderr
    );
}

/// An at-uri that resolves to nothing has already said whose account it
/// meant, so it gets no advice about naming one. Pinned because the hint
/// above is only useful while it is rare.
#[test]
fn an_at_uri_that_resolves_to_nothing_gets_no_bare_key_advice() {
    let world = published_stack("at-uri-missing");
    let missing = format!("at://{ALICE}/{PULL_NSID}/3zzzzzzzzzzzz");

    let run = world.run(&["stack", "unlink", &missing]);
    assert_ne!(run.code, Some(0));
    assert!(
        !run.stderr.contains("bare record key"),
        "an exact reference was given advice about inexact ones\n--- stderr ---\n{}",
        run.stderr
    );
}

/// **A rebase git stopped in the middle of is not a detached HEAD somebody
/// chose**, and every stack verb said it was.
///
/// A conflicted rebase leaves HEAD on no branch, so `current_branch` failed
/// with "detached HEAD … `git checkout <branch>` first" — true, useless, and
/// advice that would throw away the resolution in the working tree. Somebody
/// who has just hit a conflict is the likeliest reader of any message here,
/// and it was the one message that did not mention the rebase.
#[test]
fn a_conflicted_rebase_is_reported_as_one_and_not_as_a_detached_head() {
    let world = published_stack("mid-rebase-report");
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("one.txt", "theirs\n", "chore: conflicting change", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    world
        .run(&["stack", "rebase"])
        .refused("the rebase stopped");

    for verb in ["view", "resubmit"] {
        let run = world.run(&["stack", verb]);
        assert!(
            run.stderr.contains("a rebase is in progress"),
            "`atgc stack {verb}` did not mention the rebase\n--- stderr ---\n{}",
            run.stderr
        );
        assert!(
            run.stderr.contains("--abort"),
            "the way back was not named\n--- stderr ---\n{}",
            run.stderr
        );
    }
}

/// **A mark outside the range is silently not a cut**, and the commands that
/// act on the cut said nothing about it.
///
/// A stack recorded as grouped members reconciles as one pull request per
/// commit when the mark that grouped them is left on a commit the branch no
/// longer has — which a plain `git rebase` does, exactly when nobody is
/// thinking about marks. `stack mark` reports it when asked; `create` and
/// `resubmit` now say it where the shape is being decided.
#[test]
fn a_mark_left_outside_the_range_is_named_where_the_cut_is_made() {
    let world = published_stack("stranded-mark-warns");
    world.run(&["stack", "mark", "part1", "HEAD~1"]).success();
    world.checkout.git(&["checkout", "-q", "main"]);
    world
        .checkout
        .commit("z.txt", "z\n", "chore: move main", None);
    world.checkout.sync_remote("main");
    world.checkout.git(&["checkout", "-q", "feature"]);
    // A plain rebase, which does not carry marks.
    world.checkout.git(&["rebase", "-q", "origin/main"]);

    let run = world.run(&["stack", "resubmit", "--dry-run"]).success();
    assert!(
        run.stderr.contains("cutting nothing: part1"),
        "the stranded mark was not named where the cut was decided\n--- stderr ---\n{}",
        run.stderr
    );
    assert!(
        run.stderr.contains("stack rebase"),
        "the warning did not say what carries marks\n--- stderr ---\n{}",
        run.stderr
    );
}

/// **A state this build has not heard of must not be dropped.**
///
/// `State::Known` carries whatever string the index returned — `sources.rs`
/// puts `item["state"]` straight into it — so a state Tangled adds after
/// this build ships arrives intact, and `PullState::from_token` documents
/// that as expected rather than exceptional. It fell into the reconcile's
/// drop bucket, which `--prune` deletes: a member whose commits had left the
/// branch and whose new terminal state this build could not read would have
/// had its record removed.
///
/// `select_merge_range` already refuses an unknown state and
/// `refuse_orphaned_by_rewrite` already counts one as live. This was the one
/// site that treated it as disposable.
#[test]
fn a_state_this_build_does_not_know_is_never_dropped() {
    let world = published_stack("unknown-state-not-dropped");
    let keys = keys(&world);
    // The index reports a state from a future Tangled, for a member whose
    // commit has left the branch.
    bobbin_agrees_except(&world, &[(keys[2].as_str(), "landed-somehow")]);
    world.checkout.drop_top();

    let run = world.run(&["stack", "resubmit", "--prune"]);
    assert_ne!(run.code, Some(0), "an unreadable state was pruned");
    assert!(
        run.stderr.contains("this build does not know"),
        "the refusal did not name the unknown state\n--- stderr ---\n{}",
        run.stderr
    );
    world.with(|w| {
        assert_eq!(
            w.collection(ALICE, PULL_NSID).len(),
            3,
            "a record was deleted for a state atgc could not read"
        );
    });
}

/// **A member's patch comes from its own author's PDS, not the reader's.**
///
/// The blob for a round lives in the repository the pull record is in, so a
/// maintainer landing a contributor's stack has to ask the contributor's
/// PDS for it. `stack merge` asked its own, which made the one command that
/// exists for this the one command that could not do it — and the mock
/// served every blob to every account, so the flow's own test passed.
///
/// Pinned on the journal rather than on success alone: a merge that works
/// for some other reason would still be asking the wrong host.
#[test]
fn a_contributors_patch_is_fetched_from_the_contributors_pds() {
    let world = contributors_stack("merge-reads-authors-pds");
    bobbin_agrees_except(&world, &[]);
    world.clear_journal();

    world
        .run_as(BOB, &["stack", "merge", "--through", "1"])
        .success();

    world.with(|w| {
        let blob_reads = w.calls_to("com.atproto.sync.getBlob");
        assert!(!blob_reads.is_empty(), "the merge read no patch at all");
        for call in &blob_reads {
            assert_eq!(
                call.params.get("did").map(String::as_str),
                Some(ALICE),
                "a member authored by Alice was fetched from {:?}",
                call.params.get("did")
            );
        }
    });
}

/// **A dependent that is not yours is left where it is, and said out loud.**
///
/// `unlink` closes the chain by relinking whatever sat on the member being
/// removed. That record can belong to somebody else — the listing spans
/// authors — and rewriting it is not a thing this account can do: the write
/// went to our own repository under their record key and died with "no pull
/// record <theirs> in <us>", naming a key the user had never seen.
///
/// The unlink still stands. What changes is that the impossible half is not
/// attempted and the user is told which pull stayed attached.
#[test]
fn unlinking_under_somebody_elses_pull_leaves_theirs_alone() {
    let world = mixed_author_chain("unlink-under-a-stranger");
    // `unlink` plans over `Source::EVERY`, so the stranger's record has to be
    // in the listing for the plan to reach for it at all — which is the whole
    // situation: without an index the dependent is invisible and nothing goes
    // wrong, and with one it used to be written to the wrong repository.
    bobbin_agrees_except(&world, &[]);
    world.with(|w| {
        let bobs = w
            .get(BOB, PULL_NSID, "3bobstop00000")
            .expect("bob's member")
            .value
            .clone();
        w.bobbin_pulls.push(serde_json::json!({
            "uri": format!("at://{BOB}/{PULL_NSID}/3bobstop00000"),
            "state": "open",
            "commentCount": 0,
            "value": bobs,
        }));
    });
    let bottom = keys(&world).remove(0);
    world.clear_journal();

    let run = world.run(&["stack", "unlink", &bottom]).success();

    assert!(
        run.stderr.contains("is not yours, so it stays where it is"),
        "the pull left attached was not named\n--- stderr ---\n{}",
        run.stderr
    );
    // Alice's own record was detached...
    assert_eq!(
        parent_of(&world, &bottom),
        None,
        "the unlink did not happen"
    );
    // ...and Bob's was neither written nor fetched.
    world.with(|w| {
        assert!(
            w.get(BOB, PULL_NSID, "3bobstop00000")
                .expect("bob's record")
                .value["dependentOn"]
                .is_string(),
            "somebody else's record was rewritten"
        );
        for call in w.calls_to("com.atproto.repo.getRecord") {
            let asked_for_theirs = call.params.get("rkey").map(String::as_str)
                == Some("3bobstop00000")
                && call.params.get("repo").map(String::as_str) == Some(ALICE);
            assert!(
                !asked_for_theirs,
                "their record key was looked up in our repository"
            );
        }
    });
}

/// **The shape Tangled is actually used in: two contributors, and a repo
/// neither of them owns.**
///
/// Alice stacks against Carol's repository, Bob has his own pull on it, and
/// Carol lands Alice's stack. Three accounts, three PDSes, and no two of
/// them able to read or write each other's records — so every "whose host
/// holds this" question in the merge path is answered against a fixture
/// where a wrong answer is a 404 rather than a coincidence.
///
/// This is the case a two-account fixture cannot state at all: with the
/// acting account owning the repo, "the owner's PDS" and "my PDS" are the
/// same host and the same DID, and a command that confuses them reads as
/// correct.
#[test]
fn a_maintainer_lands_a_contributors_stack_on_a_repo_neither_contributor_owns() {
    let world = Scenario::upstream("upstream-three-accounts");
    three_commit_branch(&world);
    world.run(&["stack", "create"]).success();

    // Bob has a pull on the same repo, from his own PDS. It is not part of
    // Alice's chain and must not be dragged into the merge.
    world.with(|w| {
        w.plant(
            BOB,
            PULL_NSID,
            "3bobsown00000",
            serde_json::json!({
                "title": "bob's unrelated pull",
                "source": {"branch": "bobs-feature"},
                "target": {"repo": REPO_DID, "branch": "main"},
                "rounds": [],
                "createdAt": "2026-08-28T00:00:00Z",
            }),
        );
        // Carol owns the repo, so the knot lets her push and nobody else.
        w.knot_push_allowed = Some(vec![CAROL.to_string()]);
    });
    bobbin_agrees_except(&world, &[]);
    world.clear_journal();

    world
        .run_as(CAROL, &["stack", "merge", "--through", "1"])
        .success();

    world.with(|w| {
        // Alice's patch came from Alice's host, not Carol's and not Bob's.
        for call in w.calls_to("com.atproto.sync.getBlob") {
            assert_eq!(
                call.params.get("did").map(String::as_str),
                Some(ALICE),
                "a member of Alice's stack was fetched from {:?}",
                call.params.get("did")
            );
        }
        // The merged status is Carol's record, in Carol's repository.
        assert_eq!(
            w.collection(CAROL, PULL_STATUS_NSID).len(),
            1,
            "the maintainer's status record is not in the maintainer's repo"
        );
        // Bob's pull was untouched.
        assert!(
            w.collection(BOB, PULL_STATUS_NSID).is_empty(),
            "an unrelated contributor's pull was given a status"
        );
    });
}

/// The same three accounts, from the contributor's side: Alice reconciles
/// her own stack on Carol's repo, and nothing she does reaches for Carol's
/// or Bob's records.
#[test]
fn a_contributor_resubmits_a_stack_on_someone_elses_repo() {
    let world = Scenario::upstream("upstream-contributor-resubmit");
    three_commit_branch(&world);
    world.run(&["stack", "create"]).success();
    const FOURTH: &str = "Ifourth0000000000000000000000000000000a";
    world
        .checkout
        .commit("four.txt", "four\n", "feat: four", Some(FOURTH));
    world.with(|w| w.compare = Ok((4, knot_mailbox(&["one", "two", "three", "four"]))));
    world.clear_journal();

    world.run(&["stack", "resubmit"]).success();

    world.with(|w| {
        assert!(
            w.collection(CAROL, PULL_NSID).is_empty() && w.collection(BOB, PULL_NSID).is_empty(),
            "a contributor's resubmit wrote into another account's repository"
        );
        assert_eq!(
            w.collection(ALICE, PULL_NSID).len(),
            4,
            "the fourth commit did not become a member"
        );
    });
}

/// **A pre-rounds member can be merged, viewed and counted like any other.**
///
/// A record written before the `rounds` array existed carries its patch
/// inline, and Tangled's own backfill produced that shape. `pr view` has
/// always read it; every `stack` path refused it outright with "has no
/// rounds; nothing to read a patch from", so such a pull could be looked at
/// but never landed — and `round_count` called it one round while the reader
/// called it none. Three answers to one question.
///
/// The patch bytes are in the record, so this also asserts the merge fetches
/// no blob at all: reading one would mean the inline case was being routed
/// through the round path.
#[test]
fn a_pre_rounds_member_is_merged_from_its_inline_patch() {
    let world = Scenario::upstream("inline-patch-member");
    world.with(|w| {
        w.plant(
            ALICE,
            PULL_NSID,
            "3preround0000",
            serde_json::json!({
                "title": "feat: written before rounds existed",
                "patch": knot_mailbox(&[BOTTOM]),
                "source": {"branch": "feature"},
                "target": {"repo": REPO_DID, "branch": "main"},
                "createdAt": "2026-01-02T00:00:00Z",
            }),
        );
        w.knot_push_allowed = Some(vec![CAROL.to_string()]);
    });
    bobbin_agrees_except(&world, &[]);
    world.clear_journal();

    let uri = format!("at://{ALICE}/{PULL_NSID}/3preround0000");
    let run = world.run_as(CAROL, &["pr", "merge", &uri]).success();
    assert!(
        !run.stderr.contains("has no rounds"),
        "the inline patch was refused\n--- stderr ---\n{}",
        run.stderr
    );

    world.with(|w| {
        assert!(
            w.calls_to("com.atproto.sync.getBlob").is_empty(),
            "a record carrying its patch inline still went looking for a blob"
        );
        assert_eq!(
            w.collection(CAROL, PULL_STATUS_NSID).len(),
            1,
            "the merge wrote no status"
        );
    });
}