diff --git a/backend/app/controller/chat_controller.py b/backend/app/controller/chat_controller.py index 678ec49c..66474b69 100644 --- a/backend/app/controller/chat_controller.py +++ b/backend/app/controller/chat_controller.py @@ -16,8 +16,8 @@ import asyncio import inspect import logging import os -import re import time +from dataclasses import replace from pathlib import Path from dotenv import load_dotenv @@ -25,7 +25,7 @@ from fastapi import APIRouter, Request, Response from fastapi.responses import StreamingResponse from app.component import code -from app.component.environment import sanitize_env_path, set_user_env_path +from app.component.environment import env, sanitize_env_path, set_user_env_path from app.exception.exception import UserException from app.model.chat import ( AddTaskRequest, @@ -36,6 +36,11 @@ from app.model.chat import ( SupplementChat, sse_json, ) +from app.run_context import ( + RunContext, + apply_run_env_for_third_party, + stream_with_run_context, +) from app.service.chat_service import step_solve from app.service.task import ( Action, @@ -59,6 +64,12 @@ from app.utils.browser_launcher import ( is_cdp_url_available, normalize_cdp_url, ) +from app.utils.cdp_browser_state import ( + clear_connected_cdp_browser_for_request, + get_connected_cdp_endpoint_for_request, +) +from app.utils.workspace_paths import camel_log_root +from app.utils.workspace_resolver import get_workspace_resolver router = APIRouter() @@ -68,6 +79,9 @@ chat_logger = logging.getLogger("chat_controller") # SSE timeout configuration (60 minutes in seconds) SSE_TIMEOUT_SECONDS = 60 * 60 +# CAMEL reads this as a process-level logging toggle, not as per-run state. +os.environ.setdefault("CAMEL_MODEL_LOG_ENABLED", "true") + def _is_remote_browser_hands(request: Request | None) -> bool: hands = getattr(getattr(request, "state", None), "hands", None) @@ -93,7 +107,10 @@ async def _prepare_browser_for_request( request: Request | None, port: int, ) -> bool: - existing_cdp_url = os.environ.get("EIGENT_CDP_URL", "").strip() + existing_cdp_url = ( + get_connected_cdp_endpoint_for_request(request) + or env("EIGENT_CDP_URL", "") + ).strip() if existing_cdp_url: is_available = await asyncio.to_thread( is_cdp_url_available, existing_cdp_url @@ -102,48 +119,99 @@ async def _prepare_browser_for_request( normalized_endpoint, _, selected_port = normalize_cdp_url( existing_cdp_url ) - os.environ["EIGENT_CDP_URL"] = normalized_endpoint - os.environ["browser_port"] = str(selected_port) if request is not None: request.state.browser_available = True + request.state.cdp_url = normalized_endpoint + request.state.browser_port = selected_port return True - os.environ.pop("EIGENT_CDP_URL", None) + clear_connected_cdp_browser_for_request(request) if _is_remote_browser_hands(request): if request is not None: request.state.browser_available = True + request.state.cdp_url = None + request.state.browser_port = port return True try: endpoint = await asyncio.to_thread(ensure_cdp_browser_endpoint, port) except Exception as e: - os.environ.pop("EIGENT_CDP_URL", None) chat_logger.warning( "Could not ensure CDP browser for web mode", extra={"error": str(e), "port": port}, ) if request is not None: request.state.browser_available = False + request.state.cdp_url = None + request.state.browser_port = port return False if endpoint: - os.environ["EIGENT_CDP_URL"] = endpoint _, _, selected_port = normalize_cdp_url(endpoint) - os.environ["browser_port"] = str(selected_port) if request is not None: request.state.browser_available = True + request.state.cdp_url = endpoint + request.state.browser_port = selected_port return True - os.environ.pop("EIGENT_CDP_URL", None) chat_logger.warning( "CDP browser not available after ensure attempt", extra={"port": port}, ) if request is not None: request.state.browser_available = False + request.state.cdp_url = None + request.state.browser_port = port return False +def _build_run_context( + data: Chat, + frozen_dirs, + request: Request, + camel_log: Path, +) -> RunContext: + api_base_url = data.api_url or "https://api.openai.com/v1" + browser_port = int( + getattr(request.state, "browser_port", data.browser_port) + ) + cdp_url = getattr(request.state, "cdp_url", None) + auth_header = request.headers.get("authorization") + return RunContext( + space_id=data.space_id or data.project_id, + project_id=data.project_id, + run_id=data.run_id or data.task_id, + task_id=data.task_id, + email=data.email, + user_id=str(data.user_id) if data.user_id is not None else None, + working_directory=frozen_dirs.working_directory, + task_output_root=frozen_dirs.task_output_root, + camel_log_dir=camel_log, + binding_source=frozen_dirs.binding_source, + workdir_mode=frozen_dirs.workdir_mode or data.workdir_mode, + browser_port=browser_port, + cdp_url=cdp_url, + api_key=data.api_key, + api_base_url=api_base_url, + cloud_api_key=data.api_key if data.is_cloud() else None, + server_url=data.server_url, + auth_header=auth_header, + search_config=data.search_config or {}, + extra_env={ + "baseSnapshotId": frozen_dirs.base_snapshot_id or "", + }, + ) + + +def _camel_log_dir( + email: str, + project_id: str, + task_id: str, + user_id: str | int | None = None, +) -> Path: + return camel_log_root(email, project_id, task_id, user_id) + + async def _cleanup_task_lock_safe(task_lock, reason: str) -> bool: """Safely cleanup task lock with existence check. @@ -182,6 +250,25 @@ async def _cleanup_task_lock_safe(task_lock, reason: str) -> bool: return False +def _should_preserve_task_lock_on_cancel(task_lock) -> bool: + """Keep completed Project state alive for follow-up turns. + + The frontend closes the SSE stream after a run reaches `end`. That close is + reported to FastAPI as a cancellation, but for multi-turn Project semantics + it is not a user stop. The TaskLock carries the short-term conversation + context used by follow-up `/chat/{project_id}` requests, especially for the + single-agent harness, so completed locks with history must survive it. + """ + if not task_lock: + return False + if getattr(task_lock, "status", None) not in { + Status.done, + Status.confirming, + }: + return False + return bool(getattr(task_lock, "conversation_history", None)) + + async def timeout_stream_wrapper( stream_generator, timeout_seconds: int = SSE_TIMEOUT_SECONDS, @@ -232,6 +319,12 @@ async def timeout_stream_wrapper( chat_logger.info( "[STREAM-CANCELLED] Stream cancelled, triggering cleanup" ) + if _should_preserve_task_lock_on_cancel(task_lock): + chat_logger.info( + "[STREAM-CANCELLED] Preserving completed task lock for follow-up context", + extra={"task_id": getattr(task_lock, "id", None)}, + ) + raise if not cleanup_triggered: await _cleanup_task_lock_safe(task_lock, "CANCELLED") raise @@ -251,6 +344,9 @@ async def start_chat_stream(data: Chat, request: Request): Setup and start chat stream. Used by POST /chat and Message Router. Returns async generator of SSE chunks. """ + # TODO(brain-auth): Phase B should derive canonical user_id from + # request.state.brain_auth, then verify/replace Chat.email before any + # workspace snapshot, artifact path, or task lock is resolved. chat_logger.info( "Starting new chat session", extra={ @@ -269,48 +365,40 @@ async def start_chat_stream(data: Chat, request: Request): if safe_env_path: load_dotenv(dotenv_path=safe_env_path) - # TODO(multi-tenant): os.environ is global – concurrent sessions overwrite - # each other's API keys, file paths, and browser ports. Pass these values - # through Chat / request context instead of mutating the process environment. - os.environ["file_save_path"] = data.file_save_path() - os.environ["browser_port"] = str(data.browser_port) + resolver = get_workspace_resolver() + try: + frozen_dirs = resolver.freeze_task_directories(data, task_lock) + except ValueError as exc: + raise UserException(code.error, str(exc)) from exc + + try: + await asyncio.to_thread( + resolver.write_task_snapshot, + data.email, + frozen_dirs.snapshot, + ) + except Exception: + chat_logger.warning( + "Failed to persist task workspace snapshot", + extra={"project_id": data.project_id, "task_id": data.task_id}, + exc_info=True, + ) + # Web mode: reuse an existing CDP endpoint first, otherwise acquire browser # through RemoteHands or launch a local browser when available. if not data.cdp_browsers: await _prepare_browser_for_request(request, data.browser_port) - os.environ["OPENAI_API_KEY"] = data.api_key - os.environ["OPENAI_API_BASE_URL"] = ( - data.api_url or "https://api.openai.com/v1" - ) - os.environ["CAMEL_MODEL_LOG_ENABLED"] = "true" - # Set user-specific search engine configuration if provided - if data.search_config: - for key, value in data.search_config.items(): - if value: - os.environ[key] = value - chat_logger.debug( - f"Set search config: {key}", - extra={"project_id": data.project_id}, - ) - - email_sanitized = re.sub( - r'[\\/*?:"<>|\s]', "_", data.email.split("@")[0] - ).strip(".") - camel_log = ( - Path.home() - / ".eigent" - / email_sanitized - / ("project_" + data.project_id) - / ("task_" + data.task_id) - / "camel_logs" + camel_log = _camel_log_dir( + data.email, + data.project_id, + data.run_id or data.task_id, + data.user_id, ) camel_log.mkdir(parents=True, exist_ok=True) - - os.environ["CAMEL_LOG_DIR"] = str(camel_log) - - if data.is_cloud(): - os.environ["cloud_api_key"] = data.api_key + run_context = _build_run_context(data, frozen_dirs, request, camel_log) + apply_run_env_for_third_party(run_context) + task_lock.run_context = run_context # Set the initial current_task_id in task_lock set_current_task_id(data.project_id, data.task_id) @@ -321,6 +409,7 @@ async def start_chat_stream(data: Chat, request: Request): data=ImprovePayload( question=data.question, attaches=data.attaches or [], + project_context=data.project_context, ), new_task_id=data.task_id, ) @@ -332,10 +421,16 @@ async def start_chat_stream(data: Chat, request: Request): "project_id": data.project_id, "task_id": data.task_id, "log_dir": str(camel_log), + "working_directory": str(frozen_dirs.working_directory), + "binding_source": frozen_dirs.binding_source, }, ) return timeout_stream_wrapper( - step_solve(data, request, task_lock), task_lock=task_lock + stream_with_run_context( + step_solve(data, request, task_lock), + lambda: getattr(task_lock, "run_context", run_context), + ), + task_lock=task_lock, ) @@ -358,7 +453,12 @@ def improve(id: str, data: SupplementChat, request: Request): # Reuse an existing endpoint when possible to avoid tearing down # a browser that was manually connected through the Browser page. - port = int(os.environ.get("browser_port", "9222")) + current_context = getattr(task_lock, "run_context", None) + port = ( + current_context.browser_port + if isinstance(current_context, RunContext) + else int(env("browser_port", "9222")) + ) asyncio.run(_prepare_browser_for_request(request, port)) # Allow continuing conversation even after task is done @@ -390,39 +490,68 @@ def improve(id: str, data: SupplementChat, request: Request): new_folder_path = None if data.task_id: try: - # Get current environment values needed to construct new path - current_email = None - - # Extract email from current file_save_path if available - current_file_save_path = os.environ.get("file_save_path", "") - if current_file_save_path: - path_parts = Path(current_file_save_path).parts - if len(path_parts) >= 3 and "eigent" in path_parts: - eigent_index = path_parts.index("eigent") - if eigent_index + 1 < len(path_parts): - current_email = path_parts[eigent_index + 1] + current_email = getattr(task_lock, "email", None) # If we have the necessary info, update # the file_save_path if current_email and id: - # Create new path using the existing - # pattern: email/project_{id}/task_{id} - new_folder_path = ( - Path.home() - / "eigent" - / current_email - / f"project_{id}" - / f"task_{data.task_id}" + resolver = get_workspace_resolver() + frozen_dirs = resolver.freeze_task_directories_for( + space_id=getattr(task_lock, "space_id", id), + project_id=id, + task_id=data.task_id, + email=current_email, + task_lock=task_lock, + user_id=getattr(task_lock, "user_id", None), ) - new_folder_path.mkdir(parents=True, exist_ok=True) - os.environ["file_save_path"] = str(new_folder_path) + try: + resolver.write_task_snapshot( + current_email, frozen_dirs.snapshot + ) + except Exception: + chat_logger.warning( + "Failed to persist task workspace snapshot", + extra={"project_id": id, "task_id": data.task_id}, + exc_info=True, + ) + new_folder_path = frozen_dirs.task_output_root + camel_log = _camel_log_dir( + current_email, + id, + data.task_id, + getattr(task_lock, "user_id", None), + ) + camel_log.mkdir(parents=True, exist_ok=True) + current_context = getattr(task_lock, "run_context", None) + if isinstance(current_context, RunContext): + updated_context = replace( + current_context, + run_id=data.task_id, + task_id=data.task_id, + working_directory=frozen_dirs.working_directory, + task_output_root=frozen_dirs.task_output_root, + camel_log_dir=camel_log, + binding_source=frozen_dirs.binding_source, + browser_port=int( + getattr(request.state, "browser_port", port) + ), + cdp_url=getattr( + request.state, "cdp_url", current_context.cdp_url + ), + ) + apply_run_env_for_third_party(updated_context) + task_lock.run_context = updated_context chat_logger.info( f"Updated file_save_path to: {new_folder_path}" ) # Store the new folder path in task_lock # for potential cleanup and persistence - task_lock.new_folder_path = new_folder_path + task_lock.new_folder_path = ( + new_folder_path + if frozen_dirs.binding_source == "default" + else None + ) else: chat_logger.warning( "Could not update" @@ -445,6 +574,7 @@ def improve(id: str, data: SupplementChat, request: Request): data=ImprovePayload( question=data.question, attaches=data.attaches or [], + project_context=data.project_context, ), new_task_id=data.task_id, ) @@ -518,7 +648,17 @@ def human_reply(id: str, data: HumanReply): extra={"task_id": id, "reply_length": len(data.reply)}, ) task_lock = get_task_lock(id) - asyncio.run(task_lock.put_human_input(data.agent, data.reply)) + try: + asyncio.run(task_lock.put_human_input(data.agent, data.reply)) + except KeyError as exc: + chat_logger.warning( + "Human reply target is no longer waiting for input", + extra={"task_id": id, "agent": data.agent}, + ) + raise UserException( + code.error, + "This task is no longer waiting for a human reply. Please send a new message.", + ) from exc chat_logger.debug("Human reply processed", extra={"task_id": id}) return Response(status_code=201) diff --git a/backend/app/controller/file_controller.py b/backend/app/controller/file_controller.py index 7e961922..08b401ba 100644 --- a/backend/app/controller/file_controller.py +++ b/backend/app/controller/file_controller.py @@ -25,6 +25,7 @@ from fastapi.responses import FileResponse from app.component.environment import env from app.utils.file_utils import list_files, resolve_under_base +from app.utils.workspace_resolver import get_workspace_resolver router = APIRouter() file_logger = logging.getLogger("file_controller") @@ -189,10 +190,33 @@ def _resolve_project_root(email: str, project_id: str) -> Path: return preferred +def _resolve_file_root( + email: str, + project_id: str, + space_id: str | None = None, + user_id: str | int | None = None, +) -> Path: + if space_id: + resolver = get_workspace_resolver() + space_root = resolver.space_root( + space_id=space_id, + project_id=project_id, + email=email, + user_id=user_id, + ) + if space_root is not None: + return space_root + return _resolve_project_root(email, project_id) + + @router.get("/files") async def list_project_files( project_id: str = Query(..., description="Project ID"), email: str = Query(..., description="User email"), + space_id: str | None = Query(None, description="Optional Space ID"), + user_id: str | None = Query( + None, description="Optional canonical user ID" + ), task_id: str | None = Query( None, description="Optional task ID to scope listing" ), @@ -207,7 +231,7 @@ async def list_project_files( status_code=400, detail="project_id and email are required", ) - project_root = _resolve_project_root(email, project_id) + project_root = _resolve_file_root(email, project_id, space_id, user_id) list_dir = str(project_root) if task_id: list_dir = str(project_root / f"task_{task_id}") @@ -234,7 +258,13 @@ async def list_project_files( result.append( { "filename": Path(abs_path).name, - "url": f"/files/stream?path={path_param}&project_id={quote(project_id)}&email={quote(email)}", + "url": ( + f"/files/stream?path={path_param}" + f"&project_id={quote(project_id)}" + f"&email={quote(email)}" + + (f"&space_id={quote(space_id)}" if space_id else "") + + (f"&user_id={quote(user_id)}" if user_id else "") + ), "relativePath": rel, } ) @@ -248,6 +278,10 @@ async def stream_file( path: str = Query(..., description="Relative path from project root"), project_id: str = Query(..., description="Project ID"), email: str = Query(..., description="User email"), + space_id: str | None = Query(None, description="Optional Space ID"), + user_id: str | None = Query( + None, description="Optional canonical user ID" + ), ): """ Stream file content. Path must be relative to project root. @@ -258,7 +292,7 @@ async def stream_file( status_code=400, detail="path, project_id and email are required", ) - project_root = _resolve_project_root(email, project_id) + project_root = _resolve_file_root(email, project_id, space_id, user_id) # Resolve path and ensure it stays under project root (security) try: resolved = resolve_under_base(path, str(project_root.resolve())) @@ -285,6 +319,10 @@ async def preview_file( email: str, project_id: str, file_path: str, + space_id: str | None = Query(None, description="Optional Space ID"), + user_id: str | None = Query( + None, description="Optional canonical user ID" + ), ): """ Preview file content with a path-based URL so relative references inside @@ -296,7 +334,7 @@ async def preview_file( detail="file_path, project_id and email are required", ) - project_root = _resolve_project_root(email, project_id) + project_root = _resolve_file_root(email, project_id, space_id, user_id) try: resolved = resolve_under_base(file_path, str(project_root.resolve())) except Exception as e: diff --git a/backend/app/model/chat.py b/backend/app/model/chat.py index 559faa21..1aee1382 100644 --- a/backend/app/model/chat.py +++ b/backend/app/model/chat.py @@ -53,6 +53,12 @@ McpServers = dict[Literal["mcpServers"], dict[str, dict]] class Chat(BaseModel): task_id: str project_id: str + space_id: str | None = None + run_id: str | None = None + space_root_path: str | None = None + workdir_mode: ( + Literal["worktree", "copy", "direct-write", "artifact-only"] | None + ) = None question: str email: str attaches: list[str] = [] @@ -78,13 +84,16 @@ class Chat(BaseModel): # (e.g., GOOGLE_API_KEY, SEARCH_ENGINE_ID) search_config: dict[str, str] | None = None # User identifier for user-specific skill configurations - user_id: str | None = None + user_id: str | int | None = None # Direct server API base URL (for example http://localhost:3001/api/v1) # used by standalone Brain to sync replay steps without Electron env injection. server_url: str | None = None session_mode: Literal["workforce", "single-agent"] = "workforce" toolkit_config: dict[str, Any] | None = None remote_sub_agent_config: RemoteSubAgentConfig | None = None + # Durable Project context reconstructed from persisted runs after restart. + # In-process follow-ups still prefer TaskLock.conversation_history. + project_context: str | None = None @field_validator("model_type") @classmethod @@ -131,19 +140,40 @@ class Chat(BaseModel): ) def file_save_path(self, path: str | None = None): - email = re.sub(r'[\\/*?:"<>|\s]', "_", self.email.split("@")[0]).strip( - "." - ) + legacy_owner_key = re.sub( + r'[\\/*?:"<>|\s]', "_", self.email.split("@")[0] + ).strip(".") + if self.user_id is not None and str(self.user_id).strip(): + owner_key = "user_" + re.sub( + r'[\\/*?:"<>|\s]', "_", str(self.user_id) + ).strip(".") + else: + owner_key = legacy_owner_key + run_id = self.run_id or self.task_id # Use project-based structure: project_{project_id}/task_{task_id} - save_path = ( + project_base = ( Path.home() / "eigent" - / email + / owner_key / f"project_{self.project_id}" - / f"task_{self.task_id}" + / f"task_{run_id}" ) - if path is not None: - save_path = save_path / path + legacy_project_base = ( + Path.home() + / "eigent" + / legacy_owner_key + / f"project_{self.project_id}" + / f"task_{run_id}" + ) + if ( + owner_key != legacy_owner_key + and not project_base.exists() + and legacy_project_base.exists() + ): + # Bridge old installs whose artifacts were written under + # ~/eigent/{email_sanitized} before user_id-owned roots existed. + project_base = legacy_project_base + save_path = project_base / path if path is not None else project_base save_path.mkdir(parents=True, exist_ok=True) return str(save_path) @@ -153,6 +183,7 @@ class SupplementChat(BaseModel): question: str task_id: str | None = None attaches: list[str] = [] + project_context: str | None = None class HumanReply(BaseModel): diff --git a/backend/app/service/task.py b/backend/app/service/task.py index bc78e126..1fc1e8de 100644 --- a/backend/app/service/task.py +++ b/backend/app/service/task.py @@ -33,6 +33,7 @@ from app.model.chat import ( UpdateData, ) from app.model.enums import Status +from app.run_context import RunContext logger = logging.getLogger("task_service") @@ -79,6 +80,7 @@ class ImprovePayload(BaseModel): question: str attaches: list[str] = [] + project_context: str | None = None class ActionImproveData(BaseModel): @@ -359,12 +361,36 @@ class TaskLock: """Compressed summary of older serialized agent memory""" last_task_result: str """Store the last task execution result""" + last_task_summary: str + """Store the last generated task summary""" question_agent: Any | None """Persistent question confirmation agent""" summary_generated: bool """Track if summary has been generated for this project""" current_task_id: str | None """Current task ID to be used in SSE responses""" + run_context: RunContext | None + """Current task-scoped runtime context for this Project.""" + user_id: str | int | None + """Canonical user id when provided by the control plane.""" + working_directory: str | None + """Resolved source/work directory for the current Run.""" + task_output_root: str | None + """Resolved artifact/output directory for the current Run.""" + task_start_time: float | None + """Timestamp captured when the current Run directories were frozen.""" + email: str | None + """Legacy/display user email associated with the current Run.""" + project_id: str | None + """Project id associated with the current Run.""" + space_id: str | None + """Space id associated with the current Run.""" + workdir_mode: str | None + """Actual workdir mode used by the current Run.""" + base_snapshot_id: str | None + """Project workdir baseline snapshot id, when available.""" + new_folder_path: Any | None + """Legacy cleanup marker for default output directories.""" def __init__( self, id: str, queue: asyncio.Queue, human_input: dict @@ -384,7 +410,19 @@ class TaskLock: self.last_task_result = "" self.last_task_summary = "" self.question_agent = None + self.summary_generated = False self.current_task_id = None + self.run_context = None + self.user_id = None + self.working_directory = None + self.task_output_root = None + self.task_start_time = None + self.email = None + self.project_id = None + self.space_id = None + self.workdir_mode = None + self.base_snapshot_id = None + self.new_folder_path = None logger.info( "Task lock initialized",