From 93f6d647d4bdc4e443486a26081d2e64861559d0 Mon Sep 17 00:00:00 2001 From: Wang Qi Date: Thu, 2 Jul 2026 20:49:51 +0800 Subject: [PATCH] Fix the sandbox exec image cannot show and download (#16577) --- agent/tools/code_exec.py | 24 ++----- api/apps/restful_apis/agent_api.py | 4 +- api/apps/restful_apis/document_api.py | 94 ++++++++++++++++++--------- api/db/services/canvas_service.py | 4 +- 4 files changed, 75 insertions(+), 51 deletions(-) diff --git a/agent/tools/code_exec.py b/agent/tools/code_exec.py index 3133784e2..2cb58c23a 100644 --- a/agent/tools/code_exec.py +++ b/agent/tools/code_exec.py @@ -69,9 +69,7 @@ def select_business_output(outputs: Mapping[str, object]) -> tuple[str, object]: business_outputs = [(name, meta) for name, meta in outputs.items() if name not in SYSTEM_OUTPUT_KEYS] if len(business_outputs) != 1: - raise ContractError( - f"CodeExec contract must contain exactly one business output, got {len(business_outputs)}" - ) + raise ContractError(f"CodeExec contract must contain exactly one business output, got {len(business_outputs)}") _validate_business_output_name(business_outputs[0][0]) return business_outputs[0] @@ -124,10 +122,7 @@ def _is_number(value) -> bool: def _validate_top_level_value_domain(value) -> None: allowed = value is None or isinstance(value, (bool, str, dict, list)) or _is_number(value) if not allowed: - raise ContractError( - f"CodeExec unsupported top-level result type: {type(value).__name__}. " - "Allowed top-level values are String, Number, Boolean, Object, Array, or Null." - ) + raise ContractError(f"CodeExec unsupported top-level result type: {type(value).__name__}. Allowed top-level values are String, Number, Boolean, Object, Array, or Null.") def _normalize_expected_type(expected_type: str) -> str: @@ -161,9 +156,7 @@ def _validate_expected_type(expected_type: str, value, path: str = "") -> None: if etype.startswith("Array<") and etype.endswith(">"): inner_type = etype[6:-1].strip() if not isinstance(value, list): - raise ContractError( - f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {infer_actual_type(value)}" - ) + raise ContractError(f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {infer_actual_type(value)}") for index, item in enumerate(value): child_path = f"{path}[{index}]" if path else f"[{index}]" _validate_expected_type(inner_type, item, child_path) @@ -184,9 +177,7 @@ def _validate_expected_type(expected_type: str, value, path: str = "") -> None: raise ContractError(f"Unsupported expected type: {expected_type}") if not valid: - raise ContractError( - f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {actual_type}" - ) + raise ContractError(f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {actual_type}") def build_code_exec_contract(outputs: Mapping[str, object], raw_result) -> dict[str, object]: @@ -371,10 +362,7 @@ class CodeExec(ToolBase, ABC): reload_provider() provider_info = get_provider_info() provider_type = provider_info.get("provider_type") or "unknown" - logging.info( - f"[CodeExec]: dispatching execution to sandbox provider '{provider_type}' " - f"(language={language}, timeout={timeout_seconds}s)" - ) + logging.info(f"[CodeExec]: dispatching execution to sandbox provider '{provider_type}' (language={language}, timeout={timeout_seconds}s)") # Execute code using the provider system result = sandbox_execute_code(code=code, language=language, timeout=timeout_seconds, arguments=arguments) @@ -556,7 +544,7 @@ class CodeExec(ToolBase, ABC): settings.STORAGE_IMPL.put(SANDBOX_ARTIFACT_BUCKET, storage_name, binary) - url = f"/api/v1/documents/artifact/{storage_name}" + url = f"/api/v1/documents/artifact/{storage_name}?session_id={self._canvas.task_id}" uploaded.append( { "name": name, diff --git a/api/apps/restful_apis/agent_api.py b/api/apps/restful_apis/agent_api.py index ae28cd041..bf8840c26 100644 --- a/api/apps/restful_apis/agent_api.py +++ b/api/apps/restful_apis/agent_api.py @@ -1407,7 +1407,7 @@ async def agent_chat_completion(tenant_id, agent_id=None): dsl_str = workflow_dsl else: dsl_str = json.dumps(workflow_dsl, ensure_ascii=False) - canvas = Canvas(dsl_str, str(tenant_id), canvas_id=agent_id, custom_header=custom_header) + canvas = Canvas(dsl_str, str(tenant_id), task_id=session_id, canvas_id=agent_id, custom_header=custom_header) except Exception as exc: return server_error_response(exc) @@ -1528,7 +1528,7 @@ async def agent_chat_completion(tenant_id, agent_id=None): try: from agent.canvas import Canvas - canvas = Canvas(dsl_str, str(tenant_id), canvas_id=agent_id, custom_header=custom_header) + canvas = Canvas(dsl_str, str(tenant_id), task_id=session_id, canvas_id=agent_id, custom_header=custom_header) canvas.clear_history() except Exception as exc: return server_error_response(exc) diff --git a/api/apps/restful_apis/document_api.py b/api/apps/restful_apis/document_api.py index 846174b10..b9eecc188 100644 --- a/api/apps/restful_apis/document_api.py +++ b/api/apps/restful_apis/document_api.py @@ -20,15 +20,21 @@ import os import re from pathlib import Path -from quart import request, make_response,send_file +from quart import request, make_response, send_file from peewee import OperationalError from pydantic import ValidationError from api.apps import AUTH_JWT, AUTH_API, AUTH_BETA, current_user, login_required from api.constants import FILE_NAME_LEN_LIMIT, IMG_BASE64_PREFIX -from api.apps.services.document_api_service import validate_document_update_fields, map_doc_keys, \ - map_doc_keys_with_run_status, update_document_name_only, update_chunk_method, update_document_status_only, \ - reset_document_for_reparse +from api.apps.services.document_api_service import ( + validate_document_update_fields, + map_doc_keys, + map_doc_keys_with_run_status, + update_document_name_only, + update_chunk_method, + update_document_status_only, + reset_document_for_reparse, +) from api.db import VALID_FILE_TYPES, FileType from api.db.db_models import API4Conversation, DB from api.db.services import duplicate_name @@ -41,11 +47,24 @@ from api.db.services.knowledgebase_service import KnowledgebaseService from api.db.services.canvas_service import UserCanvasService from api.common.check_team_permission import check_kb_team_permission from api.db.services.task_service import TaskService, cancel_all_task_of -from api.utils.api_utils import construct_json_result, get_data_error_result, get_error_data_result, get_result, get_json_result, \ - server_error_response, add_tenant_id_to_kwargs, get_request_json, get_error_argument_result, check_duplicate_ids +from api.utils.api_utils import ( + construct_json_result, + get_data_error_result, + get_error_data_result, + get_result, + get_json_result, + server_error_response, + add_tenant_id_to_kwargs, + get_request_json, + get_error_argument_result, + check_duplicate_ids, +) from api.utils.pagination_utils import validate_rest_api_page_size from api.utils.validation_utils import ( - UpdateDocumentReq, format_validation_error_message, validate_and_parse_json_request, DeleteDocumentReq, + UpdateDocumentReq, + format_validation_error_message, + validate_and_parse_json_request, + DeleteDocumentReq, ) from common import settings @@ -121,7 +140,7 @@ async def upload_info(tenant_id: str): return server_error_response(e) -@manager.route("/datasets//documents/", methods=["PATCH"]) # noqa: F821 +@manager.route("/datasets//documents/", methods=["PATCH"]) # noqa: F821 @login_required @add_tenant_id_to_kwargs async def update_document(tenant_id, dataset_id, document_id): @@ -211,8 +230,7 @@ async def update_document(tenant_id, dataset_id, document_id): return error # "parser_id" provided but does not match with existing doc's file type - if "parser_id" in req and ((doc.type == FileType.VISUAL and req["parser_id"] != "picture") - or (re.search(r"\.(ppt|pptx|pages)$", doc.name) and req["parser_id"] != "presentation")): + if "parser_id" in req and ((doc.type == FileType.VISUAL and req["parser_id"] != "picture") or (re.search(r"\.(ppt|pptx|pages)$", doc.name) and req["parser_id"] != "presentation")): return get_data_error_result(message="Not supported yet!") # parser config provided (already validated in UpdateDocumentReq), update it @@ -229,7 +247,7 @@ async def update_document(tenant_id, dataset_id, document_id): if error := update_chunk_method(req, doc, tenant_id): return error - if "enabled" in req: # already checked in UpdateDocumentReq - it's int if present + if "enabled" in req: # already checked in UpdateDocumentReq - it's int if present # "enabled" flag provided, the update method will check if it's changed and then update if so if error := update_document_status_only(int(req["enabled"]), doc, kb): return error @@ -604,7 +622,10 @@ async def _upload_local_documents(kb, tenant_id): parser_config_override = None err, files = await thread_pool_exec( - FileService.upload_document, kb, file_objs, tenant_id, + FileService.upload_document, + kb, + file_objs, + tenant_id, parent_path=form.get("parent_path"), parser_config_override=parser_config_override, ) @@ -772,7 +793,7 @@ def list_docs(dataset_id, tenant_id): return get_json_result(data={"total": total, "docs": renamed_doc_list}) -def _get_docs_with_request(req, dataset_id:str): +def _get_docs_with_request(req, dataset_id: str): """Get documents with request parameters from a dataset. This function extracts filtering parameters from the request and returns @@ -839,7 +860,7 @@ def _get_docs_with_request(req, dataset_id:str): if doc_id: if not DocumentService.query(id=doc_id, kb_id=dataset_id): return RetCode.DATA_ERROR, f"You don't own the document {doc_id}.", [], 0 - doc_ids_filter = [doc_id] # id provided, ignore other filters + doc_ids_filter = [doc_id] # id provided, ignore other filters if doc_name and not DocumentService.query(name=doc_name, kb_id=dataset_id): return RetCode.DATA_ERROR, f"You don't own the document {doc_name}.", [], 0 @@ -849,8 +870,9 @@ def _get_docs_with_request(req, dataset_id:str): if len(doc_ids) > 0: doc_ids_filter = doc_ids - docs, total = DocumentService.get_by_kb_id(dataset_id, page, page_size, orderby, desc, keywords, run_status_converted, types, suffix, - name=doc_name, doc_ids=doc_ids_filter, return_empty_metadata=return_empty_metadata) + docs, total = DocumentService.get_by_kb_id( + dataset_id, page, page_size, orderby, desc, keywords, run_status_converted, types, suffix, name=doc_name, doc_ids=doc_ids_filter, return_empty_metadata=return_empty_metadata + ) # time range filter (0 means no bound) create_time_from = int(q.get("create_time_from", 0)) @@ -907,6 +929,7 @@ def _parse_run_status_filter(req_args): invalid_statuses = {status for status in converted if status not in valid_statuses} return converted, invalid_statuses + def _parse_doc_id_filter_with_metadata(req, kb_id): """Parse document ID filter based on metadata conditions from the request. @@ -976,12 +999,12 @@ def _parse_doc_id_filter_with_metadata(req, kb_id): try: metadata_condition = json.loads(req.get("metadata_condition", "{}")) except json.JSONDecodeError: - msg = f'metadata_condition must be valid JSON: {req.get("metadata_condition")}.' + msg = f"metadata_condition must be valid JSON: {req.get('metadata_condition')}." return RetCode.DATA_ERROR, msg, [], return_empty_metadata try: metadata = json.loads(req.get("metadata", "{}")) except json.JSONDecodeError: - logging.error(msg=f'metadata must be valid JSON: {req.get("metadata")}.') + logging.error(msg=f"metadata must be valid JSON: {req.get('metadata')}.") return RetCode.DATA_ERROR, "metadata must be valid JSON.", [], return_empty_metadata if isinstance(metadata, dict) and metadata.get("empty_metadata"): @@ -1106,9 +1129,7 @@ async def delete_documents(tenant_id, dataset_id): dataset_doc_ids = {doc.id for doc in DocumentService.query(kb_id=dataset_id)} invalid_ids = [doc_id for doc_id in doc_ids if doc_id not in dataset_doc_ids] if invalid_ids: - return get_error_data_result( - message=f"These documents do not belong to dataset {dataset_id} or Document not found: {', '.join(invalid_ids)}" - ) + return get_error_data_result(message=f"These documents do not belong to dataset {dataset_id} or Document not found: {', '.join(invalid_ids)}") # make sure each id is unique unique_doc_ids, duplicate_messages = check_duplicate_ids(doc_ids, "document") @@ -1128,6 +1149,7 @@ async def delete_documents(tenant_id, dataset_id): logging.exception(e) return get_error_data_result(message="Internal server error") + @manager.route("/datasets//documents//metadata/config", methods=["PUT"]) # noqa: F821 @login_required @add_tenant_id_to_kwargs @@ -1349,17 +1371,13 @@ async def update_metadata(tenant_id, dataset_id): kb_doc_ids = KnowledgebaseService.list_documents_by_ids([dataset_id]) invalid_ids = set(document_ids) - set(kb_doc_ids) if invalid_ids: - return get_error_data_result( - message=f"These documents do not belong to dataset {dataset_id}: {', '.join(invalid_ids)}" - ) + return get_error_data_result(message=f"These documents do not belong to dataset {dataset_id}: {', '.join(invalid_ids)}") target_doc_ids = set(document_ids) # Apply metadata_condition filtering if provided if metadata_condition: metas = DocMetadataService.get_flatted_meta_by_kbs([dataset_id]) - filtered_ids = set( - meta_filter(metas, convert_conditions(metadata_condition), metadata_condition.get("logic", "and")) - ) + filtered_ids = set(meta_filter(metas, convert_conditions(metadata_condition), metadata_condition.get("logic", "and"))) target_doc_ids = target_doc_ids & filtered_ids if metadata_condition.get("conditions") and not target_doc_ids: return get_result(data={"updated": 0, "matched_docs": 0}) @@ -1389,7 +1407,8 @@ async def ingest(tenant_id): logging.exception("document ingest/run failed") return server_error_response(e) -def _run_sync(user_id:str, req): + +def _run_sync(user_id: str, req): for doc_id in req["doc_ids"]: if not DocumentService.accessible(doc_id, user_id): return RetCode.AUTHENTICATION_ERROR, "No authorization." @@ -1513,6 +1532,7 @@ async def parse_documents(tenant_id, dataset_id): return get_error_data_result(message=f"Documents not found: {not_found_ids}") try: + def _run_sync(): kb_table_num_map = {} success_count = 0 @@ -1622,6 +1642,7 @@ async def stop_parse_documents(tenant_id, dataset_id): return get_error_data_result(message=f"Documents not found: {not_found_ids}") try: + def _run_sync(): success_count = 0 for doc_id in valid_doc_ids: @@ -1787,6 +1808,18 @@ def _sandbox_artifact_accessible(filename: str, user_id: str) -> bool: return False +@DB.connection_context() +def _sandbox_artifact_session_accessible(session_id: str, user_id: str) -> bool: + if not session_id: + return False + conv = API4Conversation.get_or_none(API4Conversation.id == session_id) + if not conv: + return False + if str(conv.user_id) != str(user_id) and str(conv.exp_user_id or "") != str(user_id): + return False + return UserCanvasService.accessible(conv.dialog_id, user_id) + + @manager.route("/documents/artifact/", methods=["GET"]) # noqa: F821 @login_required async def get_artifact(filename): @@ -1823,7 +1856,8 @@ async def get_artifact(filename): ext = os.path.splitext(basename)[1].lower() if ext not in ARTIFACT_CONTENT_TYPES: return get_data_error_result(message="Invalid file type.") - if not await thread_pool_exec(_sandbox_artifact_accessible, basename, current_user.id): + session_id = request.args.get("session_id", "") + if not await thread_pool_exec(_sandbox_artifact_accessible, basename, current_user.id) and not await thread_pool_exec(_sandbox_artifact_session_accessible, session_id, current_user.id): return get_data_error_result(message="Artifact not found.") data = await thread_pool_exec(settings.STORAGE_IMPL.get, bucket, basename) if not data: @@ -1959,6 +1993,7 @@ async def batch_update_document_status(tenant_id, dataset_id): return get_json_result(data=result, message="Partial failure", code=RetCode.SERVER_ERROR) return get_json_result(data=result) + @manager.route("/documents//preview", methods=["GET"]) # noqa: F821 @login_required(auth_types=[AUTH_JWT, AUTH_API, AUTH_BETA]) async def get(doc_id): @@ -2064,6 +2099,7 @@ async def download(dataset_id, document_id): mimetype=_mimetype_for_document(doc[0]), ) + @manager.route("/documents/", methods=["GET"]) # noqa: F821 @login_required async def download_document(document_id): diff --git a/api/db/services/canvas_service.py b/api/db/services/canvas_service.py index e6ff246b3..308f495c1 100644 --- a/api/db/services/canvas_service.py +++ b/api/db/services/canvas_service.py @@ -316,12 +316,12 @@ async def completion(tenant_id, agent_id, session_id=None, **kwargs): conv.message = [] if not isinstance(conv.dsl, str): conv.dsl = json.dumps(conv.dsl, ensure_ascii=False) - canvas = Canvas(conv.dsl, tenant_id, agent_id, canvas_id=agent_id, custom_header=custom_header) + canvas = Canvas(conv.dsl, tenant_id, task_id=session_id, canvas_id=agent_id, custom_header=custom_header) else: cvs, dsl = await thread_pool_exec(UserCanvasService.get_agent_dsl_with_release, agent_id, release_mode=release_mode == "true", tenant_id=tenant_id) session_id = get_uuid() - canvas = Canvas(dsl, tenant_id, agent_id, canvas_id=cvs.id, custom_header=custom_header) + canvas = Canvas(dsl, tenant_id, task_id=session_id, canvas_id=cvs.id, custom_header=custom_header) canvas.reset() # Get the version title based on release_mode version_title = await thread_pool_exec(UserCanvasVersionService.get_latest_version_title, cvs.id, release_mode=release_mode == "true")