mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-27 17:41:56 +00:00
263 lines
8 KiB
Python
263 lines
8 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. =========
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import threading
|
|
import weakref
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Literal
|
|
|
|
import httpx
|
|
|
|
from app.run_context import RunContext, get_current_run_context
|
|
from app.service.task import get_task_lock_if_exists
|
|
|
|
logger = logging.getLogger("space_overlay")
|
|
|
|
HASH_CHUNK_SIZE = 1024 * 1024
|
|
|
|
_PATH_LOCKS: weakref.WeakValueDictionary[
|
|
tuple[str, str, str, str], threading.Lock
|
|
] = weakref.WeakValueDictionary()
|
|
_PATH_LOCKS_GUARD = threading.Lock()
|
|
_OVERLAY_SYNC_FAILURES = 0
|
|
_OVERLAY_SYNC_FAILURES_GUARD = threading.Lock()
|
|
|
|
|
|
def normalize_server_url(server_url: str | None) -> str:
|
|
if not server_url:
|
|
return ""
|
|
trimmed = server_url.rstrip("/")
|
|
if trimmed.endswith("/api/v1"):
|
|
return trimmed
|
|
return f"{trimmed}/api/v1"
|
|
|
|
|
|
def sha256_of_file(path: Path) -> str | None:
|
|
if not path.exists():
|
|
return None
|
|
if path.is_symlink() or not path.is_file():
|
|
raise ValueError(f"Cannot hash non-regular file: {path}")
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(HASH_CHUNK_SIZE):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def normalize_relative_path(path: str) -> str:
|
|
normalized = PurePosixPath(path.replace("\\", "/"))
|
|
if (
|
|
not normalized.parts
|
|
or normalized.is_absolute()
|
|
or ".." in normalized.parts
|
|
):
|
|
raise ValueError("Invalid relative path")
|
|
return str(normalized)
|
|
|
|
|
|
def path_write_lock(
|
|
space_id: str,
|
|
project_id: str,
|
|
run_id: str,
|
|
rel_path: str,
|
|
) -> threading.Lock:
|
|
"""Return the per-run/path writer lock.
|
|
|
|
The lock cache is weakly held to avoid unbounded growth. Callers must keep
|
|
the returned lock strongly referenced for the whole critical section,
|
|
preferably as `with path_write_lock(...):`.
|
|
"""
|
|
|
|
key = (space_id, project_id, run_id, rel_path)
|
|
with _PATH_LOCKS_GUARD:
|
|
lock = _PATH_LOCKS.get(key)
|
|
if lock is None:
|
|
lock = threading.Lock()
|
|
_PATH_LOCKS[key] = lock
|
|
return lock
|
|
|
|
|
|
def overlay_sync_failure_count() -> int:
|
|
with _OVERLAY_SYNC_FAILURES_GUARD:
|
|
return _OVERLAY_SYNC_FAILURES
|
|
|
|
|
|
def _record_overlay_sync_failure(
|
|
*,
|
|
reason: str,
|
|
context: RunContext,
|
|
rel_path: str,
|
|
error_message: str,
|
|
) -> None:
|
|
global _OVERLAY_SYNC_FAILURES
|
|
with _OVERLAY_SYNC_FAILURES_GUARD:
|
|
_OVERLAY_SYNC_FAILURES += 1
|
|
failure_count = _OVERLAY_SYNC_FAILURES
|
|
logger.error(
|
|
"space_overlay_sync_failed",
|
|
extra={
|
|
"overlay_reason": reason,
|
|
"overlay_space_id": context.space_id,
|
|
"overlay_project_id": context.project_id,
|
|
"overlay_run_id": context.run_id,
|
|
"overlay_path": rel_path,
|
|
"overlay_failure_count": failure_count,
|
|
"overlay_error_message": error_message,
|
|
},
|
|
)
|
|
|
|
|
|
def run_context_for_task(api_task_id: str) -> RunContext | None:
|
|
context = get_current_run_context()
|
|
if context is not None:
|
|
return context
|
|
task_lock = get_task_lock_if_exists(api_task_id)
|
|
return getattr(task_lock, "run_context", None) if task_lock else None
|
|
|
|
|
|
def relative_to_workdir(
|
|
context: RunContext, path: str | Path
|
|
) -> tuple[str, Path] | None:
|
|
workdir = context.working_directory.expanduser().resolve()
|
|
target = Path(path).expanduser()
|
|
if not target.is_absolute():
|
|
target = workdir / target
|
|
target = target.resolve()
|
|
try:
|
|
rel = target.relative_to(workdir)
|
|
except ValueError:
|
|
return None
|
|
return normalize_relative_path(rel.as_posix()), target
|
|
|
|
|
|
def relative_to_artifact_root(
|
|
context: RunContext, path: str | Path
|
|
) -> str | None:
|
|
"""Return portable identity only for paths owned by the active Run.
|
|
|
|
Artifact finalization scans the task output root first and the workspace
|
|
root second. Mirror that boundary here so realtime write events can carry
|
|
the same relative identity without exposing an absolute local path as
|
|
portable identity.
|
|
"""
|
|
|
|
target = Path(path).expanduser()
|
|
if not target.is_absolute():
|
|
target = context.working_directory.expanduser().resolve() / target
|
|
target = target.resolve()
|
|
roots = (
|
|
context.task_output_root.expanduser().resolve(),
|
|
context.working_directory.expanduser().resolve(),
|
|
)
|
|
seen_roots: set[Path] = set()
|
|
for root in roots:
|
|
if root in seen_roots:
|
|
continue
|
|
seen_roots.add(root)
|
|
try:
|
|
relative_path = target.relative_to(root).as_posix()
|
|
return normalize_relative_path(relative_path)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def should_record_overlay(context: RunContext, target: Path) -> bool:
|
|
if not context.server_url or not context.auth_header:
|
|
return False
|
|
if context.workdir_mode in {"direct-write", "artifact-only"}:
|
|
return False
|
|
if (
|
|
context.working_directory.resolve()
|
|
== context.task_output_root.resolve()
|
|
):
|
|
return False
|
|
try:
|
|
target.relative_to(context.task_output_root.expanduser().resolve())
|
|
return False
|
|
except ValueError:
|
|
return True
|
|
|
|
|
|
def post_overlay_write(
|
|
context: RunContext,
|
|
rel_path: str,
|
|
target_path: Path,
|
|
*,
|
|
base_hash: str | None,
|
|
status: Literal["added", "modified", "deleted"],
|
|
file_hash: str | None = None,
|
|
size: int | None = None,
|
|
mode: int | None = None,
|
|
) -> bool:
|
|
if not should_record_overlay(context, target_path):
|
|
return True
|
|
|
|
server_url = normalize_server_url(context.server_url)
|
|
if not server_url:
|
|
return True
|
|
|
|
if status == "deleted":
|
|
file_hash = None
|
|
elif file_hash is None:
|
|
file_hash = sha256_of_file(target_path)
|
|
if (size is None or mode is None) and target_path.exists():
|
|
stat_result = target_path.stat()
|
|
size = stat_result.st_size if size is None else size
|
|
mode = stat_result.st_mode if mode is None else mode
|
|
payload = {
|
|
"run_id": context.run_id,
|
|
"path": rel_path,
|
|
"status": status,
|
|
"hash": file_hash,
|
|
"base_hash": base_hash,
|
|
"base_snapshot_id": context.extra_env.get("baseSnapshotId"),
|
|
"size": size,
|
|
"mode": mode,
|
|
"source_path": str(target_path),
|
|
"source_root": str(context.working_directory.expanduser().resolve()),
|
|
"metadata": {},
|
|
}
|
|
url = (
|
|
f"{server_url}/spaces/{context.space_id}/projects/"
|
|
f"{context.project_id}/overlays"
|
|
)
|
|
headers = {"Authorization": context.auth_header}
|
|
if context.user_id:
|
|
headers["X-User-ID"] = context.user_id
|
|
|
|
try:
|
|
with httpx.Client(timeout=5.0) as client:
|
|
response = client.post(url, json=payload, headers=headers)
|
|
if response.is_error:
|
|
_record_overlay_sync_failure(
|
|
reason=f"http_{response.status_code}",
|
|
context=context,
|
|
rel_path=rel_path,
|
|
error_message=response.text[:500],
|
|
)
|
|
return False
|
|
return True
|
|
except Exception as exc: # noqa: BLE001 - overlay sync must not fail the tool write.
|
|
_record_overlay_sync_failure(
|
|
reason="exception",
|
|
context=context,
|
|
rel_path=rel_path,
|
|
error_message=str(exc),
|
|
)
|
|
return False
|