eigent/backend/tests/app/workspace_git/test_mutation.py

# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========

from __future__ import annotations

import subprocess
from pathlib import Path

import pytest

from app.run_context import RunContext
from app.run_journal import RunEventDraft, SQLiteRunJournal
from app.workspace_git import (
    ContentRepositoryService,
    GitBackend,
    WorkspaceGitCoordinator,
    WorkspaceGitLifecycle,
    WorkspaceMutationService,
    WorkspaceSnapshotService,
)


@pytest.fixture
def journal(tmp_path):
    with SQLiteRunJournal(tmp_path / "run-journal.sqlite3") as value:
        yield value


def _services(tmp_path: Path, journal: SQLiteRunJournal):
    hooks = tmp_path / "empty-hooks"
    hooks.mkdir(exist_ok=True)
    backend = GitBackend(hooks_path=hooks)
    state_root = tmp_path / "state"
    content = ContentRepositoryService(
        journal,
        state_root=state_root,
        git_backend=backend,
    )
    coordinator = WorkspaceGitCoordinator(
        journal,
        state_root=state_root,
        git_backend=backend,
    )
    snapshots = WorkspaceSnapshotService(
        journal,
        state_root=state_root,
        git_backend=backend,
    )
    mutations = WorkspaceMutationService(
        journal,
        state_root=state_root,
        coordinator=coordinator,
        snapshots=snapshots,
        primary_checkout_enabled=False,
    )
    return content, coordinator, mutations, backend


def _context(space: Path, *, run_id: str = "run-1") -> RunContext:
    return RunContext(
        space_id="space-1",
        project_id="project-1",
        run_id=run_id,
        task_id="task-1",
        email="user@example.com",
        user_id="user-1",
        working_directory=space,
        task_output_root=space,
        camel_log_dir=space / ".logs",
        binding_source="test",
        workdir_mode="direct-write",
        browser_port=9222,
        session_mode="single-agent",
    )


def _admit(
    journal: SQLiteRunJournal,
    coordinator: WorkspaceGitCoordinator,
    *,
    run_id: str = "run-1",
):
    journal.ensure_run(
        run_id=run_id,
        project_id="project-1",
        status="pending",
    )
    admission = coordinator.admit_run(
        space_id="space-1",
        project_id="project-1",
        run_id=run_id,
        task_id="task-1",
        session_mode="single-agent",
    )
    assert admission is not None
    return admission


def _git(repository: Path, *args: str) -> str:
    return subprocess.run(
        ("git", "-C", str(repository), *args),
        check=True,
        capture_output=True,
        text=True,
        env={
            "PATH": "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin",
            "GIT_CONFIG_NOSYSTEM": "1",
            "GIT_CONFIG_GLOBAL": "/dev/null",
            "GIT_TERMINAL_PROMPT": "0",
        },
    ).stdout


def test_file_write_materializes_before_target_is_available(tmp_path, journal):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)

    prepared = mutations.prepare_file_write(
        context=_context(space),
        filename="generated.txt",
        operation_request_id="tool-call-1",
        actor_id="agent-1",
        trigger="filesystem.write",
    )

    assert prepared is not None
    assert prepared.workspace.run.materialization_state == "materialized"
    assert (
        prepared.target_path.parent == prepared.agent_workspace.agent_worktree
    )
    assert not (space / "generated.txt").exists()
    prepared.target_path.write_text("agent output", encoding="utf-8")
    commit = mutations.complete_file_write(
        prepared,
        operation_request_id="tool-call-1",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert commit == backend.current_head(prepared.workspace.run_worktree)
    assert backend.is_worktree_clean(prepared.workspace.run_worktree)
    assert not (space / "generated.txt").exists()
    items = journal.list_git_change_set_items(
        prepared.change_set.change_set_id
    )
    assert len(items) == 1
    assert items[0].relative_path == "generated.txt"
    assert items[0].change_kind == "added"
    assert items[0].item_state == "checkpointed"


def test_overlay_modified_by_agent_commits_exact_user_preimage_first(
    tmp_path,
    journal,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    user_file = space / "draft.txt"
    user_file.write_text("user preimage", encoding="utf-8")
    _admit(journal, coordinator)

    prepared = mutations.prepare_file_write(
        context=_context(space),
        filename="draft.txt",
        operation_request_id="tool-call-overlay",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert prepared is not None
    assert prepared.overlay is not None
    assert prepared.target_path.read_text() == "user preimage"
    prepared.target_path.write_text("agent result", encoding="utf-8")

    commit = mutations.complete_file_write(
        prepared,
        operation_request_id="tool-call-overlay",
        actor_id="agent-1",
        trigger="filesystem.write",
    )

    assert commit is not None
    assert (
        _git(
            prepared.workspace.run_worktree,
            "show",
            f"{commit}^2^:draft.txt",
        )
        == "user preimage"
    )
    assert (
        _git(
            prepared.workspace.run_worktree,
            "show",
            f"{commit}^2:draft.txt",
        )
        == "agent result"
    )
    assert user_file.read_text() == "user preimage"
    items = journal.list_git_change_set_items(
        prepared.change_set.change_set_id
    )
    assert items[0].preimage_digest == prepared.overlay.content_digest
    assert items[0].item_state == "checkpointed"
    entry = journal.get_workspace_overlay_entry(
        prepared.overlay.snapshot.snapshot_id,
        "draft.txt",
    )
    assert entry is not None
    assert entry.entry_state == "agent_modified"


def test_startup_reconciles_crash_after_overlay_preimage_commit(
    tmp_path,
    journal,
    monkeypatch,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    user_file = space / "draft.txt"
    user_file.write_text("user preimage", encoding="utf-8")
    _admit(journal, coordinator)

    prepared = mutations.prepare_file_write(
        context=_context(space),
        filename="draft.txt",
        operation_request_id="tool-call-crash",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert prepared is not None
    assert prepared.overlay is not None
    prepared.target_path.write_text("agent result", encoding="utf-8")
    original_update = journal.update_git_change_set_item_state

    def crash_before_state_commit(**kwargs):
        if (
            kwargs["expected_state"] == "pending"
            and kwargs["state"] == "preimage_checkpointed"
        ):
            raise RuntimeError("simulated crash before state commit")
        return original_update(**kwargs)

    with monkeypatch.context() as patch:
        patch.setattr(
            journal,
            "update_git_change_set_item_state",
            crash_before_state_commit,
        )
        with pytest.raises(RuntimeError, match="simulated crash"):
            mutations.complete_file_write(
                prepared,
                operation_request_id="tool-call-crash",
                actor_id="agent-1",
                trigger="filesystem.write",
            )

    assert prepared.target_path.read_text() == "user preimage"
    before = _git(
        prepared.workspace.run_worktree, "rev-list", "--count", "HEAD"
    )
    reconciliation = mutations.reconcile_startup()
    after = _git(
        prepared.workspace.run_worktree, "rev-list", "--count", "HEAD"
    )

    assert reconciliation.recovered_change_set_ids == (
        prepared.change_set.change_set_id,
    )
    assert reconciliation.needs_attention_change_set_ids == ()
    # The recovered history contains the User preimage checkpoint, Agent
    # result checkpoint, and one serialized merge into Run integration.
    assert int(after) == int(before) + 3
    assert prepared.target_path.read_text() == "agent result"
    assert user_file.read_text() == "user preimage"
    item = journal.list_git_change_set_items(
        prepared.change_set.change_set_id
    )[0]
    assert item.item_state == "checkpointed"
    assert item.operation_request_id == "tool-call-crash"


def test_startup_reconciles_exact_write_before_change_set_item_exists(
    tmp_path,
    journal,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)
    prepared = mutations.prepare_file_write(
        context=_context(space),
        filename="generated.txt",
        operation_request_id="tool-call-before-complete",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert prepared is not None
    prepared.target_path.write_text("durable result", encoding="utf-8")
    assert (
        journal.list_git_change_set_items(prepared.change_set.change_set_id)
        == []
    )

    reconciliation = mutations.reconcile_startup()

    assert reconciliation.recovered_change_set_ids == (
        prepared.change_set.change_set_id,
    )
    assert backend.is_worktree_clean(prepared.workspace.run_worktree)
    assert (
        _git(
            prepared.workspace.run_worktree,
            "show",
            "HEAD:generated.txt",
        )
        == "durable result"
    )
    intents = journal.list_git_mutation_intents()
    assert len(intents) == 1
    assert intents[0].status == "completed"


def test_startup_reconciles_broad_process_before_delta_scan(
    tmp_path,
    journal,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)
    prepared = mutations.prepare_broad_write(
        context=_context(space),
        operation_request_id="terminal-before-scan",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )
    assert prepared is not None
    (prepared.agent_workspace.agent_worktree / "generated.csv").write_text(
        "a,b\n1,2",
        encoding="utf-8",
    )

    reconciliation = mutations.reconcile_startup()

    assert reconciliation.recovered_change_set_ids == (
        prepared.change_set.change_set_id,
    )
    assert backend.is_worktree_clean(prepared.workspace.run_worktree)
    assert (
        _git(
            prepared.workspace.run_worktree,
            "show",
            "HEAD:generated.csv",
        )
        == "a,b\n1,2"
    )
    assert journal.list_git_mutation_intents()[0].status == "completed"


def test_checkpointed_agent_result_recovers_pending_merge_on_startup(
    tmp_path,
    journal,
    monkeypatch,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)
    prepared = mutations.prepare_file_write(
        context=_context(space),
        filename="result.txt",
        operation_request_id="checkpoint-before-merge",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert prepared is not None
    prepared.target_path.write_text("durable", encoding="utf-8")
    with monkeypatch.context() as patch:
        patch.setattr(
            mutations.workforce,
            "merge_agent_workspace",
            lambda *_args, **_kwargs: (_ for _ in ()).throw(
                RuntimeError("simulated crash before merge")
            ),
        )
        with pytest.raises(RuntimeError, match="simulated crash"):
            mutations.complete_file_write(
                prepared,
                operation_request_id="checkpoint-before-merge",
                actor_id="agent-1",
                trigger="filesystem.write",
            )

    reconciliation = mutations.reconcile_startup()

    assert prepared.change_set.change_set_id in (
        reconciliation.recovered_change_set_ids
    )
    assert (
        _git(prepared.workspace.run_worktree, "show", "HEAD:result.txt")
        == "durable"
    )


def test_same_path_can_be_checkpointed_again_without_reimporting_user_source(
    tmp_path,
    journal,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    user_file = space / "draft.txt"
    user_file.write_text("user preimage", encoding="utf-8")
    _admit(journal, coordinator)

    first = mutations.prepare_file_write(
        context=_context(space),
        filename="draft.txt",
        operation_request_id="tool-call-first",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert first is not None and first.overlay is not None
    first.target_path.write_text("agent first", encoding="utf-8")
    mutations.complete_file_write(
        first,
        operation_request_id="tool-call-first",
        actor_id="agent-1",
        trigger="filesystem.write",
    )

    second = mutations.prepare_file_write(
        context=_context(space),
        filename="draft.txt",
        operation_request_id="tool-call-second",
        actor_id="agent-1",
        trigger="filesystem.write",
    )
    assert second is not None
    assert second.overlay is None
    assert second.target_path.read_text() == "agent first"
    second.target_path.write_text("agent second", encoding="utf-8")
    commit = mutations.complete_file_write(
        second,
        operation_request_id="tool-call-second",
        actor_id="agent-1",
        trigger="filesystem.write",
    )

    assert commit is not None
    assert _git(second.workspace.run_worktree, "show", "HEAD^:draft.txt") == (
        "agent first"
    )
    assert _git(second.workspace.run_worktree, "show", "HEAD:draft.txt") == (
        "agent second"
    )
    assert user_file.read_text() == "user preimage"
    item = journal.list_git_change_set_items(second.change_set.change_set_id)[
        0
    ]
    assert item.operation_request_id == "tool-call-second"
    assert item.item_state == "checkpointed"


def test_space_without_git_keeps_legacy_write_target(tmp_path, journal):
    _, _, mutations, _ = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    journal.ensure_run(
        run_id="run-1",
        project_id="project-1",
        status="pending",
    )

    prepared = mutations.prepare_file_write(
        context=_context(space),
        filename="legacy.txt",
        operation_request_id="tool-call-legacy",
        actor_id="agent-1",
        trigger="filesystem.write",
    )

    assert prepared is None


def test_broad_write_imports_only_overlay_paths_already_pinned_by_run(
    tmp_path,
    journal,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    visible = space / "visible.txt"
    hidden = space / "not-read.txt"
    visible.write_text("visible user data", encoding="utf-8")
    hidden.write_text("not admitted", encoding="utf-8")
    _admit(journal, coordinator)
    mutations.snapshots.pin_path(
        run_id="run-1",
        relative_path="visible.txt",
    )

    prepared = mutations.prepare_broad_write(
        context=_context(space),
        operation_request_id="terminal-call-1",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )

    assert prepared is not None
    assert (
        prepared.agent_workspace.agent_worktree / "visible.txt"
    ).read_text() == ("visible user data")
    assert not (
        prepared.agent_workspace.agent_worktree / "not-read.txt"
    ).exists()
    assert [item.relative_path for item in prepared.imported_overlays] == [
        "visible.txt"
    ]

    commits = mutations.complete_broad_write(
        prepared,
        operation_request_id="terminal-call-1",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )
    assert commits == ()
    assert backend.is_worktree_clean(prepared.workspace.run_worktree)
    assert visible.read_text() == "visible user data"

    prepared_again = mutations.prepare_broad_write(
        context=_context(space),
        operation_request_id="terminal-call-retry",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )
    assert prepared_again is not None
    assert (
        prepared_again.agent_workspace.agent_worktree / "visible.txt"
    ).read_text() == "visible user data"


def test_broad_write_renews_lease_for_long_running_process(
    tmp_path,
    journal,
):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)
    prepared = mutations.prepare_broad_write(
        context=_context(space),
        operation_request_id="terminal-call-long-running",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )
    assert prepared is not None
    before = journal.get_git_agent_workspace("run-1", "developer-agent")
    assert before is not None and before.lease_until is not None

    mutations.renew_broad_write(
        prepared,
        now=before.lease_until - 1.0,
    )

    renewed = journal.get_git_agent_workspace("run-1", "developer-agent")
    assert renewed is not None and renewed.lease_until is not None
    assert renewed.lease_token == before.lease_token
    assert renewed.lease_until > before.lease_until
    mutations.complete_broad_write(
        prepared,
        operation_request_id="terminal-call-long-running",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )


def test_broad_write_checkpoints_only_actual_process_delta(tmp_path, journal):
    content, coordinator, mutations, backend = _services(tmp_path, journal)
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)
    prepared = mutations.prepare_broad_write(
        context=_context(space),
        operation_request_id="terminal-call-2",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )
    assert prepared is not None
    (prepared.agent_workspace.agent_worktree / "seed.txt").write_text(
        "updated"
    )
    (prepared.agent_workspace.agent_worktree / "generated.csv").write_text(
        "a,b\n1,2"
    )

    commits = mutations.complete_broad_write(
        prepared,
        operation_request_id="terminal-call-2",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )

    assert len(commits) == 1
    assert backend.is_worktree_clean(prepared.workspace.run_worktree)
    items = journal.list_git_change_set_items(
        prepared.change_set.change_set_id
    )
    assert [
        (item.relative_path, item.change_kind, item.item_state)
        for item in items
    ] == [
        ("generated.csv", "added", "checkpointed"),
        ("seed.txt", "modified", "checkpointed"),
    ]
    assert seed.read_text() == "seed"


def test_primary_checkout_write_commits_in_place_and_records_run_oids(
    tmp_path,
    journal,
):
    content, coordinator, _, backend = _services(tmp_path, journal)
    direct = WorkspaceMutationService(
        journal,
        state_root=tmp_path / "state",
        coordinator=coordinator,
    )
    lifecycle = WorkspaceGitLifecycle(
        journal,
        state_root=tmp_path / "state",
        coordinator=coordinator,
    )
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    base = backend.commit_paths(space, (seed,), message="seed")
    admission = _admit(journal, coordinator)

    prepared = direct.prepare_file_write(
        context=_context(space),
        filename="generated.txt",
        operation_request_id="direct-file-1",
        actor_id="developer-agent",
        trigger="filesystem.write",
    )
    assert prepared is not None
    assert prepared.workspace is None
    assert prepared.agent_workspace is None
    assert prepared.mutation_root == space
    prepared.target_path.write_text("visible now", encoding="utf-8")

    checkpoint = direct.complete_file_write(
        prepared,
        operation_request_id="direct-file-1",
        actor_id="developer-agent",
        trigger="filesystem.write",
    )

    assert checkpoint == backend.current_head(space)
    assert (space / "generated.txt").read_text() == "visible now"
    assert not (tmp_path / "state" / "worktrees").exists()
    message = _git(space, "show", "-s", "--format=%B", checkpoint or "HEAD")
    assert "Eigent-Task-ID: task-1" in message
    assert "Eigent-Run-ID: run-1" in message
    journal.append_event(
        "run-1",
        RunEventDraft(
            event_id="run-1-completed",
            event_type="run.completed",
            payload={"reason": "test"},
        ),
    )

    finalized = lifecycle.finalize_run("run-1")
    persisted = journal.get_run_git_materialization("run-1")

    assert finalized.outcome == "committed_primary"
    assert persisted is not None
    assert persisted.workspace_base_commit == base
    assert persisted.promoted_commit == checkpoint
    assert tuple(
        item.relative_path
        for item in backend.changed_paths_between(
            space,
            base_commit=base,
            target_commit=checkpoint or "HEAD",
        )
    ) == ("generated.txt",)
    assert admission.writer.request.status == "acquired"
    writer = journal.get_workspace_writer_request("workspace-writer:run-1")
    assert writer is not None and writer.status == "released"


def test_primary_checkout_preserves_dirty_user_preimage_before_agent_diff(
    tmp_path,
    journal,
):
    content, coordinator, _, backend = _services(tmp_path, journal)
    direct = WorkspaceMutationService(
        journal,
        state_root=tmp_path / "state",
        coordinator=coordinator,
    )
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    seed.write_text("user draft", encoding="utf-8")
    _admit(journal, coordinator)

    prepared = direct.prepare_file_write(
        context=_context(space),
        filename="generated.txt",
        operation_request_id="direct-file-2",
        actor_id="developer-agent",
        trigger="filesystem.write",
    )
    assert prepared is not None
    preserved_base = backend.current_head(space)
    assert preserved_base is not None
    assert _git(space, "show", f"{preserved_base}:seed.txt") == "user draft"
    prepared.target_path.write_text("agent output", encoding="utf-8")
    terminal = direct.complete_file_write(
        prepared,
        operation_request_id="direct-file-2",
        actor_id="developer-agent",
        trigger="filesystem.write",
    )

    run = journal.get_run_git_materialization("run-1")
    assert run is not None and run.workspace_base_commit == preserved_base
    assert tuple(
        item.relative_path
        for item in backend.changed_paths_between(
            space,
            base_commit=preserved_base,
            target_commit=terminal or "HEAD",
        )
    ) == ("generated.txt",)


def test_primary_checkout_broad_process_commits_only_visible_checkout(
    tmp_path,
    journal,
):
    content, coordinator, _, backend = _services(tmp_path, journal)
    direct = WorkspaceMutationService(
        journal,
        state_root=tmp_path / "state",
        coordinator=coordinator,
    )
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)

    prepared = direct.prepare_broad_write(
        context=_context(space),
        operation_request_id="direct-terminal-1",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )
    assert prepared is not None
    (prepared.mutation_root / "seed.txt").write_text(
        "updated", encoding="utf-8"
    )
    (prepared.mutation_root / "generated.csv").write_text(
        "a,b\n1,2", encoding="utf-8"
    )

    commits = direct.complete_broad_write(
        prepared,
        operation_request_id="direct-terminal-1",
        actor_id="developer-agent",
        trigger="terminal.execute",
    )

    assert len(commits) == 1
    assert backend.is_worktree_clean(space)
    assert _git(space, "show", "HEAD:seed.txt") == "updated"
    assert _git(space, "show", "HEAD:generated.csv") == "a,b\n1,2"
    assert not (tmp_path / "state" / "worktrees").exists()


def test_failed_primary_checkout_run_keeps_a_recovery_ref(tmp_path, journal):
    content, coordinator, _, backend = _services(tmp_path, journal)
    direct = WorkspaceMutationService(
        journal,
        state_root=tmp_path / "state",
        coordinator=coordinator,
    )
    lifecycle = WorkspaceGitLifecycle(
        journal,
        state_root=tmp_path / "state",
        coordinator=coordinator,
    )
    space = tmp_path / "space"
    space.mkdir()
    content.bootstrap(
        space_id="space-1",
        space_root=space,
        allow_init=True,
    )
    seed = space / "seed.txt"
    seed.write_text("seed", encoding="utf-8")
    backend.commit_paths(space, (seed,), message="seed")
    _admit(journal, coordinator)
    prepared = direct.prepare_file_write(
        context=_context(space),
        filename="partial.txt",
        operation_request_id="direct-partial-1",
        actor_id="developer-agent",
        trigger="filesystem.write",
    )
    assert prepared is not None
    prepared.target_path.write_text("recoverable", encoding="utf-8")
    terminal = direct.complete_file_write(
        prepared,
        operation_request_id="direct-partial-1",
        actor_id="developer-agent",
        trigger="filesystem.write",
    )
    journal.append_event(
        "run-1",
        RunEventDraft(
            event_id="run-1-failed",
            event_type="run.failed",
            payload={"reason": "test"},
        ),
    )

    finalized = lifecycle.finalize_run("run-1")

    assert finalized.outcome == "preserved_primary_failed"
    assert finalized.archive_ref is not None
    assert finalized.archive_ref.endswith("/recovery-failed")
    assert backend.ref_oid(space, finalized.archive_ref) == terminal