"""Tests for tools/skill_ledger.py — per-mutation audit ledger + rollback.

Covers tracker #79686 P3: ledger entries on patch/edit/delete/archive, blob
dedupe, single-entry rollback (incl. fail-closed safety capture), actor
tagging, and the skills.ledger config gate.

The first four tests are adapted from PR #50261 by @yu-xin-c (autonomous
skill history), reshaped for the all-actor JSONL ledger design.
"""

import hashlib
import json
import os
import time
from pathlib import Path

import pytest


VALID_SKILL_CONTENT = """---
name: my-skill
description: test skill
---

# My Skill

Original body.
"""


@pytest.fixture
def ledger_env(tmp_path, monkeypatch):
    """Isolated HERMES_HOME + skills dir for skill_manage and the ledger."""
    from agent import skill_utils
    from tools import skill_ledger, skill_manager_tool, skill_usage

    home = tmp_path / "home"
    skills_dir = home / "skills"
    skills_dir.mkdir(parents=True)

    monkeypatch.setattr(skill_ledger, "get_hermes_home", lambda: home)
    monkeypatch.setattr(skill_usage, "get_hermes_home", lambda: home)
    monkeypatch.setattr(skill_manager_tool, "SKILLS_DIR", skills_dir)
    monkeypatch.setattr(skill_utils, "get_all_skills_dirs", lambda: [skills_dir])
    return {"home": home, "skills": skills_dir}


def _create(name="my-skill", content=VALID_SKILL_CONTENT):
    from tools.skill_manager_tool import skill_manage

    return json.loads(skill_manage(action="create", name=name, content=content))


# ---------------------------------------------------------------------------
# Adapted from PR #50261 (@yu-xin-c)
# ---------------------------------------------------------------------------


def test_background_review_patch_ledgers_and_rolls_back(ledger_env, monkeypatch):
    """A curator-pass patch lands in the ledger tagged 'curator', and a
    single-entry rollback restores the exact pre-patch content."""
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage
    from tools.skill_provenance import (
        BACKGROUND_REVIEW,
        reset_current_write_origin,
        set_current_write_origin,
    )
    from tools.skill_manager_guards import mark_background_review_skill_read

    token = set_current_write_origin(BACKGROUND_REVIEW)
    try:
        # Created under the review fork → marked created_by: agent, so the
        # curator pass is allowed to patch it (curator invariant unchanged).
        assert _create()["success"] is True
        skill_md = ledger_env["skills"] / "my-skill" / "SKILL.md"
        original = skill_md.read_text(encoding="utf-8")
        mark_background_review_skill_read(skill_md)
        patched = json.loads(
            skill_manage(
                action="patch",
                name="my-skill",
                old_string="Original body.",
                new_string="Updated body.",
            )
        )
    finally:
        reset_current_write_origin(token)

    assert patched["success"] is True
    assert "Updated body." in skill_md.read_text(encoding="utf-8")

    rows = skill_ledger.list_entries(skill="my-skill")
    patch_rows = [r for r in rows if r["action"] == "patch"]
    assert len(patch_rows) == 1
    entry = patch_rows[0]
    assert entry["actor"] == "curator"
    assert any(i["path"].endswith("SKILL.md") for i in entry["before"])

    ok, msg = skill_ledger.rollback_entry(entry["id"])
    assert ok is True, msg
    assert skill_md.read_text(encoding="utf-8") == original


def test_foreground_patch_is_ledgered_as_agent(ledger_env):
    """Foreground skill_manage patches are ledgered too (all-actor design —
    unlike #50261's autonomous-only history) and tagged 'agent'."""
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    patched = json.loads(
        skill_manage(
            action="patch",
            name="my-skill",
            old_string="Original body.",
            new_string="Updated body.",
        )
    )
    assert patched["success"] is True

    rows = [r for r in skill_ledger.list_entries(skill="my-skill") if r["action"] == "patch"]
    assert len(rows) == 1
    assert rows[0]["actor"] == "agent"


def test_rollback_refuses_paths_outside_hermes_home(ledger_env):
    """A hand-edited ledger entry pointing outside HERMES_HOME must not
    become a write-anywhere primitive."""
    from tools import skill_ledger

    entry_id = skill_ledger.append_entry(
        "patch",
        "evil",
        before=[{"path": "/etc/passwd", "sha256": "0" * 64}],
        after=[],
    )
    assert entry_id is not None
    ok, msg = skill_ledger.rollback_entry(entry_id)
    assert ok is False
    assert "outside" in msg


def test_missing_blob_aborts_rollback_before_any_change(ledger_env):
    from tools import skill_ledger

    assert _create()["success"] is True
    skill_md = ledger_env["skills"] / "my-skill" / "SKILL.md"
    entry_id = skill_ledger.append_entry(
        "patch",
        "my-skill",
        before=[{"path": str(skill_md), "sha256": "a" * 64}],
        after=[],
    )
    current = skill_md.read_bytes()
    ok, msg = skill_ledger.rollback_entry(entry_id)
    assert ok is False
    assert "missing blob" in msg
    assert skill_md.read_bytes() == current


# ---------------------------------------------------------------------------
# New-design coverage
# ---------------------------------------------------------------------------


def test_ledger_entry_on_edit_and_delete(ledger_env):
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    edited = json.loads(
        skill_manage(
            action="edit",
            name="my-skill",
            content=VALID_SKILL_CONTENT.replace("Original body.", "Edited body."),
        )
    )
    assert edited["success"] is True
    deleted = json.loads(
        skill_manage(action="delete", name="my-skill", absorbed_into="")
    )
    assert deleted["success"] is True

    actions = [r["action"] for r in skill_ledger.list_entries(skill="my-skill")]
    assert actions == ["delete", "edit", "create"]  # newest first

    delete_entry = skill_ledger.list_entries(skill="my-skill")[0]
    # Delete intent recorded: explicit prune (absorbed_into="") + hard delete.
    assert delete_entry["evidence"]["absorbed_into"] == ""
    assert delete_entry["evidence"]["archived"] is False
    # Before-state captured, after empty (skill gone).
    assert delete_entry["before"]
    assert delete_entry["after"] == []


def test_deleted_skill_recoverable_from_ledger(ledger_env):
    """A foreground hard delete stays a hard delete — but the ledger entry
    can restore the skill's files from blobs."""
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    skill_md = ledger_env["skills"] / "my-skill" / "SKILL.md"
    original = skill_md.read_bytes()

    assert json.loads(skill_manage(action="delete", name="my-skill"))["success"]
    assert not skill_md.exists()

    entry = skill_ledger.list_entries(skill="my-skill")[0]
    ok, msg = skill_ledger.rollback_entry(entry["id"])
    assert ok is True, msg
    assert skill_md.read_bytes() == original


def test_archive_lands_in_ledger_with_curator_actor(ledger_env, monkeypatch):
    from tools import skill_ledger, skill_usage

    assert _create()["success"] is True
    # Curator auto-transition path tags the actor explicitly.
    tok = skill_ledger.set_ledger_actor("curator")
    try:
        ok, msg = skill_usage.archive_skill("my-skill")
    finally:
        skill_ledger.reset_ledger_actor(tok)
    assert ok, msg

    rows = [r for r in skill_ledger.list_entries(skill="my-skill") if r["action"] == "archive"]
    assert len(rows) == 1
    assert rows[0]["actor"] == "curator"
    assert rows[0]["before"] and rows[0]["after"]

    # And restore is ledgered as well.
    ok, msg = skill_usage.restore_skill("my-skill")
    assert ok, msg
    assert any(
        r["action"] == "restore" for r in skill_ledger.list_entries(skill="my-skill")
    )


def test_blob_dedupe_same_content_one_blob(ledger_env):
    from tools import skill_ledger

    d = ledger_env["skills"] / "dedupe-src"
    d.mkdir()
    (d / "a.md").write_text("identical content", encoding="utf-8")
    (d / "b.md").write_text("identical content", encoding="utf-8")

    manifest = skill_ledger.snapshot_paths(d)
    assert len(manifest) == 2
    hashes = {m["sha256"] for m in manifest}
    assert len(hashes) == 1  # same content → same hash
    blobs = list(skill_ledger.blobs_dir().iterdir())
    assert len(blobs) == 1  # → one blob on disk


def test_snapshot_paths_skips_transient_dirs(ledger_env):
    """Transient local artifacts (venv, node_modules, caches, .git) never reach
    the manifest or the blob store — sweeping them in grows the blob dir
    unboundedly on real installs (#107539)."""
    from tools import skill_ledger

    d = ledger_env["skills"] / "has-venv"
    d.mkdir()
    for rel, body in (("SKILL.md", "# skill"), ("scripts/run.py", "print('hi')"),
                      ("node_modules/pkg/index.js", "junk"), ("venv/bin/python", "junk"),
                      ("__pycache__/run.cpython-311.pyc", "junk"), (".git/config", "junk")):
        p = d / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(body, encoding="utf-8")

    manifest = skill_ledger.snapshot_paths(d)
    rel = {str(Path(i["path"]).relative_to(d)) for i in manifest}
    assert rel == {"SKILL.md", os.path.join("scripts", "run.py")}

    # None of the transient content was stored as a blob either.
    junk_sha = hashlib.sha256(b"junk").hexdigest()
    assert junk_sha not in {p.name for p in skill_ledger.blobs_dir().iterdir()}


def test_snapshot_paths_keeps_file_named_like_transient_dir(ledger_env):
    """The filter drops files *inside* transient dirs; a plain file whose own
    name collides with one (e.g. a ``venv`` bootstrap script) is skill content."""
    from tools import skill_ledger

    d = ledger_env["skills"] / "edge"
    d.mkdir()
    (d / "SKILL.md").write_text("# skill", encoding="utf-8")
    (d / "venv").write_text("#!/bin/sh\n", encoding="utf-8")  # a FILE, not a dir

    manifest = skill_ledger.snapshot_paths(d)
    rel = {str(Path(i["path"]).relative_to(d)) for i in manifest}
    assert "venv" in rel
    assert "SKILL.md" in rel


def test_rollback_fails_closed_when_safety_capture_fails(ledger_env, monkeypatch):
    """If the pre-rollback safety ledger entry can't be written, the rollback
    must abort with nothing changed (consistent with #63366)."""
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    skill_md = ledger_env["skills"] / "my-skill" / "SKILL.md"
    patched = json.loads(
        skill_manage(
            action="patch",
            name="my-skill",
            old_string="Original body.",
            new_string="Updated body.",
        )
    )
    assert patched["success"] is True
    entry = [r for r in skill_ledger.list_entries("my-skill") if r["action"] == "patch"][0]
    current = skill_md.read_bytes()

    monkeypatch.setattr(skill_ledger, "append_entry", lambda *a, **k: None)
    ok, msg = skill_ledger.rollback_entry(entry["id"])
    assert ok is False
    assert "safety capture failed" in msg
    assert skill_md.read_bytes() == current  # nothing changed


def test_rollback_removes_files_created_by_the_mutation(ledger_env):
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    wrote = json.loads(
        skill_manage(
            action="write_file",
            name="my-skill",
            file_path="references/extra.md",
            file_content="new supporting file",
        )
    )
    assert wrote["success"] is True
    extra = ledger_env["skills"] / "my-skill" / "references" / "extra.md"
    assert extra.exists()

    entry = [r for r in skill_ledger.list_entries("my-skill") if r["action"] == "write_file"][0]
    ok, msg = skill_ledger.rollback_entry(entry["id"])
    assert ok is True, msg
    assert not extra.exists()  # created by the mutation → removed on rollback


def test_config_gate_off_no_ledger_writes(ledger_env, monkeypatch):
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    import hermes_cli.config as _cfg

    off = {"skills": {"ledger": False}}
    monkeypatch.setattr(_cfg, "load_config", lambda *a, **k: off)
    monkeypatch.setattr(_cfg, "load_config_readonly", lambda *a, **k: off)

    assert _create()["success"] is True
    patched = json.loads(
        skill_manage(
            action="patch",
            name="my-skill",
            old_string="Original body.",
            new_string="Updated body.",
        )
    )
    assert patched["success"] is True  # mutation unaffected
    assert not skill_ledger.ledger_path().exists()
    assert not skill_ledger.blobs_dir().exists()


def test_ledger_failure_never_blocks_the_mutation(ledger_env, monkeypatch):
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    def _boom(*a, **k):
        raise OSError("disk full")

    monkeypatch.setattr(skill_ledger, "snapshot_paths", _boom)

    assert _create()["success"] is True
    patched = json.loads(
        skill_manage(
            action="patch",
            name="my-skill",
            old_string="Original body.",
            new_string="Updated body.",
        )
    )
    assert patched["success"] is True


def test_list_entries_filtering_and_limit(ledger_env):
    from tools import skill_ledger

    for i in range(5):
        skill_ledger.append_entry("patch", f"skill-{i % 2}", before=[], after=[])
    assert len(skill_ledger.list_entries(limit=3)) == 3
    only_zero = skill_ledger.list_entries(skill="skill-0")
    assert len(only_zero) == 3
    assert all(r["skill"] == "skill-0" for r in only_zero)


def test_user_actor_override(ledger_env):
    from tools import skill_ledger

    tok = skill_ledger.set_ledger_actor("user")
    try:
        entry_id = skill_ledger.append_entry("archive", "some-skill")
    finally:
        skill_ledger.reset_ledger_actor(tok)
    entry = skill_ledger.get_entry(entry_id)
    assert entry["actor"] == "user"


# ---------------------------------------------------------------------------
# Package-completeness fill from the newest curator backup (issue #96962)
# ---------------------------------------------------------------------------


def _write_skills_tarball(home: Path, files: dict, stamp: str = "2026-08-01T00-00-00Z"):
    """Write a curator-shaped ``skills.tar.gz`` under *home* (arcnames are
    relative to skills/, exactly like agent.curator_backup.snapshot_skills)."""
    import io
    import tarfile

    snap = home / "skills" / ".curator_backups" / stamp
    snap.mkdir(parents=True, exist_ok=True)
    tar_path = snap / "skills.tar.gz"
    with tarfile.open(tar_path, "w:gz") as tf:
        for rel, content in files.items():
            data = content.encode("utf-8") if isinstance(content, str) else content
            info = tarfile.TarInfo(name=rel)
            info.size = len(data)
            tf.addfile(info, io.BytesIO(data))
    return tar_path


def test_delete_after_rehome_ledgers_full_package_from_backup(ledger_env):
    """The incident shape (#96962): consolidation re-homes references/ out of
    the tree, then deletes. The delete entry must still capture the support
    file from the newest curator backup, and rollback must restore both."""
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    extra = ledger_env["skills"] / "my-skill" / "references" / "extra.md"
    wrote = json.loads(skill_manage(
        action="write_file",
        name="my-skill",
        file_path="references/extra.md",
        file_content="roadmap body",
    ))
    assert wrote["success"] is True

    # The pre-curator-run snapshot, taken while the package was whole.
    skill_md = ledger_env["skills"] / "my-skill" / "SKILL.md"
    _write_skills_tarball(
        ledger_env["home"],
        {
            "my-skill/SKILL.md": skill_md.read_text(encoding="utf-8"),
            "my-skill/references/extra.md": "roadmap body",
        },
    )

    # Re-home: the support file leaves the tree before the delete.
    extra.unlink()
    extra.parent.rmdir()

    deleted = json.loads(skill_manage(action="delete", name="my-skill"))
    assert deleted["success"] is True

    delete_entry = [
        r for r in skill_ledger.list_entries(skill="my-skill")
        if r["action"] == "delete"
    ][0]
    before_names = {Path(i["path"]).name for i in delete_entry["before"]}
    assert "SKILL.md" in before_names
    assert "extra.md" in before_names, (
        "delete ledger captured only SKILL.md after the support files were "
        "re-homed — rollback would restore a hollow skill (#96962)"
    )

    ok, msg = skill_ledger.rollback_entry(delete_entry["id"])
    assert ok is True, msg
    assert skill_md.is_file()
    assert extra.is_file()
    assert extra.read_text(encoding="utf-8") == "roadmap body"


def test_rollback_historical_hollow_entry_restores_full_package(ledger_env):
    """Entries recorded BEFORE this fix (files: 1) still restore the whole
    package: rollback-time fill from the newest curator backup."""
    from tools import skill_ledger

    skill_dir = ledger_env["skills"] / "my-skill"
    skill_dir.mkdir()
    skill_md = skill_dir / "SKILL.md"
    skill_md.write_text(VALID_SKILL_CONTENT, encoding="utf-8")
    _write_skills_tarball(
        ledger_env["home"],
        {
            "my-skill/SKILL.md": VALID_SKILL_CONTENT,
            "my-skill/references/roadmap.md": "week 1",
        },
    )
    # The mutation that made the entry: package gone, only SKILL.md captured.
    skill_md.unlink()
    skill_dir.rmdir()

    entry_id = skill_ledger.append_entry(
        "delete",
        "my-skill",
        before=[{"path": str(skill_md), "sha256": skill_ledger._store_blob(
            VALID_SKILL_CONTENT.encode("utf-8")
        )}],
        after=[],
    )
    assert entry_id is not None

    ok, msg = skill_ledger.rollback_entry(entry_id)
    assert ok is True, msg
    roadmap = skill_dir / "references" / "roadmap.md"
    assert skill_md.is_file()
    assert roadmap.is_file(), "hollow rollback: support file not restored"
    assert roadmap.read_text(encoding="utf-8") == "week 1"


def test_delete_rollback_without_backup_still_works(ledger_env):
    """No curator backup present: the fill degrades to the old behavior and
    must not break the plain delete -> rollback round trip."""
    from tools import skill_ledger
    from tools.skill_manager_tool import skill_manage

    assert _create()["success"] is True
    skill_md = ledger_env["skills"] / "my-skill" / "SKILL.md"

    deleted = json.loads(skill_manage(action="delete", name="my-skill"))
    assert deleted["success"] is True
    delete_entry = [
        r for r in skill_ledger.list_entries(skill="my-skill")
        if r["action"] == "delete"
    ][0]
    assert {Path(i["path"]).name for i in delete_entry["before"]} == {"SKILL.md"}

    ok, msg = skill_ledger.rollback_entry(delete_entry["id"])
    assert ok is True, msg
    assert skill_md.read_text(encoding="utf-8") == VALID_SKILL_CONTENT


def test_backup_fill_does_not_clobber_disk_hash(ledger_env):
    """Disk state wins: a live SKILL.md that differs from the backup copy is
    captured with the LIVE hash; the backup only fills missing paths."""
    from tools import skill_ledger

    skill_dir = ledger_env["skills"] / "my-skill"
    skill_dir.mkdir()
    skill_md = skill_dir / "SKILL.md"
    live = VALID_SKILL_CONTENT.replace("Original body.", "Live body.")
    skill_md.write_text(live, encoding="utf-8", newline="\n")
    _write_skills_tarball(
        ledger_env["home"],
        {
            "my-skill/SKILL.md": VALID_SKILL_CONTENT,
            "my-skill/references/extra.md": "from tar",
        },
    )

    captured = skill_ledger.snapshot_paths(skill_dir, complete_package=True)
    by_name = {Path(i["path"]).name: i["sha256"] for i in captured}
    live_hash = skill_ledger._store_blob(live.encode("utf-8"))
    tar_hash = skill_ledger._store_blob(VALID_SKILL_CONTENT.encode("utf-8"))
    assert by_name["SKILL.md"] == live_hash, "disk hash must win over backup"
    assert by_name["SKILL.md"] != tar_hash
    assert by_name["extra.md"] == skill_ledger._store_blob(b"from tar")


def test_backup_fill_ignores_tar_path_traversal(ledger_env):
    """Fill runs AND malicious members are rejected: a legitimate missing
    file is restored while members escaping the package prefix (absolute,
    ..) are never filled. Both assertions matter — the positive one keeps
    this test honest (a silently inert fill would pass a negatives-only
    check), the negative one pins the traversal defense."""
    from tools import skill_ledger

    skill_dir = ledger_env["skills"] / "my-skill"
    skill_dir.mkdir()
    (skill_dir / "SKILL.md").write_text(VALID_SKILL_CONTENT, encoding="utf-8")
    _write_skills_tarball(
        ledger_env["home"],
        {
            "my-skill/SKILL.md": VALID_SKILL_CONTENT,
            "my-skill/references/legit.md": "legit body",
            "../evil.md": "nope",
            "my-skill/../outside.md": "nope",
        },
    )

    captured = skill_ledger.snapshot_paths(skill_dir, complete_package=True)
    paths = [i["path"] for i in captured]
    # The legitimate missing file WAS filled — proof the fill is live.
    assert any(Path(p).as_posix().endswith("references/legit.md") for p in paths), (
        "package fill did not restore the missing support file"
    )
    # Malicious members are not.
    assert not any(p.endswith("evil.md") or p.endswith("outside.md") for p in paths)

import pytest


def _append_padded(skill_ledger, action: str, pad: str, n: int = 1) -> None:
    """Append *n* entries padded with evidence text so the ledger file grows fast."""
    for _ in range(n):
        skill_ledger.append_entry(action, "my-skill", before=[], after=[], evidence={"pad": pad})


def test_auto_compact_triggers_at_threshold(ledger_env, monkeypatch):
    """Crossing skills.ledger_max_bytes rewrites the ledger through the delta
    dedup: legacy rows that still carry identical before/after manifests (written
    before append-time ``_delta`` existed) shrink to nothing while ids and entry
    order survive, and nothing is trimmed when dedup alone reaches the cap."""
    import json

    from tools import skill_ledger

    import hermes_cli.config as _cfg

    cap = {"skills": {"ledger_max_bytes": 8192}}
    monkeypatch.setattr(_cfg, "load_config", lambda *a, **k: cap)
    monkeypatch.setattr(_cfg, "load_config_readonly", lambda *a, **k: cap)

    first_id = skill_ledger.append_entry("patch", "my-skill", before=[], after=[])
    template = json.loads(skill_ledger.ledger_path().read_text().splitlines()[0])
    with skill_ledger.ledger_path().open("a", encoding="utf-8") as fh:
        for i in range(3):  # legacy pre-delta rows: identical fat manifests on both sides
            fat = [{"path": f"my-skill/f{i}j{j}.md", "sha256": "a" * 64} for j in range(40)]
            row = dict(template, id=f"legacy{i}", before=fat, after=list(fat))
            fh.write(json.dumps(row) + "\n")
    assert skill_ledger.ledger_path().stat().st_size > 8192

    last_id = skill_ledger.append_entry("patch", "my-skill", before=[], after=[])

    # the maintenance sweep fired on that append: the file is back under the cap
    assert skill_ledger.ledger_path().stat().st_size <= 8192
    rows = skill_ledger.list_entries()
    assert {r["id"] for r in rows} == {first_id, last_id, "legacy0", "legacy1", "legacy2"}, (
        "dedup alone must reach the cap — nothing trimmed, ids survive"
    )
    assert all(r["before"] == [] and r["after"] == [] for r in rows), (
        "identical manifests must be dropped by compaction"
    )


def test_trim_oldest_when_still_over_cap(ledger_env, monkeypatch):
    """When compaction alone cannot reach the cap (every entry genuinely
    differs), the oldest lines are dropped — whatever their shape — until the
    file fits under the LOW-WATER mark (80% of the cap, so the next append does
    not immediately re-trigger the sweep). The newest entry survives, and lines
    in the retained tail are never parsed or rewritten: a malformed last line
    survives verbatim."""
    from tools import skill_ledger

    import hermes_cli.config as _cfg

    cap = {"skills": {"ledger_max_bytes": 0}}  # no sweeps while seeding
    monkeypatch.setattr(_cfg, "load_config", lambda *a, **k: cap)
    monkeypatch.setattr(_cfg, "load_config_readonly", lambda *a, **k: cap)

    newest_id = None
    for i in range(5):
        before = [{"path": f"my-skill/old{i}.md", "sha256": f"{i}" * 64}]
        after = [{"path": f"my-skill/new{i}.md", "sha256": f"{i + 1}" * 64}]
        newest_id = skill_ledger.append_entry(
            "edit", "my-skill", before=before, after=after,
            evidence={"pad": "y" * 2048})
    with open(skill_ledger.ledger_path(), "a", encoding="utf-8") as fh:
        fh.write("{not json at all\n")
    assert skill_ledger.ledger_path().stat().st_size > 8192

    cap["skills"]["ledger_max_bytes"] = 8192
    skill_ledger._maintain_size()

    rows = skill_ledger.list_entries()
    assert len(rows) < 5, "oldest entries must be trimmed when compaction is not enough"
    assert rows[0]["id"] == newest_id, "the newest entry always survives"
    # malformed lines are never parsed away — they stay in the file verbatim
    raw = skill_ledger.ledger_path().read_text(encoding="utf-8")
    assert "{not json at all" in raw
    # A sweep that fires does not stop at the cap but at the low-water mark ...
    assert skill_ledger.ledger_path().stat().st_size <= int(8192 * 0.8)
    # ... so the next append rides under the cap without paying compact+trim+gc again.
    compactions = []
    monkeypatch.setattr(skill_ledger, "compact_ledger",
                        lambda *a, **k: compactions.append(1) or (0, 0, 0))
    skill_ledger.append_entry(
        "edit", "my-skill", before=[{"path": "my-skill/z.md", "sha256": "a" * 64}],
        after=[{"path": "my-skill/z2.md", "sha256": "b" * 64}], evidence={"pad": "z" * 1000})
    assert compactions == [], "an append under the cap must not re-run the sweep"
    assert skill_ledger.ledger_path().stat().st_size <= 8192
    assert "{not json at all" in skill_ledger.ledger_path().read_text(encoding="utf-8")
    # U+2028 inside a row (ensure_ascii=False leaves it unescaped) is not a row boundary for the
    # trim: a cap that fits only the newest row keeps that row byte-for-byte, not its second half.
    u_row = (json.dumps({"id": "u2028", "skill": "my-skill", "action": "edit",
                         "evidence": {"note": "line one\u2028line two"}}, ensure_ascii=False) + "\n").encode("utf-8")
    with open(skill_ledger.ledger_path(), "ab") as fh:
        fh.write(u_row)
    assert skill_ledger._trim_oldest(len(u_row) + 8) >= 1
    assert skill_ledger.ledger_path().read_bytes() == u_row, "a retained row containing U+2028 survives intact"


def test_concurrent_appends_never_lose_a_middle_row(ledger_env, monkeypatch):
    """Two writers appending while the maintenance sweep fires on (almost) every append:
    every row each writer appended is either still in the ledger or was trimmed
    oldest-first — never silently lost from the middle of a writer's sequence. The sweep's
    read → ``os.replace`` must run under the same ``.locks/ledger.lock`` as the O_APPEND
    write, or an append landing on the replaced inode vanishes (and ``gc_blobs`` would then
    delete its blobs). The race is forced, not hoped for: writer A's first sweep pauses
    between reading the ledger and replacing it until writer B has appended (or, when the
    lock correctly blocks B, until a generous bound expires — green never depends on timing)."""
    import threading

    from tools import skill_ledger

    import hermes_cli.config as _cfg

    cap = {"skills": {"ledger_max_bytes": 4096}}  # padded rows ~600 B: a trim on nearly every append
    monkeypatch.setattr(_cfg, "load_config", lambda *a, **k: cap)
    monkeypatch.setattr(_cfg, "load_config_readonly", lambda *a, **k: cap)

    n, ids = 40, {"A": [], "B": []}
    b_go, b_done = threading.Event(), threading.Event()
    real_rewrite = skill_ledger._rewrite_ledger

    def paused_rewrite(path, lines, op):
        if threading.current_thread().name == "A" and not b_go.is_set():
            b_go.set()             # A has read the ledger; let B append now ...
            b_done.wait(1.0)       # ... and give it every chance to land before the replace
        return real_rewrite(path, lines, op)

    monkeypatch.setattr(skill_ledger, "_rewrite_ledger", paused_rewrite)
    dropped, real_trim = [], skill_ledger._trim_oldest
    monkeypatch.setattr(skill_ledger, "_trim_oldest",
                        lambda max_bytes: dropped.append(real_trim(max_bytes)) or dropped[-1])

    def writer(k: str) -> None:
        if k == "B":
            b_go.wait(10.0)
        for i in range(n):
            before = [{"path": f"my-skill/{k}-{i}.md", "sha256": "a" * 64}]
            after = [{"path": f"my-skill/{k}-{i}.md", "sha256": f"{i % 10}" * 64}]
            ids[k].append(skill_ledger.append_entry(
                "edit", "my-skill", before=before, after=after, evidence={"pad": "x" * 500}))
            if k == "B":
                b_done.set()

    seeds = 8  # seed over the cap so A's very first append sweeps
    for _ in range(seeds):
        skill_ledger.append_entry("edit", "my-skill", before=[{"path": "s", "sha256": "0" * 64}],
                                  after=[{"path": "s", "sha256": "1" * 64}], evidence={"pad": "x" * 500})
    # Unreferenced blobs: a fresh one is another process's in-flight capture (row not appended yet)
    # and must survive every sweep's blob GC; one older than the grace window is garbage and goes.
    fresh, aged = skill_ledger._store_blob(b"in-flight"), skill_ledger._store_blob(b"stale orphan")
    os.utime(skill_ledger.blobs_dir() / aged, (time.time() - 7200, time.time() - 7200))
    threads = [threading.Thread(target=writer, args=(k,), name=k) for k in ids]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert b_go.is_set(), "writer A's first append must have swept (test precondition)"
    assert all(ids["A"]) and all(ids["B"]) and len(ids["A"]) == len(ids["B"]) == n, "every append reported success"
    present = [json.loads(line)["id"] for line in
               skill_ledger.ledger_path().read_text(encoding="utf-8").splitlines() if line.strip()]
    assert present, "the newest row always survives a trim"
    assert (skill_ledger.blobs_dir() / fresh).exists() and not (skill_ledger.blobs_dir() / aged).exists()
    assert len(present) == seeds + 2 * n - sum(dropped), (
        "every row is either in the ledger or was counted as trimmed — none silently lost"
    )
    for k, seq in ids.items():
        survivors = [i for i in seq if i in set(present)]
        assert survivors == seq[len(seq) - len(survivors):], (
            f"writer {k}: rows missing from the middle — a concurrent sweep dropped an append"
        )