mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-20 14:13:32 +00:00
feat(backend): wire chat_controller into RunContext + Space workspace
This commit is contained in:
4pmtong
parent
f3c9b8ee23
commit
4fa584896a
4 changed files with 330 additions and 83 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue