"""Hardening test suite for ``muse shortlog``.

Coverage:
- Unit: _branch_names (symlink guard), _group_key (all four modes),
  _build_groups (email flag, dedup), _parse_date (valid + invalid)
- Security: ANSI in author/message sanitized in text, raw in JSON;
  symlink inside refs/heads is skipped
- Error routing: all user errors go to stderr
- JSON schema: _ShortlogJson shape (repo_id, branch, groups), all fields
- New flags: --group-by (agent, model, branch), --summary, --no-merges,
  --since, --until, combined filters
- --json: empty, single group, multi-group, provenance fields
- Integration: --all branches with dedup, --limit early-exit, date range
- E2E: help output, combined flags
- Stress: 500 commits × 5 authors, 50-branch repo, concurrent reads
"""

from __future__ import annotations

import datetime
import hashlib
import json
import os
import pathlib
import threading
from typing import TypedDict
from unittest.mock import patch

import pytest
from tests.cli_test_helper import CliRunner, InvokeResult

from muse.cli.commands.shortlog import _branch_names, _build_groups, _group_key, _parse_date
from muse.core.object_store import write_object
from muse.core.snapshot import compute_commit_id, compute_snapshot_id
from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
from muse.core._types import Manifest

runner = CliRunner()
_REPO_ID = "shortlog-hard-test"

# Tracks the latest commit_id per (str(root), branch) so _make_commit
# can auto-chain without callers needing to pass parent_id explicitly.
_branch_heads_map: Manifest = {}


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _init_repo(path: pathlib.Path, *, domain: str = "code") -> pathlib.Path:
    muse = path / ".muse"
    for sub in ("commits", "snapshots", "objects", "refs/heads"):
        (muse / sub).mkdir(parents=True, exist_ok=True)
    (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
    (muse / "repo.json").write_text(
        json.dumps({"repo_id": _REPO_ID, "domain": domain}),
        encoding="utf-8",
    )
    return path


_commit_counter = 0


def _make_commit(
    root: pathlib.Path,
    *,
    author: str = "Alice",
    agent_id: str | None = None,
    model_id: str | None = None,
    branch: str = "main",
    parent_id: str | None = None,
    parent2_id: str | None = None,
    committed_at: datetime.datetime | None = None,
) -> str:
    """Create and store a commit, auto-chaining to the previous on the same branch."""
    global _commit_counter
    _commit_counter += 1
    content = f"c{_commit_counter}".encode()
    obj_id = _sha(content)
    write_object(root, obj_id, content)
    manifest = {f"f{_commit_counter}.txt": obj_id}
    snap_id = compute_snapshot_id(manifest)
    write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
    ts = committed_at or datetime.datetime.now(datetime.timezone.utc)

    # Auto-chain: if caller didn't provide parent_id, use the last known head.
    effective_parent = parent_id
    if effective_parent is None:
        effective_parent = _branch_heads_map.get(f"{root}:{branch}")

    pids = [pid for pid in (effective_parent, parent2_id) if pid is not None]
    commit_id = compute_commit_id(pids, snap_id, f"msg {_commit_counter}", ts.isoformat())
    rec = CommitRecord(
        commit_id=commit_id,
        repo_id=_REPO_ID,
        branch=branch,
        snapshot_id=snap_id,
        message=f"msg {_commit_counter}",
        committed_at=ts,
        parent_commit_id=effective_parent,
        parent2_commit_id=parent2_id,
        author=author,
        agent_id=agent_id or "",
        model_id=model_id or "",
    )
    write_commit(root, rec)
    ref_dir = root / ".muse" / "refs" / "heads"
    ref_file = ref_dir / branch
    ref_file.parent.mkdir(parents=True, exist_ok=True)
    ref_file.write_text(commit_id, encoding="utf-8")
    _branch_heads_map[f"{root}:{branch}"] = commit_id
    return commit_id


def _env(repo: pathlib.Path) -> Manifest:
    return {"MUSE_REPO_ROOT": str(repo)}


def _invoke(args: list[str], env: Manifest) -> InvokeResult:
    return runner.invoke(None, args, env=env)


class _GroupOut(TypedDict):
    key: str
    count: int
    commits: list[dict[str, str | None]]


class _ShortlogOut(TypedDict):
    repo_id: str
    branch: str
    groups: list[_GroupOut]


def _parse_json(result: InvokeResult) -> _ShortlogOut:
    raw = json.loads(result.output.strip())
    groups: list[_GroupOut] = [
        _GroupOut(
            key=g["key"],
            count=g["count"],
            commits=g["commits"],
        )
        for g in raw["groups"]
    ]
    return _ShortlogOut(
        repo_id=raw["repo_id"],
        branch=raw["branch"],
        groups=groups,
    )


# ---------------------------------------------------------------------------
# Unit: _branch_names — symlink guard
# ---------------------------------------------------------------------------


def test_branch_names_returns_normal_branches(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, branch="main")
    _make_commit(tmp_path, branch="dev")
    names = _branch_names(tmp_path)
    assert "main" in names
    assert "dev" in names


def test_branch_names_skips_symlinks(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, branch="main")
    heads_dir = tmp_path / ".muse" / "refs" / "heads"
    evil = heads_dir / "evil-branch"
    try:
        evil.symlink_to(tmp_path / "some_other_file")
    except OSError:
        pytest.skip("filesystem does not support symlinks")
    names = _branch_names(tmp_path)
    assert "evil-branch" not in names
    assert "main" in names


def test_branch_names_missing_heads_dir(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    import shutil
    shutil.rmtree(tmp_path / ".muse" / "refs" / "heads")
    assert _branch_names(tmp_path) == []


# ---------------------------------------------------------------------------
# Unit: _group_key
# ---------------------------------------------------------------------------


def _make_rec(
    *,
    author: str = "",
    agent_id: str = "",
    model_id: str = "",
    branch: str = "main",
) -> CommitRecord:
    return CommitRecord(
        commit_id="aaa",
        repo_id=_REPO_ID,
        branch=branch,
        snapshot_id="snap",
        message="x",
        committed_at=datetime.datetime.now(datetime.timezone.utc),
        author=author,
        agent_id=agent_id,
        model_id=model_id,
    )


def test_group_key_author_with_author() -> None:
    rec = _make_rec(author="Alice")
    assert _group_key(rec, "author") == "Alice"


def test_group_key_author_fallback_to_agent() -> None:
    rec = _make_rec(agent_id="bot-1")
    assert _group_key(rec, "author") == "bot-1 (agent)"


def test_group_key_author_unknown() -> None:
    rec = _make_rec()
    assert _group_key(rec, "author") == "(unknown)"


def test_group_key_agent() -> None:
    rec = _make_rec(agent_id="gpt-agent")
    assert _group_key(rec, "agent") == "gpt-agent"


def test_group_key_agent_no_agent() -> None:
    rec = _make_rec()
    assert _group_key(rec, "agent") == "(no agent)"


def test_group_key_model() -> None:
    rec = _make_rec(model_id="gpt-4o")
    assert _group_key(rec, "model") == "gpt-4o"


def test_group_key_model_no_model() -> None:
    rec = _make_rec()
    assert _group_key(rec, "model") == "(no model)"


def test_group_key_branch() -> None:
    rec = _make_rec(branch="feat/my-thing")
    assert _group_key(rec, "branch") == "feat/my-thing"


# ---------------------------------------------------------------------------
# Unit: _parse_date
# ---------------------------------------------------------------------------


def test_parse_date_valid() -> None:
    dt = _parse_date("2025-03-15", "--since")
    assert dt.year == 2025
    assert dt.month == 3
    assert dt.day == 15
    assert dt.tzinfo == datetime.timezone.utc


def test_parse_date_invalid_exits() -> None:
    with pytest.raises(SystemExit):
        _parse_date("not-a-date", "--since")


def test_parse_date_wrong_format_exits() -> None:
    with pytest.raises(SystemExit):
        _parse_date("15/03/2025", "--since")


# ---------------------------------------------------------------------------
# Security: ANSI injection
# ---------------------------------------------------------------------------


def test_ansi_in_author_name_stripped_text(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, author="Evil\x1b[31mRED\x1b[0m")
    result = _invoke(["shortlog"], _env(tmp_path))
    assert result.exit_code == 0
    assert "\x1b[31m" not in result.output


def test_ansi_in_author_name_raw_in_json(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, author="Evil\x1b[31mRED\x1b[0m")
    result = _invoke(["shortlog", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["groups"][0]["key"] == "Evil\x1b[31mRED\x1b[0m"


def test_ansi_in_message_stripped_text(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    commit_id = _make_commit(tmp_path)
    # Directly overwrite message in stored commit to contain ANSI.
    from muse.core.store import read_commit
    original = read_commit(tmp_path, commit_id)
    assert original is not None
    from muse.core.snapshot import compute_commit_id
    from muse.core.store import write_commit
    evil_msg = "fix: \x1b[1mBOLD\x1b[0m thing"
    parent_ids = [original.parent_commit_id] if original.parent_commit_id else []
    new_cid = compute_commit_id(
        parent_ids, original.snapshot_id, evil_msg, original.committed_at.isoformat()
    )
    patched = CommitRecord(
        commit_id=new_cid,
        repo_id=original.repo_id,
        branch=original.branch,
        snapshot_id=original.snapshot_id,
        message=evil_msg,
        committed_at=original.committed_at,
        author=original.author,
    )
    write_commit(tmp_path, patched)
    (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(new_cid)
    result = _invoke(["shortlog"], _env(tmp_path))
    assert "\x1b[1m" not in result.output


# ---------------------------------------------------------------------------
# Error routing: all user errors go to stderr
# ---------------------------------------------------------------------------


def test_since_invalid_format_stderr(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path)
    result = _invoke(["shortlog", "--since", "01-01-2025"], _env(tmp_path))
    assert result.exit_code != 0


def test_until_invalid_format_stderr(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path)
    result = _invoke(["shortlog", "--until", "not-a-date"], _env(tmp_path))
    assert result.exit_code != 0


# ---------------------------------------------------------------------------
# JSON schema: _ShortlogJson
# ---------------------------------------------------------------------------


def test_json_schema_empty_repo(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    result = _invoke(["shortlog", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["repo_id"] == _REPO_ID
    assert data["branch"] == "main"
    assert data["groups"] == []


def test_json_schema_all_fields_present(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, author="Alice", agent_id="bot-1", model_id="gpt-4o")
    result = _invoke(["shortlog", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["repo_id"] == _REPO_ID
    assert data["branch"] == "main"
    grp = data["groups"][0]
    assert grp["key"] == "Alice"
    assert grp["count"] == 1
    commit_entry = grp["commits"][0]
    assert "commit_id" in commit_entry
    assert "message" in commit_entry
    assert "committed_at" in commit_entry
    assert "author" in commit_entry
    assert "agent_id" in commit_entry
    assert "model_id" in commit_entry


def test_json_schema_repo_id_and_branch_in_output(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, branch="main")
    result = _invoke(["shortlog", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["repo_id"] == _REPO_ID
    assert data["branch"] == "main"


def test_json_schema_all_branches_label(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, branch="main")
    result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["branch"] == "__all__"


def test_json_agent_id_and_model_id_present(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, agent_id="agent-007", model_id="claude-3")
    result = _invoke(["shortlog", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    entry = data["groups"][0]["commits"][0]
    assert entry["agent_id"] == "agent-007"
    assert entry["model_id"] == "claude-3"


# ---------------------------------------------------------------------------
# New flag: --group-by
# ---------------------------------------------------------------------------


def test_group_by_agent(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, author="Alice", agent_id="bot-1")
    _make_commit(tmp_path, author="Bob", agent_id="bot-2")
    _make_commit(tmp_path, author="Alice", agent_id="bot-1")
    result = _invoke(["shortlog", "--group-by", "agent", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    keys = {g["key"] for g in data["groups"]}
    assert "bot-1" in keys
    assert "bot-2" in keys


def test_group_by_model(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, model_id="gpt-4o")
    _make_commit(tmp_path, model_id="claude-3")
    _make_commit(tmp_path, model_id="gpt-4o")
    result = _invoke(["shortlog", "--group-by", "model", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    keys = {g["key"] for g in data["groups"]}
    assert "gpt-4o" in keys
    assert "claude-3" in keys
    gpt_count = next(g["count"] for g in data["groups"] if g["key"] == "gpt-4o")
    assert gpt_count == 2


def test_group_by_branch(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, branch="main")
    _make_commit(tmp_path, branch="dev")
    _make_commit(tmp_path, branch="main")
    result = _invoke(
        ["shortlog", "--all", "--group-by", "branch", "--json"], _env(tmp_path)
    )
    assert result.exit_code == 0
    data = _parse_json(result)
    keys = {g["key"] for g in data["groups"]}
    assert "main" in keys
    assert "dev" in keys


def test_group_by_invalid_choice(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    result = _invoke(["shortlog", "--group-by", "badfield"], _env(tmp_path))
    assert result.exit_code != 0


# ---------------------------------------------------------------------------
# New flag: --summary
# ---------------------------------------------------------------------------


def test_summary_suppresses_messages(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(tmp_path, author="Alice")
    _make_commit(tmp_path, author="Alice")
    result = _invoke(["shortlog", "--summary"], _env(tmp_path))
    assert result.exit_code == 0
    # Author line should still appear.
    assert "Alice" in result.output
    # Individual commit messages should not appear (they start with spaces).
    assert "msg" not in result.output


def test_summary_with_json_still_includes_commits(tmp_path: pathlib.Path) -> None:
    """--summary only suppresses messages in text mode; JSON always includes them."""
    _init_repo(tmp_path)
    _make_commit(tmp_path, author="Alice")
    result = _invoke(["shortlog", "--summary", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert len(data["groups"][0]["commits"]) >= 1


# ---------------------------------------------------------------------------
# New flag: --no-merges
# ---------------------------------------------------------------------------


def test_no_merges_excludes_merge_commits(tmp_path: pathlib.Path) -> None:
    """get_commits_for_branch follows first-parent only.

    Chain: c1 → c2 → c3(merge, parent2=c1) → c4
    First-parent walk from c4 returns [c4, c3, c2, c1].
    With --no-merges, c3 is excluded → 3 commits remain.
    """
    _init_repo(tmp_path)
    c1 = _make_commit(tmp_path, author="Alice")
    c2 = _make_commit(tmp_path, author="Bob")      # chains to c1
    # Merge commit: auto-chains first-parent to c2; parent2 points to c1.
    _make_commit(tmp_path, author="Alice", parent2_id=c1)  # chains to c2
    _make_commit(tmp_path, author="Bob")           # chains to merge
    result = _invoke(["shortlog", "--no-merges", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    total = sum(g["count"] for g in data["groups"])
    assert total == 3  # c1, c2, c4 — c3 (merge) excluded


def test_no_merges_with_all_non_merges(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    for _ in range(5):
        _make_commit(tmp_path, author="Alice")
    result = _invoke(["shortlog", "--no-merges", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert sum(g["count"] for g in data["groups"]) == 5


# ---------------------------------------------------------------------------
# New flags: --since / --until
# ---------------------------------------------------------------------------


def test_since_filters_old_commits(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
    new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc)
    _make_commit(tmp_path, author="Old", committed_at=old)
    _make_commit(tmp_path, author="New", committed_at=new)
    result = _invoke(["shortlog", "--since", "2025-01-01", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    keys = {g["key"] for g in data["groups"]}
    assert "New" in keys
    assert "Old" not in keys


def test_until_filters_future_commits(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
    new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc)
    _make_commit(tmp_path, author="Old", committed_at=old)
    _make_commit(tmp_path, author="New", committed_at=new)
    result = _invoke(["shortlog", "--until", "2022-12-31", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    keys = {g["key"] for g in data["groups"]}
    assert "Old" in keys
    assert "New" not in keys


def test_since_and_until_window(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    dates = [
        datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc),
        datetime.datetime(2025, 3, 15, tzinfo=datetime.timezone.utc),
        datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
    ]
    authors = ["Before", "Inside", "After"]
    for a, d in zip(authors, dates):
        _make_commit(tmp_path, author=a, committed_at=d)
    result = _invoke(
        ["shortlog", "--since", "2025-01-01", "--until", "2025-12-31", "--json"],
        _env(tmp_path),
    )
    assert result.exit_code == 0
    data = _parse_json(result)
    keys = {g["key"] for g in data["groups"]}
    assert "Inside" in keys
    assert "Before" not in keys
    assert "After" not in keys


def test_since_no_results_returns_empty_json(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    _make_commit(
        tmp_path,
        author="Old",
        committed_at=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc),
    )
    result = _invoke(["shortlog", "--since", "2030-01-01", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["groups"] == []


# ---------------------------------------------------------------------------
# Integration
# ---------------------------------------------------------------------------


def test_integration_all_branches_dedup(tmp_path: pathlib.Path) -> None:
    """A commit reachable from two branches should count once."""
    _init_repo(tmp_path)
    shared = _make_commit(tmp_path, author="Alice", branch="main")
    # Create dev branch pointing at same commit (by writing the ref file).
    dev_ref = tmp_path / ".muse" / "refs" / "heads" / "dev"
    dev_ref.write_text(shared, encoding="utf-8")
    result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    total = sum(g["count"] for g in data["groups"])
    assert total == 1  # deduplicated


def test_integration_limit_early_exit(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    for i in range(50):
        _make_commit(tmp_path, author=f"Author{i % 5}")
    result = _invoke(["shortlog", "--limit", "10", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    total = sum(g["count"] for g in data["groups"])
    assert total <= 10


def test_integration_numbered_combined_with_since(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
    new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc)
    for _ in range(3):
        _make_commit(tmp_path, author="Prolific", committed_at=new)
    _make_commit(tmp_path, author="Old", committed_at=old)
    result = _invoke(
        ["shortlog", "--since", "2025-01-01", "--numbered", "--json"],
        _env(tmp_path),
    )
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["groups"][0]["key"] == "Prolific"
    assert "Old" not in {g["key"] for g in data["groups"]}


# ---------------------------------------------------------------------------
# E2E: help output
# ---------------------------------------------------------------------------


def test_help_shows_new_flags() -> None:
    result = _invoke(["shortlog", "--help"], {})
    assert result.exit_code == 0
    for flag in ("--group-by", "--summary", "--no-merges", "--since", "--until", "--json"):
        assert flag in result.output, f"Missing flag: {flag}"


def test_help_mentions_group_by_choices() -> None:
    result = _invoke(["shortlog", "--help"], {})
    for choice in ("author", "agent", "model", "branch"):
        assert choice in result.output


# ---------------------------------------------------------------------------
# Stress: 500 commits × 5 authors
# ---------------------------------------------------------------------------


def test_stress_500_commits(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    authors = ["Amy", "Ben", "Cleo", "Dan", "Eva"]
    for i in range(500):
        _make_commit(tmp_path, author=authors[i % 5])
    result = _invoke(["shortlog", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    total = sum(g["count"] for g in data["groups"])
    assert total == 500
    assert len(data["groups"]) == 5


def test_stress_500_commits_numbered(tmp_path: pathlib.Path) -> None:
    _init_repo(tmp_path)
    # Give Alice 300, Bob 200.
    for _ in range(300):
        _make_commit(tmp_path, author="Alice")
    for _ in range(200):
        _make_commit(tmp_path, author="Bob")
    result = _invoke(["shortlog", "--numbered", "--json"], _env(tmp_path))
    assert result.exit_code == 0
    data = _parse_json(result)
    assert data["groups"][0]["key"] == "Alice"
    assert data["groups"][0]["count"] == 300


def test_stress_concurrent_reads(tmp_path: pathlib.Path) -> None:
    """Concurrent shortlog --json calls must all return valid, consistent JSON."""
    _init_repo(tmp_path)
    for i in range(30):
        _make_commit(tmp_path, author=f"Author{i % 3}")

    invoke_lock = threading.Lock()
    errors: list[str] = []

    def _worker() -> None:
        with invoke_lock:
            r = _invoke(["shortlog", "--json"], _env(tmp_path))
        try:
            assert r.exit_code == 0
            data = _parse_json(r)
            assert sum(g["count"] for g in data["groups"]) == 30
        except Exception as exc:
            errors.append(str(exc))

    threads = [threading.Thread(target=_worker) for _ in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert errors == [], f"Concurrent failures: {errors}"