"""Real Git local-work safety: caller divergence, restore faults and rescue retention."""
import contextlib
from pathlib import Path
import subprocess
from unittest.mock import patch

import pytest

from hermes_cli import main as hermes_main, update_cmd
from tests.hermes_cli.test_update_target_identity import git, update_tree  # noqa: F401


@pytest.mark.parametrize('history,failure,keep', [
    ('ordinary', None, False), ('ordinary', None, True),
    ('ordinary', 'reset', False), ('ordinary', 'reset', True),
    ('orphan', None, False), ('orphan', 'reset', False),
    ('orphan', 'ref', False), ('orphan', 'head', False),
])
def test_update_preserves_local_work_and_rescues_orphan_before_reset(
    update_tree, monkeypatch, capsys, history, failure, keep,
):
    t = update_tree
    git(t.clone, 'checkout', '-q', 'main')
    if history == 'orphan':
        git(t.clone, 'checkout', '--orphan', 'fresh')
        git(t.clone, 'branch', '-D', 'main')
        git(t.clone, 'branch', '-m', 'main')
    (t.clone / 'local.txt').write_text('committed\n', encoding='utf-8')
    git(t.clone, 'add', 'local.txt')
    git(t.clone, '-c', 'commit.gpgsign=false', 'commit', '-qm', 'local history')
    before = git(t.clone, 'rev-parse', 'HEAD')
    (t.clone / 'untracked.txt').write_text('local edit\n', encoding='utf-8')
    t.args.channel, t.args.keep_stash = 'main', keep
    monkeypatch.setattr(hermes_main, '_sync_with_upstream_if_needed', update_cmd._sync_with_upstream_if_needed)
    monkeypatch.setattr(update_cmd, '_UPDATE_CRITICAL_MODULES', ())
    original = subprocess.run
    resets = []

    def fault(command, *args, **kwargs):
        if 'reset' in command and '--hard' in command:
            refs = original(['git', 'for-each-ref', '--format=%(objectname)',
                             'refs/hermes-update-backups/'], cwd=t.clone,
                            check=True, capture_output=True, text=True).stdout.split()
            assert refs == ([before] if failure not in {'ref', 'head'} else [])
            resets.append(command)
        if ((failure == 'ref' and 'update-ref' in command and '-d' not in command)
                or (failure == 'reset' and 'reset' in command and '--hard' in command)):
            return subprocess.CompletedProcess(command, 128, stdout='', stderr='fixture I/O refusal')
        return original(command, *args, **kwargs)

    monkeypatch.setattr(subprocess, 'run', fault)
    if failure == 'head':
        monkeypatch.setattr(update_cmd, '_capture_head_sha', lambda *_: None)
    if failure == 'reset':
        with pytest.raises(SystemExit) as error:
            hermes_main.cmd_update(t.args)
        assert error.value.code == 1
        assert not t.requests
        assert git(t.clone, 'rev-parse', 'HEAD') == before
        assert not (t.clone / 'untracked.txt').exists()
    else:
        hermes_main.cmd_update(t.args)
        assert len(t.requests) == 1
        assert git(t.clone, 'rev-parse', 'HEAD') == t.newer
        assert (t.clone / 'untracked.txt').exists() is (not keep)
    assert len(resets) == 1
    stashes = git(t.clone, 'stash', 'list')
    assert bool(stashes) is (keep or failure == 'reset')
    if stashes:
        assert git(t.clone, 'show', 'stash@{0}^3:untracked.txt') == 'local edit'
    output = capsys.readouterr().out
    if failure == 'ref':
        assert 'backup write failed' in output and 'backed up current HEAD' not in output
    if failure == 'reset':
        assert 'preserved in stash' in output
    if failure not in {'ref', 'head'}:
        assert f'expires after {update_cmd._ORPHAN_RESCUE_REF_MAX_AGE_DAYS} days' in output
        kind = 'orphan' if history == 'orphan' else 'diverged'
        assert f'refs/hermes-update-backups/{kind}-main-' in output
        if kind == 'diverged':
            assert 'commit(s) not on origin/main leave the branch' in output


@pytest.mark.parametrize('mode', ['count', 'age', 'unparseable'])
def test_rescue_retention_uses_real_refs(tmp_path, monkeypatch, mode):
    from datetime import datetime, timedelta, timezone

    git(tmp_path, 'init', '-q', '-b', 'main')
    git(tmp_path, '-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid',
        '-c', 'commit.gpgsign=false', 'commit', '--allow-empty', '-qm', 'base')
    now = datetime.now(timezone.utc)
    prefix = 'refs/hermes-update-backups/orphan-main-'
    if mode == 'count':
        refs = [prefix + (now - timedelta(hours=20-i)).strftime('%Y%m%d-%H%M%S') + '-abc'
                for i in range(update_cmd._ORPHAN_RESCUE_REFS_TO_KEEP + 2)]
        removed = set(refs[:2])
    elif mode == 'age':
        refs = [prefix + '20200101-000000-abc', prefix + now.strftime('%Y%m%d-%H%M%S') + '-abc']
        removed = {refs[0]}
    else:
        refs, removed = [prefix + 'not-a-timestamp-abc'], set()
    for ref in refs:
        git(tmp_path, 'update-ref', ref, 'HEAD')
    update_cmd._prune_orphan_rescue_refs(['git'], tmp_path, 'main')
    assert set(git(tmp_path, 'for-each-ref', '--format=%(refname)', 'refs/hermes-update-backups/').splitlines()) == set(refs) - removed


@pytest.mark.parametrize('fault,body,modules,message', [
    ('syntax', '<<<<<<< Updated upstream\nVALUE = 2\n', (), 'made the Hermes agent unexecutable'),
    ('import', "raise RuntimeError('restored local failure')\n", ('consumer',), 'restored local failure'),
    ('preexisting', 'VALUE = 2\n', ('first',), None),
    ('later', "raise RuntimeError('restored later failure')\n", ('first', 'consumer'), 'restored later failure'),
    ('exit', "raise SystemExit('restored exit')\n", ('first', 'consumer'), 'restored exit'),
    ('terminated', 'import os\nos._exit(7)\n', ('consumer',), 'exit code 7'),
    ('paths', 'VALUE = 2\n', (), 'restored Python source discovery'),
])
def test_restore_validates_real_stash_and_each_import(probe_root, monkeypatch, capsys, fault, body, modules, message):
    import hermes_cli.update_cmd_stash as stash

    tmp_path = probe_root

    git(tmp_path, 'init', '-q', '-b', 'main')
    (tmp_path / 'first.py').write_text("raise RuntimeError('missing local config')\n", encoding='utf-8')
    source = tmp_path / 'consumer.py'
    source.write_text('VALUE = 1\n', encoding='utf-8')
    git(tmp_path, 'add', 'first.py', 'consumer.py', 'hermes_bootstrap.py')
    git(tmp_path, '-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid',
        '-c', 'commit.gpgsign=false', 'commit', '-qm', 'base')
    source.write_text(body, encoding='utf-8')
    ref = hermes_main._stash_local_changes_if_needed(['git'], tmp_path)
    assert ref
    monkeypatch.setattr(update_cmd, '_UPDATE_CRITICAL_MODULES', modules)
    if fault == 'paths':
        monkeypatch.setattr(stash, '_restored_python_paths', lambda *_: None)
        monkeypatch.setattr(update_cmd, '_restored_python_paths', lambda *_: None)
    if message is None:
        assert hermes_main._restore_stashed_changes(['git'], tmp_path, ref, prompt_user=False)
        assert source.read_text(encoding='utf-8') == body
        assert not git(tmp_path, 'stash', 'list')
    else:
        with pytest.raises(SystemExit) as error:
            hermes_main._restore_stashed_changes(['git'], tmp_path, ref, prompt_user=False)
        assert error.value.code == 1
        assert source.read_text(encoding='utf-8') == 'VALUE = 1\n'
        assert not git(tmp_path, 'status', '--porcelain')
        assert git(tmp_path, 'stash', 'list')
        output = capsys.readouterr().out
        assert message in output and 'gateway was not restarted' in output
        assert f'git stash apply {ref}' in output


@pytest.mark.parametrize('error', [EOFError(), UnicodeDecodeError('utf-8', b'\xff', 0, 1, 'invalid')])
def test_unreadable_stash_prompt_keeps_work(tmp_path, monkeypatch, capsys, error):
    def unreadable(*_):
        raise error
    monkeypatch.setattr('builtins.input', unreadable)
    assert hermes_main._restore_stashed_changes(['git'], tmp_path, 'stash@{0}', prompt_user=True) is False
    assert 'git stash apply stash@{0}' in capsys.readouterr().out


def test_unreadable_upstream_prompt_does_not_add_remote(tmp_path, monkeypatch):
    git(tmp_path, 'init', '-q', '-b', 'main')
    def unreadable(*_):
        raise UnicodeDecodeError('utf-8', b'\xff', 0, 1, 'invalid')
    monkeypatch.setattr('builtins.input', unreadable)
    monkeypatch.setattr(update_cmd, '_should_skip_upstream_prompt', lambda *_: False)
    update_cmd._sync_with_upstream_if_needed(['git'], tmp_path)
    assert not git(tmp_path, 'remote')


def test_update_parser_accepts_keep_stash():
    """The flag parses and defaults off."""
    import argparse

    from hermes_cli.subcommands.update import build_update_parser

    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers()
    build_update_parser(subparsers, cmd_update=lambda args: None)

    args = parser.parse_args(["update", "--keep-stash"])
    assert args.keep_stash is True
    args = parser.parse_args(["update"])
    assert args.keep_stash is False



def test_bootstrap_marker_not_autostashed_by_update(tmp_path):
    """#38529: the Desktop bootstrap marker must be git-ignored so that
    ``hermes update``'s ``git stash push --include-untracked`` does not sweep it
    into an autostash on every run.

    Behavioral + hermetic: build a throwaway repo that adopts the project's real
    ``.gitignore`` (the contract under test), drop the marker, and confirm the
    same stash invocation the updater uses leaves it untouched.
    """
    import shutil
    import subprocess

    if shutil.which("git") is None:
        pytest.skip("git not available")

    repo_gitignore = Path(hermes_main.__file__).resolve().parents[1] / ".gitignore"

    def git(*args):
        return subprocess.run(
            ["git", *args], cwd=tmp_path, capture_output=True, text=True, check=True
        )

    git("init", "-q")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    (tmp_path / ".gitignore").write_text(repo_gitignore.read_text())
    (tmp_path / "tracked.txt").write_text("x\n")
    git("add", "-A")
    git("commit", "-qm", "init")

    marker = tmp_path / ".hermes-bootstrap-complete"
    marker.write_text("")

    # Exact flags used by hermes update (hermes_cli/main.py).
    git("stash", "push", "--include-untracked", "-m", "hermes-update-autostash")

    assert marker.exists(), (
        ".hermes-bootstrap-complete was swept into the update autostash — it must "
        "be listed in .gitignore so `git stash -u` skips it (#38529)."
    )
    # It must not even register as a dirty/untracked change.
    status = subprocess.run(
        ["git", "status", "--porcelain"], cwd=tmp_path, capture_output=True, text=True
    ).stdout
    assert ".hermes-bootstrap-complete" not in status



def test_update_autostash_survives_undeletable_untracked_dir(tmp_path):
    """Behavioral E2E of the whole permission-denied class with real git:
    root-owned-style undeletable untracked dir → stash succeeds, update-style
    reset works, restore round-trips, nothing lost. (#70127 follow-up)"""
    import contextlib
    import os
    import shutil
    import subprocess

    if shutil.which("git") is None:
        pytest.skip("git not available")
    if os.name == "nt":
        pytest.skip("POSIX permission semantics")
    if os.geteuid() == 0:
        pytest.skip("root ignores directory write bits")

    def git(*args, check=True):
        return subprocess.run(
            ["git", *args], cwd=tmp_path, capture_output=True, text=True, check=check
        )

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    (tmp_path / "tracked.txt").write_text("v1\n")
    git("add", "-A")
    git("commit", "-qm", "init")

    (tmp_path / "tracked.txt").write_text("v2 local change\n")
    pkg = tmp_path / "packaging" / "homebrew"
    pkg.mkdir(parents=True)
    (pkg / "hermes-agent.rb").write_text("formula\n")
    os.chmod(pkg, 0o555)  # undeletable contents, like a root-owned dir
    try:
        stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)
        assert stash_ref

        # The tracked change is stashed; simulate the updater's checkout window.
        assert (tmp_path / "tracked.txt").read_text() == "v1\n"

        restored = hermes_main._restore_stashed_changes(
            ["git"], tmp_path, stash_ref, prompt_user=False
        )
        assert restored is True
        assert (tmp_path / "tracked.txt").read_text() == "v2 local change\n"
        assert (pkg / "hermes-agent.rb").read_text() == "formula\n"
    finally:
        os.chmod(pkg, 0o755)


def test_stash_selector_is_a_bare_index_never_a_brace_selector(tmp_path):
    """The updater drops its autostash through a selector read back from ``git stash list``; on
    native Windows MSYS strips the braces from ``stash@{N}`` in git.exe's argv, so the selector
    must be the bare index git accepts everywhere (#87542)."""
    import subprocess

    import hermes_cli.update_cmd_stash as stash_mod

    def git(*args):
        return subprocess.run(["git", *args], cwd=tmp_path, capture_output=True, text=True, check=True)

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    (tmp_path / "f.txt").write_text("v1\n")
    git("add", "-A")
    git("commit", "-qm", "init")
    (tmp_path / "f.txt").write_text("older\n")
    git("stash", "push", "-q", "-m", "older")
    target_sha = git("rev-parse", "refs/stash").stdout.strip()
    (tmp_path / "f.txt").write_text("newer\n")
    git("stash", "push", "-q", "-m", "newer")

    selector = stash_mod._resolve_stash_selector(["git"], tmp_path, target_sha)

    assert selector == "1"
    git("stash", "drop", selector)
    assert target_sha not in git("stash", "list", "--format=%H").stdout


def test_restore_stays_parked_when_untracked_baseline_is_unknown(
    monkeypatch, tmp_path, capsys
):
    """Unknown cleanup scope must not turn into a destructive empty baseline."""
    from hermes_cli import update_cmd
    import hermes_cli.update_cmd_stash as update_cmd_stash

    monkeypatch.setattr(update_cmd, "_git_untracked_paths", lambda *_args: None)
    monkeypatch.setattr(update_cmd_stash, "_git_untracked_paths", lambda *_args: None)

    restored = hermes_main._restore_stashed_changes(
        ["git"], tmp_path, "stash@{0}", prompt_user=False
    )

    assert restored is False
    output = capsys.readouterr().out
    assert "cleanup baseline is unknown" in output
    assert "git stash apply stash@{0}" in output



def test_reject_does_not_claim_cleanup_when_git_state_is_unknown(
    monkeypatch, tmp_path, capsys
):
    """Cleanup failures must not be reported as a restored clean tree."""
    from hermes_cli import update_cmd
    import hermes_cli.update_cmd_stash as update_cmd_stash

    monkeypatch.setattr(update_cmd, "_git_untracked_paths", lambda *_args: None)
    monkeypatch.setattr(update_cmd_stash, "_git_untracked_paths", lambda *_args: None)

    with pytest.raises(SystemExit):
        update_cmd._reject_unsafe_stash_restore(
            ["git"], tmp_path, "stash@{0}", set(), "consumer.py", "invalid"
        )

    output = capsys.readouterr().out
    assert "could not be fully restored automatically" in output
    assert "The clean updated tree has been restored" not in output



def test_gateway_restore_prompt_defaults_to_keep_stash(tmp_path, capsys):
    prompts = []

    restored = hermes_main._restore_stashed_changes(
        ["git"],
        tmp_path,
        "stash@{0}",
        prompt_user=True,
        input_fn=lambda prompt, default: prompts.append((prompt, default)) or "",
    )

    assert restored is False
    assert prompts == [("Restore local changes now? [y/N]", "n")]
    assert "still preserved in git stash" in capsys.readouterr().out



def test_prune_orphan_rescue_refs_with_real_git_unpins_objects(tmp_path):
    """End-to-end with real git: an orphan rescue ref pins a snapshot's
    objects against gc; pruning the ref (age-expired) makes them collectable.
    This is the sabotage/size test for the #87745 bounded-growth mitigation."""
    import shutil
    import subprocess

    if shutil.which("git") is None:
        pytest.skip("git not available")

    def git(*args, check=True):
        return subprocess.run(
            ["git", *args], cwd=tmp_path, capture_output=True, text=True, check=check
        )

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    git("config", "gc.auto", "0")
    (tmp_path / "f.txt").write_text("base\n")
    git("add", "-A")
    git("commit", "-qm", "init")

    # Snapshot commit carrying a "large" payload (scaled down for CI).
    import os

    (tmp_path / "big.bin").write_bytes(os.urandom(512 * 1024))
    git("add", "-A")
    git("commit", "-qm", "snapshot")
    snap_sha = git("rev-parse", "HEAD").stdout.strip()

    # Rewind, park the snapshot behind an AGE-EXPIRED rescue ref.
    git("reset", "-q", "--hard", "HEAD~1")
    old_ref = "refs/hermes-update-backups/orphan-main-20200101-000000-" + snap_sha[:12]
    git("update-ref", old_ref, snap_sha)

    # With the ref present, gc cannot drop the snapshot objects.
    git("reflog", "expire", "--expire=now", "--all")
    git("gc", "-q", "--prune=now")
    assert git("cat-file", "-e", snap_sha, check=False).returncode == 0

    # Prune (the ref's 2020 timestamp is way past the age window) → ref gone.
    update_cmd._prune_orphan_rescue_refs(["git"], tmp_path, "main")
    remaining = git("for-each-ref", "refs/hermes-update-backups/").stdout
    assert old_ref not in remaining

    # And gc can now reclaim the snapshot's objects.
    git("gc", "-q", "--prune=now")
    assert git("cat-file", "-e", snap_sha, check=False).returncode != 0


def test_autostash_survives_intent_to_add_entries(tmp_path):
    """An index entry from `git add -N` must not block the update autostash.

    Reported: `hermes update` aborted with "Entry 'tests/...' not uptodate. Cannot merge." because
    `git add -N` records a path with the empty blob and zeroed stat data, which `git stash push`
    refuses outright. Editors that show new files in diffs leave exactly that state behind, and the
    update must not require the user to repair their index by hand.
    """
    import subprocess

    def git(*args, check=True):
        return subprocess.run(
            ["git", *args], cwd=tmp_path, capture_output=True, text=True, check=check
        )

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    (tmp_path / "tracked.txt").write_text("v1\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "init")

    # The reported shape: a new local file recorded with `git add -N`, alongside a normal edit.
    (tmp_path / "tracked.txt").write_text("v2 local\n", encoding="utf-8")
    local_test = tmp_path / "tests" / "test_live_custom_provider_poll.py"
    local_test.parent.mkdir()
    body = "def test_poll():\n    assert True\n"
    local_test.write_text(body, encoding="utf-8")
    git("add", "-N", "tests/test_live_custom_provider_poll.py")
    # Precondition: git reports it as " A" (present in the worktree, absent from the index) - the
    # intent-to-add shape that `git stash push` refuses.
    assert " A tests/test_live_custom_provider_poll.py" in git("status", "--porcelain").stdout.splitlines()

    stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)

    assert stash_ref, "the update must be able to stash an intent-to-add entry"
    # The stash must have taken everything, so the pull cannot be blocked by a dirty tree.
    assert git("status", "--porcelain").stdout == ""
    assert hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False)
    assert local_test.read_text(encoding="utf-8") == body
    assert (tmp_path / "tracked.txt").read_text(encoding="utf-8") == "v2 local\n"


class _ReceiptProbe:
    """Minimal stand-in for the active update receipt: records steps.

    ``record_step`` clones the active receipt before mutating it (copy-on-write per context),
    so the steps list is shared by reference for the probe to observe.
    """

    def __init__(self):
        self.steps = []
        self.data = {}  # copied per record; ``steps`` is shared by reference

    def step(self, name, ok, detail=""):
        self.steps.append({"name": name, "ok": ok, "detail": detail})


@contextlib.contextmanager
def _active_receipt(probe):
    from hermes_cli import update_receipt

    token = update_receipt._current.set(probe)
    try:
        yield
    finally:
        update_receipt._current.reset(token)


def test_conflicted_restore_records_parked_step_in_receipt(monkeypatch, tmp_path):
    import subprocess
    from hermes_cli import update_receipt

    def git(*args, check=True):
        return subprocess.run(["git", *args], cwd=tmp_path, capture_output=True, text=True, check=check)

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    source = tmp_path / "tools" / "terminal_tool.py"
    source.parent.mkdir()
    source.write_text("VALUE = 1\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "init")

    # Local edit whose restore will conflict with the pulled change.
    source.write_text("VALUE = 2\n", encoding="utf-8")
    stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)
    assert stash_ref
    # Simulate the pull moving the same lines: the stash apply now conflicts.
    git("checkout", "HEAD")
    source.write_text("VALUE = 3\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "pulled change")

    probe = _ReceiptProbe()
    with _active_receipt(probe):
        restored = hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False)

    assert restored is False
    disposition = [s for s in probe.steps if s["name"] == "local_changes_stash"]
    assert len(disposition) == 1
    assert disposition[0]["ok"] is False
    assert "parked" in disposition[0]["detail"]
    assert stash_ref in disposition[0]["detail"]
    assert git("stash", "list").stdout.strip(), "stash must survive for manual recovery"


def test_clean_restore_records_restored_step_in_receipt(monkeypatch, tmp_path):
    import subprocess
    from hermes_cli import update_receipt

    def git(*args, check=True):
        return subprocess.run(["git", *args], cwd=tmp_path, capture_output=True, text=True, check=check)

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    source = tmp_path / "tools" / "terminal_tool.py"
    source.parent.mkdir()
    source.write_text("VALUE = 1\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "init")

    source.write_text("VALUE = 2\n", encoding="utf-8")
    stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)
    assert stash_ref

    probe = _ReceiptProbe()
    with _active_receipt(probe):
        restored = hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False)

    assert restored is True
    disposition = [s for s in probe.steps if s["name"] == "local_changes_stash"]
    assert len(disposition) == 1
    assert disposition[0]["ok"] is True
    assert "restored" in disposition[0]["detail"]
    assert stash_ref in disposition[0]["detail"]


def test_keep_stash_park_records_parked_step_in_receipt(capsys):
    probe = _ReceiptProbe()
    import hermes_cli.update_cmd_stash as stash_mod
    from hermes_cli import update_receipt

    with _active_receipt(probe):
        stash_mod._park_stashed_changes("deadbeefcafe")

    out = capsys.readouterr().out
    assert "--keep-stash" in out
    disposition = [s for s in probe.steps if s["name"] == "local_changes_stash"]
    assert len(disposition) == 1
    assert disposition[0]["ok"] is False
    assert "parked" in disposition[0]["detail"]


def _repo_with_stash(tmp_path, local_source):
    """A repo whose autostash holds a tracked ``mod.py`` edit and an untracked ``notes.md``."""
    import subprocess

    def git(*args, check=True):
        return subprocess.run(["git", *args], cwd=tmp_path, capture_output=True, text=True, check=check)

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    (tmp_path / "mod.py").write_text("X = 1\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "init")
    (tmp_path / "mod.py").write_text(local_source, encoding="utf-8")
    (tmp_path / "notes.md").write_text("my private notes\n", encoding="utf-8")
    stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)
    assert stash_ref
    return git, stash_ref


@pytest.mark.parametrize("local_source", ["X = 2\n", "X = (\n"], ids=["healthy", "breaks-hermes"])
def test_untracked_file_replaced_by_the_update_keeps_the_stash(tmp_path, local_source):
    """#124641: the update adds a file where the user had an untracked file of the same name.
    ``stash apply`` refuses it ("already exists, no checkout"); the stash is the only copy of the
    user's version, so it is never dropped. The restored tree still gets the health check: a restore
    that breaks Hermes resets the tree and exits 1."""
    git, stash_ref = _repo_with_stash(tmp_path, local_source)
    # The pull adds its own notes.md.
    (tmp_path / "notes.md").write_text("upstream notes\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "upstream adds notes.md")

    probe = _ReceiptProbe()
    with _active_receipt(probe):
        if local_source == "X = 2\n":
            restored = hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False)
            assert restored is False
        else:
            with pytest.raises(SystemExit) as exc:
                hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False)
            assert exc.value.code == 1

    healthy = local_source == "X = 2\n"
    assert (tmp_path / "mod.py").read_text(encoding="utf-8") == (local_source if healthy else "X = 1\n")
    assert (tmp_path / "notes.md").read_text(encoding="utf-8") == "upstream notes\n"
    assert git("show", f"{stash_ref}^3:notes.md").stdout == "my private notes\n"
    assert git("stash", "list").stdout.strip(), "the stash is the only copy of the user's notes.md"
    if healthy:
        disposition = [s for s in probe.steps if s["name"] == "local_changes_stash"]
        assert len(disposition) == 1 and disposition[0]["ok"] is False
        assert "parked" in disposition[0]["detail"] and "notes.md" in disposition[0]["detail"]


def test_untracked_file_the_update_does_not_track_is_never_reported_replaced(tmp_path, capsys):
    """#70127: an untracked file still in the tree after the stash (it could not be deleted) and
    changed since is not the update's file; HEAD does not track it, so the restore completes."""
    git, stash_ref = _repo_with_stash(tmp_path, "X = 2\n")
    # The occupant survived the stash and was edited during the update window.
    (tmp_path / "notes.md").write_text("edited while locked\n", encoding="utf-8")

    assert hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False) is True

    assert (tmp_path / "mod.py").read_text(encoding="utf-8") == "X = 2\n"
    assert "The update added" not in capsys.readouterr().out


def test_unrequested_park_is_never_reported_as_update_complete(monkeypatch, tmp_path):
    """A restore that conflicts leaves the patch parked: the completion line must say so (#122557);
    a park the user asked for (--keep-stash) stays a normal completion."""
    import hermes_cli.update_cmd_stash as stash_mod

    monkeypatch.setattr(stash_mod, "_pending_autostash", None)

    def git(*args):
        return subprocess.run(["git", *args], cwd=tmp_path, capture_output=True, text=True, check=True)

    git("init", "-q", "-b", "main")
    git("config", "user.email", "t@example.com")
    git("config", "user.name", "t")
    source = tmp_path / "notes.txt"
    source.write_text("VALUE = 1\n", encoding="utf-8")
    git("add", "-A")
    git("commit", "-qm", "init")
    source.write_text("VALUE = 2\n", encoding="utf-8")
    stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)
    source.write_text("VALUE = 3\n", encoding="utf-8")
    git("commit", "-qam", "pulled change")

    assert hermes_main._restore_stashed_changes(["git"], tmp_path, stash_ref, prompt_user=False) is False
    notice = stash_mod._unrestored_autostash_notice()
    assert notice and not notice.startswith("\u2713") and stash_ref in notice

    stash_mod._park_stashed_changes(stash_ref)
    assert stash_mod._unrestored_autostash_notice() is None