feat(backend): introduce RunContext + workspace_resolver runtime

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"):

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",
+]

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

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)]

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()))
 

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

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)
+    )

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

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")