diff --git a/backend/app/component/environment.py b/backend/app/component/environment.py index 64812a24..385db29e 100644 --- a/backend/app/component/environment.py +++ b/backend/app/component/environment.py @@ -213,6 +213,23 @@ def env(key: str, default=None): Security: Re-validates path at point of use to ensure integrity. """ + # Run-scoped values are the first source of truth for mutable runtime + # settings. This keeps legacy `env("file_save_path")` call sites working + # without relying on process-global os.environ during concurrent runs. + try: + # Inline import avoids a startup cycle: run_context imports no env + # helpers, but many early modules import env before the runtime package. + from app.run_context import get_run_env_override + + run_value = get_run_env_override(key) + if run_value is not None: + logger.debug( + f"Environment variable retrieved from RunContext: key={key}" + ) + return run_value + except ImportError: + pass + # If we have a user-specific environment path, try to reload it # to get latest values. if hasattr(_thread_local, "env_path"): diff --git a/backend/app/run_context/__init__.py b/backend/app/run_context/__init__.py new file mode 100644 index 00000000..e075b5fc --- /dev/null +++ b/backend/app/run_context/__init__.py @@ -0,0 +1,33 @@ +# ========= 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 app.run_context.context import ( + RunContext, + apply_run_env_for_third_party, + current_run_context, + get_current_run_context, + get_run_env_override, + run_context_scope, + stream_with_run_context, +) + +__all__ = [ + "RunContext", + "apply_run_env_for_third_party", + "current_run_context", + "get_current_run_context", + "get_run_env_override", + "run_context_scope", + "stream_with_run_context", +] diff --git a/backend/app/run_context/context.py b/backend/app/run_context/context.py new file mode 100644 index 00000000..38e2bf27 --- /dev/null +++ b/backend/app/run_context/context.py @@ -0,0 +1,141 @@ +# ========= 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. ========= + +import os +from collections.abc import AsyncIterator, Callable, Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass, field +from pathlib import Path + +THIRD_PARTY_OS_ENV_KEYS = ("CAMEL_LOG_DIR", "CAMEL_WORKDIR") + + +@dataclass(frozen=True) +class RunContext: + """Task-scoped runtime values that must not live in process globals.""" + + space_id: str + project_id: str + run_id: str + task_id: str + email: str + user_id: str | None + working_directory: Path + task_output_root: Path + camel_log_dir: Path + binding_source: str + workdir_mode: str | None + browser_port: int + cdp_url: str | None = None + api_key: str | None = None + api_base_url: str | None = None + cloud_api_key: str | None = None + server_url: str | None = None + auth_header: str | None = None + search_config: dict[str, str] = field(default_factory=dict) + extra_env: dict[str, str] = field(default_factory=dict) + + def env_overrides(self) -> dict[str, str]: + values: dict[str, str] = { + "file_save_path": str(self.task_output_root), + "browser_port": str(self.browser_port), + "CAMEL_LOG_DIR": str(self.camel_log_dir), + "CAMEL_WORKDIR": str(self.task_output_root), + } + if self.cdp_url: + values["EIGENT_CDP_URL"] = self.cdp_url + if self.api_key: + values["OPENAI_API_KEY"] = self.api_key + if self.api_base_url: + values["OPENAI_API_BASE_URL"] = self.api_base_url + if self.cloud_api_key: + values["cloud_api_key"] = self.cloud_api_key + if self.server_url: + values["SERVER_URL"] = self.server_url + values.update( + {key: value for key, value in self.search_config.items() if value} + ) + values.update( + {key: value for key, value in self.extra_env.items() if value} + ) + return values + + def env_value(self, key: str) -> str | None: + return self.env_overrides().get(key) + + +current_run_context: ContextVar[RunContext | None] = ContextVar( + "current_run_context", default=None +) + + +def get_current_run_context() -> RunContext | None: + return current_run_context.get() + + +def get_run_env_override(key: str) -> str | None: + context = get_current_run_context() + if context is None: + return None + return context.env_value(key) + + +def apply_run_env_for_third_party(context: RunContext) -> None: + """Publish the small env subset third-party libraries read directly. + + First-party code must use app.component.environment.env(), which reads + RunContext before os.environ. Some third-party libraries, notably CAMEL's + model logging, call os.environ.get(...) themselves and cannot see our + ContextVar. Keep this shim intentionally tiny and auditable. + """ + + overrides = context.env_overrides() + for key in THIRD_PARTY_OS_ENV_KEYS: + value = overrides.get(key) + if value: + os.environ[key] = value + + +@contextmanager +def run_context_scope(context: RunContext) -> Iterator[RunContext]: + token = current_run_context.set(context) + try: + yield context + finally: + current_run_context.reset(token) + + +async def stream_with_run_context( + stream: AsyncIterator[str], + context_getter: Callable[[], RunContext | None], +) -> AsyncIterator[str]: + iterator = stream.__aiter__() + while True: + context = context_getter() + if context is None: + try: + yield await iterator.__anext__() + except StopAsyncIteration: + return + continue + + token = current_run_context.set(context) + try: + item = await iterator.__anext__() + except StopAsyncIteration: + return + finally: + current_run_context.reset(token) + yield item diff --git a/backend/app/utils/cdp_browser_state.py b/backend/app/utils/cdp_browser_state.py new file mode 100644 index 00000000..8d3f2458 --- /dev/null +++ b/backend/app/utils/cdp_browser_state.py @@ -0,0 +1,167 @@ +# ========= 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. ========= + +import logging +import time +from typing import Any + +from fastapi import Request + +from app.component.environment import env +from app.utils.browser_launcher import ( + _is_cdp_available, + is_cdp_url_available, + is_local_cdp_host, + normalize_cdp_url, +) + +logger = logging.getLogger("cdp_browser_state") + +_web_cdp_browser_meta_by_owner: dict[str, dict[str, Any]] = {} + + +def browser_owner_key(request: Request | None) -> str: + auth = getattr(getattr(request, "state", None), "brain_auth", None) + user_id = getattr(auth, "user_id", None) + if user_id and user_id != "local": + return str(user_id) + + # Bridge release fallback: NoneAuth still emits "local", while the + # frontend already sends X-User-ID. Phase B auth will make this token-only. + if request is not None: + header_user_id = request.headers.get("x-user-id") + if header_user_id: + return header_user_id + return "local" + + +def build_web_cdp_browser( + endpoint: str, + *, + is_external: bool, + name: str | None = None, + added_at: int | None = None, + resource_session_id: str | None = None, + managed_by: str = "local", +) -> dict[str, Any]: + normalized_endpoint, host, port = normalize_cdp_url(endpoint) + default_location = ( + str(port) if is_local_cdp_host(host) else f"{host}:{port}" + ) + browser_name = name or ( + f"External Browser ({default_location})" + if is_external + else f"Managed Browser ({default_location})" + ) + browser_id = resource_session_id or ( + f"web-cdp-{port}" + if is_local_cdp_host(host) + else f"web-cdp-{host.replace('.', '-')}-{port}" + ) + return { + "id": browser_id, + "port": port, + "endpoint": normalized_endpoint, + "host": host, + "isExternal": is_external, + "name": browser_name, + "addedAt": added_at or int(time.time() * 1000), + "resourceSessionId": resource_session_id, + "managedBy": managed_by, + } + + +def get_connected_cdp_endpoint(owner_key: str) -> str | None: + if owner_key in _web_cdp_browser_meta_by_owner: + return _web_cdp_browser_meta_by_owner[owner_key].get("endpoint") + cdp_url = env("EIGENT_CDP_URL") + if cdp_url: + return cdp_url + return None + + +def get_connected_cdp_endpoint_for_request( + request: Request | None, +) -> str | None: + return get_connected_cdp_endpoint(browser_owner_key(request)) + + +def get_connected_cdp_meta(owner_key: str) -> dict[str, Any] | None: + return _web_cdp_browser_meta_by_owner.get(owner_key) + + +def get_connected_cdp_port(owner_key: str) -> int | None: + cdp_url = get_connected_cdp_endpoint(owner_key) + if not cdp_url: + return None + try: + _, _, port = normalize_cdp_url(cdp_url) + return port + except Exception: + logger.warning("Invalid EIGENT_CDP_URL: %s", cdp_url) + return None + + +def set_connected_cdp_browser( + owner_key: str, + endpoint: str, + *, + is_external: bool, + name: str | None = None, + resource_session_id: str | None = None, + managed_by: str = "local", +) -> dict[str, Any]: + normalized_endpoint, _, _ = normalize_cdp_url(endpoint) + browser = build_web_cdp_browser( + normalized_endpoint, + is_external=is_external, + name=name, + resource_session_id=resource_session_id, + managed_by=managed_by, + ) + _web_cdp_browser_meta_by_owner[owner_key] = browser + return browser + + +def clear_connected_cdp_browser(owner_key: str) -> None: + _web_cdp_browser_meta_by_owner.pop(owner_key, None) + + +def clear_connected_cdp_browser_for_request(request: Request | None) -> None: + clear_connected_cdp_browser(browser_owner_key(request)) + + +def is_cdp_endpoint_available(endpoint: str) -> bool: + _, host, port = normalize_cdp_url(endpoint) + if is_local_cdp_host(host): + return _is_cdp_available(port) + + return is_cdp_url_available(endpoint) + + +def list_connected_cdp_browsers(owner_key: str) -> list[dict[str, Any]]: + meta = _web_cdp_browser_meta_by_owner.get(owner_key) + endpoint = get_connected_cdp_endpoint(owner_key) + if endpoint is None: + return [] + + if not is_cdp_endpoint_available(endpoint): + if meta and meta.get("endpoint") == endpoint: + clear_connected_cdp_browser(owner_key) + return [] + + if meta and meta.get("endpoint") == endpoint: + return [meta] + + return [build_web_cdp_browser(endpoint, is_external=True)] diff --git a/backend/app/utils/file_utils.py b/backend/app/utils/file_utils.py index 7ffad28a..4c3bcacf 100644 --- a/backend/app/utils/file_utils.py +++ b/backend/app/utils/file_utils.py @@ -23,6 +23,7 @@ from pathlib import Path from app.component.environment import env from app.exception.exception import PathEscapesBaseError from app.model.chat import Chat +from app.run_context import get_current_run_context logger = logging.getLogger("file_utils") @@ -31,7 +32,7 @@ MAX_PATH_LENGTH_WIN = 260 MAX_PATH_LENGTH_UNIX = 4096 # Default directory names to skip when listing (list_files) DEFAULT_SKIP_DIRS = frozenset( - {".git", "node_modules", "__pycache__", "venv", ".venv"} + {".git", "node_modules", "__pycache__", "venv", ".venv", "camel_logs"} ) # Default file extensions to skip when listing (list_files) DEFAULT_SKIP_EXTENSIONS: tuple[str, ...] = (".pyc", ".tmp", ".temp") @@ -228,7 +229,7 @@ def list_files( except OSError: return [] base_real = os.path.realpath(resolve_base) - skip_dirs = set(DEFAULT_SKIP_DIRS) if skip_dirs is None else skip_dirs + skip_dirs = set(DEFAULT_SKIP_DIRS).union(skip_dirs or set()) result: list[str] = [] try: for root, dirs, files in os.walk(resolved_dir, followlinks=False): @@ -280,6 +281,12 @@ def get_working_directory(options: Chat, task_lock=None) -> str: and task_lock.new_folder_path ): raw = Path(task_lock.new_folder_path) + elif task_lock and getattr(task_lock, "working_directory", None): + raw = Path(task_lock.working_directory) + elif ( + context := get_current_run_context() + ) is not None and context.project_id == options.project_id: + raw = context.working_directory else: raw = Path(env("file_save_path", options.file_save_path())) diff --git a/backend/app/utils/space_overlay_client.py b/backend/app/utils/space_overlay_client.py new file mode 100644 index 00000000..0d364deb --- /dev/null +++ b/backend/app/utils/space_overlay_client.py @@ -0,0 +1,231 @@ +# ========= 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 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 diff --git a/backend/app/utils/workspace_paths.py b/backend/app/utils/workspace_paths.py new file mode 100644 index 00000000..86790e2f --- /dev/null +++ b/backend/app/utils/workspace_paths.py @@ -0,0 +1,169 @@ +# ========= 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. ========= + +import re +from pathlib import Path + +from app.component.environment import env + + +def get_workspace_root() -> Path: + return Path(env("EIGENT_WORKSPACE", "~/.eigent/workspace")).expanduser() + + +def get_eigent_root() -> Path: + eigent = Path.home() / "eigent" + if eigent.exists(): + return eigent + dot_eigent = Path.home() / ".eigent" + if dot_eigent.exists(): + return dot_eigent + return eigent + + +def sanitize_email(email: str) -> str: + return re.sub(r'[\\/*?:"<>|\s]', "_", email.split("@")[0]).strip(".") + + +def sanitize_identity(value: str | int | None) -> str: + if value is None: + return "" + return re.sub(r'[\\/*?:"<>|\s]', "_", str(value)).strip(".") + + +def runtime_owner_key(email: str, user_id: str | int | None = None) -> str: + user_key = sanitize_identity(user_id) + if user_key: + return f"user_{user_key}" + return sanitize_email(email) + + +def project_root( + email: str, project_id: str, user_id: str | int | None = None +) -> Path: + return ( + get_eigent_root() + / runtime_owner_key(email, user_id) + / f"project_{project_id}" + ) + + +def project_task_root( + email: str, + project_id: str, + task_id: str, + user_id: str | int | None = None, +) -> Path: + return project_root(email, project_id, user_id) / f"task_{task_id}" + + +def legacy_task_root( + email: str, task_id: str, user_id: str | int | None = None +) -> Path: + return ( + get_eigent_root() + / runtime_owner_key(email, user_id) + / f"task_{task_id}" + ) + + +def camel_log_root( + email: str, + project_id: str, + task_id: str, + user_id: str | int | None = None, +) -> Path: + return ( + Path.home() + / ".eigent" + / runtime_owner_key(email, user_id) + / f"project_{project_id}" + / f"task_{task_id}" + / "camel_logs" + ) + + +def legacy_camel_log_root( + email: str, task_id: str, user_id: str | int | None = None +) -> Path: + return ( + Path.home() + / ".eigent" + / runtime_owner_key(email, user_id) + / f"task_{task_id}" + / "camel_logs" + ) + + +def runtime_task_root( + email: str, + project_id: str, + task_id: str, + user_id: str | int | None = None, +) -> Path: + return ( + Path.home() + / ".eigent" + / runtime_owner_key(email, user_id) + / "runtime" + / f"project_{project_id}" + / f"task_{task_id}" + ) + + +def run_output_root( + email: str, + space_id: str, + project_id: str, + run_id: str, + user_id: str | int | None = None, +) -> Path: + return ( + Path.home() + / ".eigent" + / runtime_owner_key(email, user_id) + / "spaces" + / space_id + / "projects" + / project_id + / "runs" + / run_id + ) + + +def project_workdir_root( + email: str, + space_id: str, + project_id: str, + user_id: str | int | None = None, +) -> Path: + return ( + Path.home() + / ".eigent" + / runtime_owner_key(email, user_id) + / "spaces" + / space_id + / "projects" + / project_id + / "workdir" + ) + + +def workspace_state_root(email: str, user_id: str | int | None = None) -> Path: + return ( + Path.home() + / ".eigent" + / "workspaces" + / runtime_owner_key(email, user_id) + ) diff --git a/backend/app/utils/workspace_resolver.py b/backend/app/utils/workspace_resolver.py new file mode 100644 index 00000000..f3081fea --- /dev/null +++ b/backend/app/utils/workspace_resolver.py @@ -0,0 +1,635 @@ +# ========= 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. ========= + +import json +import logging +import shutil +import time +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Literal +from uuid import uuid4 + +from app.model.chat import Chat +from app.router_layer.hands_resolver import get_environment_hands +from app.utils.workspace_paths import ( + camel_log_root, + legacy_camel_log_root, + legacy_task_root, + project_root, + project_task_root, + project_workdir_root, + run_output_root, + workspace_state_root, +) + +logger = logging.getLogger("workspace_resolver") +BindingSource = Literal["space_local_brain", "default"] +WORKDIR_MARKER = ".eigent-workdir.json" +COPY_IGNORE_DIRS = { + ".git", + ".hg", + ".svn", + "node_modules", + ".venv", + "venv", + "dist", + "build", + ".next", + ".cache", + "__pycache__", +} +MAX_COPY_FILE_SIZE = 25 * 1024 * 1024 + + +def _binding_enabled_for_current_environment() -> bool: + hands = get_environment_hands() + get_manifest = getattr(hands, "get_capability_manifest", None) + if get_manifest is None: + return False + try: + manifest = get_manifest() + except Exception: + logger.warning( + "Failed to read hands capability manifest for workspace binding", + exc_info=True, + ) + return False + if not isinstance(manifest, dict): + return False + return manifest.get("deployment") == "local" + + +def _same_workspace_path(left: str, right: str) -> bool: + try: + return ( + Path(left).expanduser().resolve() + == Path(right).expanduser().resolve() + ) + except (OSError, RuntimeError): + return False + + +def _folder_fingerprint(path: Path) -> dict[str, Any]: + stat = path.stat() + return { + "kind": "local_folder", + "path": str(path), + "device": stat.st_dev, + "inode": stat.st_ino, + "mtime_ns": stat.st_mtime_ns, + "ctime_ns": stat.st_ctime_ns, + } + + +def _read_workdir_marker(workdir: Path) -> dict[str, Any] | None: + marker = workdir / WORKDIR_MARKER + if not marker.exists(): + return None + try: + return json.loads(marker.read_text(encoding="utf-8")) + except Exception: + logger.warning("Failed to read Project workdir marker: %s", marker) + return None + + +def _copy_space_baseline(source_root: Path, workdir: Path) -> str: + existing_marker = _read_workdir_marker(workdir) + if existing_marker and existing_marker.get("base_snapshot_id"): + return str(existing_marker["base_snapshot_id"]) + + workdir.mkdir(parents=True, exist_ok=True) + _copy_tree_limited(source_root, workdir) + + base_snapshot_id = f"snapshot_{uuid4().hex}" + marker = workdir / WORKDIR_MARKER + marker.write_text( + json.dumps( + { + "base_snapshot_id": base_snapshot_id, + "source_root": str(source_root), + "created_at": datetime.now(UTC).isoformat(), + "copy_ignore_dirs": sorted(COPY_IGNORE_DIRS), + "max_copy_file_size": MAX_COPY_FILE_SIZE, + }, + indent=2, + ), + encoding="utf-8", + ) + return base_snapshot_id + + +def _copy_tree_limited(source_root: Path, target_root: Path) -> None: + target_root.mkdir(parents=True, exist_ok=True) + for item in source_root.iterdir(): + if item.name in COPY_IGNORE_DIRS or item.name == WORKDIR_MARKER: + continue + target = target_root / item.name + try: + if item.is_symlink(): + continue + if item.is_dir(): + _copy_tree_limited(item, target) + continue + if item.is_file() and item.stat().st_size <= MAX_COPY_FILE_SIZE: + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(item, target) + except OSError: + logger.warning( + "Failed to copy Space baseline item into Project workdir: %s", + item, + exc_info=True, + ) + + +@dataclass(frozen=True) +class WorkspaceBinding: + space_id: str + workspace_root: str + source: str + created_at: str + updated_at: str + root_fingerprint: dict[str, Any] | None = None + version: int = 2 + + +@dataclass(frozen=True) +class TaskSnapshot: + task_id: str + project_id: str + space_id: str + user_id: str | int | None + working_directory: str + task_output_root: str + task_start_time: float + binding_source: BindingSource + created_at: str + workdir_mode: str | None = None + base_snapshot_id: str | None = None + version: int = 2 + + +class WorkspaceStore: + def _state_roots( + self, email: str, user_id: str | int | None = None + ) -> tuple[Path, ...]: + primary = workspace_state_root(email, user_id) + legacy = workspace_state_root(email) + if user_id is not None and primary != legacy: + return (primary, legacy) + return (primary,) + + def _space_path( + self, email: str, space_id: str, user_id: str | int | None = None + ) -> Path: + return ( + self._state_roots(email, user_id)[0] + / "spaces" + / f"{space_id}.json" + ) + + def _space_paths( + self, email: str, space_id: str, user_id: str | int | None = None + ) -> tuple[Path, ...]: + return tuple( + root / "spaces" / f"{space_id}.json" + for root in self._state_roots(email, user_id) + ) + + def _task_path( + self, email: str, task_id: str, user_id: str | int | None = None + ) -> Path: + return ( + self._state_roots(email, user_id)[0] / "tasks" / f"{task_id}.json" + ) + + def _task_paths( + self, email: str, task_id: str, user_id: str | int | None = None + ) -> tuple[Path, ...]: + return tuple( + root / "tasks" / f"{task_id}.json" + for root in self._state_roots(email, user_id) + ) + + def get_binding( + self, email: str, space_id: str, user_id: str | int | None = None + ) -> WorkspaceBinding | None: + for path in self._space_paths(email, space_id, user_id): + if not path.exists(): + continue + try: + data = json.loads(path.read_text(encoding="utf-8")) + if "space_id" not in data and "project_id" in data: + data["space_id"] = data.pop("project_id") + return WorkspaceBinding(**data) + except Exception: + logger.warning("Failed to read workspace binding: %s", path) + return None + + def save_binding( + self, + email: str, + space_id: str, + workspace_root: str, + *, + user_id: str | int | None = None, + root_fingerprint: dict[str, Any] | None = None, + ) -> WorkspaceBinding: + now = datetime.now(UTC).isoformat() + existing = self.get_binding(email, space_id, user_id) + binding = WorkspaceBinding( + space_id=space_id, + workspace_root=workspace_root, + source="space_local_brain", + created_at=existing.created_at if existing else now, + updated_at=now, + root_fingerprint=root_fingerprint, + ) + primary_path = self._space_path(email, space_id, user_id) + self._atomic_write(primary_path, asdict(binding)) + for legacy_path in self._space_paths(email, space_id, user_id)[1:]: + if legacy_path != primary_path and legacy_path.exists(): + legacy_path.unlink() + return binding + + def _promote_binding_to_user_path( + self, + email: str, + user_id: str | int | None, + binding: WorkspaceBinding, + ) -> None: + if user_id is None: + return + primary_path = self._space_path(email, binding.space_id, user_id) + self._atomic_write(primary_path, asdict(binding)) + for legacy_path in self._space_paths(email, binding.space_id, user_id)[ + 1: + ]: + if legacy_path != primary_path and legacy_path.exists(): + legacy_path.unlink() + + def delete_binding( + self, email: str, space_id: str, user_id: str | int | None = None + ) -> None: + for path in self._space_paths(email, space_id, user_id): + if path.exists(): + path.unlink() + + def reconcile_bindings( + self, + email: str, + active_space_ids: set[str], + user_id: str | int | None = None, + ) -> list[WorkspaceBinding]: + if not active_space_ids: + logger.warning( + "Skipping workspace binding reconciliation with no active Space ids", + extra={"email": email}, + ) + return [] + + removed: list[WorkspaceBinding] = [] + for binding in self.list_bindings(email, user_id): + if binding.space_id in active_space_ids: + self._promote_binding_to_user_path(email, user_id, binding) + continue + self.delete_binding(email, binding.space_id, user_id) + removed.append(binding) + return removed + + def list_bindings( + self, email: str, user_id: str | int | None = None + ) -> list[WorkspaceBinding]: + bindings_by_space: dict[str, WorkspaceBinding] = {} + for root in self._state_roots(email, user_id): + spaces_dir = root / "spaces" + if not spaces_dir.exists(): + continue + for path in spaces_dir.glob("*.json"): + try: + data = json.loads(path.read_text(encoding="utf-8")) + if "space_id" not in data and "project_id" in data: + data["space_id"] = data.pop("project_id") + binding = WorkspaceBinding(**data) + bindings_by_space.setdefault(binding.space_id, binding) + except Exception: + logger.warning( + "Failed to read workspace binding: %s", path + ) + return list(bindings_by_space.values()) + + def get_snapshot( + self, email: str, task_id: str, user_id: str | int | None = None + ) -> TaskSnapshot | None: + for path in self._task_paths(email, task_id, user_id): + if not path.exists(): + continue + try: + data = json.loads(path.read_text(encoding="utf-8")) + if "space_id" not in data: + data["space_id"] = data.get("project_id", "") + data.setdefault("user_id", None) + data.setdefault("workdir_mode", None) + data.setdefault("base_snapshot_id", None) + return TaskSnapshot(**data) + except Exception: + logger.warning("Failed to read task snapshot: %s", path) + return None + + def save_snapshot(self, email: str, snapshot: TaskSnapshot) -> None: + self._atomic_write( + self._task_path(email, snapshot.task_id, snapshot.user_id), + asdict(snapshot), + ) + + def _atomic_write(self, path: Path, data: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + tmp_path.write_text( + json.dumps(data, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + tmp_path.replace(path) + + +@dataclass(frozen=True) +class FrozenTaskDirectories: + working_directory: Path + task_output_root: Path + task_start_time: float + binding_source: BindingSource + workdir_mode: str | None + base_snapshot_id: str | None + snapshot: TaskSnapshot + + +class WorkspaceResolver: + def __init__(self, store: WorkspaceStore | None = None) -> None: + self.store = store or WorkspaceStore() + + def ensure_space_binding( + self, + email: str, + space_id: str, + root_path: str, + user_id: str | int | None = None, + ) -> WorkspaceBinding: + if not _binding_enabled_for_current_environment(): + raise ValueError( + "Workspace folder binding is disabled in this deployment" + ) + + resolved = Path(root_path).expanduser().resolve() + if not resolved.exists() or not resolved.is_dir(): + raise ValueError("Space root_path is not a readable directory") + + existing = self.store.get_binding(email, space_id, user_id) + if existing is not None: + if _same_workspace_path(existing.workspace_root, str(resolved)): + return existing + raise ValueError("Space is already bound to a different folder") + + return self.store.save_binding( + email, + space_id, + str(resolved), + user_id=user_id, + root_fingerprint=_folder_fingerprint(resolved), + ) + + def space_root( + self, + space_id: str, + project_id: str, + email: str, + user_id: str | int | None = None, + ) -> Path | None: + binding = self.store.get_binding(email, space_id, user_id) + if binding: + bound_path = Path(binding.workspace_root).expanduser() + if bound_path.is_dir(): + return bound_path + + legacy_project = project_root(email, project_id, user_id) + if user_id is not None and not legacy_project.exists(): + legacy_project = project_root(email, project_id) + if legacy_project.exists(): + return legacy_project + return None + + def task_output_root( + self, + space_id: str, + project_id: str, + task_id: str, + email: str, + user_id: str | int | None = None, + ) -> Path: + snapshot = self.store.get_snapshot(email, task_id, user_id) + if snapshot: + return Path(snapshot.task_output_root).expanduser() + + old_project_task = project_task_root( + email, project_id, task_id, user_id + ) + if old_project_task.exists(): + return old_project_task + if user_id is not None: + legacy_project_task = project_task_root(email, project_id, task_id) + if legacy_project_task.exists(): + return legacy_project_task + + old_legacy_task = legacy_task_root(email, task_id, user_id) + if old_legacy_task.exists(): + return old_legacy_task + if user_id is not None: + legacy_email_task = legacy_task_root(email, task_id) + if legacy_email_task.exists(): + return legacy_email_task + + binding = self.store.get_binding(email, space_id, user_id) + if binding: + bound_path = Path(binding.workspace_root).expanduser() + if bound_path.is_dir(): + return run_output_root( + email, space_id, project_id, task_id, user_id + ) + + return old_project_task + + def log_root( + self, + project_id: str, + task_id: str, + email: str, + user_id: str | int | None = None, + ) -> Path: + root = camel_log_root(email, project_id, task_id, user_id) + if root.exists(): + return root + if user_id is not None: + legacy_project_root = camel_log_root(email, project_id, task_id) + if legacy_project_root.exists(): + return legacy_project_root + legacy = legacy_camel_log_root(email, task_id, user_id) + if legacy.exists(): + return legacy + if user_id is not None: + legacy_email = legacy_camel_log_root(email, task_id) + if legacy_email.exists(): + return legacy_email + return root + + def freeze_task_directories( + self, options: Chat, task_lock + ) -> FrozenTaskDirectories: + space_id = options.space_id or options.project_id + task_lock.workdir_mode = options.workdir_mode + if options.space_root_path: + self.ensure_space_binding( + options.email, + space_id, + options.space_root_path, + user_id=options.user_id, + ) + return self.freeze_task_directories_for( + space_id=space_id, + project_id=options.project_id, + task_id=options.task_id, + email=options.email, + task_lock=task_lock, + fallback_task_root=options.file_save_path(), + user_id=options.user_id, + ) + + def freeze_task_directories_for( + self, + space_id: str, + project_id: str, + task_id: str, + email: str, + task_lock, + fallback_task_root: str | Path | None = None, + user_id: str | int | None = None, + ) -> FrozenTaskDirectories: + binding = self.store.get_binding(email, space_id, user_id) + if binding and Path(binding.workspace_root).expanduser().is_dir(): + source_root = Path(binding.workspace_root).expanduser().resolve() + task_output = run_output_root( + email, space_id, project_id, task_id, user_id + ) + workdir_mode = getattr(task_lock, "workdir_mode", None) + if workdir_mode == "artifact-only": + working_directory = task_output + base_snapshot_id = None + elif workdir_mode == "direct-write": + working_directory = source_root + base_snapshot_id = None + else: + working_directory = project_workdir_root( + email, + space_id, + project_id, + user_id, + ) + base_snapshot_id = _copy_space_baseline( + source_root, working_directory + ) + workdir_mode = workdir_mode or "copy" + binding_source: BindingSource = "space_local_brain" + elif fallback_task_root is not None: + working_directory = Path(fallback_task_root).expanduser() + task_output = working_directory + binding_source = "default" + workdir_mode = getattr(task_lock, "workdir_mode", None) + base_snapshot_id = None + else: + working_directory = project_task_root( + email, project_id, task_id, user_id + ) + task_output = working_directory + binding_source = "default" + workdir_mode = getattr(task_lock, "workdir_mode", None) + base_snapshot_id = None + + working_directory.mkdir(parents=True, exist_ok=True) + task_output.mkdir(parents=True, exist_ok=True) + task_start_time = time.time() + snapshot = TaskSnapshot( + task_id=task_id, + project_id=project_id, + space_id=space_id, + user_id=user_id, + working_directory=str(working_directory), + task_output_root=str(task_output), + task_start_time=task_start_time, + binding_source=binding_source, + workdir_mode=workdir_mode, + base_snapshot_id=base_snapshot_id, + created_at=datetime.now(UTC).isoformat(), + ) + + task_lock.working_directory = str(working_directory) + task_lock.task_output_root = str(task_output) + task_lock.task_start_time = task_start_time + task_lock.email = email + task_lock.user_id = user_id + task_lock.project_id = project_id + task_lock.space_id = space_id + task_lock.current_task_id = task_id + task_lock.workdir_mode = workdir_mode + task_lock.base_snapshot_id = base_snapshot_id + + return FrozenTaskDirectories( + working_directory=working_directory, + task_output_root=task_output, + task_start_time=task_start_time, + binding_source=binding_source, + workdir_mode=workdir_mode, + base_snapshot_id=base_snapshot_id, + snapshot=snapshot, + ) + + def write_task_snapshot(self, email: str, snapshot: TaskSnapshot) -> None: + self.store.save_snapshot(email, snapshot) + + def refresh_project_workdir( + self, + *, + space_id: str, + project_id: str, + email: str, + user_id: str | int | None = None, + ) -> str: + binding = self.store.get_binding(email, space_id, user_id) + if binding is None: + raise ValueError("Space is not bound in this Brain") + source_root = Path(binding.workspace_root).expanduser().resolve() + if not source_root.is_dir(): + raise ValueError("Bound Space root is not available") + workdir = project_workdir_root(email, space_id, project_id, user_id) + if workdir.exists(): + shutil.rmtree(workdir) + return _copy_space_baseline(source_root, workdir) + + +_resolver: WorkspaceResolver | None = None + + +def get_workspace_resolver() -> WorkspaceResolver: + global _resolver + if _resolver is None: + _resolver = WorkspaceResolver() + return _resolver diff --git a/backend/main.py b/backend/main.py index df13432c..304ce42d 100644 --- a/backend/main.py +++ b/backend/main.py @@ -47,6 +47,11 @@ from app.router import register_routers from app.utils.event_loop_utils import set_main_event_loop os.environ["PYTHONIOENCODING"] = "utf-8" +_fallback_camel_log_dir = ( + pathlib.Path.home() / ".eigent" / "fallback" / "camel_logs" +) +_fallback_camel_log_dir.mkdir(parents=True, exist_ok=True) +os.environ.setdefault("CAMEL_LOG_DIR", str(_fallback_camel_log_dir)) app_logger = logging.getLogger("main")