eigent/backend/app/controller/workspace_git_controller.py
4pmtong 1b4820e54d
Some checks are pending
Pre-commit / pre-commit (push) Waiting to run
Test / Run Web + Local Brain Smoke (push) Waiting to run
Test / Run Frontend Guardrails (push) Waiting to run
Test / Run Python Tests (push) Waiting to run
feat: run tasks on durable project git checkouts
2026-08-21 22:32:52 +08:00

1603 lines
52 KiB
Python

# ========= 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. =========
"""Authenticated Desktop-local Content Repository API."""
from __future__ import annotations
from pathlib import Path, PurePosixPath
from typing import Annotated, Literal
from fastapi import APIRouter, Depends, HTTPException, Query, Response
from pydantic import BaseModel, Field, field_validator
from app.auth import require_local_control_principal
from app.run_journal import (
IdempotencyConflictError,
InvalidRunTransitionError,
OptimisticConcurrencyError,
configured_run_journal_path,
get_default_run_journal,
)
from app.utils.workspace_resolver import get_workspace_resolver
from app.workspace_git import (
AdvancedGitApprovalRequired,
AdvancedGitCommandRejected,
AdvancedGitError,
AdvancedGitOutcomeUnknown,
AdvancedGitService,
ContentRepositoryConsentRequired,
ContentRepositoryError,
ContentRepositoryService,
GitBackendError,
NestedRepositoryError,
NoCheckpointChangesError,
RepositoryStateChangedError,
RunWorkspaceEditService,
WorkspaceGitCoordinator,
WorkspaceSnapshotError,
WorkspaceSnapshotService,
)
from app.workspace_git.backend import RepositoryDiagnostics
router = APIRouter(dependencies=[Depends(require_local_control_principal)])
_PROJECT_CHANGE_MAX_FILES = 500
_PROJECT_CHANGE_MAX_BYTES = 2_000_000
class GitBootstrapBody(BaseModel):
email: str = Field(min_length=1)
user_id: str | int | None = None
allow_init: bool = False
eigent_owned_space: bool = False
class GitCheckpointBody(BaseModel):
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
expected_repo_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
paths: list[str] = Field(min_length=1, max_length=500)
path_sources: dict[str, str]
target_role: Literal["user", "project", "run", "agent"]
target_id: str = Field(min_length=1, max_length=256)
actor_id: str = Field(min_length=1, max_length=200)
trigger: str = Field(min_length=1, max_length=200)
message: str = Field(min_length=1, max_length=500)
workspace_source: Literal["user", "run"] = "user"
run_id: str | None = Field(default=None, max_length=256)
@field_validator("paths")
@classmethod
def validate_relative_paths(cls, paths: list[str]) -> list[str]:
normalized: list[str] = []
for value in paths:
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("Git checkpoint paths must be relative")
normalized.append(path.as_posix())
if len(set(normalized)) != len(normalized):
raise ValueError("Git checkpoint paths must be unique")
return normalized
class GitSavePointBody(BaseModel):
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
expected_repo_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
actor_id: str = Field(min_length=1, max_length=200)
message: str = Field(default="Save progress", min_length=1, max_length=500)
class GitRestoreBody(BaseModel):
email: str = Field(min_length=1)
user_id: str | int | None = None
checkpoint_id: str = Field(
min_length=1,
max_length=128,
pattern=r"^checkpoint_[0-9a-f]{32}$",
)
operation_request_id: str = Field(min_length=1, max_length=128)
expected_repo_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
class GitMaterializeRunBody(BaseModel):
space_id: str = Field(min_length=1, max_length=256)
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
expected_repo_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
expected_project_version: int = Field(ge=0)
expected_project_head: str | None = Field(
default=None,
pattern=r"^[0-9a-f]{40,64}$",
)
class GitPromoteRunBody(BaseModel):
space_id: str = Field(min_length=1, max_length=256)
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
expected_run_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
expected_project_version: int = Field(ge=0)
expected_project_head: str = Field(pattern=r"^[0-9a-f]{40,64}$")
expected_run_head: str = Field(pattern=r"^[0-9a-f]{40,64}$")
class GitRefreshProjectBody(BaseModel):
space_id: str = Field(min_length=1, max_length=256)
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
expected_projection_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
expected_project_version: int = Field(ge=0)
expected_integration_head: str = Field(pattern=r"^[0-9a-f]{40,64}$")
expected_projected_head: str = Field(pattern=r"^[0-9a-f]{40,64}$")
class GitSnapshotBody(BaseModel):
space_id: str = Field(min_length=1, max_length=256)
email: str = Field(min_length=1)
user_id: str | int | None = None
expected_user_working_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
class GitRunWorkspaceEditBody(BaseModel):
space_id: str = Field(min_length=1, max_length=256)
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
editor_session_id: str = Field(min_length=1, max_length=128)
relative_path: str = Field(min_length=1, max_length=4096)
content: str = Field(max_length=4 * 1024 * 1024)
expected_content_digest: str | None = Field(
default=None,
pattern=r"^[0-9a-f]{64}$",
)
actor_id: str = Field(min_length=1, max_length=200)
class AdvancedGitPreviewBody(BaseModel):
email: str = Field(min_length=1)
user_id: str | int | None = None
operation_request_id: str = Field(min_length=1, max_length=128)
argv: list[str] = Field(min_length=1, max_length=128)
class AdvancedGitExecuteBody(AdvancedGitPreviewBody):
expected_repo_state_digest: str | None = Field(
default=None,
pattern=r"^[0-9a-f]{64}$",
)
confirmed_action_digest: str | None = Field(
default=None,
pattern=r"^[0-9a-f]{64}$",
)
actor_id: str = Field(min_length=1, max_length=200)
def _service() -> ContentRepositoryService:
return ContentRepositoryService(
get_default_run_journal(),
state_root=configured_run_journal_path().parent / "workspace-git",
)
def _coordinator() -> WorkspaceGitCoordinator:
service = _service()
return WorkspaceGitCoordinator(
service.journal,
state_root=service.state_root,
git_backend=service.git,
)
def _snapshot_service() -> WorkspaceSnapshotService:
service = _service()
return WorkspaceSnapshotService(
service.journal,
state_root=service.state_root,
git_backend=service.git,
)
def _run_edit_service() -> RunWorkspaceEditService:
service = _service()
return RunWorkspaceEditService(
service.journal,
state_root=service.state_root,
coordinator=_coordinator(),
)
def _advanced_service() -> AdvancedGitService:
service = _service()
return AdvancedGitService(
service.journal,
content=service,
git_backend=service.git,
)
def _binding_root(
*,
space_id: str,
email: str,
user_id: str | int | None,
) -> Path:
binding = get_workspace_resolver().store.get_binding(
email,
space_id,
user_id,
)
if binding is None:
raise HTTPException(
status_code=404,
detail={
"code": "workspace_binding_not_found",
"message": "The Space has no local workspace binding.",
},
)
root = Path(binding.workspace_root).expanduser()
if not root.is_dir():
raise HTTPException(
status_code=409,
detail={
"code": "workspace_binding_unavailable",
"message": "The bound workspace folder is unavailable.",
},
)
return root.resolve()
def _assert_repository_binding(repository, root: Path) -> None:
if Path(repository.root_path).expanduser().resolve() != root:
raise HTTPException(
status_code=409,
detail={
"code": "git_repository_binding_mismatch",
"message": (
"The persisted Content Repository no longer matches the "
"Space binding. Reconciliation is required."
),
},
)
def _git_error(exc: Exception) -> HTTPException:
if isinstance(exc, HTTPException):
return exc
if isinstance(exc, ContentRepositoryConsentRequired):
return HTTPException(
status_code=409,
detail={
"code": "git_init_consent_required",
"message": str(exc),
},
)
if isinstance(exc, WorkspaceSnapshotError):
detail = {
"code": exc.code,
"message": str(exc),
"retryable": exc.retryable,
"refresh_available": exc.refresh_available,
"automatic_retry_limit": exc.automatic_retry_limit,
}
return HTTPException(
status_code=(
404 if exc.code == "workspace_path_not_found" else 409
),
detail=detail,
)
if isinstance(exc, AdvancedGitApprovalRequired):
return HTTPException(
status_code=409,
detail={
"code": exc.code,
"message": str(exc),
"action_digest": exc.action_digest,
},
)
if isinstance(exc, AdvancedGitCommandRejected):
return HTTPException(
status_code=422,
detail={
"code": exc.code,
"reason_code": exc.reason_code,
"message": str(exc),
"remediation": exc.remediation,
"human_interaction_required": (exc.human_interaction_required),
},
)
if isinstance(exc, AdvancedGitOutcomeUnknown):
return HTTPException(
status_code=409,
detail={
"code": exc.code,
"message": str(exc),
"retryable": False,
},
)
if isinstance(exc, AdvancedGitError):
return HTTPException(
status_code=409,
detail={
"code": getattr(exc, "code", "advanced_git_error"),
"message": str(exc),
},
)
if isinstance(exc, RepositoryStateChangedError):
return HTTPException(
status_code=409,
detail={"code": "repo_state_changed", "message": str(exc)},
)
if isinstance(exc, OptimisticConcurrencyError):
return HTTPException(
status_code=409,
detail={"code": "project_git_state_changed", "message": str(exc)},
)
if isinstance(exc, NoCheckpointChangesError):
return HTTPException(
status_code=409,
detail={"code": "git_no_changes", "message": str(exc)},
)
if isinstance(
exc,
(IdempotencyConflictError, InvalidRunTransitionError),
):
return HTTPException(
status_code=409,
detail={"code": "git_operation_conflict", "message": str(exc)},
)
if isinstance(exc, NestedRepositoryError):
return HTTPException(
status_code=409,
detail={
"code": "nested_repository_requires_binding",
"message": str(exc),
},
)
if isinstance(exc, (ValueError, GitBackendError)):
return HTTPException(
status_code=422,
detail={"code": "invalid_git_operation", "message": str(exc)},
)
if isinstance(exc, ContentRepositoryError):
return HTTPException(
status_code=409,
detail={"code": "git_needs_attention", "message": str(exc)},
)
return HTTPException(
status_code=500,
detail={"code": "git_operation_failed", "message": "Git failed"},
)
def _diagnostics_payload(value: RepositoryDiagnostics) -> dict:
return {
"healthy": value.healthy,
"issues": list(value.issues),
"has_submodules": value.has_submodules,
"has_remotes": value.has_remotes,
"repo_state": {
"head_oid": value.state_token.head_oid,
"branch_or_detached_head": (
value.state_token.branch_or_detached_head
),
"index_digest": value.state_token.index_digest,
"operation_state": value.state_token.operation_state,
"digest": value.state_token.digest,
},
}
def _advanced_preview_payload(preview) -> dict:
return {
"classification": preview.classification.operation,
"subcommand": preview.classification.subcommand,
"safety_class": preview.classification.safety_class.value,
"external_side_effect": preview.classification.external_side_effect,
"risk_tags": list(preview.classification.risk_tags),
"action_digest": preview.action_digest,
"effect": preview.effect.value,
"reason": preview.reason,
"requires_confirmation": preview.requires_confirmation,
"display_argv": list(preview.display_argv),
}
@router.get("/spaces/{space_id}/git/status")
async def git_status(
space_id: str,
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
try:
if repository is None:
inspection = service.inspect(root)
return {
"space_id": space_id,
"enabled": False,
"enablement": inspection.enablement,
"consent_required": inspection.consent_required,
"existing_repository": inspection.probe.is_repository,
"nested_in_parent": inspection.probe.nested_in_parent,
"diagnostics": (
_diagnostics_payload(inspection.diagnostics)
if inspection.diagnostics is not None
else None
),
}
_assert_repository_binding(repository, root)
status = service.status(repository.repository_id)
return {
"space_id": space_id,
"enabled": True,
"repository_id": repository.repository_id,
"state": repository.state,
"ownership": repository.ownership,
"version_coverage": repository.version_coverage,
"hooks_mode": repository.hooks_mode,
"managed_paths": list(status.managed_paths),
"pending_managed_paths": list(status.pending_managed_paths),
"pending_managed_paths_truncated": (
status.pending_managed_paths_truncated
),
"diagnostics": _diagnostics_payload(status.diagnostics),
}
except Exception as exc:
raise _git_error(exc) from exc
@router.post("/spaces/{space_id}/git/bootstrap")
async def git_bootstrap(space_id: str, body: GitBootstrapBody):
root = _binding_root(
space_id=space_id,
email=body.email,
user_id=body.user_id,
)
try:
result = _service().bootstrap(
space_id=space_id,
space_root=root,
allow_init=body.allow_init,
eigent_owned_space=body.eigent_owned_space,
)
except Exception as exc:
raise _git_error(exc) from exc
return {
"space_id": space_id,
"repository_id": result.repository.repository_id,
"initialized": result.initialized,
"ownership": result.repository.ownership,
"state": result.repository.state,
"version_coverage": result.repository.version_coverage,
"diagnostics": _diagnostics_payload(result.diagnostics),
}
@router.get("/spaces/{space_id}/git/history")
async def git_history(
space_id: str,
limit: int = Query(50, ge=1, le=200),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, root)
try:
return _advanced_service().history(
repository_id=repository.repository_id,
limit=limit,
)
except Exception as exc:
raise _git_error(exc) from exc
@router.post("/spaces/{space_id}/git/operations:preview")
async def git_advanced_preview(
space_id: str,
body: AdvancedGitPreviewBody,
):
root = _binding_root(
space_id=space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, root)
try:
preview = _advanced_service().preview(
space_id=space_id,
repository_id=repository.repository_id,
argv=tuple(body.argv),
operation_request_id=body.operation_request_id,
)
return _advanced_preview_payload(preview)
except Exception as exc:
raise _git_error(exc) from exc
@router.post("/spaces/{space_id}/git/operations")
async def git_advanced_execute(
space_id: str,
body: AdvancedGitExecuteBody,
):
root = _binding_root(
space_id=space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, root)
try:
return _advanced_service().execute(
space_id=space_id,
repository_id=repository.repository_id,
argv=tuple(body.argv),
operation_request_id=body.operation_request_id,
expected_repo_state_digest=body.expected_repo_state_digest,
confirmed_action_digest=body.confirmed_action_digest,
actor_id=body.actor_id,
)
except Exception as exc:
raise _git_error(exc) from exc
@router.get("/spaces/{space_id}/git/diff")
async def git_diff(
space_id: str,
paths: Annotated[list[str], Query(min_length=1, max_length=500)],
source_commit: str | None = Query(None),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
bound_root = _binding_root(
space_id=space_id,
email=email,
user_id=user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, bound_root)
try:
root = Path(repository.root_path)
diff = service.diff(
repository.repository_id,
paths=tuple(root / path for path in paths),
source_commit=source_commit,
)
return {"repository_id": repository.repository_id, "diff": diff}
except Exception as exc:
raise _git_error(exc) from exc
@router.get("/spaces/{space_id}/git/checkpoints")
async def git_checkpoints(
space_id: str,
limit: int = Query(100, ge=1, le=500),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
bound_root = _binding_root(
space_id=space_id,
email=email,
user_id=user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, bound_root)
checkpoints = service.journal.list_git_checkpoints(
repository.repository_id,
limit=limit,
)
return {
"repository_id": repository.repository_id,
"checkpoints": [
{
"checkpoint_id": item.checkpoint_id,
"target_role": item.target_role,
"target_id": item.target_id,
"commit_oid": item.commit_oid,
"parent_oid": item.parent_oid,
"paths": list(item.paths),
"actor_id": item.actor_id,
"trigger": item.trigger,
"message": item.message,
"created_at": item.created_at,
}
for item in checkpoints
],
}
@router.post("/spaces/{space_id}/git/checkpoints", status_code=201)
async def git_checkpoint(space_id: str, body: GitCheckpointBody):
root = _binding_root(
space_id=space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, root)
worktree_root = None
if body.workspace_source == "run":
if not body.run_id:
raise HTTPException(
status_code=422,
detail={
"code": "run_id_required",
"message": "run_id is required for a Run checkpoint.",
},
)
run = service.journal.get_run_git_materialization(body.run_id)
if (
run is None
or run.repository_id != repository.repository_id
or run.materialization_state != "materialized"
or not run.worktree_path
):
raise HTTPException(
status_code=409,
detail={
"code": "run_workspace_not_materialized",
"message": "The Run workspace is not materialized.",
},
)
worktree_root = Path(run.worktree_path)
try:
checkpoint = service.checkpoint(
repository.repository_id,
operation_request_id=body.operation_request_id,
expected_repo_state_digest=body.expected_repo_state_digest,
paths=tuple((worktree_root or root) / path for path in body.paths),
path_sources=body.path_sources,
target_role=body.target_role,
target_id=body.target_id,
actor_id=body.actor_id,
trigger=body.trigger,
message=body.message,
worktree_root=worktree_root,
)
except Exception as exc:
raise _git_error(exc) from exc
return {
"checkpoint_id": checkpoint.checkpoint_id,
"repository_id": checkpoint.repository_id,
"commit_oid": checkpoint.commit_oid,
"parent_oid": checkpoint.parent_oid,
"paths": list(checkpoint.paths),
"created_at": checkpoint.created_at,
}
@router.post("/spaces/{space_id}/git/save-point", status_code=201)
async def git_save_point(space_id: str, body: GitSavePointBody):
"""Checkpoint the current managed delta and nothing else."""
root = _binding_root(
space_id=space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
if repository is None:
raise HTTPException(status_code=404, detail="Git is not enabled")
_assert_repository_binding(repository, root)
try:
status = service.status(repository.repository_id)
if not status.pending_managed_paths:
raise NoCheckpointChangesError(
"No managed path has a pending change"
)
paths = status.pending_managed_paths
checkpoint = service.checkpoint(
repository.repository_id,
operation_request_id=body.operation_request_id,
expected_repo_state_digest=body.expected_repo_state_digest,
paths=tuple(root / path for path in paths),
path_sources={path: "user_selected" for path in paths},
target_role="user",
target_id=space_id,
actor_id=body.actor_id,
trigger="user.save_point",
message=body.message,
)
except Exception as exc:
raise _git_error(exc) from exc
return {
"checkpoint_id": checkpoint.checkpoint_id,
"repository_id": checkpoint.repository_id,
"commit_oid": checkpoint.commit_oid,
"parent_oid": checkpoint.parent_oid,
"paths": list(checkpoint.paths),
"remaining_managed_changes": status.pending_managed_paths_truncated,
"created_at": checkpoint.created_at,
}
@router.post("/spaces/{space_id}/git/restore", status_code=201)
async def git_restore_candidate(
space_id: str,
body: GitRestoreBody,
):
bound_root = _binding_root(
space_id=space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
checkpoint = service.journal.get_git_checkpoint(body.checkpoint_id)
repository = service.journal.get_space_git_repository(space_id=space_id)
if (
repository is None
or checkpoint is None
or checkpoint.repository_id != repository.repository_id
):
raise HTTPException(status_code=404, detail="Checkpoint not found")
_assert_repository_binding(repository, bound_root)
try:
candidate = service.prepare_restore_candidate(
body.checkpoint_id,
operation_request_id=body.operation_request_id,
expected_repo_state_digest=body.expected_repo_state_digest,
)
except Exception as exc:
raise _git_error(exc) from exc
return {
"checkpoint_id": body.checkpoint_id,
"repository_id": repository.repository_id,
"candidate_ref": candidate.ref_name,
"commit_oid": candidate.commit_oid,
"applied_to_user_worktree": False,
}
def _project_workspace_payload(
project,
run=None,
*,
projection_state_digest: str | None = None,
) -> dict:
return {
"project_id": project.project_id,
"repository_id": project.repository_id,
"state": project.state,
"version": project.version,
"integration_ref": project.integration_ref,
"integration_head": project.integration_head,
"projected_head": project.projected_head,
"freshness": (
"current"
if project.integration_head == project.projected_head
else "stale"
),
"pending_apply": project.pending_apply,
"materialized": project.integration_ref is not None,
"projection_state_digest": projection_state_digest,
"run": (
None
if run is None
else {
"run_id": run.run_id,
"workspace_base_ref": run.workspace_base_ref,
"workspace_base_commit": run.workspace_base_commit,
"materialization_state": run.materialization_state,
"run_ref": run.run_ref,
"promoted_commit": run.promoted_commit,
"version": run.version,
}
),
}
def _project_change_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 HTTPException(status_code=422, detail="Invalid change path")
return path.as_posix()
def _project_change_context(
*,
project_id: str,
space_id: str,
email: str,
user_id: str | None,
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
project = service.journal.get_project_git_state(project_id)
if (
repository is None
or project is None
or project.repository_id != repository.repository_id
):
raise HTTPException(
status_code=404, detail="Project Git state not found"
)
_assert_repository_binding(repository, root)
base_commit = project.last_synced_user_head
if base_commit is None:
runs = sorted(
(
run
for run in service.journal.list_run_git_materializations()
if run.project_id == project_id
and run.repository_id == repository.repository_id
and run.workspace_base_commit is not None
),
key=lambda run: (run.created_at, run.run_id),
)
base_commit = runs[0].workspace_base_commit if runs else None
return service, repository, project, base_commit, project.integration_head
def _git_change_side(
service: ContentRepositoryService,
root: Path,
commit: str,
path: str,
) -> tuple[str | None, int | None]:
oid = service.git.blob_oid_at_path(root, commit, path)
if oid is None:
return None, None
return oid, service.git.object_size(root, oid)
def _read_git_change_side(
service: ContentRepositoryService,
root: Path,
oid: str | None,
size: int | None,
) -> dict:
if oid is None or size is None:
return {
"content": None,
"size": None,
"binary": False,
"too_large": False,
}
if size > _PROJECT_CHANGE_MAX_BYTES:
return {
"content": None,
"size": size,
"binary": False,
"too_large": True,
}
data = (
b""
if size == 0
else service.git.read_blob_range(
root,
oid,
start_offset=0,
max_bytes=size,
)
)
try:
content = data.decode("utf-8")
except UnicodeDecodeError:
content = None
binary = b"\0" in data[:8000] or content is None
return {
"content": None if binary else content,
"size": size,
"binary": binary,
"too_large": False,
}
def _run_change_context(
*,
run_id: str,
space_id: str,
email: str,
user_id: str | None,
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
run = service.journal.get_run_git_materialization(run_id)
if (
repository is None
or run is None
or run.repository_id != repository.repository_id
):
raise HTTPException(status_code=404, detail="Run Git state not found")
_assert_repository_binding(repository, root)
canonical_run = service.journal.get_run(run_id)
terminal = canonical_run is not None and canonical_run.status in {
"completed",
"failed",
"cancelled",
}
if not terminal and (
run.workspace_base_commit is None or run.promoted_commit is None
):
raise HTTPException(
status_code=409,
detail="Run changes are not finalized yet",
)
return (
service,
repository,
run,
run.workspace_base_commit,
run.promoted_commit,
)
def _git_changes_payload(
*,
service: ContentRepositoryService,
repository,
identity: dict,
base_commit: str | None,
target_commit: str | None,
) -> dict:
if base_commit is None or target_commit is None:
return {
"repository_id": repository.repository_id,
**identity,
"base_commit": base_commit,
"target_commit": target_commit,
"files": [],
"totals": {"added": 0, "removed": 0},
"truncated": False,
}
root = Path(repository.root_path)
changes = service.git.changed_paths_between(
root,
base_commit=base_commit,
target_commit=target_commit,
)
stats = {
item.relative_path: item
for item in service.git.path_line_stats_between(
root,
base_commit=base_commit,
target_commit=target_commit,
)
}
visible_changes = changes[:_PROJECT_CHANGE_MAX_FILES]
visible_paths = tuple(item.relative_path for item in visible_changes)
before_blobs = {
item.relative_path: item
for item in service.git.blobs_at_paths(
root, base_commit, visible_paths
)
}
after_blobs = {
item.relative_path: item
for item in service.git.blobs_at_paths(
root, target_commit, visible_paths
)
}
files = []
for change in visible_changes:
before_blob = before_blobs.get(change.relative_path)
after_blob = after_blobs.get(change.relative_path)
line_stat = stats.get(change.relative_path)
files.append(
{
"path": change.relative_path,
"status": {
"A": "added",
"M": "modified",
"D": "deleted",
"T": "modified",
}[change.status],
"before_size": (
before_blob.size_bytes if before_blob else None
),
"after_size": (after_blob.size_bytes if after_blob else None),
"binary": (
line_stat is not None and line_stat.added_lines is None
),
"added_lines": (line_stat.added_lines if line_stat else None),
"removed_lines": (
line_stat.removed_lines if line_stat else None
),
}
)
text_stats = [
item for item in stats.values() if item.added_lines is not None
]
return {
"repository_id": repository.repository_id,
**identity,
"base_commit": base_commit,
"target_commit": target_commit,
"files": files,
"totals": {
"added": sum(item.added_lines or 0 for item in text_stats),
"removed": sum(item.removed_lines or 0 for item in text_stats),
},
"truncated": len(changes) > _PROJECT_CHANGE_MAX_FILES,
}
def _git_change_content_payload(
*,
service: ContentRepositoryService,
repository,
identity: dict,
path: str,
base_commit: str,
target_commit: str,
) -> dict:
relative_path = _project_change_path(path)
root = Path(repository.root_path)
changed_paths = {
item.relative_path
for item in service.git.changed_paths_between(
root,
base_commit=base_commit,
target_commit=target_commit,
)
}
if relative_path not in changed_paths:
raise HTTPException(status_code=404, detail="Change not found")
before_oid, before_size = _git_change_side(
service, root, base_commit, relative_path
)
after_oid, after_size = _git_change_side(
service, root, target_commit, relative_path
)
return {
**identity,
"path": relative_path,
"base_commit": base_commit,
"target_commit": target_commit,
"before": _read_git_change_side(
service, root, before_oid, before_size
),
"after": _read_git_change_side(service, root, after_oid, after_size),
}
def _snapshot_payload(snapshot) -> dict:
return {
"snapshot_id": snapshot.snapshot_id,
"run_id": snapshot.run_id,
"project_id": snapshot.project_id,
"repository_id": snapshot.repository_id,
"generation": snapshot.generation,
"project_base_commit": snapshot.project_base_commit,
"project_state_version": snapshot.project_state_version,
"user_head": snapshot.user_head,
"user_working_state_digest": snapshot.user_working_state_digest,
"overlay_manifest_digest": snapshot.overlay_manifest_digest,
"state": snapshot.state,
"created_at": snapshot.created_at,
"updated_at": snapshot.updated_at,
}
def _assert_snapshot_owner(
*,
run_id: str,
space_id: str,
email: str,
user_id: str | int | None,
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
run = service.journal.get_run_git_materialization(run_id)
if (
repository is None
or run is None
or run.repository_id != repository.repository_id
):
raise HTTPException(status_code=404, detail="Run Git state not found")
_assert_repository_binding(repository, root)
return repository
@router.get("/projects/{project_id}/git/workspace")
async def project_git_workspace(
project_id: str,
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
project = service.journal.get_project_git_state(project_id)
if (
repository is None
or project is None
or project.repository_id != repository.repository_id
):
raise HTTPException(
status_code=404, detail="Project Git state not found"
)
_assert_repository_binding(repository, root)
projection_digest = (
service.git.repo_state_token(Path(project.worktree_path)).digest
if project.worktree_path
else None
)
return _project_workspace_payload(
project,
projection_state_digest=projection_digest,
)
@router.get("/projects/{project_id}/git/changes")
async def project_git_changes(
project_id: str,
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
"""List authoritative Project changes without reading file contents."""
service, repository, _project, base_commit, target_commit = (
_project_change_context(
project_id=project_id,
space_id=space_id,
email=email,
user_id=user_id,
)
)
try:
return _git_changes_payload(
service=service,
repository=repository,
identity={"project_id": project_id},
base_commit=base_commit,
target_commit=target_commit,
)
except Exception as exc:
raise _git_error(exc) from exc
@router.get("/projects/{project_id}/git/changes/content")
async def project_git_change_content(
project_id: str,
path: str = Query(..., min_length=1, max_length=4096),
base_commit: str = Query(..., min_length=1),
target_commit: str = Query(..., min_length=1),
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
"""Read one changed file's before/after text from pinned Git commits."""
service, repository, _project, current_base, current_target = (
_project_change_context(
project_id=project_id,
space_id=space_id,
email=email,
user_id=user_id,
)
)
if current_base != base_commit or current_target != target_commit:
raise HTTPException(
status_code=409,
detail="Project changed; refresh the change review",
)
try:
return _git_change_content_payload(
service=service,
repository=repository,
identity={},
path=path,
base_commit=base_commit,
target_commit=target_commit,
)
except HTTPException:
raise
except Exception as exc:
raise _git_error(exc) from exc
@router.get("/runs/{run_id}/git/changes")
async def run_git_changes(
run_id: str,
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
"""List one finalized Run's authoritative Git changes lazily."""
service, repository, run, base_commit, target_commit = _run_change_context(
run_id=run_id,
space_id=space_id,
email=email,
user_id=user_id,
)
try:
return _git_changes_payload(
service=service,
repository=repository,
identity={"run_id": run_id, "project_id": run.project_id},
base_commit=base_commit,
target_commit=target_commit,
)
except Exception as exc:
raise _git_error(exc) from exc
@router.get("/runs/{run_id}/git/changes/content")
async def run_git_change_content(
run_id: str,
path: str = Query(..., min_length=1, max_length=4096),
base_commit: str = Query(..., min_length=1),
target_commit: str = Query(..., min_length=1),
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
"""Read one finalized Run change from its pinned Git commits."""
service, repository, run, current_base, current_target = (
_run_change_context(
run_id=run_id,
space_id=space_id,
email=email,
user_id=user_id,
)
)
if current_base != base_commit or current_target != target_commit:
raise HTTPException(
status_code=409,
detail="Run changed; refresh the change review",
)
try:
return _git_change_content_payload(
service=service,
repository=repository,
identity={"run_id": run_id, "project_id": run.project_id},
path=path,
base_commit=base_commit,
target_commit=target_commit,
)
except HTTPException:
raise
except Exception as exc:
raise _git_error(exc) from exc
@router.get("/runs/{run_id}/git/workspace")
async def run_git_workspace(
run_id: str,
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
root = _binding_root(space_id=space_id, email=email, user_id=user_id)
service = _service()
repository = service.journal.get_space_git_repository(space_id=space_id)
run = service.journal.get_run_git_materialization(run_id)
project = (
service.journal.get_project_git_state(run.project_id)
if run is not None
else None
)
if (
repository is None
or run is None
or project is None
or run.repository_id != repository.repository_id
):
raise HTTPException(status_code=404, detail="Run Git state not found")
_assert_repository_binding(repository, root)
projection_digest = (
service.git.repo_state_token(Path(project.worktree_path)).digest
if project.worktree_path
else None
)
return _project_workspace_payload(
project,
run,
projection_state_digest=projection_digest,
)
@router.post("/runs/{run_id}/git/workspace/files:save", status_code=201)
async def save_run_git_workspace_file(
run_id: str,
body: GitRunWorkspaceEditBody,
):
_assert_snapshot_owner(
run_id=run_id,
space_id=body.space_id,
email=body.email,
user_id=body.user_id,
)
try:
result = _run_edit_service().save_text(
run_id=run_id,
relative_path=body.relative_path,
content=body.content,
operation_request_id=body.operation_request_id,
editor_session_id=body.editor_session_id,
actor_id=body.actor_id,
expected_content_digest=body.expected_content_digest,
)
except Exception as exc:
raise _git_error(exc) from exc
return {
"run_id": result.run_id,
"relative_path": result.relative_path,
"content_digest": result.content_digest,
"checkpoint_id": result.checkpoint.checkpoint_id,
"commit_oid": result.checkpoint.commit_oid,
"created_at": result.checkpoint.created_at,
}
@router.get("/runs/{run_id}/git/snapshot")
async def run_git_snapshot(
run_id: str,
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
_assert_snapshot_owner(
run_id=run_id,
space_id=space_id,
email=email,
user_id=user_id,
)
snapshot = _snapshot_service().get_snapshot(run_id)
return {
"run_id": run_id,
"materialized": snapshot is not None,
"snapshot": (
_snapshot_payload(snapshot) if snapshot is not None else None
),
}
@router.post("/runs/{run_id}/git/snapshot:refresh")
async def refresh_run_git_snapshot(
run_id: str,
body: GitSnapshotBody,
):
_assert_snapshot_owner(
run_id=run_id,
space_id=body.space_id,
email=body.email,
user_id=body.user_id,
)
try:
snapshot = _snapshot_service().refresh_snapshot(
run_id,
expected_user_working_state_digest=(
body.expected_user_working_state_digest
),
)
except Exception as exc:
raise _git_error(exc) from exc
return {"snapshot": _snapshot_payload(snapshot)}
@router.get("/runs/{run_id}/git/snapshot/files")
async def read_run_git_snapshot_file(
run_id: str,
path: str = Query(..., min_length=1, max_length=4096),
start_offset: int = Query(0, ge=0),
max_bytes: int = Query(256 * 1024, ge=1, le=4 * 1024 * 1024),
space_id: str = Query(..., min_length=1),
email: str = Query(..., min_length=1),
user_id: str | None = Query(None),
):
_assert_snapshot_owner(
run_id=run_id,
space_id=space_id,
email=email,
user_id=user_id,
)
try:
result = _snapshot_service().read_range(
run_id=run_id,
relative_path=path,
start_offset=start_offset,
max_bytes=max_bytes,
)
except Exception as exc:
raise _git_error(exc) from exc
headers = {
"Accept-Ranges": "bytes",
"X-Eigent-Snapshot-Id": result.snapshot.snapshot_id,
"X-Eigent-Snapshot-Source": result.source_kind,
"X-Content-SHA256": result.content_digest,
}
if result.end_offset > result.start_offset:
headers["Content-Range"] = (
f"bytes {result.start_offset}-{result.end_offset - 1}/"
f"{result.size_bytes}"
)
else:
headers["Content-Range"] = f"bytes */{result.size_bytes}"
return Response(
content=result.content,
status_code=206,
media_type="application/octet-stream",
headers=headers,
)
@router.post("/runs/{run_id}/git/workspace:materialize")
async def materialize_run_git_workspace(
run_id: str,
body: GitMaterializeRunBody,
):
root = _binding_root(
space_id=body.space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(
space_id=body.space_id
)
run = service.journal.get_run_git_materialization(run_id)
if (
repository is None
or run is None
or run.repository_id != repository.repository_id
):
raise HTTPException(status_code=404, detail="Run Git state not found")
_assert_repository_binding(repository, root)
try:
workspace = _coordinator().ensure_run_materialized(
run_id=run_id,
operation_request_id=body.operation_request_id,
expected_repo_state_digest=body.expected_repo_state_digest,
expected_project_version=body.expected_project_version,
expected_project_head=body.expected_project_head,
)
except Exception as exc:
raise _git_error(exc) from exc
return _project_workspace_payload(
workspace.project,
workspace.run,
projection_state_digest=_coordinator()
.git.repo_state_token(workspace.project_worktree)
.digest,
)
@router.post("/runs/{run_id}/git/workspace:promote")
async def promote_run_git_workspace(
run_id: str,
body: GitPromoteRunBody,
):
root = _binding_root(
space_id=body.space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(
space_id=body.space_id
)
run = service.journal.get_run_git_materialization(run_id)
if (
repository is None
or run is None
or run.repository_id != repository.repository_id
):
raise HTTPException(status_code=404, detail="Run Git state not found")
_assert_repository_binding(repository, root)
try:
workspace = _coordinator().promote_run(
run_id=run_id,
operation_request_id=body.operation_request_id,
expected_run_state_digest=body.expected_run_state_digest,
expected_project_version=body.expected_project_version,
expected_project_head=body.expected_project_head,
expected_run_head=body.expected_run_head,
)
except Exception as exc:
raise _git_error(exc) from exc
return _project_workspace_payload(
workspace.project,
workspace.run,
projection_state_digest=_coordinator()
.git.repo_state_token(workspace.project_worktree)
.digest,
)
@router.post("/projects/{project_id}/git/workspace:refresh")
async def refresh_project_git_workspace(
project_id: str,
body: GitRefreshProjectBody,
):
root = _binding_root(
space_id=body.space_id,
email=body.email,
user_id=body.user_id,
)
service = _service()
repository = service.journal.get_space_git_repository(
space_id=body.space_id
)
project = service.journal.get_project_git_state(project_id)
if (
repository is None
or project is None
or project.repository_id != repository.repository_id
):
raise HTTPException(
status_code=404, detail="Project Git state not found"
)
_assert_repository_binding(repository, root)
coordinator = _coordinator()
try:
project = coordinator.refresh_project_projection(
project_id=project_id,
operation_request_id=body.operation_request_id,
expected_projection_state_digest=(
body.expected_projection_state_digest
),
expected_project_version=body.expected_project_version,
expected_integration_head=body.expected_integration_head,
expected_projected_head=body.expected_projected_head,
)
except Exception as exc:
raise _git_error(exc) from exc
if not project.worktree_path:
raise HTTPException(status_code=409, detail="Project worktree missing")
return _project_workspace_payload(
project,
projection_state_digest=coordinator.git.repo_state_token(
Path(project.worktree_path)
).digest,
)