    shim.chmod(shim.stat().st_mode | stat.S_IEXEC)
    return f


ON_PATH = "# agent-journal: the journal command"


def put_on_path(bin_: Path) -> str:
    if str(bin_) in os.environ.get("PATH", "").split(os.pathsep):
        return ""
    shell = Path(os.environ.get("SHELL", "")).name
    profile = Path.home() / {"zsh": ".zshrc", "bash": ".bash_profile"}.get(shell, ".profile")
    if ON_PATH not in (profile.read_text() if profile.is_file() else ""):
        with profile.open("a") as written:
            written.write(f'\n{ON_PATH}\nexport PATH="$HOME/.local/bin:$PATH"\n')
    return f"{bin_} added to your PATH in {profile}: open a new terminal, then type journal"


def install(project: Path, root: Path | None = None) -> list[str]:
    root = root or project / ".journal"
    refresh(PACKAGE, code(root))
    done = configure(project, root)
    retire(root)
    return done


def old_git_hook(project: Path) -> list[str]:
    hook = project / ".git" / "hooks" / "post-commit"
    if hook.is_file() and "agent-journal:" in hook.read_text(errors="replace"):
        hook.unlink()
        return ["the version 1 post-commit git hook removed"]
    return []


def configure(project: Path, root: Path) -> list[str]:
    done = old_git_hook(project)
    present = []
    for name, cls in LOADED.providers.items():
        provider = cls()
        if not provider.present(project):
            continue
        f = provider.wire(project, LOADED.hook_command(code(root) / "hook.sh", name, root))
        done.append(f"{name}: hooks in {f.relative_to(project)}")
        present.append(name)
    if not present:
        return [*done, f"no agent found here: neither {' nor '.join(name.capitalize() for name in LOADED.providers)}"]
    written, linked = LOADED.publish(project, tuple(present))
    done.append(f"{len(written)} skills in {LOADED.library}" + (f", linked from {', '.join(LOADED.linked[a] for a in present if a in LOADED.linked)}" if linked else ""))
    record = LOADED.record(root, LOADED.default_env(root))
    briefing = LOADED.brief(project, record)
    named = ' and '.join(sorted(cls.briefing_file for cls in LOADED.providers.values() if cls.briefing_file))
    done.append(f"the journal's block in {', '.join(f.name for f in briefing.written) or named}")
    done.extend(briefing.left)
    written = LOADED.agent_types(project, record)
    if written:
        done.append(f"agent types: {', '.join(f.stem for f in written)}")
    done.append(f"the journal command: {alias(project, root).relative_to(project)}")
    told = put_on_path(Path.home() / ".local" / "bin")
    if told:
        done.append(told)
    return done


def token() -> str:
    if not shutil.which("gh"):
        return ""
    try:
        got = subprocess.run(["gh", "auth", "token"], capture_output=True, text=True, timeout=20)
    except (OSError, subprocess.SubprocessError):
        return ""
    return got.stdout.strip() if got.returncode == 0 else ""


def with_token(repository: str, secret: str) -> str:
    return repository.replace("https://", f"https://x-access-token:{secret}@", 1) if secret and repository.startswith("https://github.com/") else repository


def redacted(text: str, secret: str) -> str:
    return text.replace(secret, "the token") if secret else text


def version_key(version: str) -> tuple:
    return tuple(int(part) if part.isdigit() else 0 for part in str(version).split("."))


def released(repository: str = REPOSITORY) -> str:
    try:
        listed = subprocess.run(["git", "ls-remote", "--tags", "--refs", repository, "v*"], capture_output=True, text=True, timeout=10)
    except (OSError, subprocess.TimeoutExpired):
        return ""
    versions = [line.rsplit("/v", 1)[1] for line in listed.stdout.splitlines() if "/v" in line] if not listed.returncode else []
    return max(versions, key=version_key) if versions else ""


def fetch(into: Path, repository: str = "", ref: str = "") -> tuple[str, str]:
    wanted = repository or os.environ.get(REPOSITORY_ENV, REPOSITORY)
    secret = token() if wanted.startswith("https://github.com/") else ""
    source = with_token(wanted, secret)
    into.mkdir(parents=True, exist_ok=True)
    try:
        for step in (["init", "-q"], ["fetch", "-q", "--depth", "1", source, ref or "HEAD"], ["checkout", "-q", "FETCH_HEAD"]):
            done = subprocess.run(["git", *step], cwd=into, capture_output=True, text=True, timeout=120)
            if done.returncode:
                return "", redacted(done.stderr.strip() or f"git {step[0]} failed", secret)
        return subprocess.run(["git", "rev-parse", "HEAD"], cwd=into, capture_output=True, text=True, timeout=30).stdout.strip(), ""
    except (OSError, subprocess.TimeoutExpired) as error:
        return "", str(error)


def keep_copy(root: Path) -> str:
    if not (root / "environments").is_dir():
        return ""
    version = version_in(code(root), "unknown")
    attic = root / "attic"
    attic.mkdir(parents=True, exist_ok=True)
    copy = attic / f"before-{version}-{int(time.time())}.tar.gz"
    with tarfile.open(copy, "w:gz") as archive:
        for entry in sorted(root.iterdir()):
            if entry.name not in NOT_RECORD and not entry.name.startswith(("journal-", ARCHIVE)):
                archive.add(entry, arcname=entry.name)
    for old in sorted(attic.glob("before-*.tar.gz"), key=lambda f: f.stat().st_mtime, reverse=True)[KEPT_COPIES:]:
        old.unlink(missing_ok=True)
    return f"a copy of the record is kept in {copy.relative_to(root.parent)}"


def half_done(root: Path) -> bool:
    return (code(root) / "__main__.py").is_file() and (root / ARCHIVE).exists()


def upgrade(project: Path, root: Path | None = None) -> list[str]:
    root = root or project / ".journal"
from dataclasses import replace
import inspect
import json


import features
from controllers.types import Agents, Works
from runner.gate import gated
from runner.hooks import handle
from engine.gates import AFTERWARDS, CANCELERS, LONG_COMMAND, POLICIES, HookCall, cancelled, gate_file
from engine.gates import held
from engine.wording import APPENDS
from providers import PROVIDERS
from providers.payload import Hook
from resources.base import AGENT, SYSTEM
from resources.types import SUBAGENT
from tests.conftest import fresh



def test_a_write_is_refused_until_work_is_open_for_every_provider():
    features.load()
    REFUSED = features.FEATURES["work_tracking"].line("undeclared held", {})[0]
    record = fresh()
    root, env = record.root, record.env

    for name, provider_cls in PROVIDERS.items():
        provider = provider_cls()
        session = f"{name}-7"

        def hook(event, tool="", cwd="", **tool_input):
            return handle(provider, root, env, {"hook_event_name": event, "session_id": session, "tool_name": tool, "tool_input": tool_input, "cwd": cwd})

        hook("SessionStart")
        assert held(record, session) == REFUSED, f"{name}: a fresh session with nothing open: the flag says refused"
        assert hook("PreToolUse", "Read", file_path="x.py") == {}, f"{name}: a read passes"
        assert hook("PreToolUse", "Bash", command="cat x.py | grep y") == {}, f"{name}: a Bash read passes"
        assert hook("PreToolUse", "Edit", file_path="x.py") == provider.blocking(REFUSED), \
            f"{name}: an edit is refused, in the harness's shape"
        assert hook("PreToolUse", "Bash", command="git commit -m x") == provider.blocking(REFUSED), \
            f"{name}: a writing command is refused"
        assert hook("PreToolUse", "Bash", command="echo x > out.txt") == provider.blocking(REFUSED), \
            f"{name}: a redirect is a write"
        assert hook("PreToolUse", "Bash", command="make > /dev/null") == {}, f"{name}: a redirect to /dev/null is not"
        assert hook("PreToolUse", "Bash", command="python3 tests/x.py 2>&1 | tail -1") == {}, f"{name}: joining stderr is not a write"
        project = str(root.parent)
        assert hook("PreToolUse", "Write", cwd=project, file_path="web/x.js") == provider.blocking(REFUSED), \
            f"{name}: a project file is gated"
        assert hook("PreToolUse", "Write", cwd=project, file_path=".journal/environments/main/notes.md") == {}, \
            f"{name}: a file inside the journal is not project work"
        assert hook("PreToolUse", "Edit", cwd=project, file_path="/tmp/elsewhere/memory.md") == {}, \
            f"{name}: a file outside the project is not project work"
        assert hook("PreToolUse", "Bash", command="journal work start \"x\" >/dev/null; .journal/journal todo add x") == {}, \
            f"{name}: a journal command is never gated, it is how work opens"
        work = Works(record, actor=AGENT).create(f"the header for {name}")
        assert held(record, session) == "", f"{name}: work open: the flag flips to allowed"
        assert hook("PreToolUse", "Edit", file_path="x.py") == {}, f"{name}: the same edit passes"
        Works(record, actor=AGENT).complete(work.n, "done")
        assert hook("PreToolUse", "Write", file_path="y.py") == provider.blocking(REFUSED), \
            f"{name}: work ended, nothing open: refused again"
        assert json.loads(gate_file(root, env, session).read_text())["work_tracking"] == {"why": REFUSED, "reach": "main"}, \
            f"{name}: the flag is a file per environment and session, with the why"


def test_a_hook_that_crashes_is_told_to_the_agent_for_every_provider(monkeypatch):
    from commands.dispatch import dispatch
    import commands.http  # noqa: F401
    from controllers.types import Notices, Nudges
    from resources.base import SYSTEM
    from tests.kit import report
    features.load()
    record = fresh()
    report(record, "working", "PreToolUse")

    for name, provider_cls in PROVIDERS.items():
        def crash(self, hook, root):
            raise TypeError(f"{name} crashed")
        monkeypatch.setattr(provider_cls, "facts", crash)
        body = {"hook_event_name": "PreToolUse", "session_id": "claude-1", "tool_name": "Read", "tool_input": {"file_path": "x.py"}}
        dispatch("POST", f"/api/hook/{name}", record.root, {"root": str(record.root), "env": record.env}, body)
        lines = [f"{n.title} {n.brief}" for n in Nudges(record, actor=SYSTEM).rows.every()]
        assert any("hit an error" in line and f"TypeError: {name} crashed" in line for line in lines), \
            f"{name}: a crash inside the hook reaches the agent, with the error"
        dispatch("POST", f"/api/hook/{name}", record.root, {"root": str(record.root), "env": record.env}, body)
        assert len([line for line in lines if "crashed" in line]) == len([n for n in Nudges(record, actor=SYSTEM).rows.every() if "crashed" in n.brief]), \
            f"{name}: the same error again is not told twice"
        for notice in Notices(record, actor=SYSTEM).rows.standing():
            Notices(record, actor=SYSTEM).complete(notice.n, "fixed")
    monkeypatch.undo()
    from engine import runtime
    runtime.hook_failures(record.root).write_text(f"1790000000 000 claude {record.env}\n1790000001 500 claude {record.env}\n")
    commands.http.unanswered(record.root)
    lines = [f"{n.title} {n.brief}" for n in Nudges(record, actor=SYSTEM).rows.every()]
    assert any("no answer from the server 2 times (codes 000, 500)" in line for line in lines), \
        "hooks the server never answered are told once it answers again"
    assert not runtime.hook_failures(record.root).exists(), "and are told only once"
    monkeypatch.setattr(runtime, "STARTED", [1790000100.0])
    runtime.hook_failures(record.root).write_text(f"1790000095 000 claude {record.env}\n")
    commands.http.unanswered(record.root)
    assert len([n for n in Nudges(record, actor=SYSTEM).rows.every() if "no answer from the server" in n.brief]) == 1, \
        "a hook missed while the server was restarting is not an error"
    runtime.restarting(record.root).write_text("1790000060")
    runtime.hook_failures(record.root).write_text(f"1790000070 000 claude {record.env}\n")
    commands.http.unanswered(record.root)
    assert len([n for n in Nudges(record, actor=SYSTEM).rows.every() if "no answer from the server" in n.brief]) == 1, \
        "however long a restart the journal began itself takes, the hooks it missed are not an error"


def test_a_command_runs_as_the_session_its_own_shell_names_for_every_provider(monkeypatch):
    import os
    from commands.cli import context
    from commands.parser import parser
    from engine.sessions import Sessions
    record = fresh()
    sessions = Sessions(record.root)
    for session in ("first-agent", "second-agent"):
        sessions.bind(session, record.env, pid=os.getpid(), provider="claude")
    for provider in PROVIDERS.values():
        if not provider.session_variable:
            continue
        monkeypatch.setenv("JOURNAL_SESSION_VARIABLE", provider.session_variable)
        monkeypatch.setenv(provider.session_variable, "second-agent")
        monkeypatch.delenv("JOURNAL_SESSION", raising=False)
        args = vars(parser("todo").parse_args(["--root", str(record.root), "todo", "all"]))
        assert context(args)["session"] == "second-agent", \
            f"{provider.name}: with two agents in one environment, a command runs as the agent whose shell ran it"


def values_for(feature, key: str, line) -> dict:
    appended = [p for append in APPENDS.each(key=f"{feature.name}.{key}") for p in inspect.signature(append).parameters.values()
                if p.default is p.empty and p.kind is p.POSITIONAL_OR_KEYWORD and p.name != "record"]
    return {**{p.name: [1] if p.annotation is list else "1" for p in appended}, **{name: "1" for name in line.placeholders()}}


def spied(guard, asked: list):
    def ask(*given):
        asked.append(guard)
        return ""
    ask.guard = guard
    return ask


def test_every_line_and_guard_reaches_exactly_the_agents_its_reach_names():
    features.load()
    record = fresh()
    agents = Agents(record, actor=SYSTEM)
    main = agents.by_session("claude-1")
    helper = agents.update(agents.by_session("claude-helper").n, status=SUBAGENT)
    for feature in features.FEATURES.values():
        for key, line in feature.lines.items():
            for row in (main, helper):
                arrived = feature.journal.whisper(record, row, key, **values_for(feature, key, line)) is not None
                assert arrived == line.reach.reaches(row.subagent), f"{feature.name}.{key} is {line.reach}; it arrived at a subagent row: {row.subagent}"

    provider = PROVIDERS["claude"]()
    asked: list = []
    points = (POLICIES, AFTERWARDS, CANCELERS)
    kept = [list(point.entries) for point in points]
    for point, entries in zip(points, kept):
        point.entries = [replace(entry, value=spied(entry.value.guard, asked)) for entry in entries]
    every = [entry.value.guard for entries in kept for entry in entries]
    dispatch = {"subagent_type": "general-purpose", "model": "haiku", "description": "look", "prompt": "look"}
    try:
        for subagent in (False, True):
            asked.clear()
            called = {"hook_event_name": "PreToolUse", "session_id": "claude-1", "tool_name": "Agent", "tool_input": dispatch}
            hook = Hook.read({**called, "agent_id": "helper"} if subagent else called, provider.tool_kinds)
            call = HookCall(provider, record, hook, main)
            gated(call)
            cancelled(LONG_COMMAND, call, {})
            wanted = sorted((g for g in every if g.reaches(subagent)), key=repr)
            assert sorted(asked, key=repr) == wanted, f"a {'subagent' if subagent else 'main agent'}'s call asks exactly the guards that reach it"
    finally:
        for point, entries in zip(points, kept):
            point.entries = entries
import json
import os
import subprocess
import sys
import time
import zipfile
from pathlib import Path

from engine.heal import broken
from engine.package import point
from engine.sessions import hold_build
from install import STUBS
from providers import DRIVERS
from scripts.boot_guard import PROJECT, WAIT, launches
from scripts.checks.imports import imports, missing

HERE = Path(__file__).resolve().parents[1]
CODE = HERE / "src"


def installed(place: Path) -> Path:
    (place / PROJECT / ".claude").mkdir(parents=True)
    env = {**os.environ, "HOME": str(place / "home"), "AGENT_JOURNAL_BOOTSTRAPPED": "1"}
    subprocess.run([sys.executable, str(CODE / "install.py"), "upgrade", str(place / PROJECT)], env=env, capture_output=True, timeout=120)
    return place / PROJECT / ".journal"


def test_every_import_in_the_package_resolves():
    assert [f"{path.name}:{node.lineno}" for path, node in imports() for alias in node.names if missing(node.module, alias.name)] == []


def test_every_agent_launches_under_the_journal_and_exits_cleanly(tmp_path):
    for name in DRIVERS:
        launches(tmp_path / name, CODE / "journal.py", name)


def test_a_restart_brings_the_agent_back_under_the_same_supervisor(tmp_path):
    from engine import runtime
    from agents.terminal import relaunch
    root = tmp_path / PROJECT / ".journal"
    moved = []

    def restarted():
        folder = next((root / "runtime" / "sessions").glob("claude-*"))
        before = json.loads((folder / "launched.json").read_text())["pid"]
        relaunch(root, runtime.env(root), folder.name, "")
        began = time.time()
        while time.time() - began < WAIT and json.loads((folder / "launched.json").read_text())["pid"] == before:
            time.sleep(0.2)
        moved.append(json.loads((folder / "launched.json").read_text())["pid"] != before)

    launches(tmp_path, CODE / "journal.py", "claude", during=restarted)
    assert moved == [True], "the supervisor stops the agent and starts it again in the same session, and nothing is left running after"


def test_every_agent_launches_from_an_installed_zip(tmp_path):
    place = tmp_path
    (place / PROJECT).mkdir()
    env = {**os.environ, "HOME": str(place / "home"), "AGENT_JOURNAL_BOOTSTRAPPED": "1"}
    installed = subprocess.run([sys.executable, str(CODE / "install.py"), "upgrade", str(place / PROJECT)], env=env, capture_output=True, text=True, timeout=120)
    root = place / PROJECT / ".journal"
    left = sorted(f.relative_to(root / "src").as_posix() for f in (root / "src").rglob("*.py"))
    assert ((root / "journal.pyz").is_file(), left) == (True, sorted(STUBS)), f"the Python is packed into one zip, a stub left at each old entry:\n{installed.stdout}{installed.stderr}"
    from engine import runtime
    assert not runtime.upgrading(root), "an upgrade that has finished leaves no mark, so the supervisor may reload"
    (root / "runtime" / "upgrading").touch()
    assert runtime.upgrading(root), "while one is under way, the mark holds the supervisor's reload back"
    (root / "runtime" / "upgrading").unlink()
    journal = [sys.executable, str(root / "journal.py"), "--root", str(root)]
    subprocess.run([*journal, "doc", "create", "Kept across upgrades"], cwd=place / PROJECT, env=env, capture_output=True, timeout=WAIT)
    again = subprocess.run([sys.executable, str(CODE / "install.py"), "upgrade", str(place / PROJECT)], env=env, capture_output=True, text=True, timeout=120)
    listed = subprocess.run([*journal, "doc", "all"], cwd=place / PROJECT, env=env, capture_output=True, text=True, timeout=WAIT).stdout
    assert "Kept across upgrades" in listed, f"a project record survives an upgrade:\n{again.stdout}{again.stderr}"
    repository = place / "release"
    shipped = subprocess.run(["git", "ls-files", "-co", "--exclude-standard"], cwd=HERE, capture_output=True, text=True, timeout=WAIT).stdout.split()
    for name in shipped:
        if (HERE / name).is_file():
            (repository / name).parent.mkdir(parents=True, exist_ok=True)
            (repository / name).write_bytes((HERE / name).read_bytes())
    for step in (["init", "-q"], ["add", "-A"], ["-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "-m", "release"]):
        subprocess.run(["git", *step], cwd=repository, capture_output=True, timeout=WAIT)
    itself = subprocess.run([*journal, "upgrade"], cwd=place / PROJECT, env={**env, "AGENT_JOURNAL_REPO": str(repository), "AGENT_JOURNAL_BOOTSTRAPPED": ""},
                            capture_output=True, text=True, timeout=180)
    listed = subprocess.run([*journal, "doc", "all"], cwd=place / PROJECT, env=env, capture_output=True, text=True, timeout=WAIT).stdout
    assert ("Traceback" not in itself.stdout + itself.stderr, "Kept across upgrades" in listed, (root / "journal.pyz").resolve().name.startswith("journal-")) == (True, True, True), \
        f"an installed journal upgrades itself from a release, keeps its records, and runs from a versioned build:\n{itself.stdout}{itself.stderr}"
    for name in DRIVERS:
        launches(place, root / "journal.py", name)
    (repository / "src" / "channel.py").write_text((repository / "src" / "channel.py").read_text() + f"\nRELEASE = {os.urandom(4000).hex()!r}\n")
    subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qam", "a new build"], cwd=repository, capture_output=True, timeout=WAIT)

    moved = []

    def upgraded():
        subprocess.run([*journal, "upgrade"], cwd=place / PROJECT, env={**env, "AGENT_JOURNAL_REPO": str(repository), "AGENT_JOURNAL_BOOTSTRAPPED": ""},
                       capture_output=True, timeout=180)
        newest, began = (root / "journal.pyz").resolve().name, time.time()
        while not moved and time.time() - began < WAIT:
            moved.extend(marker for marker in (root / "runtime" / "builds").glob("*") if marker.read_text() == newest)
            time.sleep(0.2)

    launches(place, root / "journal.py", "claude", during=upgraded)
    assert moved, "the launcher carried its running agent over to the new build"


def test_the_journal_starts_on_a_record_with_a_damaged_row(tmp_path):
    place = tmp_path
    root = place / ".journal"
    journal = [sys.executable, str(CODE / "journal.py"), "--root", str(root)]
    subprocess.run([*journal, "todo", "create", "a row"], cwd=place, capture_output=True, timeout=WAIT)
    (root / "environments" / "main" / "todo" / "002.md").write_text("")
    (root / "environments" / "main" / "todo" / "003.md").write_text('---\n{"n": 3, "title": "odd", "unknown_field": 1}\n---\nbody\n')
    ran = subprocess.run([*journal, "status"], cwd=place, capture_output=True, text=True, timeout=WAIT)
    assert (ran.returncode, "Traceback" in ran.stderr) == (0, False), f"a damaged row stopped the journal:\n{ran.stderr}"
    subprocess.run([*journal, "todo", "all"], cwd=place, capture_output=True, timeout=WAIT)
    notices = subprocess.run([*journal, "notice", "all"], cwd=place, capture_output=True, text=True, timeout=WAIT).stdout
    assert "could not be read" in notices, "a row that cannot be read is named in a notice, never dropped in silence"


def test_an_upgrade_keeps_a_build_a_live_session_runs_from(tmp_path):
    root = installed(tmp_path)
    good = (root / "journal.pyz").resolve()
    old = [root / f"journal-0.0.{i}-old000000{i}.pyz" for i in range(3)]
    for i, build in enumerate(old):
        build.write_bytes(good.read_bytes())
        os.utime(build, (i, i))
    hold_build(root, old[0])
    subprocess.run([sys.executable, str(CODE / "install.py"), "upgrade", str(root.parent)], env={**os.environ, "HOME": str(tmp_path / "home"), "AGENT_JOURNAL_BOOTSTRAPPED": "1"},
                   capture_output=True, timeout=120)
    kept = sorted(build.name for build in root.glob("journal-*.pyz") if build != old[0])
    assert old[0].is_file() and not old[1].is_file() and len(kept) <= 2, f"the build a live process runs from is kept; of the others only the two newest stay: {kept}"


def heals(root: Path, good: Path):
    def healed() -> None:
        began = time.time()
        while (root / "journal.pyz").resolve() != good and time.time() - began < WAIT:
            time.sleep(0.2)
    return healed


def test_a_build_whose_supervisor_dies_on_start_goes_back_to_the_last_good_one(tmp_path):
    place, root = tmp_path, installed(tmp_path)
    good = (root / "journal.pyz").resolve()
    bad = root / "journal-99.0.0-broken0000.pyz"
    with zipfile.ZipFile(good) as source, zipfile.ZipFile(bad, "w") as target:
        for item in source.infolist():
            if not item.filename.startswith("runner/worker."):
                target.writestr(item, source.read(item))
        target.writestr("runner/worker.py", "raise SystemExit(1)\n")
    point(root, bad)
    launches(place, root / "journal.py", "codex", during=heals(root, good))
    assert ((root / "journal.pyz").resolve(), broken(root)) == (good, [bad.name]), "the journal went back to the build that works and remembers the broken one"


def test_a_build_whose_server_dies_on_start_goes_back_to_the_last_good_one(tmp_path):
    place, root = tmp_path, installed(tmp_path)
    good = (root / "journal.pyz").resolve()
    bad = root / "journal-99.0.0-broken0000.pyz"
    with zipfile.ZipFile(good) as source, zipfile.ZipFile(bad, "w") as target:
        for item in source.infolist():
            if not item.filename.startswith("serve."):
                target.writestr(item, source.read(item))
        target.writestr("serve.py", source.read("serve.py").decode().replace("def run(", "def run(*_, **__):\n    raise SystemExit(3)\n\n\ndef unused(", 1))
    point(root, bad)
    launches(place, root / "journal.py", "codex", during=heals(root, good))
    assert ((root / "journal.pyz").resolve(), broken(root)) == (good, [bad.name]), "the journal went back to the build that works and remembers the broken one"


def test_a_killed_server_is_reaped_so_a_new_one_starts(tmp_path, monkeypatch):
    import signal
    from engine import viewer
    from engine.sessions import alive
    root = installed(tmp_path)
    monkeypatch.setenv("HOME", str(tmp_path / "home"))
    starter = subprocess.Popen([sys.executable, "-c", f"import sys, time; sys.path.insert(0, {str(CODE)!r}); from pathlib import Path; from engine import viewer; "
                                f"root = Path({str(root)!r}); print(viewer.launch(root, root.parent)[0], flush=True); time.sleep({WAIT})"],
                               stdout=subprocess.PIPE, text=True)
    try:
        assert starter.stdout.readline().strip(), "the server answers"
        killed = viewer.last(root).pid
        os.kill(killed, signal.SIGKILL)
        began = time.time()
        while alive(killed) and time.time() - began < WAIT / 3:
            time.sleep(0.1)
        assert not alive(killed), "the process that started the server reaps it, so it is not left a zombie that looks alive"
    finally:
        starter.kill()
        starter.wait(WAIT)
    url = viewer.launch(root, root.parent)[0]
    try:
        assert url, "a new server starts where the killed one was"
    finally:
        os.kill(viewer.last(root).pid, signal.SIGTERM)


def test_a_message_shown_while_the_server_is_down_reaches_the_chat_once_it_is_back(tmp_path):
    root = installed(tmp_path)
    env = {**os.environ, "HOME": str(tmp_path / "home"), "AGENT_JOURNAL_ACTIVE": "1", "JOURNAL_ENV": ""}
    journal = [sys.executable, str(root / "journal.py"), "--root", str(root)]
    wired = json.loads((root.parent / ".claude" / "settings.local.json").read_text())["hooks"]["SessionStart"][0]["hooks"][0]["command"]
    hook = lambda body: subprocess.run(["sh", "-c", wired], input=json.dumps(body), text=True, env=env, capture_output=True, timeout=WAIT)
    channel = json.loads((root.parent / ".mcp.json").read_text())["mcpServers"]["journal"]
    assert channel == {"command": sys.executable, "args": [str(root / "journal.py"), "-m", "channel", str(root)]}, \
        "the channel runs the interpreter that installed it, with a path that has a space in it kept whole"
    shown = {"hook_event_name": "MessageDisplay", "session_id": "s1", "message_id": "m1", "index": 0, "final": True, "delta": "said while the server was down"}

    def serving():
        server = subprocess.Popen([*journal, "serve", "--port", "0"], cwd=root.parent, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        began = time.time()
        while time.time() - began < WAIT and not (root / "runtime" / "heartbeat").is_file():
            time.sleep(0.1)
        return server

    server = serving()
    try:
        hook({"hook_event_name": "SessionStart", "session_id": "s1", "cwd": str(root.parent), "source": "startup"})
    finally:
        server.terminate()
        server.wait(WAIT)
    (root / "runtime" / "heartbeat").unlink(missing_ok=True)
    hook(shown)
    assert list((root / "runtime" / "unsent").glob("*.json")), "the display hook keeps what the server could not take"
    server = serving()
    try:
        began, listed = time.time(), ""
        while "said while the server was down" not in listed and time.time() - began < WAIT:
            time.sleep(0.2)
            listed = subprocess.run([*journal, "message", "all"], cwd=root.parent, env=env, capture_output=True, text=True, timeout=WAIT).stdout
    finally:
        server.terminate()
        server.wait(WAIT)
    assert "said while the server was down" in listed, "a message shown while the server was down reaches the chat once it is back"


def test_a_migration_that_fails_leaves_the_record_as_it_was(tmp_path, monkeypatch):
    import threading
    import types
    import migrations
    from engine.stored import append_text, write_text
    root = tmp_path / ".journal"
    (root / "environments" / "main" / "todo").mkdir(parents=True)
    kept = root / "environments" / "main" / "todo" / "001.md"
    added = root / "environments" / "main" / "todo" / "002.md"
    events = root / "environments" / "main" / "events.jsonl"
    started = threading.Event()
    writers = []
    kept.write_text("the user's row")
    touched, broke = types.ModuleType("migrations.m9998_touch"), types.ModuleType("migrations.m9999_break")
    touched.run = lambda r: kept.write_text("changed halfway") or "touched"

    def breaking(r):
        def writing():
            started.set()
            append_text(events, "an event emitted while migrating\n")
            write_text(added, "written while migrating")
        writer = threading.Thread(target=writing)
        writers.append(writer)
        writer.start()
        assert started.wait(timeout=1), "the record write starts during the migration"
        assert writer.is_alive(), "a record write waits for the migration to finish"
        raise RuntimeError("the migration broke")
    broke.run = breaking
    monkeypatch.setitem(sys.modules, "migrations.m9998_touch", touched)
    monkeypatch.setitem(sys.modules, "migrations.m9999_break", broke)
    monkeypatch.setattr(migrations, "names", lambda: ["m9998_touch", "m9999_break"])
    try:
        migrations.run(root)
    except RuntimeError:
        pass
    writers[0].join(timeout=1)
    left = lambda: list(root.glob(f".{migrations.BACKUP}-*"))
    assert (kept.read_text(), added.read_text(), events.read_text(), migrations.applied(root), left()) == \
           ("the user's row", "written while migrating", "an event emitted while migrating\n", {}, []), \
        "a failed migration restores its backup before a waiting record write lands"
    monkeypatch.setattr(migrations, "names", lambda: ["m9998_touch"])
    assert migrations.run(root) == ["m9998_touch"] and not left(), "a run that succeeds deletes its backup"


def test_an_installer_left_with_only_itself_fetches_the_package_and_finishes(tmp_path):
    code = tmp_path / ".journal" / "src"
    code.mkdir(parents=True)
    (code / "install.py").write_bytes((HERE / "install.py").read_bytes())
    ran = subprocess.run([sys.executable, str(code / "install.py"), "finish", str(tmp_path)], cwd=tmp_path, capture_output=True, text=True,
                         timeout=WAIT * 4, env={**os.environ, "AGENT_JOURNAL_REPO": str(HERE)})
    assert ran.returncode == 0, f"an older installer copies only install.py and runs it; it must heal:\n{ran.stderr[-2000:]}"
    assert (code / "engine").is_dir() and (tmp_path / ".journal" / "journal.pyz").is_file(), "the package is back and packed"
import subprocess
import sys
import threading
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path

from controllers.types import Agents, Nudges
from engine.record import Record
from resources.base import AGENT, SYSTEM
from commands.cli import captured  # noqa: F401
from commands.cli import run  # noqa: F401
import commands.http  # noqa: F401
from commands.dispatch import dispatch  # noqa: F401
from commands.launch import asked_for  # noqa: F401
from commands.launch import asked_history  # noqa: F401
from commands.launch import asked_prompts  # noqa: F401
from commands.launch import asked_resume  # noqa: F401
from commands.launch import defaults  # noqa: F401
from commands.queries import ended  # noqa: F401
from features.plans.controller import Plans  # noqa: F401
from features.plugins.manifest import MANIFEST  # noqa: F401
from features.plugins.manifest import read  # noqa: F401
from features.tickets.controller import Tickets  # noqa: F401
from runner import engine as engine_module  # noqa: F401
from runner import engines  # noqa: F401
from runner.engine import Engine  # noqa: F401
from runner.gate import PAUSED  # noqa: F401
from runner.hooks import answer  # noqa: F401
from runner.hooks import handle  # noqa: F401


def report(record, status, event, session="claude-1", **more):
    agents = Agents(record, actor=SYSTEM)
    row = agents.by_session(session)
    uses = int(row.data.get("uses") or 0) + (event == "PreToolUse")
    agents.saw(row.n, {"hook": event, "session": session, "cause": AGENT}, **{**row.data, "uses": uses, **more, "status": status, "event": event, "at": time.time()})


def tick(record, session="claude-1"):
    from runner.engine import emit_clock
    emit_clock(record, session)


def idle(record, **more):
    report(record, "working", "PreToolUse", **more)
    report(record, "idle", "Stop", **more)


def nudges(record):
    return [n.title for n in Nudges(record).all()]


def nudges_with_briefs(record):
    return [(n.title, n.brief) for n in Nudges(record).all()]


def git(where: Path, *args: str) -> str:
    return subprocess.run(["git", *args], cwd=where, check=True, capture_output=True, text=True, timeout=30).stdout.strip()


def commit(where: Path, name: str, text: str) -> str:
    (where / name).write_text(text)
    git(where, "add", name)
    git(where, "commit", "-q", "-m", f"write {name}")
    return git(where, "rev-parse", "HEAD")


@dataclass(frozen=True)
class Repo:
    record: Record
    project: Path


def project_on(branch: str) -> Repo:
    from tests.conftest import fresh
    record = fresh()
    record.root.mkdir(parents=True, exist_ok=True)
    project = record.root.resolve().parent
    git(project, "init", "-q", "-b", "main")
    git(project, "config", "user.email", "t@t")
    git(project, "config", "user.name", "t")
    (project / ".gitignore").write_text("/.journal\n/.claude/worktrees/\n")
    git(project, "add", ".gitignore")
    git(project, "commit", "-q", "-m", "start")
    git(project, "checkout", "-q", "-b", branch)
    commit(project, "shared.txt", "one\n")
    return Repo(record, project)


AUDITED = {"open": "opened", "os.scandir": "scanned", "os.listdir": "scanned"}
ACTIVE = threading.local()
INSTALLED = []


@dataclass
class Work:
    opened: list = field(default_factory=list)
    scanned: list = field(default_factory=list)


def recorded(event: str, args: tuple) -> None:
    work = getattr(ACTIVE, "work", None)
    if work is not None and event in AUDITED:
        getattr(work, AUDITED[event]).append(str(args[0]))


@contextmanager
def counted():
    if not INSTALLED:
        sys.addaudithook(recorded)
        INSTALLED.append(recorded)
    ACTIVE.work = Work()
    try:
        yield ACTIVE.work
    finally:
        ACTIVE.work = None
