eigent/backend/app/workspace_git/edit.py

"""Durable user edits inside a private Run integration worktree."""

from __future__ import annotations

import hashlib
import os
import time
import uuid
from dataclasses import dataclass
from pathlib import Path, PurePosixPath

from app.run_journal import (
    GitCheckpointRecord,
    SQLiteRunJournal,
    configured_run_journal_path,
    get_default_run_journal,
)
from app.workspace_config import canonical_digest
from app.workspace_git.content import (
    ContentRepositoryError,
    ContentRepositoryService,
    RepositoryStateChangedError,
)
from app.workspace_git.coordinator import WorkspaceGitCoordinator


@dataclass(frozen=True)
class RunWorkspaceEditResult:
    run_id: str
    relative_path: str
    content_digest: str
    checkpoint: GitCheckpointRecord


class RunWorkspaceEditService:
    """Serialize UI edits with Agent merges and checkpoint before success."""

    def __init__(
        self,
        journal: SQLiteRunJournal,
        *,
        state_root: Path,
        coordinator: WorkspaceGitCoordinator | None = None,
        max_content_bytes: int = 4 * 1024 * 1024,
    ) -> None:
        if max_content_bytes < 1:
            raise ValueError("Run workspace edit limit must be positive")
        self.journal = journal
        self.state_root = state_root.expanduser().resolve()
        self.coordinator = coordinator or WorkspaceGitCoordinator(
            journal,
            state_root=self.state_root,
        )
        self.git = self.coordinator.git
        self.content: ContentRepositoryService = self.coordinator.content
        self.max_content_bytes = max_content_bytes

    def save_text(
        self,
        *,
        run_id: str,
        relative_path: str,
        content: str,
        operation_request_id: str,
        editor_session_id: str,
        actor_id: str,
        expected_content_digest: str | None = None,
        now: float | None = None,
    ) -> RunWorkspaceEditResult:
        timestamp = now if now is not None else time.time()
        if not editor_session_id.strip():
            raise ValueError("Editor session id is required")
        encoded = content.encode("utf-8")
        if len(encoded) > self.max_content_bytes:
            raise ValueError(
                "Run workspace text edit exceeds the bounded save limit"
            )
        normalized = self._relative_path(relative_path)
        desired_digest = hashlib.sha256(encoded).hexdigest()
        run = self.journal.get_run_git_materialization(run_id)
        if (
            run is None
            or run.materialization_state != "materialized"
            or not run.worktree_path
        ):
            raise ContentRepositoryError(
                "Run workspace must be materialized before editing"
            )
        repository = self.journal.get_git_repository(run.repository_id)
        if repository is None:
            raise ContentRepositoryError("Run repository is unavailable")
        root = Path(run.worktree_path).expanduser().resolve()
        target = root / normalized
        self._assert_safe_target(root, target)
        operation_id = (
            "gitop_"
            + canonical_digest(
                {
                    "repository_id": run.repository_id,
                    "request_id": operation_request_id,
                }
            )[:32]
        )
        payload = {
            "run_id": run_id,
            "relative_path": normalized,
            "content_digest": desired_digest,
            "expected_content_digest": expected_content_digest,
            "editor_session_id": editor_session_id,
            "actor_id": actor_id,
        }
        lock_path = self.content.repository_lock_path(repository.space_id)
        with self.content.repository_lock(lock_path):
            before = self.git.repo_state_token(root)
            existing_operation = self.journal.get_git_operation(operation_id)
            persisted_expected_digest = (
                existing_operation.expected_repo_state_digest
                if existing_operation is not None
                else before.digest
            )
            operation = self.journal.begin_git_operation(
                operation_id=operation_id,
                repository_id=run.repository_id,
                request_id=operation_request_id,
                operation_type="run.workspace_edit",
                payload_digest=canonical_digest(payload),
                expected_repo_state_digest=persisted_expected_digest,
            )
            if operation.status == "completed":
                return self._completed_result(
                    operation.result,
                    run_id=run_id,
                    relative_path=normalized,
                    content_digest=desired_digest,
                )
            current_digest = self._digest_file(target)
            if operation.status == "prepared":
                if current_digest != expected_content_digest:
                    self.journal.fail_git_operation(
                        operation_id,
                        error_code="workspace_content_changed",
                        error_message=(
                            "Run workspace content changed before save"
                        ),
                    )
                    raise RepositoryStateChangedError(
                        "Run workspace file changed; refresh before saving"
                    )
                self.journal.mark_git_operation_dispatched(
                    operation_id,
                    observed_repo_state_digest=before.digest,
                    now=timestamp,
                )
            elif operation.status != "dispatched":
                raise ContentRepositoryError(
                    f"Run workspace edit is {operation.status!r}"
                )
            if current_digest != desired_digest:
                self._atomic_replace(target, encoded)
            checkpoint_request_id = f"{operation_request_id}:checkpoint"
            try:
                checkpoint = self.content.checkpoint(
                    run.repository_id,
                    operation_request_id=checkpoint_request_id,
                    expected_repo_state_digest=self.git.repo_state_token(
                        root
                    ).digest,
                    paths=(target,),
                    path_sources={normalized: "user_selected"},
                    target_role="run",
                    target_id=run_id,
                    actor_id=actor_id,
                    trigger="run_workspace.user_edit",
                    message=f"Save Run workspace edit {normalized}",
                    worktree_root=root,
                    repository_lock_held=True,
                )
            except Exception:
                # The edit operation stays dispatched. Retrying the same
                # request resumes from the intended content and the
                # checkpoint operation is independently idempotent.
                raise
            result = {
                "run_id": run_id,
                "relative_path": normalized,
                "content_digest": desired_digest,
                "checkpoint_id": checkpoint.checkpoint_id,
                "commit_oid": checkpoint.commit_oid,
            }
            self.journal.complete_git_operation(
                operation_id,
                result=result,
                observed_repo_state_digest=self.git.repo_state_token(
                    root
                ).digest,
                now=timestamp,
            )
        return RunWorkspaceEditResult(
            run_id=run_id,
            relative_path=normalized,
            content_digest=desired_digest,
            checkpoint=checkpoint,
        )

    def _completed_result(
        self,
        result: dict | None,
        *,
        run_id: str,
        relative_path: str,
        content_digest: str,
    ) -> RunWorkspaceEditResult:
        checkpoint_id = str((result or {}).get("checkpoint_id") or "")
        checkpoint = self.journal.get_git_checkpoint(checkpoint_id)
        if checkpoint is None:
            raise ContentRepositoryError(
                "Completed Run workspace edit has no checkpoint"
            )
        return RunWorkspaceEditResult(
            run_id=run_id,
            relative_path=relative_path,
            content_digest=content_digest,
            checkpoint=checkpoint,
        )

    @staticmethod
    def _relative_path(value: str) -> str:
        path = PurePosixPath(value)
        if (
            not value
            or path.is_absolute()
            or ".." in path.parts
            or value.startswith(("~/", "\\\\"))
            or (len(value) > 1 and value[1] == ":")
        ):
            raise ValueError("Run workspace edit path must be relative")
        return path.as_posix()

    @staticmethod
    def _assert_safe_target(root: Path, target: Path) -> None:
        resolved = target.resolve(strict=False)
        try:
            resolved.relative_to(root)
        except ValueError as exc:
            raise ContentRepositoryError(
                "Run workspace edit target escapes its worktree"
            ) from exc

    @staticmethod
    def _digest_file(path: Path) -> str | None:
        if not path.is_file():
            return None
        digest = hashlib.sha256()
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
        return digest.hexdigest()

    @staticmethod
    def _atomic_replace(target: Path, content: bytes) -> None:
        target.parent.mkdir(parents=True, exist_ok=True)
        temporary = target.with_name(f".{target.name}.{uuid.uuid4().hex}.tmp")
        try:
            with temporary.open("wb") as handle:
                handle.write(content)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, target)
        finally:
            temporary.unlink(missing_ok=True)


def get_default_run_workspace_edit_service() -> RunWorkspaceEditService:
    return RunWorkspaceEditService(
        get_default_run_journal(),
        state_root=configured_run_journal_path().parent / "workspace-git",
    )