Merge branch 'feat/agent-v2' into feat/agent-v2-e2e-test

Co-authored-by: CodingOnStar <[email protected]>

.github/workflows/web-e2e.yml

@@ -1,6 +1,10 @@
 name: Web Full-Stack E2E
 
 on:
+  push:
+    branches:
+      - feat/agent-v2-e2e-test
+
   workflow_call:
     inputs:
       run-external-runtime:
@@ -59,7 +63,7 @@ jobs:
         run: vp run e2e:full
 
       - name: Run external runtime E2E tests
-        if: ${{ inputs.run-external-runtime }}
+        if: ${{ inputs.run-external-runtime || github.ref == 'refs/heads/feat/agent-v2-e2e-test' }}
         working-directory: ./e2e
         env:
           E2E_ADMIN_EMAIL: [email protected]

api/.env.example

@@ -663,6 +663,9 @@ PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
 
+# Dify Agent backend
+AGENT_BACKEND_BASE_URL=http://localhost:5050
+
 # Marketplace configuration
 MARKETPLACE_ENABLED=true
 MARKETPLACE_API_URL=https://marketplace.dify.ai

api/commands/__init__.py

@@ -26,7 +26,6 @@ from .rbac import migrate_dataset_permissions_to_rbac, migrate_member_roles_to_r
 from .retention import (
     archive_workflow_runs,
     archive_workflow_runs_plan,
-    backfill_workflow_run_archive_bundles,
     clean_expired_messages,
     clean_workflow_runs,
     cleanup_orphaned_draft_variables,
@@ -55,7 +54,6 @@ __all__ = [
     "archive_workflow_runs",
     "archive_workflow_runs_plan",
     "backfill_plugin_auto_upgrade",
-    "backfill_workflow_run_archive_bundles",
     "clean_expired_messages",
     "clean_workflow_runs",
     "cleanup_orphaned_draft_variables",

api/commands/retention.py

@@ -56,15 +56,6 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
     return sorted(set(parsed))
 
 
-def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[str] | None:
-    if raw_ids is None:
-        return None
-    parsed = sorted({raw_id.strip() for raw_id in raw_ids.split(",") if raw_id.strip()})
-    if not parsed:
-        raise click.BadParameter(f"{param_name} must not be empty")
-    return parsed
-
-
 def _get_archive_candidate_tenant_ids_by_prefix(
     prefix: str,
     *,
@@ -648,82 +639,6 @@ def archive_workflow_runs(
     )
 
 
-@click.command(
-    "backfill-workflow-run-archive-bundles",
-    help="Backfill workflow-run archive bundle DB index from object-storage manifests.",
-)
-@click.option("--tenant-ids", default=None, help="Optional comma-separated tenant IDs.")
-@click.option(
-    "--tenant-prefixes",
-    default=None,
-    help="Optional comma-separated tenant ID first hex digits, e.g. 0,1,a,f.",
-)
-@click.option("--year", default=None, type=click.IntRange(min=1, max=9999), help="Optional archive year filter.")
-@click.option("--month", default=None, type=click.IntRange(min=1, max=12), help="Optional archive month filter.")
-@click.option("--limit", default=None, type=click.IntRange(min=1), help="Maximum number of manifests to process.")
-@click.option("--dry-run", is_flag=True, help="Preview without writing workflow_run_archive_bundles.")
-def backfill_workflow_run_archive_bundles(
-    tenant_ids: str | None,
-    tenant_prefixes: str | None,
-    year: int | None,
-    month: int | None,
-    limit: int | None,
-    dry_run: bool,
-) -> None:
-    """
-    Reconcile `workflow_run_archive_bundles` from V2 archive manifests.
-
-    This command is meant for bootstrapping the listing/download index after deploy or repairing index drift. The R2
-    manifests remain the source of truth; this command only mirrors their query metadata into the database.
-    """
-    from services.retention.workflow_run.archive_bundle_index import WorkflowRunArchiveBundleIndexBackfill
-
-    if tenant_ids and tenant_prefixes:
-        raise click.UsageError("Choose either --tenant-ids or --tenant-prefixes, not both.")
-    if month is not None and year is None:
-        raise click.UsageError("--month must be used with --year.")
-
-    parsed_tenant_ids = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")
-    parsed_tenant_prefixes = _parse_tenant_prefixes(tenant_prefixes)
-    if not parsed_tenant_ids and not parsed_tenant_prefixes:
-        click.echo(
-            click.style(
-                "No tenant scope supplied; scanning the full workflow-runs/v2/ archive prefix.",
-                fg="yellow",
-            )
-        )
-
-    started_at = datetime.datetime.now(datetime.UTC)
-    click.echo(click.style(f"Starting archive bundle index backfill at {started_at.isoformat()}.", fg="white"))
-
-    backfill = WorkflowRunArchiveBundleIndexBackfill()
-    summary = backfill.run(
-        tenant_ids=parsed_tenant_ids,
-        tenant_prefixes=parsed_tenant_prefixes or None,
-        year=year,
-        month=month,
-        limit=limit,
-        dry_run=dry_run,
-    )
-    status = "completed with failures" if summary.bundles_failed else "completed successfully"
-    fg = "red" if summary.bundles_failed else "green"
-    action = "would_upsert" if dry_run else "upserted"
-    action_count = summary.bundles_processed if dry_run else summary.bundles_upserted
-    click.echo(
-        click.style(
-            f"Backfill {status}. manifests_found={summary.manifests_found} "
-            f"bundles_processed={summary.bundles_processed} {action}={action_count} "
-            f"bundles_failed={summary.bundles_failed} runs={summary.workflow_run_count} rows={summary.row_count} "
-            f"archive_bytes={summary.archive_bytes} duration={summary.elapsed_time:.2f}s",
-            fg=fg,
-        )
-    )
-    for error in summary.errors[:10]:
-        click.echo(click.style(f"  failed {error}", fg="red"))
-    if len(summary.errors) > 10:
-        click.echo(click.style(f"  ... and {len(summary.errors) - 10} more failures", fg="red"))
-
-
 def _echo_bundle_archive_operation_summary(summary) -> None:
     status = "completed successfully" if summary.bundles_failed == 0 else "completed with failures"
     fg = "green" if summary.bundles_failed == 0 else "red"

api/configs/extra/agent_backend_config.py

@@ -25,11 +25,10 @@ class AgentBackendConfig(BaseSettings):
     AGENT_SHELL_ENABLED: bool = Field(
         description=(
             "Inject the dify.shell layer (sandboxed bash workspace) into Agent runs. "
-            "Requires the agent backend to be wired with a shellctl entrypoint; keep it "
-            "off until shellctl is deployed, otherwise every agent run that includes the "
-            "shell layer will fail."
+            "Requires the agent backend to be wired with a shellctl entrypoint before "
+            "shell-using Agent runs are executed."
         ),
-        default=False,
+        default=True,
     )
 
     AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS: NonNegativeFloat = Field(

api/configs/feature/__init__.py

@@ -363,7 +363,10 @@ class FileAccessConfig(BaseSettings):
     INTERNAL_FILES_URL: str = Field(
         description="Internal base URL for file access within Docker network,"
         " used for plugin daemon and internal service communication."
-        " Falls back to FILES_URL if not specified.",
+        " Explicit INTERNAL_FILES_URL takes precedence; otherwise SERVER_CONSOLE_API_URL is used,"
+        " then FILES_URL.",
+        validation_alias=AliasChoices("INTERNAL_FILES_URL", "SERVER_CONSOLE_API_URL"),
+        alias_priority=1,
         default="",
     )
 

api/controllers/console/__init__.py

@@ -43,7 +43,6 @@ from . import (
     setup,
     spec,
     version,
-    workflow_run_archive,
 )
 from .agent import composer as agent_composer
 from .agent import roster as agent_roster
@@ -239,7 +238,6 @@ __all__ = [
     "workflow_draft_variable",
     "workflow_node_output_inspector",
     "workflow_run",
-    "workflow_run_archive",
     "workflow_statistic",
     "workflow_trigger",
     "workspace",

api/controllers/console/workflow_run_archive.py

@@ -1,221 +0,0 @@
-import datetime
-from http import HTTPStatus
-
-from flask import redirect
-from flask_restx import Resource
-from pydantic import BaseModel, Field
-from werkzeug.exceptions import Conflict, NotFound
-
-from controllers.common.fields import RedirectResponse
-from controllers.common.schema import register_response_schema_models, register_schema_models
-from controllers.console import console_ns
-from controllers.console.wraps import (
-    RBACPermission,
-    RBACResourceScope,
-    account_initialization_required,
-    is_admin_or_owner_required,
-    rbac_permission_required,
-    setup_required,
-)
-from extensions.ext_database import db
-from fields.base import ResponseModel
-from libs.archive_storage import get_export_storage
-from libs.helper import dump_response
-from libs.login import current_account_with_tenant, login_required
-from services.retention.workflow_run.archive_download_preparation import ARCHIVE_DOWNLOAD_MIME_TYPE
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-)
-from services.retention.workflow_run.archive_log_service import (
-    WorkflowRunArchiveDownloadNotReadyError,
-    WorkflowRunArchiveDownloadTaskNotFoundError,
-    WorkflowRunArchiveNotFoundError,
-    create_workflow_run_archive_download_task,
-    get_ready_workflow_run_archive_download_task,
-    get_workflow_run_archive_download_task,
-    list_workflow_run_archives,
-)
-
-
-class WorkflowRunArchiveDownloadPayload(BaseModel):
-    """Request body for preparing one monthly workflow-run archive download."""
-
-    year: int = Field(ge=1)
-    month: int = Field(ge=1, le=12)
-
-
-class WorkflowRunArchiveSummaryResponse(ResponseModel):
-    archived_month_count: int
-    workflow_run_count: int
-    archive_bytes: int
-    latest_archived_at: datetime.datetime | None = None
-
-
-class WorkflowRunArchiveDownloadTaskResponse(ResponseModel):
-    download_id: str
-    year: int
-    month: int
-    bundle_count: int
-    archive_bytes: int
-    status: WorkflowRunArchiveDownloadStatus
-    file_name: str | None = None
-    file_size_bytes: int | None = None
-    error: str | None = None
-    created_at: datetime.datetime
-    updated_at: datetime.datetime
-    expires_at: datetime.datetime
-    started_at: datetime.datetime | None = None
-    finished_at: datetime.datetime | None = None
-
-
-class WorkflowRunArchiveMonthResponse(ResponseModel):
-    year: int
-    month: int
-    bundle_count: int
-    workflow_run_count: int
-    row_count: int
-    archive_bytes: int
-    latest_archived_at: datetime.datetime
-    download_task: WorkflowRunArchiveDownloadTaskResponse | None = None
-
-
-class WorkflowRunArchiveListResponse(ResponseModel):
-    summary: WorkflowRunArchiveSummaryResponse
-    months: list[WorkflowRunArchiveMonthResponse]
-
-
-register_schema_models(console_ns, WorkflowRunArchiveDownloadPayload)
-register_response_schema_models(
-    console_ns,
-    WorkflowRunArchiveSummaryResponse,
-    WorkflowRunArchiveMonthResponse,
-    WorkflowRunArchiveListResponse,
-    WorkflowRunArchiveDownloadTaskResponse,
-    RedirectResponse,
-)
-
-
-def _current_ids() -> tuple[str, str]:
-    """Return current `(tenant_id, account_id)` or raise when no workspace is selected."""
-    current_user, current_tenant_id = current_account_with_tenant()
-    if not current_tenant_id:
-        raise NotFound("Current workspace not found")
-    return current_tenant_id, current_user.id
-
-
-def _presigned_url_expires_in(expires_at: datetime.datetime) -> int:
-    """Keep the storage URL no longer-lived than the Redis task and cap it for browser downloads."""
-    expires_at_utc = expires_at if expires_at.tzinfo else expires_at.replace(tzinfo=datetime.UTC)
-    remaining_seconds = int((expires_at_utc - datetime.datetime.now(datetime.UTC)).total_seconds())
-    return max(1, min(3600, remaining_seconds))
-
-
-@console_ns.route("/workflow-run-archives")
-class WorkflowRunArchivesApi(Resource):
-    @console_ns.doc("list_workflow_run_archives")
-    @console_ns.doc(description="List monthly workflow-run archive metadata for the current workspace")
-    @console_ns.response(200, "Success", console_ns.models[WorkflowRunArchiveListResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
-    def get(self):
-        tenant_id, _ = _current_ids()
-        return dump_response(WorkflowRunArchiveListResponse, list_workflow_run_archives(db.session(), tenant_id))
-
-
-@console_ns.route("/workflow-run-archives/downloads")
-class WorkflowRunArchiveDownloadsApi(Resource):
-    @console_ns.doc("create_workflow_run_archive_download")
-    @console_ns.doc(description="Create or return a temporary workflow-run archive download task")
-    @console_ns.expect(console_ns.models[WorkflowRunArchiveDownloadPayload.__name__])
-    @console_ns.response(
-        HTTPStatus.ACCEPTED,
-        "Download task accepted",
-        console_ns.models[WorkflowRunArchiveDownloadTaskResponse.__name__],
-    )
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
-    def post(self):
-        tenant_id, account_id = _current_ids()
-        payload = WorkflowRunArchiveDownloadPayload.model_validate(console_ns.payload or {})
-        try:
-            task = create_workflow_run_archive_download_task(
-                db.session(),
-                tenant_id=tenant_id,
-                requested_by=account_id,
-                year=payload.year,
-                month=payload.month,
-            )
-        except WorkflowRunArchiveNotFoundError as exc:
-            raise NotFound(str(exc)) from exc
-        return dump_response(WorkflowRunArchiveDownloadTaskResponse, task), HTTPStatus.ACCEPTED
-
-
-@console_ns.route("/workflow-run-archives/downloads/<string:download_id>")
-class WorkflowRunArchiveDownloadApi(Resource):
-    @console_ns.doc("get_workflow_run_archive_download")
-    @console_ns.doc(description="Get a temporary workflow-run archive download task")
-    @console_ns.response(200, "Success", console_ns.models[WorkflowRunArchiveDownloadTaskResponse.__name__])
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
-    def get(self, download_id: str):
-        tenant_id, _ = _current_ids()
-        try:
-            task = get_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
-        except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
-            raise NotFound(str(exc)) from exc
-        return dump_response(WorkflowRunArchiveDownloadTaskResponse, task)
-
-
-@console_ns.route("/workflow-run-archives/downloads/<string:download_id>/file")
-class WorkflowRunArchiveDownloadFileApi(Resource):
-    @console_ns.doc("download_workflow_run_archive_file")
-    @console_ns.doc(description="Redirect to a prepared workflow-run archive ZIP file")
-    @console_ns.response(
-        302,
-        "Redirect to pre-signed archive storage URL",
-        console_ns.models[RedirectResponse.__name__],
-    )
-    @console_ns.response(409, "Download task is not ready")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
-    def get(self, download_id: str):
-        tenant_id, _ = _current_ids()
-        try:
-            task = get_ready_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
-        except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
-            raise NotFound(str(exc)) from exc
-        except WorkflowRunArchiveDownloadNotReadyError as exc:
-            raise Conflict(str(exc)) from exc
-
-        storage_key = task.storage_key
-        if storage_key is None:
-            raise Conflict(f"Workflow run archive download is not ready: {download_id}")
-
-        storage = get_export_storage()
-        presigned_url = storage.generate_presigned_url(
-            storage_key,
-            expires_in=_presigned_url_expires_in(task.expires_at),
-            filename=task.file_name,
-            content_type=ARCHIVE_DOWNLOAD_MIME_TYPE,
-        )
-        return redirect(presigned_url, code=HTTPStatus.FOUND)

api/controllers/files/upload.py

@@ -1,5 +1,3 @@
-from mimetypes import guess_extension
-
 from flask import request
 from flask_restx import Resource
 from flask_restx.api import HTTPStatus
@@ -8,7 +6,7 @@ from werkzeug.exceptions import Forbidden
 
 import services
 from core.tools.signature import verify_plugin_file_signature
-from core.tools.tool_file_manager import ToolFileManager
+from core.tools.tool_file_manager import ToolFileManager, resolve_extension
 from core.workflow.file_reference import build_file_reference
 from fields.file_fields import FileResponse
 
@@ -110,7 +108,7 @@ class PluginUploadFileApi(Resource):
                 conversation_id=args.conversation_id,
             )
 
-            extension = guess_extension(tool_file.mimetype) or ".bin"
+            extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
             preview_url = ToolFileManager.sign_file(tool_file_id=tool_file.id, extension=extension)
 
             # Create a dictionary with all the necessary attributes

api/controllers/inner_api/plugin/plugin.py

@@ -476,6 +476,7 @@ class PluginDownloadFileRequestApi(Resource):
             user_from=payload.user_from,
             invoke_from=payload.invoke_from,
             file_mapping=payload.file.model_dump(mode="python", exclude_none=True),
+            for_external=payload.for_external,
         )
         return BaseBackwardsInvocationResponse(
             data={

api/core/app/apps/agent_app/app_runner.py

@@ -27,6 +27,7 @@ from clients.agent_backend import (
     AgentBackendInternalEventType,
     AgentBackendRunClient,
     AgentBackendRunEventAdapter,
+    AgentBackendRunFailedInternalEvent,
     AgentBackendRunSucceededInternalEvent,
     AgentBackendStreamInternalEvent,
     extract_runtime_layer_specs,
@@ -57,6 +58,14 @@ from core.workflow.nodes.agent_v2.ask_human_resume import build_deferred_tool_re
 from extensions.ext_database import db
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
 from graphon.model_runtime.entities.message_entities import AssistantPromptMessage, PromptMessage, UserPromptMessage
+from graphon.model_runtime.errors.invoke import (
+    InvokeAuthorizationError,
+    InvokeBadRequestError,
+    InvokeConnectionError,
+    InvokeError,
+    InvokeRateLimitError,
+    InvokeServerUnavailableError,
+)
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import CreatorUserRole
 from models.model import MessageAgentThought
@@ -71,6 +80,22 @@ class _DefaultSessionScopeSnapshotId:
 
 _DEFAULT_SESSION_SCOPE_SNAPSHOT_ID = _DefaultSessionScopeSnapshotId()
 
+_AGENT_BACKEND_INVOKE_ERROR_BY_REASON: Mapping[str, type[InvokeError]] = {
+    "InvokeAuthorizationError": InvokeAuthorizationError,
+    "InvokeBadRequestError": InvokeBadRequestError,
+    "CredentialsValidateFailedError": InvokeBadRequestError,
+    "InvokeConnectionError": InvokeConnectionError,
+    "InvokeRateLimitError": InvokeRateLimitError,
+    "InvokeServerUnavailableError": InvokeServerUnavailableError,
+}
+
+
+def _agent_backend_failure_to_exception(event: AgentBackendRunFailedInternalEvent) -> Exception:
+    err_cls = _AGENT_BACKEND_INVOKE_ERROR_BY_REASON.get(event.reason or "")
+    if err_cls is not None:
+        return err_cls(event.error)
+    return AgentBackendError(event.error or "Agent backend run did not complete successfully.")
+
 
 def _prompt_messages_from_query(user_query: str | None) -> list[PromptMessage]:
     if not user_query:
@@ -412,12 +437,15 @@ class _AgentProcessRecorder:
     def _lookup_tool_thought(self, *, index: int, tool_call_id: str | None) -> str | None:
         if tool_call_id and tool_call_id in self._tool_by_call_id:
             return self._tool_by_call_id[tool_call_id]
+        if index < 0:
+            return None
         return self._tool_by_index.get(index)
 
     def _remember_tool_thought(
         self, *, index: int, tool_call_id: str | None, tool_name: str | None, thought_id: str
     ) -> None:
-        self._tool_by_index[index] = thought_id
+        if index >= 0:
+            self._tool_by_index[index] = thought_id
         if tool_call_id:
             self._tool_by_call_id[tool_call_id] = thought_id
         if tool_name:
@@ -433,6 +461,10 @@ class _AgentProcessRecorder:
         return None
 
     def _mark_tool_observed(self, thought_id: str) -> None:
+        self._tool_by_index = {index: value for index, value in self._tool_by_index.items() if value != thought_id}
+        self._tool_by_call_id = {
+            tool_call_id: value for tool_call_id, value in self._tool_by_call_id.items() if value != thought_id
+        }
         for open_thought_ids in self._open_tool_by_name.values():
             open_thought_ids.discard(thought_id)
 
@@ -530,7 +562,12 @@ def _event_index(data: dict[str, Any]) -> int:
 
 
 def _string_or_none(value: Any) -> str | None:
-    return value if isinstance(value, str) and value else None
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip()
+    if not normalized or normalized.lower() in {"none", "null"}:
+        return None
+    return normalized
 
 
 def _json_or_text(value: Any) -> str | None:
@@ -652,8 +689,9 @@ class AgentAppRunner:
             return
 
         if not isinstance(terminal, AgentBackendRunSucceededInternalEvent):
-            error = getattr(terminal, "error", None) or "Agent backend run did not complete successfully."
-            raise AgentBackendError(str(error))
+            if isinstance(terminal, AgentBackendRunFailedInternalEvent):
+                raise _agent_backend_failure_to_exception(terminal)
+            raise AgentBackendError("Agent backend run did not complete successfully.")
 
         answer = self._terminal_output_to_answer(terminal.output)
         try:

api/core/app/apps/base_app_generate_response_converter.py

@@ -8,7 +8,7 @@ from pydantic import JsonValue
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError, InvokeRateLimitError
 
 logger = logging.getLogger(__name__)
 
@@ -127,6 +127,7 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
             },
             ModelCurrentlyNotSupportError: {"code": "model_currently_not_support", "status": 400},
             InvokeError: {"code": "completion_request_error", "status": 400},
+            InvokeRateLimitError: {"code": "rate_limit_error", "status": 429},
         }
 
         # Determine the response based on the type of exception

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -420,6 +420,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
         message.total_price = usage.total_price
         message.currency = usage.currency
         self._task_state.llm_result.usage.latency = message.provider_response_latency
+        self._task_state.metadata.usage = self._task_state.llm_result.usage
         message.message_metadata = self._task_state.metadata.model_dump_json()
 
         if trace_manager:

api/core/plugin/entities/request.py

@@ -276,6 +276,7 @@ class RequestRequestDownloadFile(BaseModel):
         "validation",
     ]
     file: RequestDownloadFileMapping
+    for_external: bool = True
 
     model_config = ConfigDict(extra="forbid")
 

api/core/tools/tool_file_manager.py

@@ -4,6 +4,7 @@ import hmac
 import logging
 import os
 import time
+import urllib.parse
 from collections.abc import Generator
 from mimetypes import guess_extension, guess_type
 from uuid import uuid4
@@ -26,7 +27,7 @@ logger = logging.getLogger(__name__)
 class ToolFileManager:
     @staticmethod
     def _build_graph_file_reference(tool_file: ToolFile) -> File:
-        extension = guess_extension(tool_file.mimetype) or ".bin"
+        extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
         return File(
             file_type=get_file_type_by_mime_type(tool_file.mimetype),
             transfer_method=FileTransferMethod.TOOL_FILE,
@@ -70,7 +71,7 @@ class ToolFileManager:
         mimetype: str,
         filename: str | None = None,
     ) -> ToolFile:
-        extension = guess_extension(mimetype) or ".bin"
+        extension = resolve_extension(filename=filename, mimetype=mimetype)
         unique_name = uuid4().hex
         unique_filename = f"{unique_name}{extension}"
         # default just as before
@@ -120,7 +121,8 @@ class ToolFileManager:
             or response.headers.get("Content-Type", "").split(";")[0].strip()
             or "application/octet-stream"
         )
-        extension = guess_extension(mimetype) or ".bin"
+        url_filename = os.path.basename(urllib.parse.urlparse(file_url).path)
+        extension = resolve_extension(filename=url_filename, mimetype=mimetype)
         unique_name = uuid4().hex
         filename = f"{unique_name}{extension}"
         filepath = f"tools/{tenant_id}/{filename}"
@@ -220,4 +222,11 @@ def _factory() -> ToolFileManager:
     return ToolFileManager()
 
 
+def resolve_extension(*, filename: str | None, mimetype: str) -> str:
+    filename_extension = os.path.splitext(filename or "")[1].lower()
+    if filename_extension:
+        return filename_extension
+    return guess_extension(mimetype) or ".bin"
+
+
 set_tool_file_manager_factory(_factory)

api/core/tools/utils/message_transformer.py

@@ -3,7 +3,6 @@ import re
 from collections.abc import Generator
 from datetime import date, datetime
 from decimal import Decimal
-from mimetypes import guess_extension
 from typing import Any
 from uuid import UUID
 
@@ -11,7 +10,7 @@ import numpy as np
 import pytz
 
 from core.tools.entities.tool_entities import ToolInvokeMessage
-from core.tools.tool_file_manager import ToolFileManager
+from core.tools.tool_file_manager import ToolFileManager, resolve_extension
 from core.workflow.file_reference import parse_file_reference
 from graphon.file import File, FileTransferMethod, FileType
 from libs.login import current_user
@@ -91,7 +90,8 @@ class ToolFileMessageTransformer:
                         conversation_id=conversation_id,
                     )
 
-                    url = f"/files/tools/{tool_file.id}{guess_extension(tool_file.mimetype) or '.png'}"
+                    extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
+                    url = cls.get_tool_file_url(tool_file_id=tool_file.id, extension=extension)
                     meta = cls._with_tool_file_meta(
                         message.meta,
                         tool_file_id=str(tool_file.id),
@@ -136,7 +136,8 @@ class ToolFileMessageTransformer:
                     filename=filename,
                 )
 
-                url = cls.get_tool_file_url(tool_file_id=tool_file.id, extension=guess_extension(tool_file.mimetype))
+                extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
+                url = cls.get_tool_file_url(tool_file_id=tool_file.id, extension=extension)
                 meta = cls._with_tool_file_meta(meta, tool_file_id=str(tool_file.id))
 
                 # check if file is image

api/core/workflow/nodes/agent_v2/agent_node.py

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any, override
 from agenton.compositor import CompositorSessionSnapshot
 
 from clients.agent_backend import (
+    AgentBackendAgentMessageDeltaInternalEvent,
     AgentBackendDeferredToolCallInternalEvent,
     AgentBackendError,
     AgentBackendHTTPError,
@@ -481,6 +482,10 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                         if isinstance(internal_event, AgentBackendStreamInternalEvent):
                             self._record_stream_metadata(metadata, internal_event)
                         continue
+                    if internal_event.type == AgentBackendInternalEventType.AGENT_MESSAGE_DELTA:
+                        if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
+                            self._record_agent_message_delta_metadata(metadata, internal_event)
+                        continue
                     metadata["agent_backend"] = {
                         **dict(metadata.get("agent_backend") or {}),
                         "stream_event_count": stream_event_count,
@@ -734,6 +739,17 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 agent_backend["usage"] = dict(usage)
         metadata["agent_backend"] = agent_backend
 
+    @staticmethod
+    def _record_agent_message_delta_metadata(
+        metadata: dict[str, Any], event: AgentBackendAgentMessageDeltaInternalEvent
+    ) -> None:
+        agent_backend = dict(metadata.get("agent_backend") or {})
+        agent_backend["agent_message_delta_count"] = int(agent_backend.get("agent_message_delta_count") or 0) + 1
+        agent_backend["agent_message_delta_length"] = int(agent_backend.get("agent_message_delta_length") or 0) + len(
+            event.delta
+        )
+        metadata["agent_backend"] = agent_backend
+
     @classmethod
     @override
     def _extract_variable_selector_to_variable_mapping(

api/core/workflow/nodes/agent_v2/output_file_rebacker.py

@@ -10,13 +10,13 @@ trustworthy metadata.
 
 from __future__ import annotations
 
-from mimetypes import guess_extension
 from uuid import UUID
 
 from sqlalchemy import select
 from sqlalchemy.exc import DataError, SQLAlchemyError
 
 from core.db.session_factory import session_factory
+from core.tools.tool_file_manager import resolve_extension
 from core.workflow.file_reference import build_file_reference
 from graphon.file import File, FileTransferMethod, get_file_type_by_mime_type
 from models.tools import ToolFile
@@ -46,7 +46,7 @@ def reback_tool_file_output(*, tenant_id: str, tool_file_id: str) -> File | None
         return None
 
     mime_type = tool_file.mimetype or ""
-    extension = guess_extension(mime_type) or ".bin"
+    extension = resolve_extension(filename=tool_file.name, mimetype=mime_type)
     return File(
         type=get_file_type_by_mime_type(mime_type),
         transfer_method=FileTransferMethod.TOOL_FILE,

api/extensions/ext_celery.py

@@ -156,7 +156,6 @@ def init_app(app: DifyApp) -> Celery:
         "tasks.generate_summary_index_task",  # summary index generation
         "tasks.regenerate_summary_index_task",  # summary index regeneration
         "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
-        "tasks.workflow_run_archive_download_tasks",  # workflow-run archive download preparation
     ]
     day = dify_config.CELERY_BEAT_SCHEDULER_TIME
 

api/extensions/ext_commands.py

@@ -7,7 +7,6 @@ def init_app(app: DifyApp):
         archive_workflow_runs,
         archive_workflow_runs_plan,
         backfill_plugin_auto_upgrade,
-        backfill_workflow_run_archive_bundles,
         clean_expired_messages,
         clean_workflow_runs,
         cleanup_orphaned_draft_variables,
@@ -78,7 +77,6 @@ def init_app(app: DifyApp):
         install_rag_pipeline_plugins,
         archive_workflow_runs_plan,
         archive_workflow_runs,
-        backfill_workflow_run_archive_bundles,
         delete_archived_workflow_runs,
         restore_workflow_runs,
         clean_workflow_runs,

api/factories/file_factory/builders.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import mimetypes
+import os
 import uuid
 from collections.abc import Mapping, Sequence
 from typing import Any, Literal, NotRequired, TypedDict, assert_never, cast
@@ -285,7 +286,7 @@ def _build_from_remote_url(
         raise ValueError("Invalid file url")
 
     mime_type, filename, file_size = get_remote_file_info(url)
-    extension = mimetypes.guess_extension(mime_type) or ("." + filename.split(".")[-1] if "." in filename else ".bin")
+    extension = os.path.splitext(filename)[1].lower() or mimetypes.guess_extension(mime_type) or ".bin"
     detected_file_type = standardize_file_type(extension=extension, mime_type=mime_type)
     file_type = _resolve_file_type(
         detected_file_type=detected_file_type,
@@ -326,7 +327,12 @@ def _build_from_tool_file(
         if tool_file is None:
             raise ValueError(f"ToolFile {tool_file_id} not found")
 
-        extension = "." + tool_file.file_key.split(".")[-1] if "." in tool_file.file_key else ".bin"
+        extension = (
+            os.path.splitext(tool_file.name)[1].lower()
+            or mimetypes.guess_extension(tool_file.mimetype)
+            or os.path.splitext(tool_file.file_key)[1].lower()
+            or ".bin"
+        )
         detected_file_type = standardize_file_type(extension=extension, mime_type=tool_file.mimetype)
         file_type = _resolve_file_type(
             detected_file_type=detected_file_type,

api/fields/message_fields.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 from datetime import datetime
+from decimal import Decimal
 from uuid import uuid4
 
-from pydantic import Field, field_validator
+from pydantic import Field, computed_field, field_validator
 
 from core.entities.execution_extra_content import ExecutionExtraContentDomainModel
 from fields.base import ResponseModel
@@ -55,10 +56,19 @@ class MessageListItem(ResponseModel):
     created_at: int | None = None
     agent_thoughts: list[AgentThought]
     message_files: list[MessageFile]
+    message_tokens: int = 0
+    answer_tokens: int = 0
+    provider_response_latency: float = 0
+    total_price: Decimal | None = None
+    currency: str | None = None
     status: str
     error: str | None = None
     extra_contents: list[ExecutionExtraContentDomainModel]
 
+    @computed_field
+    def total_tokens(self) -> int:
+        return self.message_tokens + self.answer_tokens
+
     @field_validator("inputs", mode="before")
     @classmethod
     def _normalize_inputs(cls, value: JSONValueType) -> JSONValueType:

api/libs/archive_storage.py

@@ -11,7 +11,6 @@ import hashlib
 import logging
 from collections.abc import Generator
 from typing import Any, cast
-from urllib.parse import quote
 
 import boto3
 import orjson
@@ -198,22 +197,13 @@ class ArchiveStorage:
         except ClientError as e:
             raise ArchiveStorageError(f"Failed to delete object '{key}': {e}")
 
-    def generate_presigned_url(
-        self,
-        key: str,
-        expires_in: int = 3600,
-        *,
-        filename: str | None = None,
-        content_type: str | None = None,
-    ) -> str:
+    def generate_presigned_url(self, key: str, expires_in: int = 3600) -> str:
         """
         Generate a pre-signed URL for downloading an object.
 
         Args:
             key: Object key (path) within the bucket
             expires_in: URL validity duration in seconds (default: 1 hour)
-            filename: Optional browser download filename
-            content_type: Optional response content type
 
         Returns:
             Pre-signed URL string.
@@ -221,15 +211,10 @@ class ArchiveStorage:
         Raises:
             ArchiveStorageError: If generation fails
         """
-        params = {"Bucket": self.bucket, "Key": key}
-        if filename:
-            params["ResponseContentDisposition"] = f"attachment; filename*=UTF-8''{quote(filename)}"
-        if content_type:
-            params["ResponseContentType"] = content_type
         try:
             return self.client.generate_presigned_url(
                 ClientMethod="get_object",
-                Params=params,
+                Params={"Bucket": self.bucket, "Key": key},
                 ExpiresIn=expires_in,
             )
         except ClientError as e:

api/libs/token.py

@@ -22,7 +22,6 @@ logger = logging.getLogger(__name__)
 
 CSRF_WHITE_LIST = [
     re.compile(r"/console/api/apps/[a-f0-9-]+/workflows/draft"),
-    re.compile(r"/console/api/workflow-run-archives/downloads/[a-f0-9]+/file"),
 ]
 
 

api/migrations/versions/2026_06_25_1500-7a1c2d9e4b60_add_workflow_run_archive_index_tables.py

@@ -1,61 +0,0 @@
-"""add workflow run archive bundle index table
-
-Revision ID: 7a1c2d9e4b60
-Revises: c3d4e5f6a7b8
-Create Date: 2026-06-25 15:00:00.000000
-
-"""
-
-import sqlalchemy as sa
-from alembic import op
-
-import models
-
-# revision identifiers, used by Alembic.
-revision = "7a1c2d9e4b60"
-down_revision = "c3d4e5f6a7b8"
-branch_labels = None
-depends_on = None
-
-
-def _uuid_column(name: str, **kwargs):
-    if op.get_bind().dialect.name == "postgresql":
-        kwargs.setdefault("server_default", sa.text("uuidv7()"))
-    return sa.Column(name, models.types.StringUUID(), **kwargs)
-
-
-def upgrade() -> None:
-    op.create_table(
-        "workflow_run_archive_bundles",
-        _uuid_column("id", nullable=False),
-        sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
-        sa.Column("year", sa.Integer(), nullable=False),
-        sa.Column("month", sa.Integer(), nullable=False),
-        sa.Column("shard", sa.String(length=32), nullable=False),
-        sa.Column("bundle_id", sa.String(length=64), nullable=False),
-        sa.Column("workflow_run_count", sa.Integer(), nullable=False),
-        sa.Column("row_count", sa.BigInteger(), nullable=False),
-        sa.Column("archive_bytes", sa.BigInteger(), nullable=False),
-        sa.Column("archived_at", sa.DateTime(), nullable=False),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.PrimaryKeyConstraint("id", name="workflow_run_archive_bundle_pkey"),
-        sa.UniqueConstraint(
-            "tenant_id",
-            "year",
-            "month",
-            "shard",
-            "bundle_id",
-            name="workflow_run_archive_bundle_identity_uq",
-        ),
-    )
-    op.create_index(
-        "workflow_run_archive_bundle_tenant_month_idx",
-        "workflow_run_archive_bundles",
-        ["tenant_id", "year", "month"],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("workflow_run_archive_bundle_tenant_month_idx", table_name="workflow_run_archive_bundles")
-    op.drop_table("workflow_run_archive_bundles")

api/models/__init__.py

@@ -144,7 +144,6 @@ from .workflow import (
     WorkflowNodeExecutionTriggeredFrom,
     WorkflowPause,
     WorkflowRun,
-    WorkflowRunArchiveBundle,
     WorkflowType,
     resolve_workflow_kind,
 )
@@ -283,7 +282,6 @@ __all__ = [
     "WorkflowNodeExecutionTriggeredFrom",
     "WorkflowPause",
     "WorkflowRun",
-    "WorkflowRunArchiveBundle",
     "WorkflowRunTriggeredFrom",
     "WorkflowSchedulePlan",
     "WorkflowToolProvider",

api/models/workflow.py

@@ -1446,40 +1446,6 @@ class WorkflowArchiveLog(TypeBase):
         }
 
 
-class WorkflowRunArchiveBundle(DefaultFieldsDCMixin, TypeBase):
-    """
-    Query index for one immutable V2 workflow-run archive bundle.
-
-    R2 manifest objects remain the recoverable archive source of truth. This table stores the small subset needed to
-    list tenant/month archives and locate bundles without listing object storage online. Missing rows can be rebuilt
-    from existing manifests by a backfill/reconciliation command.
-    """
-
-    __tablename__ = "workflow_run_archive_bundles"
-    __table_args__ = (
-        sa.PrimaryKeyConstraint("id", name="workflow_run_archive_bundle_pkey"),
-        sa.UniqueConstraint(
-            "tenant_id",
-            "year",
-            "month",
-            "shard",
-            "bundle_id",
-            name="workflow_run_archive_bundle_identity_uq",
-        ),
-        sa.Index("workflow_run_archive_bundle_tenant_month_idx", "tenant_id", "year", "month"),
-    )
-
-    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    year: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    month: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    shard: Mapped[str] = mapped_column(String(32), nullable=False)
-    bundle_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    workflow_run_count: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    row_count: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
-    archive_bytes: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
-    archived_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
-
-
 class ConversationVariable(TypeBase):
     __tablename__ = "workflow_conversation_variables"
 

api/openapi/markdown/console-openapi.md

@@ -9690,61 +9690,6 @@ Suggest example workflow-generator instructions for the tenant
 | 200 | Suggestions generated successfully | **application/json**: [GeneratorResponse](#generatorresponse)<br> |
 | 400 | Invalid request parameters |  |
 
-### [GET] /workflow-run-archives
-List monthly workflow-run archive metadata for the current workspace
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 200 | Success | **application/json**: [WorkflowRunArchiveListResponse](#workflowrunarchivelistresponse)<br> |
-
-### [POST] /workflow-run-archives/downloads
-Create or return a temporary workflow-run archive download task
-
-#### Request Body
-
-| Required | Schema |
-| -------- | ------ |
-|  Yes | **application/json**: [WorkflowRunArchiveDownloadPayload](#workflowrunarchivedownloadpayload)<br> |
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 202 | Download task accepted | **application/json**: [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse)<br> |
-
-### [GET] /workflow-run-archives/downloads/{download_id}
-Get a temporary workflow-run archive download task
-
-#### Parameters
-
-| Name | Located in | Description | Required | Schema |
-| ---- | ---------- | ----------- | -------- | ------ |
-| download_id | path |  | Yes | string |
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 200 | Success | **application/json**: [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse)<br> |
-
-### [GET] /workflow-run-archives/downloads/{download_id}/file
-Redirect to a prepared workflow-run archive ZIP file
-
-#### Parameters
-
-| Name | Located in | Description | Required | Schema |
-| ---- | ---------- | ----------- | -------- | ------ |
-| download_id | path |  | Yes | string |
-
-#### Responses
-
-| Code | Description | Schema |
-| ---- | ----------- | ------ |
-| 302 | Redirect to pre-signed archive storage URL | **application/json**: [RedirectResponse](#redirectresponse)<br> |
-| 409 | Download task is not ready |  |
-
 ### [GET] /workflow/{workflow_run_id}/events
 **Get workflow execution events stream after resume**
 
@@ -17669,19 +17614,25 @@ Built-in tool icons are URL strings; API-based tool icons are provider-defined p
 | ---- | ---- | ----------- | -------- |
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | Yes |
 | answer | string |  | Yes |
+| answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
+| currency | string |  | No |
 | error | string |  | No |
 | extra_contents | [ [HumanInputContent](#humaninputcontent) ] |  | Yes |
 | feedback | [SimpleFeedback](#simplefeedback) |  | No |
 | id | string |  | Yes |
 | inputs | object |  | Yes |
 | message_files | [ [MessageFile](#messagefile) ] |  | Yes |
+| message_tokens | integer |  | No |
 | metadata | [JSONValueType](#jsonvaluetype) |  | No |
 | parent_message_id | string |  | No |
+| provider_response_latency | number |  | No |
 | query | string |  | Yes |
 | retriever_resources | [ [RetrieverResource](#retrieverresource) ] |  | Yes |
 | status | string |  | Yes |
+| total_price | string |  | No |
+| total_tokens | integer |  | Yes |
 
 #### ExternalApiTemplateListQuery
 
@@ -23349,71 +23300,6 @@ tenant's default model. The underlying generator never raises — an empty
 | result | string |  | Yes |
 | updated_at | integer |  | Yes |
 
-#### WorkflowRunArchiveDownloadPayload
-
-Request body for preparing one monthly workflow-run archive download.
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| month | integer |  | Yes |
-| year | integer |  | Yes |
-
-#### WorkflowRunArchiveDownloadStatus
-
-Lifecycle state for an asynchronous archive download request.
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| WorkflowRunArchiveDownloadStatus | string | Lifecycle state for an asynchronous archive download request. |  |
-
-#### WorkflowRunArchiveDownloadTaskResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| archive_bytes | integer |  | Yes |
-| bundle_count | integer |  | Yes |
-| created_at | dateTime |  | Yes |
-| download_id | string |  | Yes |
-| error | string |  | No |
-| expires_at | dateTime |  | Yes |
-| file_name | string |  | No |
-| file_size_bytes | integer |  | No |
-| finished_at | string |  | No |
-| month | integer |  | Yes |
-| started_at | string |  | No |
-| status | [WorkflowRunArchiveDownloadStatus](#workflowrunarchivedownloadstatus) |  | Yes |
-| updated_at | dateTime |  | Yes |
-| year | integer |  | Yes |
-
-#### WorkflowRunArchiveListResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| months | [ [WorkflowRunArchiveMonthResponse](#workflowrunarchivemonthresponse) ] |  | Yes |
-| summary | [WorkflowRunArchiveSummaryResponse](#workflowrunarchivesummaryresponse) |  | Yes |
-
-#### WorkflowRunArchiveMonthResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| archive_bytes | integer |  | Yes |
-| bundle_count | integer |  | Yes |
-| download_task | [WorkflowRunArchiveDownloadTaskResponse](#workflowrunarchivedownloadtaskresponse) |  | No |
-| latest_archived_at | dateTime |  | Yes |
-| month | integer |  | Yes |
-| row_count | integer |  | Yes |
-| workflow_run_count | integer |  | Yes |
-| year | integer |  | Yes |
-
-#### WorkflowRunArchiveSummaryResponse
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| archive_bytes | integer |  | Yes |
-| archived_month_count | integer |  | Yes |
-| latest_archived_at | string |  | No |
-| workflow_run_count | integer |  | Yes |
-
 #### WorkflowRunCountQuery
 
 | Name | Type | Description | Required |

api/openapi/markdown/service-openapi.md

@@ -3467,18 +3467,24 @@ Model class for i18n object.
 | ---- | ---- | ----------- | -------- |
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | Yes |
 | answer | string |  | Yes |
+| answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
+| currency | string |  | No |
 | error | string |  | No |
 | extra_contents | [ [HumanInputContent](#humaninputcontent) ] |  | Yes |
 | feedback | [SimpleFeedback](#simplefeedback) |  | No |
 | id | string |  | Yes |
 | inputs | object |  | Yes |
 | message_files | [ [MessageFile](#messagefile) ] |  | Yes |
+| message_tokens | integer |  | No |
 | parent_message_id | string |  | No |
+| provider_response_latency | number |  | No |
 | query | string |  | Yes |
 | retriever_resources | [ [RetrieverResource](#retrieverresource) ] |  | Yes |
 | status | string |  | Yes |
+| total_price | string |  | No |
+| total_tokens | integer |  | Yes |
 
 #### MessageListQuery
 

api/openapi/markdown/web-openapi.md

@@ -1685,19 +1685,25 @@ in form definiton, or a variable while the workflow is running.
 | ---- | ---- | ----------- | -------- |
 | agent_thoughts | [ [AgentThought](#agentthought) ] |  | Yes |
 | answer | string |  | Yes |
+| answer_tokens | integer |  | No |
 | conversation_id | string |  | Yes |
 | created_at | integer |  | No |
+| currency | string |  | No |
 | error | string |  | No |
 | extra_contents | [ [HumanInputContent](#humaninputcontent) ] |  | Yes |
 | feedback | [SimpleFeedback](#simplefeedback) |  | No |
 | id | string |  | Yes |
 | inputs | object |  | Yes |
 | message_files | [ [MessageFile](#messagefile) ] |  | Yes |
+| message_tokens | integer |  | No |
 | metadata | [JSONValueType](#jsonvaluetype) |  | No |
 | parent_message_id | string |  | No |
+| provider_response_latency | number |  | No |
 | query | string |  | Yes |
 | retriever_resources | [ [RetrieverResource](#retrieverresource) ] |  | Yes |
 | status | string |  | Yes |
+| total_price | string |  | No |
+| total_tokens | integer |  | Yes |
 
 #### WebModelConfigResponse
 

api/providers/trace/trace-aliyun/tests/unit_tests/aliyun_trace/test_aliyun_trace.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 from datetime import UTC, datetime
 from types import SimpleNamespace
 from typing import cast
@@ -259,14 +260,16 @@ def test_get_project_url_success(trace_instance: AliyunDataTrace):
     assert trace_instance.get_project_url() == "project-url"
 
 
-def test_get_project_url_error(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
+def test_get_project_url_error(
+    trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+):
     monkeypatch.setattr(trace_instance.trace_client, "get_project_url", MagicMock(side_effect=Exception("boom")))
-    logger_mock = MagicMock()
-    monkeypatch.setattr(aliyun_trace_module, "logger", logger_mock)
 
+    caplog.set_level(logging.INFO, logger=aliyun_trace_module.logger.name)
     with pytest.raises(ValueError, match=r"Aliyun get project url failed: boom"):
         trace_instance.get_project_url()
-    logger_mock.info.assert_called()
+
+    assert "Aliyun get project url failed: boom" in caplog.text
 
 
 def test_workflow_trace_adds_workflow_and_node_spans(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):

api/providers/trace/trace-tencent/tests/unit_tests/tencent_trace/test_client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import sys
 import types
 from types import SimpleNamespace
@@ -87,7 +88,6 @@ class PatchedCoreComponents(TypedDict):
     tracer: MagicMock
     span: MagicMock
     tracer_provider: MagicMock
-    logger: MagicMock
     trace_api: Any
 
 
@@ -148,9 +148,6 @@ def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> PatchedCoreCompone
     resource = MagicMock(name="resource")
     monkeypatch.setattr(client_module, "Resource", MagicMock(return_value=resource))
 
-    logger_mock = MagicMock(name="tencent_logger")
-    monkeypatch.setattr(client_module, "logger", logger_mock)
-
     trace_api_stub = SimpleNamespace(
         set_span_in_context=MagicMock(name="set_span_in_context", return_value="trace-context"),
         NonRecordingSpan=MagicMock(name="non_recording_span", side_effect=lambda ctx: f"non-{ctx}"),
@@ -174,7 +171,6 @@ def patch_core_components(monkeypatch: pytest.MonkeyPatch) -> PatchedCoreCompone
         "tracer": tracer,
         "span": span,
         "tracer_provider": tracer_provider,
-        "logger": logger_mock,
         "trace_api": trace_api_stub,
     }
 
@@ -268,14 +264,15 @@ def test_record_methods_skip_when_histogram_missing() -> None:
     client.record_trace_duration(0.5)
 
 
-def test_record_llm_duration_handles_exceptions(patch_core_components: PatchedCoreComponents) -> None:
+def test_record_llm_duration_handles_exceptions(caplog: pytest.LogCaptureFixture) -> None:
     client = _build_client()
     client.hist_llm_duration = MagicMock(name="hist_llm_duration")
     client.hist_llm_duration.record.side_effect = RuntimeError("boom")
 
+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
     client.record_llm_duration(0.2)
-    logger = patch_core_components["logger"]
-    logger.debug.assert_called()
+
+    assert "[Tencent APM] Failed to record LLM duration" in caplog.text
 
 
 def test_create_and_export_span_sets_attributes(patch_core_components: PatchedCoreComponents) -> None:
@@ -328,12 +325,15 @@ def test_create_and_export_span_uses_parent_context(patch_core_components: Patch
     trace_api.set_span_in_context.assert_called_once()
 
 
-def test_create_and_export_span_exception_logs_error(patch_core_components: PatchedCoreComponents) -> None:
+def test_create_and_export_span_exception_logs_error(
+    patch_core_components: PatchedCoreComponents, caplog: pytest.LogCaptureFixture
+) -> None:
     client = _build_client()
     span = patch_core_components["span"]
     span.get_span_context.return_value = _make_span_context(span_id=2)
     client.tracer.start_span.side_effect = RuntimeError("boom")
 
+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
     client._create_and_export_span(
         SpanData(
             trace_id=1,
@@ -346,8 +346,10 @@ def test_create_and_export_span_exception_logs_error(patch_core_components: Patc
             end_time=1,
         )
     )
-    logger = patch_core_components["logger"]
-    logger.exception.assert_called_once()
+
+    error_records = [record for record in caplog.records if record.levelno == logging.ERROR]
+    assert len(error_records) == 1
+    assert error_records[0].getMessage() == "[Tencent APM] Error creating span: span"
 
 
 def test_api_check_connects_successfully(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -423,23 +425,18 @@ def test_shutdown_flushes_all_components(patch_core_components: PatchedCoreCompo
     metric_reader.shutdown.assert_called_once()
 
 
-def test_shutdown_logs_when_meter_provider_fails(patch_core_components: PatchedCoreComponents) -> None:
+def test_shutdown_logs_when_meter_provider_fails(caplog: pytest.LogCaptureFixture) -> None:
     client = _build_client()
     meter_provider = meter_provider_instances[-1]
     meter_provider.shutdown.side_effect = RuntimeError("boom")
     assert client.metric_reader is not None
     client.metric_reader.shutdown.side_effect = RuntimeError("boom")
 
+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
     client.shutdown()
-    logger = patch_core_components["logger"]
-    logger.debug.assert_any_call(
-        "[Tencent APM] Error shutting down meter provider",
-        exc_info=True,
-    )
-    logger.debug.assert_any_call(
-        "[Tencent APM] Error shutting down metric reader",
-        exc_info=True,
-    )
+
+    assert "[Tencent APM] Error shutting down meter provider" in caplog.text
+    assert "[Tencent APM] Error shutting down metric reader" in caplog.text
 
 
 def test_metrics_initialization_failure_sets_histogram_attributes(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -456,10 +453,11 @@ def test_metrics_initialization_failure_sets_histogram_attributes(monkeypatch: p
     assert client.metric_reader is None
 
 
-def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_components: PatchedCoreComponents) -> None:
+def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
     client = _build_client()
     monkeypatch.setattr(client, "_create_and_export_span", MagicMock(side_effect=RuntimeError("boom")))
 
+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
     client.add_span(
         SpanData(
             trace_id=1,
@@ -473,8 +471,9 @@ def test_add_span_logs_exception(monkeypatch: pytest.MonkeyPatch, patch_core_com
         )
     )
 
-    logger = patch_core_components["logger"]
-    logger.exception.assert_called_once()
+    error_records = [record for record in caplog.records if record.levelno == logging.ERROR]
+    assert len(error_records) == 1
+    assert error_records[0].getMessage() == "[Tencent APM] Failed to create span: span"
 
 
 def test_create_and_export_span_converts_attribute_types(patch_core_components: PatchedCoreComponents) -> None:
@@ -535,16 +534,20 @@ def test_record_trace_duration_converts_attributes() -> None:
     ],
 )
 def test_record_methods_handle_exceptions(
-    method: str, attr_name: str, args: tuple[object, ...], patch_core_components: PatchedCoreComponents
+    method: str, attr_name: str, args: tuple[object, ...], caplog: pytest.LogCaptureFixture
 ) -> None:
     client = _build_client()
     hist_mock = MagicMock(name=attr_name)
     hist_mock.record.side_effect = RuntimeError("boom")
     setattr(client, attr_name, hist_mock)
 
+    caplog.set_level(logging.DEBUG, logger=client_module.logger.name)
     getattr(client, method)(*args)
-    logger = patch_core_components["logger"]
-    logger.debug.assert_called()
+
+    assert any(
+        record.levelno == logging.DEBUG and record.getMessage().startswith("[Tencent APM] Failed to record")
+        for record in caplog.records
+    )
 
 
 def test_metrics_initializes_grpc_metric_exporter() -> None:

api/services/file_request_service.py

@@ -45,6 +45,7 @@ class FileRequestService:
         user_from: UserFrom | str,
         invoke_from: InvokeFrom | str,
         file_mapping: Mapping[str, Any],
+        for_external: bool = True,
     ) -> DownloadFileRequestResult:
         """Resolve one file mapping into signed download metadata.
 
@@ -61,7 +62,7 @@ class FileRequestService:
         )
         with bind_file_access_scope(scope):
             file = self._build_file(mapping=file_mapping, tenant_id=tenant_id)
-            download_url = file_helpers.resolve_file_url(file, for_external=True)
+            download_url = file_helpers.resolve_file_url(file, for_external=for_external)
 
         if not download_url:
             raise ValueError("file does not support signed download")

api/services/retention/workflow_run/archive_bundle_index.py

@@ -1,347 +0,0 @@
-"""
-Workflow-run archive bundle index helpers.
-
-Archive manifests in object storage remain the recoverable source of truth. This module mirrors their small query
-surface into `workflow_run_archive_bundles` so console listing and download jobs can avoid listing R2 on request.
-The backfill path is intentionally idempotent: every manifest is decoded, checked against the V2 schema markers, and
-upserted by immutable bundle identity.
-"""
-
-import datetime
-import json
-import logging
-import time
-from collections.abc import Sequence
-from dataclasses import dataclass, field
-from typing import TypedDict, cast
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-
-from extensions.ext_database import db
-from libs.archive_storage import ArchiveStorage, get_archive_storage
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.constants import (
-    ARCHIVE_BUNDLE_FORMAT,
-    ARCHIVE_BUNDLE_MANIFEST_NAME,
-    ARCHIVE_BUNDLE_SCHEMA_VERSION,
-)
-
-logger = logging.getLogger(__name__)
-
-ARCHIVE_BUNDLE_ROOT_PREFIX = "workflow-runs/v2/"
-
-
-class ArchiveBundleTableManifestEntry(TypedDict):
-    row_count: int
-    checksum: str
-    size_bytes: int
-    object_key: str
-
-
-class ArchiveBundleManifest(TypedDict):
-    schema_version: str
-    archive_format: str
-    tenant_id: str
-    tenant_prefix: str
-    year: int
-    month: int
-    shard: str
-    bundle_id: str
-    object_prefix: str
-    workflow_run_count: int
-    workflow_node_execution_count: int
-    min_created_at: str
-    max_created_at: str
-    min_run_id: str
-    max_run_id: str
-    archived_at: str
-    tables: dict[str, ArchiveBundleTableManifestEntry]
-    run_ids: list[str]
-
-
-@dataclass(frozen=True)
-class ArchiveBundleIndexValues:
-    """Computed DB-index values derived from one manifest."""
-
-    row_count: int
-    archive_bytes: int
-    archived_at: datetime.datetime
-
-
-@dataclass
-class ArchiveBundleIndexBackfillSummary:
-    """Aggregate result for a manifest-to-DB-index reconciliation run."""
-
-    manifests_found: int = 0
-    bundles_processed: int = 0
-    bundles_upserted: int = 0
-    bundles_failed: int = 0
-    workflow_run_count: int = 0
-    row_count: int = 0
-    archive_bytes: int = 0
-    elapsed_time: float = 0.0
-    errors: list[str] = field(default_factory=list)
-
-
-def decode_archive_bundle_manifest(manifest_data: bytes) -> ArchiveBundleManifest:
-    """Decode raw manifest bytes into the V2 archive manifest shape."""
-    return cast(ArchiveBundleManifest, json.loads(manifest_data.decode("utf-8")))
-
-
-def parse_archive_manifest_datetime(value: str) -> datetime.datetime:
-    """Parse manifest datetimes and normalize timezone-aware values to naive UTC for DB storage."""
-    parsed = datetime.datetime.fromisoformat(value)
-    if parsed.tzinfo is None:
-        return parsed
-    return parsed.astimezone(datetime.UTC).replace(tzinfo=None)
-
-
-def calculate_archive_bundle_index_values(
-    manifest: ArchiveBundleManifest,
-    manifest_size_bytes: int,
-) -> ArchiveBundleIndexValues:
-    """Calculate row count, stored bytes, and archived timestamp for the DB index."""
-    _validate_archive_bundle_manifest(manifest)
-    row_count = sum(entry["row_count"] for entry in manifest["tables"].values())
-    archive_bytes = manifest_size_bytes + sum(entry["size_bytes"] for entry in manifest["tables"].values())
-    return ArchiveBundleIndexValues(
-        row_count=row_count,
-        archive_bytes=archive_bytes,
-        archived_at=parse_archive_manifest_datetime(manifest["archived_at"]),
-    )
-
-
-def upsert_archive_bundle_index_from_manifest(
-    session: Session,
-    manifest: ArchiveBundleManifest,
-    manifest_size_bytes: int,
-) -> WorkflowRunArchiveBundle:
-    """
-    Persist one archive manifest into `workflow_run_archive_bundles`.
-
-    The caller owns transaction boundaries. Re-running this function for the same manifest is safe and refreshes the
-    mutable metrics derived from object sizes and row counts.
-    """
-    values = calculate_archive_bundle_index_values(manifest, manifest_size_bytes)
-    existing = session.scalar(
-        select(WorkflowRunArchiveBundle).where(
-            WorkflowRunArchiveBundle.tenant_id == manifest["tenant_id"],
-            WorkflowRunArchiveBundle.year == manifest["year"],
-            WorkflowRunArchiveBundle.month == manifest["month"],
-            WorkflowRunArchiveBundle.shard == manifest["shard"],
-            WorkflowRunArchiveBundle.bundle_id == manifest["bundle_id"],
-        )
-    )
-    if existing is None:
-        bundle = WorkflowRunArchiveBundle(
-            tenant_id=manifest["tenant_id"],
-            year=manifest["year"],
-            month=manifest["month"],
-            shard=manifest["shard"],
-            bundle_id=manifest["bundle_id"],
-            workflow_run_count=manifest["workflow_run_count"],
-            row_count=values.row_count,
-            archive_bytes=values.archive_bytes,
-            archived_at=values.archived_at,
-        )
-        session.add(bundle)
-        return bundle
-
-    existing.workflow_run_count = manifest["workflow_run_count"]
-    existing.row_count = values.row_count
-    existing.archive_bytes = values.archive_bytes
-    existing.archived_at = values.archived_at
-    return existing
-
-
-class WorkflowRunArchiveBundleIndexBackfill:
-    """
-    Rebuild the DB bundle index by scanning object-store manifests.
-
-    Tenant IDs are the cheapest scope because they map directly to the object prefix. Tenant prefixes are supported for
-    rollout reconciliation, but they still require listing all tenants under that prefix and filtering keys locally.
-    """
-
-    storage: ArchiveStorage | None
-    session_factory: sessionmaker[Session]
-
-    def __init__(
-        self,
-        *,
-        storage: ArchiveStorage | None = None,
-        session_factory: sessionmaker[Session] | None = None,
-    ) -> None:
-        self.storage = storage
-        self.session_factory = session_factory or sessionmaker(bind=db.engine, expire_on_commit=False)
-
-    def run(
-        self,
-        *,
-        tenant_ids: Sequence[str] | None = None,
-        tenant_prefixes: Sequence[str] | None = None,
-        year: int | None = None,
-        month: int | None = None,
-        limit: int | None = None,
-        dry_run: bool = False,
-    ) -> ArchiveBundleIndexBackfillSummary:
-        """Scan matching manifest objects and idempotently upsert their DB index rows."""
-        start_time = time.time()
-        summary = ArchiveBundleIndexBackfillSummary()
-        storage = self.storage or get_archive_storage()
-        manifest_keys = self._list_manifest_keys(
-            storage,
-            tenant_ids=tenant_ids,
-            tenant_prefixes=tenant_prefixes,
-            year=year,
-            month=month,
-        )
-        summary.manifests_found = len(manifest_keys)
-
-        if limit is not None:
-            manifest_keys = manifest_keys[:limit]
-
-        for manifest_key in manifest_keys:
-            try:
-                manifest_data = storage.get_object(manifest_key)
-                manifest = decode_archive_bundle_manifest(manifest_data)
-                self._validate_manifest_scope(
-                    manifest,
-                    manifest_key=manifest_key,
-                    tenant_ids=tenant_ids,
-                    tenant_prefixes=tenant_prefixes,
-                    year=year,
-                    month=month,
-                )
-                values = calculate_archive_bundle_index_values(manifest, len(manifest_data))
-                summary.bundles_processed += 1
-                summary.workflow_run_count += manifest["workflow_run_count"]
-                summary.row_count += values.row_count
-                summary.archive_bytes += values.archive_bytes
-                if dry_run:
-                    continue
-
-                with self.session_factory() as session:
-                    upsert_archive_bundle_index_from_manifest(session, manifest, len(manifest_data))
-                    session.commit()
-                summary.bundles_upserted += 1
-            except Exception as exc:
-                logger.warning("Failed to backfill workflow archive bundle index from %s", manifest_key, exc_info=True)
-                summary.bundles_failed += 1
-                summary.errors.append(f"{manifest_key}: {exc}")
-
-        summary.elapsed_time = time.time() - start_time
-        return summary
-
-    @classmethod
-    def _list_manifest_keys(
-        cls,
-        storage: ArchiveStorage,
-        *,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> list[str]:
-        prefixes = cls._list_prefixes(tenant_ids=tenant_ids, tenant_prefixes=tenant_prefixes, year=year, month=month)
-        keys: list[str] = []
-        for prefix in prefixes:
-            keys.extend(storage.list_objects(prefix))
-        return sorted(
-            key
-            for key in keys
-            if key.endswith(f"/{ARCHIVE_BUNDLE_MANIFEST_NAME}")
-            and cls._manifest_key_matches_scope(
-                key,
-                tenant_ids=tenant_ids,
-                tenant_prefixes=tenant_prefixes,
-                year=year,
-                month=month,
-            )
-        )
-
-    @staticmethod
-    def _list_prefixes(
-        *,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> list[str]:
-        if tenant_ids:
-            prefixes = []
-            for tenant_id in sorted(set(tenant_ids)):
-                prefix = f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix={tenant_id[0].lower()}/tenant_id={tenant_id}/"
-                if year is not None:
-                    prefix += f"year={year:04d}/"
-                    if month is not None:
-                        prefix += f"month={month:02d}/"
-                prefixes.append(prefix)
-            return prefixes
-
-        if tenant_prefixes:
-            return [
-                f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix={tenant_prefix}/"
-                for tenant_prefix in sorted(set(tenant_prefixes))
-            ]
-
-        return [ARCHIVE_BUNDLE_ROOT_PREFIX]
-
-    @staticmethod
-    def _manifest_key_matches_scope(
-        key: str,
-        *,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> bool:
-        if tenant_ids and _extract_key_part(key, "tenant_id") not in set(tenant_ids):
-            return False
-        if tenant_prefixes and _extract_key_part(key, "tenant_prefix") not in set(tenant_prefixes):
-            return False
-        if year is not None and _extract_key_part(key, "year") != f"{year:04d}":
-            return False
-        if month is not None and _extract_key_part(key, "month") != f"{month:02d}":
-            return False
-        return True
-
-    @staticmethod
-    def _validate_manifest_scope(
-        manifest: ArchiveBundleManifest,
-        *,
-        manifest_key: str,
-        tenant_ids: Sequence[str] | None,
-        tenant_prefixes: Sequence[str] | None,
-        year: int | None,
-        month: int | None,
-    ) -> None:
-        expected_object_prefix = manifest_key.removesuffix(f"/{ARCHIVE_BUNDLE_MANIFEST_NAME}")
-        if manifest["object_prefix"] != expected_object_prefix:
-            raise ValueError(
-                f"manifest object_prefix mismatch: expected={expected_object_prefix}, "
-                f"actual={manifest['object_prefix']}"
-            )
-        if tenant_ids and manifest["tenant_id"] not in tenant_ids:
-            raise ValueError(f"manifest tenant_id is outside requested scope: {manifest['tenant_id']}")
-        if tenant_prefixes and manifest["tenant_prefix"] not in tenant_prefixes:
-            raise ValueError(f"manifest tenant_prefix is outside requested scope: {manifest['tenant_prefix']}")
-        if year is not None and manifest["year"] != year:
-            raise ValueError(f"manifest year is outside requested scope: {manifest['year']}")
-        if month is not None and manifest["month"] != month:
-            raise ValueError(f"manifest month is outside requested scope: {manifest['month']}")
-
-
-def _validate_archive_bundle_manifest(manifest: ArchiveBundleManifest) -> None:
-    if manifest["schema_version"] != ARCHIVE_BUNDLE_SCHEMA_VERSION:
-        raise ValueError(f"unsupported archive bundle schema version: {manifest['schema_version']}")
-    if manifest["archive_format"] != ARCHIVE_BUNDLE_FORMAT:
-        raise ValueError(f"unsupported archive bundle format: {manifest['archive_format']}")
-
-
-def _extract_key_part(key: str, name: str) -> str | None:
-    prefix = f"{name}="
-    for part in key.split("/"):
-        if part.startswith(prefix):
-            return part[len(prefix) :]
-    return None

api/services/retention/workflow_run/archive_download_preparation.py

@@ -1,325 +0,0 @@
-"""
-Prepare monthly workflow-run archive downloads.
-
-Console requests create a short-lived Redis task and Celery runs this module in the background. The DB bundle index is
-the online lookup source: this preparer never lists archive storage, and it validates the indexed bundle set against the
-stable download id before packaging archive Parquet objects into one user-facing CSV ZIP file.
-"""
-
-import datetime
-import hashlib
-import io
-import logging
-import zipfile
-from collections.abc import Sequence
-from typing import cast
-
-import pyarrow.csv as pa_csv
-import pyarrow.parquet as pq
-from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
-
-from extensions.ext_database import db
-from libs.archive_storage import ArchiveStorage, get_archive_storage, get_export_storage
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.archive_bundle_index import (
-    ARCHIVE_BUNDLE_ROOT_PREFIX,
-    ArchiveBundleManifest,
-    ArchiveBundleTableManifestEntry,
-    decode_archive_bundle_manifest,
-)
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-)
-from services.retention.workflow_run.constants import (
-    ARCHIVE_BUNDLE_FORMAT,
-    ARCHIVE_BUNDLE_MANIFEST_NAME,
-    ARCHIVE_BUNDLE_SCHEMA_VERSION,
-)
-
-logger = logging.getLogger(__name__)
-
-ARCHIVE_DOWNLOAD_ROOT_PREFIX = "workflow-runs/downloads/v1/"
-ARCHIVE_DOWNLOAD_MIME_TYPE = "application/zip"
-
-
-class WorkflowRunArchiveDownloadPreparer:
-    """
-    Build one ready-to-download CSV ZIP for a Redis archive download task.
-
-    The output object is deterministic for a given `download_id`, so retrying a failed task overwrites the same
-    temporary object instead of creating unbounded duplicate files. Source archive bundles are read from the archive
-    bucket, while the prepared ZIP is written to the export bucket so object lifecycle policies can expire downloads
-    without touching long-lived archives.
-    """
-
-    archive_storage: ArchiveStorage | None
-    download_storage: ArchiveStorage | None
-    cache: WorkflowRunArchiveDownloadTaskCache
-    session_factory: sessionmaker[Session]
-
-    def __init__(
-        self,
-        *,
-        storage: ArchiveStorage | None = None,
-        archive_storage: ArchiveStorage | None = None,
-        download_storage: ArchiveStorage | None = None,
-        cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-        session_factory: sessionmaker[Session] | None = None,
-    ) -> None:
-        self.archive_storage = archive_storage or storage
-        self.download_storage = download_storage or storage
-        self.cache = cache or WorkflowRunArchiveDownloadTaskCache()
-        self.session_factory = session_factory or sessionmaker(bind=db.engine, expire_on_commit=False)
-
-    def prepare(self, *, tenant_id: str, download_id: str) -> WorkflowRunArchiveDownloadTask | None:
-        """Prepare a ZIP for an existing Redis task and persist terminal task state."""
-        task = self.cache.get(tenant_id=tenant_id, download_id=download_id)
-        if task is None:
-            logger.info("Workflow run archive download task expired before preparation: %s", download_id)
-            return None
-        if task.status == WorkflowRunArchiveDownloadStatus.READY:
-            return task
-        if task.status == WorkflowRunArchiveDownloadStatus.FAILED:
-            logger.info("Skipping failed workflow run archive download task: %s", download_id)
-            return task
-
-        processing_task = self._mark_processing(task)
-        try:
-            archive_storage = self.archive_storage or get_archive_storage()
-            download_storage = self.download_storage or get_export_storage()
-            bundles = self._get_task_bundles(processing_task)
-            payload = self._build_zip_payload(archive_storage, processing_task, bundles)
-            storage_key = build_archive_download_storage_key(processing_task)
-            download_storage.put_object(storage_key, payload)
-            return self._mark_ready(processing_task, storage_key=storage_key, file_size_bytes=len(payload))
-        except Exception as exc:
-            logger.exception("Failed to prepare workflow run archive download %s", download_id)
-            return self._mark_failed(processing_task, error=str(exc))
-
-    def _get_task_bundles(self, task: WorkflowRunArchiveDownloadTask) -> list[WorkflowRunArchiveBundle]:
-        with self.session_factory() as session:
-            return _list_task_bundles(session, task)
-
-    def _build_zip_payload(
-        self,
-        storage: ArchiveStorage,
-        task: WorkflowRunArchiveDownloadTask,
-        bundles: Sequence[WorkflowRunArchiveBundle],
-    ) -> bytes:
-        zip_root = f"workflow-run-logs-{task.year:04d}-{task.month:02d}"
-        csv_buffers: dict[str, io.BytesIO] = {}
-        csv_headers_written: set[str] = set()
-
-        for bundle in bundles:
-            object_prefix = _build_archive_bundle_object_prefix(task, bundle)
-            _, manifest = _load_and_validate_manifest(storage, task, bundle, object_prefix)
-            for table_name in sorted(manifest["tables"]):
-                entry = manifest["tables"][table_name]
-                object_key = entry["object_key"]
-                table_payload = storage.get_object(object_key)
-                _validate_table_payload(object_key=object_key, entry=entry, payload=table_payload)
-                csv_payload = _parquet_payload_to_csv(
-                    table_payload,
-                    include_header=table_name not in csv_headers_written,
-                )
-                if not csv_payload:
-                    continue
-                csv_buffers.setdefault(table_name, io.BytesIO()).write(csv_payload)
-                csv_headers_written.add(table_name)
-
-        buffer = io.BytesIO()
-        with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as archive:
-            for table_name, csv_buffer in sorted(csv_buffers.items()):
-                archive.writestr(f"{zip_root}/{table_name}.csv", csv_buffer.getvalue())
-        return buffer.getvalue()
-
-    def _mark_processing(self, task: WorkflowRunArchiveDownloadTask) -> WorkflowRunArchiveDownloadTask:
-        now = datetime.datetime.now(datetime.UTC)
-        processing_task = task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.PROCESSING,
-                "error": None,
-                "updated_at": now,
-                "started_at": task.started_at or now,
-            }
-        )
-        self.cache.save(processing_task)
-        return processing_task
-
-    def _mark_ready(
-        self,
-        task: WorkflowRunArchiveDownloadTask,
-        *,
-        storage_key: str,
-        file_size_bytes: int,
-    ) -> WorkflowRunArchiveDownloadTask:
-        now = datetime.datetime.now(datetime.UTC)
-        ready_task = task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.READY,
-                "file_name": build_archive_download_file_name(task),
-                "storage_key": storage_key,
-                "file_size_bytes": file_size_bytes,
-                "error": None,
-                "updated_at": now,
-                "finished_at": now,
-            }
-        )
-        self.cache.save(ready_task)
-        return ready_task
-
-    def _mark_failed(self, task: WorkflowRunArchiveDownloadTask, *, error: str) -> WorkflowRunArchiveDownloadTask:
-        now = datetime.datetime.now(datetime.UTC)
-        failed_task = task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.FAILED,
-                "error": error,
-                "updated_at": now,
-                "finished_at": now,
-            }
-        )
-        self.cache.save(failed_task)
-        return failed_task
-
-
-def build_archive_download_file_name(task: WorkflowRunArchiveDownloadTask) -> str:
-    """Return the browser download filename for one monthly archive."""
-    return f"workflow-run-logs-{task.year:04d}-{task.month:02d}.zip"
-
-
-def build_archive_download_storage_key(task: WorkflowRunArchiveDownloadTask) -> str:
-    """Return the deterministic object-store key for a prepared download ZIP."""
-    return (
-        f"{ARCHIVE_DOWNLOAD_ROOT_PREFIX}tenant_prefix={task.tenant_id[0].lower()}/tenant_id={task.tenant_id}/"
-        f"year={task.year:04d}/month={task.month:02d}/{task.download_id}.zip"
-    )
-
-
-def _list_task_bundles(session: Session, task: WorkflowRunArchiveDownloadTask) -> list[WorkflowRunArchiveBundle]:
-    stmt = (
-        select(WorkflowRunArchiveBundle)
-        .where(
-            WorkflowRunArchiveBundle.tenant_id == task.tenant_id,
-            WorkflowRunArchiveBundle.year == task.year,
-            WorkflowRunArchiveBundle.month == task.month,
-        )
-        .order_by(WorkflowRunArchiveBundle.shard, WorkflowRunArchiveBundle.bundle_id)
-    )
-    indexed_bundles = list(session.scalars(stmt))
-    if task.bundle_refs:
-        requested_refs = [(ref.shard, ref.bundle_id) for ref in task.bundle_refs]
-    else:
-        requested_bundle_ids = set(task.bundle_ids)
-        requested_refs = [
-            (bundle.shard, bundle.bundle_id) for bundle in indexed_bundles if bundle.bundle_id in requested_bundle_ids
-        ]
-
-    bundle_by_ref = {(bundle.shard, bundle.bundle_id): bundle for bundle in indexed_bundles}
-    missing_refs = [ref for ref in requested_refs if ref not in bundle_by_ref]
-    if missing_refs:
-        raise ValueError(f"archive bundle index is missing requested bundles: {missing_refs}")
-
-    bundles = [bundle_by_ref[ref] for ref in requested_refs]
-    if len(bundles) != task.bundle_count:
-        raise ValueError(f"archive bundle count changed: expected={task.bundle_count}, actual={len(bundles)}")
-
-    download_id = build_archive_download_id(
-        tenant_id=task.tenant_id,
-        year=task.year,
-        month=task.month,
-        bundle_refs=requested_refs,
-    )
-    if download_id != task.download_id:
-        raise ValueError("archive download id no longer matches indexed bundle set")
-
-    return bundles
-
-
-def _build_archive_bundle_object_prefix(
-    task: WorkflowRunArchiveDownloadTask,
-    bundle: WorkflowRunArchiveBundle,
-) -> str:
-    return (
-        f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix={task.tenant_id[0].lower()}/tenant_id={task.tenant_id}/"
-        f"year={task.year:04d}/month={task.month:02d}/shard={bundle.shard}/bundle={bundle.bundle_id}"
-    )
-
-
-def _load_and_validate_manifest(
-    storage: ArchiveStorage,
-    task: WorkflowRunArchiveDownloadTask,
-    bundle: WorkflowRunArchiveBundle,
-    object_prefix: str,
-) -> tuple[bytes, ArchiveBundleManifest]:
-    manifest_key = f"{object_prefix}/{ARCHIVE_BUNDLE_MANIFEST_NAME}"
-    manifest_data = storage.get_object(manifest_key)
-    manifest = decode_archive_bundle_manifest(manifest_data)
-    _validate_manifest(task=task, bundle=bundle, manifest=manifest, object_prefix=object_prefix)
-    return manifest_data, manifest
-
-
-def _validate_manifest(
-    *,
-    task: WorkflowRunArchiveDownloadTask,
-    bundle: WorkflowRunArchiveBundle,
-    manifest: ArchiveBundleManifest,
-    object_prefix: str,
-) -> None:
-    if manifest["schema_version"] != ARCHIVE_BUNDLE_SCHEMA_VERSION:
-        raise ValueError(f"unsupported archive bundle schema version: {manifest['schema_version']}")
-    if manifest["archive_format"] != ARCHIVE_BUNDLE_FORMAT:
-        raise ValueError(f"unsupported archive bundle format: {manifest['archive_format']}")
-    if manifest["tenant_id"] != task.tenant_id:
-        raise ValueError(f"manifest tenant_id mismatch: expected={task.tenant_id}, actual={manifest['tenant_id']}")
-    if manifest["year"] != task.year:
-        raise ValueError(f"manifest year mismatch: expected={task.year}, actual={manifest['year']}")
-    if manifest["month"] != task.month:
-        raise ValueError(f"manifest month mismatch: expected={task.month}, actual={manifest['month']}")
-    if manifest["shard"] != bundle.shard:
-        raise ValueError(f"manifest shard mismatch: expected={bundle.shard}, actual={manifest['shard']}")
-    if manifest["bundle_id"] != bundle.bundle_id:
-        raise ValueError(f"manifest bundle_id mismatch: expected={bundle.bundle_id}, actual={manifest['bundle_id']}")
-    if manifest["object_prefix"] != object_prefix:
-        raise ValueError(
-            f"manifest object_prefix mismatch: expected={object_prefix}, actual={manifest['object_prefix']}"
-        )
-    if not manifest["tables"]:
-        raise ValueError("manifest tables must not be empty")
-    for table_name, raw_entry in manifest["tables"].items():
-        entry = cast(ArchiveBundleTableManifestEntry, raw_entry)
-        expected_object_key = f"{object_prefix}/{table_name}.parquet"
-        if entry["object_key"] != expected_object_key:
-            raise ValueError(
-                f"manifest object_key mismatch for {table_name}: "
-                f"expected={expected_object_key}, actual={entry['object_key']}"
-            )
-
-
-def _validate_table_payload(
-    *,
-    object_key: str,
-    entry: ArchiveBundleTableManifestEntry,
-    payload: bytes,
-) -> None:
-    if len(payload) != entry["size_bytes"]:
-        raise ValueError(f"archive object size mismatch for {object_key}")
-    checksum = hashlib.md5(payload).hexdigest()
-    if checksum != entry["checksum"]:
-        raise ValueError(f"archive object checksum mismatch for {object_key}")
-
-
-def _parquet_payload_to_csv(payload: bytes, *, include_header: bool) -> bytes:
-    table = pq.read_table(io.BytesIO(payload))
-    if table.num_columns == 0:
-        return b""
-    buffer = io.BytesIO()
-    pa_csv.write_csv(
-        table,
-        buffer,
-        write_options=pa_csv.WriteOptions(include_header=include_header),
-    )
-    return buffer.getvalue()

api/services/retention/workflow_run/archive_download_task_cache.py

@@ -1,178 +0,0 @@
-"""Redis-backed temporary state for workflow-run archive downloads."""
-
-import datetime
-import hashlib
-import json
-import logging
-from collections.abc import Sequence
-from enum import StrEnum
-
-from pydantic import BaseModel, ConfigDict, Field
-
-from extensions.ext_redis import RedisClientWrapper, redis_client
-
-logger = logging.getLogger(__name__)
-
-ARCHIVE_DOWNLOAD_FORMAT_VERSION = "v1"
-DEFAULT_ARCHIVE_DOWNLOAD_TASK_TTL_SECONDS = 24 * 60 * 60
-_CACHE_KEY_PREFIX = "workflow_run_archive_download"
-
-
-class WorkflowRunArchiveDownloadStatus(StrEnum):
-    """Lifecycle state for an asynchronous archive download request."""
-
-    PENDING = "pending"
-    PROCESSING = "processing"
-    READY = "ready"
-    FAILED = "failed"
-
-
-class WorkflowRunArchiveBundleRef(BaseModel):
-    """Immutable object-store identity for one bundle included in a download task."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    shard: str
-    bundle_id: str
-
-
-class WorkflowRunArchiveDownloadTask(BaseModel):
-    """Temporary Redis payload for a monthly archive download request."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    download_id: str
-    tenant_id: str
-    requested_by: str
-    year: int = Field(ge=1)
-    month: int = Field(ge=1, le=12)
-    bundle_ids: list[str]
-    bundle_refs: list[WorkflowRunArchiveBundleRef] = Field(default_factory=list)
-    bundle_count: int = Field(ge=0)
-    archive_bytes: int = Field(ge=0)
-    status: WorkflowRunArchiveDownloadStatus
-    file_name: str | None = None
-    storage_key: str | None = None
-    file_size_bytes: int | None = Field(default=None, ge=0)
-    celery_task_id: str | None = None
-    error: str | None = None
-    created_at: datetime.datetime
-    updated_at: datetime.datetime
-    expires_at: datetime.datetime
-    started_at: datetime.datetime | None = None
-    finished_at: datetime.datetime | None = None
-
-
-class WorkflowRunArchiveDownloadTaskCache:
-    """Store ephemeral archive download task state in Redis with a TTL."""
-
-    _redis: RedisClientWrapper
-
-    def __init__(self, redis: RedisClientWrapper = redis_client) -> None:
-        self._redis = redis
-
-    def get(self, *, tenant_id: str, download_id: str) -> WorkflowRunArchiveDownloadTask | None:
-        raw = self._redis.get(self._cache_key(tenant_id=tenant_id, download_id=download_id))
-        if raw is None:
-            return None
-        data = raw.decode("utf-8") if isinstance(raw, bytes | bytearray) else raw
-        try:
-            return WorkflowRunArchiveDownloadTask.model_validate_json(data)
-        except ValueError:
-            logger.warning("Malformed workflow run archive download task cache entry: %s", download_id)
-            return None
-
-    def save(self, task: WorkflowRunArchiveDownloadTask) -> None:
-        ttl_seconds = self._ttl_seconds(task.expires_at)
-        self._redis.setex(
-            self._cache_key(tenant_id=task.tenant_id, download_id=task.download_id),
-            ttl_seconds,
-            task.model_dump_json(),
-        )
-
-    def create_if_absent(self, task: WorkflowRunArchiveDownloadTask) -> bool:
-        ttl_seconds = self._ttl_seconds(task.expires_at)
-        result = self._redis.set(
-            self._cache_key(tenant_id=task.tenant_id, download_id=task.download_id),
-            task.model_dump_json(),
-            ex=ttl_seconds,
-            nx=True,
-        )
-        return bool(result)
-
-    def delete(self, *, tenant_id: str, download_id: str) -> None:
-        self._redis.delete(self._cache_key(tenant_id=tenant_id, download_id=download_id))
-
-    @staticmethod
-    def _cache_key(*, tenant_id: str, download_id: str) -> str:
-        return f"{_CACHE_KEY_PREFIX}:{tenant_id}:{download_id}"
-
-    @staticmethod
-    def _ttl_seconds(expires_at: datetime.datetime) -> int:
-        expires_at_utc = expires_at if expires_at.tzinfo else expires_at.replace(tzinfo=datetime.UTC)
-        remaining = expires_at_utc - datetime.datetime.now(datetime.UTC)
-        return max(int(remaining.total_seconds()), 1)
-
-
-def build_pending_archive_download_task(
-    *,
-    tenant_id: str,
-    requested_by: str,
-    year: int,
-    month: int,
-    bundle_ids: Sequence[str],
-    bundle_refs: Sequence[tuple[str, str]] = (),
-    archive_bytes: int,
-    download_id: str,
-    ttl_seconds: int = DEFAULT_ARCHIVE_DOWNLOAD_TASK_TTL_SECONDS,
-    now: datetime.datetime | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """Create the Redis payload stored when the console starts an archive download."""
-    created_at = now or datetime.datetime.now(datetime.UTC)
-    if created_at.tzinfo is None:
-        created_at = created_at.replace(tzinfo=datetime.UTC)
-    normalized_bundle_ids = list(bundle_ids)
-    normalized_bundle_refs = [
-        WorkflowRunArchiveBundleRef(shard=shard, bundle_id=bundle_id) for shard, bundle_id in bundle_refs
-    ]
-    return WorkflowRunArchiveDownloadTask(
-        download_id=download_id,
-        tenant_id=tenant_id,
-        requested_by=requested_by,
-        year=year,
-        month=month,
-        bundle_ids=normalized_bundle_ids,
-        bundle_refs=normalized_bundle_refs,
-        bundle_count=len(normalized_bundle_ids),
-        archive_bytes=archive_bytes,
-        status=WorkflowRunArchiveDownloadStatus.PENDING,
-        created_at=created_at,
-        updated_at=created_at,
-        expires_at=created_at + datetime.timedelta(seconds=ttl_seconds),
-    )
-
-
-def build_archive_download_id(
-    *,
-    tenant_id: str,
-    year: int,
-    month: int,
-    bundle_refs: Sequence[tuple[str, str]],
-    download_format_version: str = ARCHIVE_DOWNLOAD_FORMAT_VERSION,
-) -> str:
-    """Build a stable id for the exact archive download content."""
-    if not bundle_refs:
-        raise ValueError("bundle_refs must not be empty")
-    normalized_refs = sorted(f"{shard}:{bundle_id}" for shard, bundle_id in bundle_refs)
-    payload = json.dumps(
-        {
-            "tenant_id": tenant_id,
-            "year": year,
-            "month": month,
-            "bundle_refs": normalized_refs,
-            "download_format_version": download_format_version,
-        },
-        sort_keys=True,
-        separators=(",", ":"),
-    )
-    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:32]

api/services/retention/workflow_run/archive_log_service.py

@@ -1,293 +0,0 @@
-"""
-Console-facing workflow-run archive queries.
-
-The object store remains the recoverable archive source of truth. This module only reads the DB bundle index and writes
-temporary Redis download-task state, so console requests never list R2 online.
-"""
-
-import datetime
-import logging
-import uuid
-from collections.abc import Callable
-from dataclasses import dataclass
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-    build_pending_archive_download_task,
-)
-
-logger = logging.getLogger(__name__)
-
-ArchiveDownloadTaskDispatcher = Callable[
-    [WorkflowRunArchiveDownloadTask, WorkflowRunArchiveDownloadTaskCache],
-    WorkflowRunArchiveDownloadTask,
-]
-
-
-@dataclass(frozen=True)
-class WorkflowRunArchiveMonth:
-    """Aggregated archive metadata for one tenant/month."""
-
-    year: int
-    month: int
-    bundle_count: int
-    workflow_run_count: int
-    row_count: int
-    archive_bytes: int
-    latest_archived_at: datetime.datetime
-    download_task: WorkflowRunArchiveDownloadTask | None
-
-
-@dataclass(frozen=True)
-class WorkflowRunArchiveSummary:
-    """Top-level archive totals shown on the console page."""
-
-    archived_month_count: int
-    workflow_run_count: int
-    archive_bytes: int
-    latest_archived_at: datetime.datetime | None
-
-
-@dataclass(frozen=True)
-class WorkflowRunArchiveList:
-    """Console response model before controller serialization."""
-
-    summary: WorkflowRunArchiveSummary
-    months: list[WorkflowRunArchiveMonth]
-
-
-class WorkflowRunArchiveNotFoundError(Exception):
-    """Raised when no archive bundles exist for a requested tenant/month."""
-
-
-class WorkflowRunArchiveDownloadTaskNotFoundError(Exception):
-    """Raised when the temporary Redis task has expired or never existed."""
-
-
-class WorkflowRunArchiveDownloadNotReadyError(Exception):
-    """Raised when a cached download task has not produced a file yet."""
-
-
-def list_workflow_run_archives(
-    session: Session,
-    tenant_id: str,
-    *,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-) -> WorkflowRunArchiveList:
-    """Return monthly archive metadata for one tenant from the DB bundle index."""
-    stmt = (
-        select(WorkflowRunArchiveBundle)
-        .where(WorkflowRunArchiveBundle.tenant_id == tenant_id)
-        .order_by(
-            WorkflowRunArchiveBundle.year.desc(),
-            WorkflowRunArchiveBundle.month.desc(),
-            WorkflowRunArchiveBundle.shard,
-            WorkflowRunArchiveBundle.bundle_id,
-        )
-    )
-    month_bundles: dict[tuple[int, int], list[WorkflowRunArchiveBundle]] = {}
-    for bundle in session.scalars(stmt):
-        month_bundles.setdefault((bundle.year, bundle.month), []).append(bundle)
-
-    task_cache = cache or WorkflowRunArchiveDownloadTaskCache()
-    months: list[WorkflowRunArchiveMonth] = []
-    for (year, month), bundles in month_bundles.items():
-        bundle_refs = [(bundle.shard, bundle.bundle_id) for bundle in bundles]
-        months.append(
-            WorkflowRunArchiveMonth(
-                year=year,
-                month=month,
-                bundle_count=len(bundles),
-                workflow_run_count=sum(bundle.workflow_run_count for bundle in bundles),
-                row_count=sum(bundle.row_count for bundle in bundles),
-                archive_bytes=sum(bundle.archive_bytes for bundle in bundles),
-                latest_archived_at=max(bundle.archived_at for bundle in bundles),
-                download_task=_get_cached_month_download_task(
-                    task_cache,
-                    tenant_id=tenant_id,
-                    year=year,
-                    month=month,
-                    bundle_refs=bundle_refs,
-                ),
-            )
-        )
-    latest_archived_at = max((month.latest_archived_at for month in months), default=None)
-    return WorkflowRunArchiveList(
-        summary=WorkflowRunArchiveSummary(
-            archived_month_count=len(months),
-            workflow_run_count=sum(month.workflow_run_count for month in months),
-            archive_bytes=sum(month.archive_bytes for month in months),
-            latest_archived_at=latest_archived_at,
-        ),
-        months=months,
-    )
-
-
-def _get_cached_month_download_task(
-    cache: WorkflowRunArchiveDownloadTaskCache,
-    *,
-    tenant_id: str,
-    year: int,
-    month: int,
-    bundle_refs: list[tuple[str, str]],
-) -> WorkflowRunArchiveDownloadTask | None:
-    if not bundle_refs:
-        return None
-    download_id = build_archive_download_id(
-        tenant_id=tenant_id,
-        year=year,
-        month=month,
-        bundle_refs=bundle_refs,
-    )
-    try:
-        return cache.get(tenant_id=tenant_id, download_id=download_id)
-    except Exception:
-        logger.warning("Failed to read cached workflow run archive download task: %s", download_id, exc_info=True)
-        return None
-
-
-def create_workflow_run_archive_download_task(
-    session: Session,
-    *,
-    tenant_id: str,
-    requested_by: str,
-    year: int,
-    month: int,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-    dispatcher: ArchiveDownloadTaskDispatcher | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """
-    Create or return the idempotent Redis task for downloading one tenant/month archive.
-
-    The task identity is based on the exact ordered bundle set currently indexed for the month. If the month receives a
-    new bundle later, the next request gets a different download id and prepares a fresh file.
-    """
-    bundles = _list_archive_bundles(session, tenant_id=tenant_id, year=year, month=month)
-    if not bundles:
-        raise WorkflowRunArchiveNotFoundError(f"Workflow run archive not found: {year:04d}-{month:02d}")
-
-    bundle_refs = [(bundle.shard, bundle.bundle_id) for bundle in bundles]
-    download_id = build_archive_download_id(
-        tenant_id=tenant_id,
-        year=year,
-        month=month,
-        bundle_refs=bundle_refs,
-    )
-    task = build_pending_archive_download_task(
-        tenant_id=tenant_id,
-        requested_by=requested_by,
-        year=year,
-        month=month,
-        bundle_ids=[bundle.bundle_id for bundle in bundles],
-        bundle_refs=bundle_refs,
-        archive_bytes=sum(bundle.archive_bytes for bundle in bundles),
-        download_id=download_id,
-    )
-    task_cache = cache or WorkflowRunArchiveDownloadTaskCache()
-    dispatch = dispatcher or _dispatch_workflow_run_archive_download_task
-    if task_cache.create_if_absent(task):
-        return dispatch(task, task_cache)
-
-    existing = task_cache.get(tenant_id=tenant_id, download_id=download_id)
-    if existing is not None:
-        if existing.status == WorkflowRunArchiveDownloadStatus.FAILED:
-            task_cache.save(task)
-            return dispatch(task, task_cache)
-        if existing.status == WorkflowRunArchiveDownloadStatus.PENDING and not existing.celery_task_id:
-            return dispatch(existing, task_cache)
-        return existing
-
-    task_cache.save(task)
-    return dispatch(task, task_cache)
-
-
-def get_workflow_run_archive_download_task(
-    *,
-    tenant_id: str,
-    download_id: str,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """Return a cached archive download task or raise when the TTL has expired."""
-    task_cache = cache or WorkflowRunArchiveDownloadTaskCache()
-    task = task_cache.get(tenant_id=tenant_id, download_id=download_id)
-    if task is None:
-        raise WorkflowRunArchiveDownloadTaskNotFoundError(f"Workflow run archive download not found: {download_id}")
-    return task
-
-
-def get_ready_workflow_run_archive_download_task(
-    *,
-    tenant_id: str,
-    download_id: str,
-    cache: WorkflowRunArchiveDownloadTaskCache | None = None,
-) -> WorkflowRunArchiveDownloadTask:
-    """Return a ready cached archive download task or raise when the file is not available."""
-    task = get_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id, cache=cache)
-    if task.status != WorkflowRunArchiveDownloadStatus.READY or not task.storage_key or not task.file_name:
-        raise WorkflowRunArchiveDownloadNotReadyError(f"Workflow run archive download is not ready: {download_id}")
-    return task
-
-
-def _list_archive_bundles(
-    session: Session,
-    *,
-    tenant_id: str,
-    year: int,
-    month: int,
-) -> list[WorkflowRunArchiveBundle]:
-    stmt = (
-        select(WorkflowRunArchiveBundle)
-        .where(
-            WorkflowRunArchiveBundle.tenant_id == tenant_id,
-            WorkflowRunArchiveBundle.year == year,
-            WorkflowRunArchiveBundle.month == month,
-        )
-        .order_by(WorkflowRunArchiveBundle.shard, WorkflowRunArchiveBundle.bundle_id)
-    )
-    return list(session.scalars(stmt))
-
-
-def _dispatch_workflow_run_archive_download_task(
-    task: WorkflowRunArchiveDownloadTask,
-    cache: WorkflowRunArchiveDownloadTaskCache,
-) -> WorkflowRunArchiveDownloadTask:
-    """
-    Enqueue background ZIP preparation and persist the Celery id before the worker can start.
-
-    The Redis task key is the idempotency boundary. We generate the Celery id in the API process, save it on the task,
-    then submit with that exact id so duplicate console requests keep seeing one logical download request.
-    """
-    from tasks.workflow_run_archive_download_tasks import prepare_workflow_run_archive_download_task
-
-    now = datetime.datetime.now(datetime.UTC)
-    celery_task_id = uuid.uuid4().hex
-    queued_task = task.model_copy(update={"celery_task_id": celery_task_id, "updated_at": now})
-    cache.save(queued_task)
-
-    try:
-        prepare_workflow_run_archive_download_task.apply_async(
-            args=(queued_task.tenant_id, queued_task.download_id),
-            task_id=celery_task_id,
-        )
-    except Exception:
-        failure_time = datetime.datetime.now(datetime.UTC)
-        failed_task = queued_task.model_copy(
-            update={
-                "status": WorkflowRunArchiveDownloadStatus.FAILED,
-                "error": "Failed to enqueue archive download task.",
-                "updated_at": failure_time,
-                "finished_at": failure_time,
-            }
-        )
-        cache.save(failed_task)
-        logger.exception("Failed to enqueue workflow run archive download task %s", queued_task.download_id)
-        return failed_task
-
-    return queued_task

api/services/retention/workflow_run/archive_paid_plan_workflow_run.py

@@ -5,8 +5,8 @@ This service archives workflow run logs for paid plan users older than the confi
 90 days) to S3-compatible storage.
 
 Archive V2 writes bundle-level Parquet objects. A bundle contains many workflow runs and their related table rows.
-Bundle metadata lives in the object-store manifest as the recoverable source of truth. Completed bundles are also
-mirrored into a small database index so console listing and download jobs do not list object storage online.
+Bundle metadata lives in the object-store manifest instead of a database table, so archive/delete/restore does not move
+the large-table retention problem into another OLTP table.
 
 Archive campaigns should use fixed absolute UTC windows for every tenant-prefix/shard execution. Relative windows are
 evaluated at process start and are not safe for multi-day rollout because each command would scan a different window.
@@ -64,12 +64,6 @@ from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowN
 from repositories.api_workflow_run_repository import APIWorkflowRunRepository
 from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
 from services.billing_service import BillingService
-from services.retention.workflow_run.archive_bundle_index import (
-    ArchiveBundleManifest,
-    ArchiveBundleTableManifestEntry,
-    decode_archive_bundle_manifest,
-    upsert_archive_bundle_index_from_manifest,
-)
 from services.retention.workflow_run.constants import (
     ARCHIVE_BUNDLE_FORMAT,
     ARCHIVE_BUNDLE_INDEX_NAME,
@@ -581,7 +575,6 @@ class WorkflowRunArchiver:
                     raise ArchiveStorageNotConfiguredError("Archive storage not configured")
                 if storage.object_exists(self._get_manifest_object_key(identity)):
                     self._write_bundle_index(storage, identity)
-                    self._sync_existing_bundle_index(session, storage, identity)
                     result.success = True
                     result.skipped = True
                     result.error = "bundle already archived"
@@ -605,7 +598,6 @@ class WorkflowRunArchiver:
                 result.run_count = len(runs)
                 if storage.object_exists(self._get_manifest_object_key(identity)):
                     self._write_bundle_index(storage, identity)
-                    self._sync_existing_bundle_index(session, storage, identity)
                     result.success = True
                     result.skipped = True
                     result.error = "filtered bundle already archived"
@@ -637,8 +629,6 @@ class WorkflowRunArchiver:
                     storage.put_object(self._get_table_object_key(identity, table_name), payload)
                 storage.put_object(self._get_manifest_object_key(identity), manifest_data)
                 self._merge_bundle_manifest_into_index(storage, identity, [run.id for run in runs])
-                manifest = decode_archive_bundle_manifest(manifest_data)
-                upsert_archive_bundle_index_from_manifest(session, manifest, len(manifest_data))
                 session.commit()
 
                 logger.info(
@@ -662,23 +652,6 @@ class WorkflowRunArchiver:
         result.elapsed_time = time.time() - start_time
         return result
 
-    def _sync_existing_bundle_index(
-        self,
-        session: Session,
-        storage: ArchiveStorage,
-        identity: ArchiveBundleIdentity,
-    ) -> None:
-        """Best-effort DB index sync for a bundle whose manifest already exists in archive storage."""
-        manifest_key = self._get_manifest_object_key(identity)
-        try:
-            manifest_data = storage.get_object(manifest_key)
-            manifest = decode_archive_bundle_manifest(manifest_data)
-            upsert_archive_bundle_index_from_manifest(session, manifest, len(manifest_data))
-            session.commit()
-        except Exception:
-            session.rollback()
-            logger.warning("Failed to sync workflow archive bundle index for %s", manifest_key, exc_info=True)
-
     def _lock_runs_for_archive(
         self,
         session: Session,
@@ -806,9 +779,9 @@ class WorkflowRunArchiver:
         identity: ArchiveBundleIdentity,
         runs: Sequence[WorkflowRun],
         table_stats: list[TableStats],
-    ) -> ArchiveBundleManifest:
+    ) -> ArchiveManifestDict:
         """Generate a manifest for the archived workflow run bundle."""
-        tables: dict[str, ArchiveBundleTableManifestEntry] = {
+        tables: dict[str, TableStatsManifestEntry] = {
             stat.table_name: {
                 "row_count": stat.row_count,
                 "checksum": stat.checksum,
@@ -821,9 +794,6 @@ class WorkflowRunArchiver:
         end_before = self.end_before
         if end_before is None:
             raise ValueError("archive window end must be set")
-        archive_window_end = self._format_window_datetime(end_before)
-        if archive_window_end is None:
-            raise ValueError("archive window end must be set")
         return ArchiveManifestDict(
             schema_version=ARCHIVE_BUNDLE_SCHEMA_VERSION,
             archive_format=ARCHIVE_BUNDLE_FORMAT,
@@ -843,7 +813,7 @@ class WorkflowRunArchiver:
             archived_at=datetime.datetime.now(datetime.UTC).isoformat(),
             campaign_id=self.campaign_id,
             archive_window_start=self._format_window_datetime(self.start_from),
-            archive_window_end=archive_window_end,
+            archive_window_end=end_before.isoformat(),
             run_shard=identity.shard,
             tables=tables,
             run_ids=[run.id for run in sorted_runs],

api/services/retention/workflow_run/bundle_archive_maintenance.py

@@ -1,10 +1,9 @@
 """
 Maintain V2 workflow-run archive bundles.
 
-Archive V2 keeps object-store manifests as the recoverable bundle source of truth. This maintenance module still
-discovers delete/restore targets by listing `manifest.json` objects and uses object-store marker files for
-delete/restore state. The separate database bundle index is intended for console listing and download jobs, not as the
-source of truth for destructive maintenance.
+Archive V2 keeps bundle metadata in object-store manifests, not in a database table. This module discovers bundles by
+listing `manifest.json` objects, uses object-store marker files for delete/restore state, and only touches the database
+for source-table validation, deletion, and restoration.
 
 Each bundle is processed in its own database transaction. A failed bundle leaves source rows unchanged unless the
 transaction has already committed; marker handling makes the next run able to reconcile the common committed-but-marker

api/tasks/workflow_run_archive_download_tasks.py

@@ -1,18 +0,0 @@
-"""Celery tasks for preparing workflow-run archive downloads."""
-
-import logging
-
-from celery import shared_task
-
-from services.retention.workflow_run.archive_download_preparation import WorkflowRunArchiveDownloadPreparer
-
-logger = logging.getLogger(__name__)
-
-WORKFLOW_RUN_ARCHIVE_DOWNLOAD_QUEUE = "workflow_archive"
-
-
-@shared_task(queue=WORKFLOW_RUN_ARCHIVE_DOWNLOAD_QUEUE)
-def prepare_workflow_run_archive_download_task(tenant_id: str, download_id: str) -> None:
-    """Prepare a cached workflow-run archive download in the background."""
-    logger.info("Preparing workflow run archive download: tenant=%s download_id=%s", tenant_id, download_id)
-    WorkflowRunArchiveDownloadPreparer().prepare(tenant_id=tenant_id, download_id=download_id)

api/tests/integration_tests/services/retention/test_workflow_run_archiver.py

@@ -7,7 +7,6 @@ import pyarrow as pa
 import pyarrow.parquet as pq
 import pytest
 
-from models.workflow import WorkflowRunArchiveBundle
 from services.retention.workflow_run.archive_paid_plan_workflow_run import (
     ArchiveSummary,
     WorkflowRunArchiver,
@@ -401,37 +400,6 @@ class TestArchiveRunIdempotency:
         assert result.skipped is True
         assert result.error == "bundle already archived"
 
-    def test_successful_bundle_persists_archive_index(self):
-        archiver = WorkflowRunArchiver(days=90)
-        run = MagicMock()
-        run.id = str(uuid.uuid4())
-        run.tenant_id = str(uuid.uuid4())
-        run.created_at = datetime.datetime(2025, 3, 15, 10, 0, 0)
-        session = MagicMock()
-        session.scalar.return_value = None
-        storage = MagicMock()
-        storage.object_exists.return_value = False
-        table_data = {
-            "workflow_runs": [{"id": run.id, "tenant_id": run.tenant_id}],
-            "workflow_node_executions": [{"id": str(uuid.uuid4()), "workflow_run_id": run.id}],
-        }
-
-        with (
-            patch.object(archiver, "_lock_runs_for_archive", return_value=[run]),
-            patch.object(archiver, "_extract_bundle_data", return_value=table_data),
-        ):
-            result = archiver._archive_bundle(session, storage, [run])
-
-        archived_bundle = session.add.call_args.args[0]
-        assert result.success is True
-        assert isinstance(archived_bundle, WorkflowRunArchiveBundle)
-        assert archived_bundle.tenant_id == run.tenant_id
-        assert archived_bundle.year == 2025
-        assert archived_bundle.month == 3
-        assert archived_bundle.workflow_run_count == 1
-        assert archived_bundle.row_count == 2
-        session.commit.assert_called_once()
-
     def test_index_skips_all_already_archived_runs(self):
         archiver = WorkflowRunArchiver(days=90)
         run = MagicMock()

api/tests/unit_tests/configs/test_dify_config.py

@@ -75,6 +75,7 @@ def test_dify_config(monkeypatch: pytest.MonkeyPatch):
     # default values
     assert config.EDITION == "SELF_HOSTED"
     assert config.API_COMPRESSION_ENABLED is False
+    assert config.AGENT_SHELL_ENABLED is True
     assert config.SENTRY_TRACES_SAMPLE_RATE == 1.0
     assert config.TEMPLATE_TRANSFORM_MAX_LENGTH == 400_000
 
@@ -110,6 +111,25 @@ def test_http_timeout_defaults(monkeypatch: pytest.MonkeyPatch):
     assert config.HTTP_REQUEST_MAX_WRITE_TIMEOUT == 600
 
 
+def test_internal_files_url_falls_back_to_server_console_api_url(monkeypatch: pytest.MonkeyPatch):
+    os.environ.clear()
+    monkeypatch.setenv("SERVER_CONSOLE_API_URL", "http://api:5001")
+
+    config = DifyConfig(_env_file=None)
+
+    assert config.INTERNAL_FILES_URL == "http://api:5001"
+
+
+def test_internal_files_url_prefers_explicit_value(monkeypatch: pytest.MonkeyPatch):
+    os.environ.clear()
+    monkeypatch.setenv("INTERNAL_FILES_URL", "http://files-internal:5001")
+    monkeypatch.setenv("SERVER_CONSOLE_API_URL", "http://api:5001")
+
+    config = DifyConfig(_env_file=None)
+
+    assert config.INTERNAL_FILES_URL == "http://files-internal:5001"
+
+
 # NOTE: If there is a `.env` file in your Workspace, this test might not succeed as expected.
 # This is due to `pymilvus` loading all the variables from the `.env` file into `os.environ`.
 def test_flask_configs(monkeypatch: pytest.MonkeyPatch):

api/tests/unit_tests/controllers/console/explore/test_message.py

@@ -49,6 +49,8 @@ def make_message():
     msg.query = "hello"
     msg.re_sign_file_url_answer = ""
     msg.user_feedback = MagicMock(rating=None)
+    msg.total_price = None
+    msg.currency = None
     msg.status = "normal"
     msg.error = None
     return msg

api/tests/unit_tests/controllers/files/test_upload.py

@@ -34,11 +34,11 @@ class DummyFile:
 
 
 class DummyToolFile:
-    def __init__(self):
+    def __init__(self, name="test.txt", mimetype="text/plain"):
         self.id = "file-id"
-        self.name = "test.txt"
+        self.name = name
         self.size = 10
-        self.mimetype = "text/plain"
+        self.mimetype = mimetype
         self.original_url = "http://original"
         self.user_id = "user-1"
         self.tenant_id = "tenant-1"
@@ -56,7 +56,7 @@ class TestPluginUploadFileApi:
         mock_get_user,
         mock_verify_signature,
     ):
-        dummy_file = DummyFile()
+        dummy_file = DummyFile(filename="report.docx", mimetype="application/octet-stream")
 
         module.request = fake_request(
             {
@@ -71,7 +71,10 @@ class TestPluginUploadFileApi:
         )
 
         tool_file_manager_instance = mock_tool_file_manager.return_value
-        tool_file_manager_instance.create_file_by_raw.return_value = DummyToolFile()
+        tool_file_manager_instance.create_file_by_raw.return_value = DummyToolFile(
+            name="report.docx",
+            mimetype="application/octet-stream",
+        )
 
         mock_tool_file_manager.sign_file.return_value = "signed-url"
 
@@ -84,10 +87,12 @@ class TestPluginUploadFileApi:
         assert result["id"] == "file-id"
         assert result["reference"] == build_file_reference(record_id="file-id")
         assert result["preview_url"] == "signed-url"
+        assert result["extension"] == ".docx"
         mock_verify_signature.assert_called_once()
         assert mock_verify_signature.call_args.kwargs["conversation_id"] == "conversation-1"
         tool_file_manager_instance.create_file_by_raw.assert_called_once()
         assert tool_file_manager_instance.create_file_by_raw.call_args.kwargs["conversation_id"] == "conversation-1"
+        mock_tool_file_manager.sign_file.assert_called_once_with(tool_file_id="file-id", extension=".docx")
 
     def test_missing_file(self):
         module.request = fake_request(

api/tests/unit_tests/controllers/inner_api/plugin/test_plugin.py

@@ -318,6 +318,7 @@ class TestPluginDownloadFileRequestApi:
         mock_payload.user_id = "user-id"
         mock_payload.user_from = "account"
         mock_payload.invoke_from = "debugger"
+        mock_payload.for_external = False
         reference = build_file_reference(record_id="tool-file-1")
         mock_payload.file.model_dump.return_value = {
             "transfer_method": "tool_file",
@@ -333,6 +334,7 @@ class TestPluginDownloadFileRequestApi:
             user_from="account",
             invoke_from="debugger",
             file_mapping={"transfer_method": "tool_file", "reference": reference},
+            for_external=False,
         )
         assert result["data"] == {
             "filename": "report.pdf",

api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py

@@ -37,6 +37,7 @@ from pydantic_ai.messages import (
 from clients.agent_backend import (
     AgentBackendError,
     AgentBackendRunEventAdapter,
+    AgentBackendRunFailedInternalEvent,
     AgentBackendStreamInternalEvent,
     FakeAgentBackendRunClient,
     FakeAgentBackendScenario,
@@ -54,6 +55,7 @@ from core.app.entities.queue_entities import (
     QueueMessageEndEvent,
 )
 from core.workflow.nodes.agent_v2.ask_human_resume import AskHumanResumeOutcome
+from graphon.model_runtime.errors.invoke import InvokeRateLimitError
 from models.agent_config_entities import AgentSoulConfig
 from models.model import MessageAgentThought
 
@@ -1039,6 +1041,130 @@ def test_tool_result_without_call_id_matches_unique_open_tool_name(monkeypatch):
     assert rows[0].observation == "Knowledge base search results: browser skill"
 
 
+def test_repeated_tool_calls_without_call_id_or_index_create_distinct_rows(monkeypatch):
+    fake_session = _FakeDbSession()
+    monkeypatch.setattr(app_runner_module.db, "session", fake_session)
+    qm = _FakeQueueManager()
+    recorder = app_runner_module._AgentProcessRecorder(
+        dify_context=_dify_ctx(),
+        message_id="msg-1",
+        queue_manager=qm,  # type: ignore[arg-type]
+    )
+
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_call",
+                "part": {
+                    "part_kind": "tool-call",
+                    "tool_name": "shell_run",
+                    "args": {"script": "lookup find"},
+                },
+            },
+        )
+    )
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_result",
+                "part": {
+                    "part_kind": "tool-return",
+                    "tool_name": "shell_run",
+                    "content": "find output",
+                },
+            },
+        )
+    )
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_call",
+                "part": {
+                    "part_kind": "tool-call",
+                    "tool_name": "shell_run",
+                    "args": {"script": "lookup out"},
+                },
+            },
+        )
+    )
+    recorder.handle_stream_event(
+        AgentBackendStreamInternalEvent(
+            run_id="run-1",
+            data={
+                "event_kind": "function_tool_result",
+                "part": {
+                    "part_kind": "tool-return",
+                    "tool_name": "shell_run",
+                    "content": "out output",
+                },
+            },
+        )
+    )
+
+    rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
+    assert len(rows) == 2
+    assert rows[0].tool == "shell_run"
+    assert rows[0].tool_input == '{"script": "lookup find"}'
+    assert rows[0].observation == "find output"
+    assert rows[1].tool == "shell_run"
+    assert rows[1].tool_input == '{"script": "lookup out"}'
+    assert rows[1].observation == "out output"
+
+
+def test_repeated_tool_calls_with_placeholder_call_id_and_reused_index_create_distinct_rows(monkeypatch):
+    fake_session = _FakeDbSession()
+    monkeypatch.setattr(app_runner_module.db, "session", fake_session)
+    qm = _FakeQueueManager()
+    recorder = app_runner_module._AgentProcessRecorder(
+        dify_context=_dify_ctx(),
+        message_id="msg-1",
+        queue_manager=qm,  # type: ignore[arg-type]
+    )
+
+    for script, output in (("lookup find", "find output"), ("lookup out", "out output")):
+        recorder.handle_stream_event(
+            AgentBackendStreamInternalEvent(
+                run_id="run-1",
+                data={
+                    "event_kind": "function_tool_call",
+                    "index": 0,
+                    "part": {
+                        "part_kind": "tool-call",
+                        "tool_name": "shell_run",
+                        "tool_call_id": "None",
+                        "args": {"script": script},
+                    },
+                },
+            )
+        )
+        recorder.handle_stream_event(
+            AgentBackendStreamInternalEvent(
+                run_id="run-1",
+                data={
+                    "event_kind": "function_tool_result",
+                    "part": {
+                        "part_kind": "tool-return",
+                        "tool_name": "shell_run",
+                        "tool_call_id": "None",
+                        "content": output,
+                    },
+                },
+            )
+        )
+
+    rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
+    assert len(rows) == 2
+    assert rows[0].tool == "shell_run"
+    assert rows[0].tool_input == '{"script": "lookup find"}'
+    assert rows[0].observation == "find output"
+    assert rows[1].tool == "shell_run"
+    assert rows[1].tool_input == '{"script": "lookup out"}'
+    assert rows[1].observation == "out output"
+
+
 def test_prior_session_snapshot_is_threaded_into_request():
     prior = CompositorSessionSnapshot(layers=[])
     client = FakeAgentBackendRunClient()
@@ -1088,6 +1214,19 @@ def test_failed_run_raises_agent_backend_error():
     assert store.saved == []
 
 
+def test_agent_backend_failure_to_exception_maps_rate_limit_reason():
+    err = app_runner_module._agent_backend_failure_to_exception(
+        AgentBackendRunFailedInternalEvent(
+            run_id="run-1",
+            error="quota exceeded",
+            reason="InvokeRateLimitError",
+        )
+    )
+
+    assert isinstance(err, InvokeRateLimitError)
+    assert str(err) == "quota exceeded"
+
+
 def test_stopped_task_cancels_agent_backend_run_and_skips_session_save():
     client = _RecordingFakeAgentBackendRunClient()
     store = _FakeSessionStore()

api/tests/unit_tests/core/app/task_pipeline/test_based_generate_task_pipeline.py

@@ -3,10 +3,11 @@ from unittest.mock import Mock
 
 import pytest
 
+from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.queue_entities import QueueErrorEvent
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.errors.error import QuotaExceededError
-from graphon.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError, InvokeRateLimitError
 from models.enums import MessageStatus
 
 
@@ -68,6 +69,11 @@ class TestBasedGenerateTaskPipeline:
         assert error_response.task_id == "task-1"
         assert ping_response.task_id == "task-1"
 
+    def test_stream_converter_maps_invoke_rate_limit_error(self):
+        data = AppGenerateResponseConverter._error_to_stream_response(InvokeRateLimitError("quota exceeded"))
+
+        assert data == {"code": "rate_limit_error", "status": 429, "message": "quota exceeded"}
+
     def test_handle_output_moderation_when_flagged(self, pipeline):
         handler = Mock()
         handler.moderation_completion.return_value = ("filtered", True)

api/tests/unit_tests/core/plugin/entities/test_request.py

@@ -22,6 +22,7 @@ def test_request_download_file_accepts_tool_file_reference() -> None:
 
     assert payload.file.transfer_method == "tool_file"
     assert payload.file.reference == reference
+    assert payload.for_external is True
 
 
 def test_request_download_file_accepts_remote_url() -> None:
@@ -42,6 +43,25 @@ def test_request_download_file_accepts_remote_url() -> None:
     assert payload.file.url == "https://example.com/report.pdf"
 
 
+def test_request_download_file_accepts_internal_download_request() -> None:
+    reference = build_file_reference(record_id="tool-file-1")
+    payload = RequestRequestDownloadFile.model_validate(
+        {
+            "tenant_id": "tenant-1",
+            "user_id": "user-1",
+            "user_from": "account",
+            "invoke_from": "debugger",
+            "file": {
+                "transfer_method": "tool_file",
+                "reference": reference,
+            },
+            "for_external": False,
+        }
+    )
+
+    assert payload.for_external is False
+
+
 def test_request_download_file_rejects_remote_url_without_url() -> None:
     with pytest.raises(ValidationError, match="url is required"):
         _ = RequestRequestDownloadFile.model_validate(

api/tests/unit_tests/core/tools/test_tool_file_manager.py

@@ -60,6 +60,37 @@ def test_create_file_by_raw_stores_file_and_persists_record() -> None:
     session.refresh.assert_called_once_with(file_model)
 
 
+def test_create_file_by_raw_prefers_filename_extension_over_mimetype() -> None:
+    manager = ToolFileManager()
+    session = Mock()
+    session.refresh.side_effect = lambda model: setattr(model, "id", "tf-docx")
+
+    def tool_file_factory(**kwargs):
+        return SimpleNamespace(**kwargs)
+
+    with (
+        patch("core.tools.tool_file_manager.storage") as storage,
+        patch("core.tools.tool_file_manager.ToolFile", side_effect=tool_file_factory),
+        patch("core.tools.tool_file_manager.uuid4", return_value=SimpleNamespace(hex="abc")),
+        _patch_session_factory(session),
+    ):
+        file_model = manager.create_file_by_raw(
+            user_id="u1",
+            tenant_id="t1",
+            conversation_id="c1",
+            file_binary=b"docx",
+            mimetype="application/octet-stream",
+            filename="report.docx",
+        )
+
+    assert file_model.name == "report.docx"
+    assert file_model.file_key == "tools/t1/abc.docx"
+    storage.save.assert_called_once_with("tools/t1/abc.docx", b"docx")
+    session.add.assert_called_once_with(file_model)
+    session.commit.assert_called_once()
+    session.refresh.assert_called_once_with(file_model)
+
+
 def test_create_file_by_url_downloads_and_persists_record() -> None:
     manager = ToolFileManager()
     response = Mock()
@@ -88,6 +119,32 @@ def test_create_file_by_url_downloads_and_persists_record() -> None:
     session.refresh.assert_called_once_with(file_model)
 
 
+def test_create_file_by_url_prefers_url_extension_over_mimetype() -> None:
+    manager = ToolFileManager()
+    response = Mock()
+    response.content = b"docx"
+    response.headers = {"Content-Type": "application/octet-stream"}
+    response.raise_for_status.return_value = None
+    session = Mock()
+
+    def tool_file_factory(**kwargs):
+        return SimpleNamespace(**kwargs)
+
+    session.refresh.side_effect = lambda model: setattr(model, "id", "tf-docx")
+    with (
+        patch("core.tools.tool_file_manager.storage") as storage,
+        patch("core.tools.tool_file_manager.ToolFile", side_effect=tool_file_factory),
+        patch("core.tools.tool_file_manager.uuid4", return_value=SimpleNamespace(hex="urlabc")),
+        _patch_session_factory(session),
+        patch("core.tools.tool_file_manager.remote_fetcher.make_request", return_value=response),
+    ):
+        file_model = manager.create_file_by_url("u1", "t1", "https://example.com/report.docx?download=1", "c1")
+
+    assert file_model.file_key == "tools/t1/urlabc.docx"
+    assert file_model.name == "urlabc.docx"
+    storage.save.assert_called_once_with("tools/t1/urlabc.docx", b"docx")
+
+
 def test_create_file_by_url_raises_on_timeout() -> None:
     manager = ToolFileManager()
 

api/tests/unit_tests/core/tools/utils/test_message_transformer.py

@@ -7,9 +7,10 @@ from core.tools.entities.tool_entities import ToolInvokeMessage
 
 
 class _FakeToolFile:
-    def __init__(self, mimetype: str):
+    def __init__(self, mimetype: str, name: str | None):
         self.id = "fake-tool-file-id"
         self.mimetype = mimetype
+        self.name = name or "fake-tool-file.bin"
 
 
 class _FakeToolFileManager:
@@ -38,7 +39,7 @@ class _FakeToolFileManager:
             "mimetype": mimetype,
             "filename": filename,
         }
-        return _FakeToolFile(mimetype)
+        return _FakeToolFile(mimetype, filename)
 
 
 @pytest.fixture(autouse=True)
@@ -89,6 +90,29 @@ def test_transform_tool_invoke_messages_mimetype_key_present_but_none():
     assert o.meta["tool_file_id"] == "fake-tool-file-id"
 
 
+def test_transform_tool_invoke_messages_prefers_filename_extension_over_mimetype():
+    msg = ToolInvokeMessage(
+        type=ToolInvokeMessage.MessageType.BLOB,
+        message=ToolInvokeMessage.BlobMessage(blob=b"docx"),
+        meta={"mime_type": "application/octet-stream", "filename": "report.docx"},
+    )
+
+    out = list(
+        mt.ToolFileMessageTransformer.transform_tool_invoke_messages(
+            messages=_gen([msg]),
+            user_id="u1",
+            tenant_id="t1",
+            conversation_id="c1",
+        )
+    )
+
+    assert _FakeToolFileManager.last_call is not None
+    assert _FakeToolFileManager.last_call["filename"] == "report.docx"
+    assert len(out) == 1
+    assert isinstance(out[0].message, ToolInvokeMessage.TextMessage)
+    assert out[0].message.text.endswith(".docx")
+
+
 def test_transform_tool_invoke_messages_parses_existing_tool_file_link_meta():
     msg = ToolInvokeMessage(
         type=ToolInvokeMessage.MessageType.IMAGE_LINK,

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_agent_node.py

@@ -1,10 +1,12 @@
+from datetime import UTC, datetime
 from types import SimpleNamespace
 from typing import cast
 from unittest.mock import MagicMock, patch
 
 from agenton.compositor import CompositorSessionSnapshot
 from dify_agent.layers.ask_human import AskHumanToolResult
-from dify_agent.protocol import RunStartedEvent, RunSucceededEvent, RunSucceededEventData
+from dify_agent.protocol import PydanticAIStreamRunEvent, RunStartedEvent, RunSucceededEvent, RunSucceededEventData
+from pydantic_ai.messages import PartDeltaEvent, TextPartDelta
 
 from clients.agent_backend import (
     AgentBackendRunEventAdapter,
@@ -190,6 +192,30 @@ class FileOutputBackendClient(FakeAgentBackendRunClient):
         )
 
 
+class AgentMessageDeltaBackendClient(FakeAgentBackendRunClient):
+    def _events(self, run_id: str):
+        created_at = datetime(2026, 1, 1, tzinfo=UTC)
+        return (
+            RunStartedEvent(id="1-0", run_id=run_id, created_at=created_at),
+            PydanticAIStreamRunEvent(
+                id="2-0",
+                run_id=run_id,
+                created_at=created_at,
+                data=PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="hello ")),
+                agent_message_delta="hello ",
+            ),
+            RunSucceededEvent(
+                id="3-0",
+                run_id=run_id,
+                created_at=created_at,
+                data=RunSucceededEventData(
+                    output={"text": "hello agent"},
+                    session_snapshot=CompositorSessionSnapshot(layers=[]),
+                ),
+            ),
+        )
+
+
 def _node(
     *,
     scenario: FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCESS,
@@ -277,6 +303,19 @@ def test_agent_node_run_maps_successful_agent_backend_run_to_node_result():
     assert layers["llm"]["config"]["credentials"] == "[REDACTED]"
 
 
+def test_agent_node_run_ignores_agent_message_delta_until_terminal_result():
+    events = list(_node(agent_backend_client=AgentMessageDeltaBackendClient())._run())
+
+    assert len(events) == 1
+    result = cast(StreamCompletedEvent, events[0]).node_run_result
+    assert result.status == WorkflowNodeExecutionStatus.SUCCEEDED
+    assert result.outputs == {"text": "hello agent"}
+    agent_backend = result.metadata[WorkflowNodeExecutionMetadataKey.AGENT_LOG]["agent_backend"]
+    assert agent_backend["status"] == "succeeded"
+    assert agent_backend["agent_message_delta_count"] == 1
+    assert agent_backend["agent_message_delta_length"] == len("hello ")
+
+
 def test_agent_node_run_normalizes_declared_file_output_with_canonical_mapping():
     tool_reference = build_file_reference(record_id="tool-file-1")
     with patch(

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_output_file_rebacker.py

@@ -57,6 +57,17 @@ def test_reback_resolves_tenant_tool_file_to_file():
     assert file.extension == ".png"
 
 
+def test_reback_prefers_filename_extension_over_mimetype():
+    tf = _seed(mimetype="application/octet-stream", name="report.docx", size=99)
+    file = reback_tool_file_output(tenant_id=TENANT, tool_file_id=tf)
+
+    assert file is not None
+    assert file.filename == "report.docx"
+    assert file.mime_type == "application/octet-stream"
+    assert file.extension == ".docx"
+    assert file.type == FileType.CUSTOM
+
+
 def test_reback_other_tenant_returns_none():
     tf = _seed()
     assert reback_tool_file_output(tenant_id="33333333-3333-3333-3333-333333333333", tool_file_id=tf) is None

api/tests/unit_tests/factories/test_build_from_mapping.py

@@ -165,6 +165,20 @@ def test_build_from_mapping_accepts_opaque_related_id_for_tool_file(mock_tool_fi
     assert file.storage_key == "tool_file.pdf"
 
 
+def test_build_from_mapping_prefers_tool_filename_extension_over_mimetype(mock_tool_file):
+    mock_tool_file.name = "report.docx"
+    mock_tool_file.file_key = "tools/test_tenant_id/file.bin"
+    mock_tool_file.mimetype = "application/octet-stream"
+    mapping = tool_file_mapping(file_type="document")
+
+    file = build_from_mapping(mapping=mapping, tenant_id=TEST_TENANT_ID)
+
+    assert file.extension == ".docx"
+    assert file.filename == "report.docx"
+    assert file.mime_type == "application/octet-stream"
+    assert file.storage_key == "tools/test_tenant_id/file.bin"
+
+
 @pytest.mark.parametrize(
     ("file_type", "should_pass", "expected_error"),
     [
@@ -213,6 +227,25 @@ def test_build_from_remote_url(mock_http_head):
     assert file.size == 2048
 
 
+def test_build_from_remote_url_prefers_filename_extension_over_mimetype():
+    mapping = {
+        "transfer_method": "remote_url",
+        "url": TEST_REMOTE_URL,
+        "type": "document",
+    }
+
+    with patch(
+        "factories.file_factory.builders.get_remote_file_info",
+        return_value=("application/octet-stream", "report.docx", 99),
+    ):
+        file = build_from_mapping(mapping=mapping, tenant_id=TEST_TENANT_ID)
+
+    assert file.filename == "report.docx"
+    assert file.extension == ".docx"
+    assert file.mime_type == "application/octet-stream"
+    assert file.size == 99
+
+
 @pytest.mark.parametrize(
     ("file_type", "should_pass", "expected_error"),
     [

api/tests/unit_tests/fields/test_message_fields.py

@@ -1,4 +1,6 @@
-from fields.message_fields import ExploreMessageListItem, MessageListItem
+from decimal import Decimal
+
+from fields.message_fields import ExploreMessageListItem, MessageListItem, WebMessageListItem
 
 
 def _base_kwargs():
@@ -34,3 +36,33 @@ class TestExploreMessageListItem:
         # Guard the public service-API contract: the base item must not leak metadata.
         payload = MessageListItem(**_base_kwargs()).model_dump(mode="json")
         assert "metadata" not in payload
+
+    def test_message_list_item_exposes_usage_fields(self):
+        payload = MessageListItem(
+            **_base_kwargs(),
+            message_tokens=7,
+            answer_tokens=11,
+            provider_response_latency=1.25,
+            total_price=Decimal("0.0001234"),
+            currency="USD",
+        ).model_dump(mode="json")
+
+        assert payload["message_tokens"] == 7
+        assert payload["answer_tokens"] == 11
+        assert payload["total_tokens"] == 18
+        assert payload["provider_response_latency"] == 1.25
+        assert payload["total_price"] == "0.0001234"
+        assert payload["currency"] == "USD"
+
+    def test_web_message_list_item_exposes_usage_and_metadata(self):
+        payload = WebMessageListItem(
+            **_base_kwargs(),
+            metadata={"usage": {"total_tokens": 18}},
+            message_tokens=7,
+            answer_tokens=11,
+        ).model_dump(mode="json")
+
+        assert payload["metadata"] == {"usage": {"total_tokens": 18}}
+        assert payload["message_tokens"] == 7
+        assert payload["answer_tokens"] == 11
+        assert payload["total_tokens"] == 18

api/tests/unit_tests/libs/test_archive_storage.py

@@ -247,32 +247,6 @@ def test_generate_presigned_url(monkeypatch: pytest.MonkeyPatch):
     assert url == "http://signed-url"
 
 
-def test_generate_presigned_url_with_download_headers(monkeypatch: pytest.MonkeyPatch):
-    _configure_storage(monkeypatch)
-    client, _ = _mock_client(monkeypatch)
-    client.generate_presigned_url.return_value = "http://signed-url"
-    storage = ArchiveStorage(bucket=BUCKET_NAME)
-
-    url = storage.generate_presigned_url(
-        "key",
-        expires_in=123,
-        filename="workflow-run-logs-2025-03.zip",
-        content_type="application/zip",
-    )
-
-    client.generate_presigned_url.assert_called_once_with(
-        ClientMethod="get_object",
-        Params={
-            "Bucket": "archive-bucket",
-            "Key": "key",
-            "ResponseContentDisposition": "attachment; filename*=UTF-8''workflow-run-logs-2025-03.zip",
-            "ResponseContentType": "application/zip",
-        },
-        ExpiresIn=123,
-    )
-    assert url == "http://signed-url"
-
-
 def test_generate_presigned_url_error(monkeypatch: pytest.MonkeyPatch):
     _configure_storage(monkeypatch)
     client, _ = _mock_client(monkeypatch)

api/tests/unit_tests/libs/test_token.py

@@ -3,7 +3,6 @@ from unittest.mock import MagicMock
 
 import pytest
 from flask import Request
-from werkzeug.exceptions import Unauthorized
 from werkzeug.wrappers import Response
 
 from constants import COOKIE_NAME_ACCESS_TOKEN, COOKIE_NAME_WEBAPP_ACCESS_TOKEN
@@ -12,17 +11,10 @@ from libs.token import extract_access_token, extract_webapp_access_token, set_cs
 
 
 class MockRequest:
-    def __init__(
-        self,
-        headers: dict[str, str],
-        cookies: dict[str, str],
-        args: dict[str, str],
-        path: str = "/console/api/test",
-    ):
+    def __init__(self, headers: dict[str, str], cookies: dict[str, str], args: dict[str, str]):
         self.headers: dict[str, str] = headers
         self.cookies: dict[str, str] = cookies
         self.args: dict[str, str] = args
-        self.path = path
 
 
 def test_extract_access_token():
@@ -71,24 +63,3 @@ def test_set_csrf_cookie_includes_domain_when_configured(monkeypatch: pytest.Mon
     assert any("csrf_token=abc123" in c for c in cookies)
     assert any("Domain=example.com" in c for c in cookies)
     assert all("__Host-" not in c for c in cookies)
-
-
-def test_workflow_run_archive_download_file_bypasses_csrf():
-    request = cast(
-        Request,
-        MockRequest(
-            headers={},
-            cookies={},
-            args={},
-            path="/console/api/workflow-run-archives/downloads/5923ce20291444af45f0580fb49f1cc9/file",
-        ),
-    )
-
-    token.check_csrf_token(request, "account-1")
-
-
-def test_non_whitelisted_path_requires_csrf():
-    request = cast(Request, MockRequest(headers={}, cookies={}, args={}, path="/console/api/test"))
-
-    with pytest.raises(Unauthorized):
-        token.check_csrf_token(request, "account-1")

api/tests/unit_tests/services/retention/workflow_run/test_archive_bundle_index.py

@@ -1,197 +0,0 @@
-import datetime
-import json
-from collections.abc import Iterator
-from typing import cast
-from unittest.mock import MagicMock
-
-from services.retention.workflow_run.archive_bundle_index import (
-    ARCHIVE_BUNDLE_ROOT_PREFIX,
-    ArchiveBundleManifest,
-    WorkflowRunArchiveBundleIndexBackfill,
-    calculate_archive_bundle_index_values,
-    decode_archive_bundle_manifest,
-    upsert_archive_bundle_index_from_manifest,
-)
-from services.retention.workflow_run.constants import ARCHIVE_BUNDLE_FORMAT, ARCHIVE_BUNDLE_SCHEMA_VERSION
-
-TENANT_ID = "1251fe32-c0c7-4fe2-a7bd-a8105267faf5"
-BUNDLE_ID = "bundle-a"
-OBJECT_PREFIX = (
-    f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix=1/tenant_id={TENANT_ID}/"
-    f"year=2025/month=03/shard=00-of-01/bundle={BUNDLE_ID}"
-)
-MANIFEST_KEY = f"{OBJECT_PREFIX}/manifest.json"
-
-
-class FakeArchiveStorage:
-    listed_prefixes: list[str]
-    objects: dict[str, bytes]
-
-    def __init__(self, objects: dict[str, bytes]) -> None:
-        self.objects = objects
-        self.listed_prefixes = []
-
-    def list_objects(self, prefix: str) -> list[str]:
-        self.listed_prefixes.append(prefix)
-        return sorted(key for key in self.objects if key.startswith(prefix))
-
-    def get_object(self, key: str) -> bytes:
-        return self.objects[key]
-
-
-class FakeSessionContext:
-    session: MagicMock
-
-    def __init__(self, session: MagicMock) -> None:
-        self.session = session
-
-    def __enter__(self) -> MagicMock:
-        return self.session
-
-    def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
-        return None
-
-
-class FakeSessionFactory:
-    session: MagicMock
-
-    def __init__(self, session: MagicMock) -> None:
-        self.session = session
-
-    def __call__(self) -> FakeSessionContext:
-        return FakeSessionContext(self.session)
-
-
-class FailingSessionFactory:
-    def __call__(self) -> Iterator[MagicMock]:
-        raise AssertionError("dry-run should not open a database session")
-
-
-def _manifest(*, object_prefix: str = OBJECT_PREFIX, month: int = 3) -> ArchiveBundleManifest:
-    return ArchiveBundleManifest(
-        schema_version=ARCHIVE_BUNDLE_SCHEMA_VERSION,
-        archive_format=ARCHIVE_BUNDLE_FORMAT,
-        tenant_id=TENANT_ID,
-        tenant_prefix="1",
-        year=2025,
-        month=month,
-        shard="00-of-01",
-        bundle_id=BUNDLE_ID,
-        object_prefix=object_prefix,
-        workflow_run_count=2,
-        workflow_node_execution_count=3,
-        min_created_at="2025-03-01T00:00:00+00:00",
-        max_created_at="2025-03-02T00:00:00+00:00",
-        min_run_id="run-a",
-        max_run_id="run-b",
-        archived_at="2026-06-25T08:00:00+00:00",
-        tables={
-            "workflow_runs": {
-                "row_count": 2,
-                "checksum": "checksum-a",
-                "size_bytes": 100,
-                "object_key": f"{object_prefix}/workflow_runs.parquet",
-            },
-            "workflow_node_executions": {
-                "row_count": 3,
-                "checksum": "checksum-b",
-                "size_bytes": 200,
-                "object_key": f"{object_prefix}/workflow_node_executions.parquet",
-            },
-        },
-        run_ids=["run-a", "run-b"],
-    )
-
-
-def _manifest_bytes(manifest: ArchiveBundleManifest | None = None) -> bytes:
-    return json.dumps(manifest or _manifest()).encode("utf-8")
-
-
-def test_decode_and_calculate_archive_bundle_index_values() -> None:
-    data = _manifest_bytes()
-
-    manifest = decode_archive_bundle_manifest(data)
-    values = calculate_archive_bundle_index_values(manifest, len(data))
-
-    assert manifest["tenant_id"] == TENANT_ID
-    assert values.row_count == 5
-    assert values.archive_bytes == len(data) + 300
-    assert values.archived_at == datetime.datetime(2026, 6, 25, 8, 0)
-
-
-def test_upsert_archive_bundle_index_inserts_new_bundle() -> None:
-    session = MagicMock()
-    session.scalar.return_value = None
-    data = _manifest_bytes()
-
-    bundle = upsert_archive_bundle_index_from_manifest(session, decode_archive_bundle_manifest(data), len(data))
-
-    assert bundle.tenant_id == TENANT_ID
-    assert bundle.year == 2025
-    assert bundle.month == 3
-    assert bundle.workflow_run_count == 2
-    assert bundle.row_count == 5
-    assert bundle.archive_bytes == len(data) + 300
-    session.add.assert_called_once_with(bundle)
-
-
-def test_upsert_archive_bundle_index_updates_existing_bundle() -> None:
-    existing = MagicMock()
-    session = MagicMock()
-    session.scalar.return_value = existing
-    data = _manifest_bytes()
-
-    bundle = upsert_archive_bundle_index_from_manifest(session, decode_archive_bundle_manifest(data), len(data))
-
-    assert bundle is existing
-    assert existing.workflow_run_count == 2
-    assert existing.row_count == 5
-    assert existing.archive_bytes == len(data) + 300
-    assert existing.archived_at == datetime.datetime(2026, 6, 25, 8, 0)
-    session.add.assert_not_called()
-
-
-def test_backfill_lists_tenant_month_prefix_and_upserts_bundle_index() -> None:
-    storage = FakeArchiveStorage({MANIFEST_KEY: _manifest_bytes()})
-    session = MagicMock()
-    session.scalar.return_value = None
-    backfill = WorkflowRunArchiveBundleIndexBackfill(
-        storage=cast(MagicMock, storage),
-        session_factory=cast(MagicMock, FakeSessionFactory(session)),
-    )
-
-    summary = backfill.run(tenant_ids=[TENANT_ID], year=2025, month=3)
-
-    assert storage.listed_prefixes == [
-        f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix=1/tenant_id={TENANT_ID}/year=2025/month=03/"
-    ]
-    assert summary.manifests_found == 1
-    assert summary.bundles_processed == 1
-    assert summary.bundles_upserted == 1
-    assert summary.bundles_failed == 0
-    session.add.assert_called_once()
-    session.commit.assert_called_once()
-
-
-def test_backfill_dry_run_filters_by_year_month_without_database_write() -> None:
-    other_month_prefix = OBJECT_PREFIX.replace("month=03", "month=04")
-    storage = FakeArchiveStorage(
-        {
-            MANIFEST_KEY: _manifest_bytes(),
-            f"{other_month_prefix}/manifest.json": _manifest_bytes(
-                _manifest(object_prefix=other_month_prefix, month=4)
-            ),
-        }
-    )
-    backfill = WorkflowRunArchiveBundleIndexBackfill(
-        storage=cast(MagicMock, storage),
-        session_factory=cast(MagicMock, FailingSessionFactory()),
-    )
-
-    summary = backfill.run(tenant_prefixes=["1"], year=2025, month=3, dry_run=True)
-
-    assert storage.listed_prefixes == [f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix=1/"]
-    assert summary.manifests_found == 1
-    assert summary.bundles_processed == 1
-    assert summary.bundles_upserted == 0
-    assert summary.archive_bytes > 0

api/tests/unit_tests/services/retention/workflow_run/test_archive_download_preparation.py

@@ -1,273 +0,0 @@
-import datetime
-import hashlib
-import io
-import json
-import zipfile
-from types import SimpleNamespace
-from typing import cast
-
-import pyarrow as pa
-import pyarrow.parquet as pq
-from sqlalchemy.orm import Session, sessionmaker
-
-from libs.archive_storage import ArchiveStorage
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.archive_bundle_index import ARCHIVE_BUNDLE_ROOT_PREFIX, ArchiveBundleManifest
-from services.retention.workflow_run.archive_download_preparation import (
-    WorkflowRunArchiveDownloadPreparer,
-    build_archive_download_storage_key,
-)
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-    build_pending_archive_download_task,
-)
-from services.retention.workflow_run.constants import ARCHIVE_BUNDLE_FORMAT, ARCHIVE_BUNDLE_SCHEMA_VERSION
-
-TENANT_ID = "1251fe32-c0c7-4fe2-a7bd-a8105267faf5"
-BUNDLE_ID = "bundle-a"
-SHARD = "00-of-01"
-OBJECT_PREFIX = (
-    f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix=1/tenant_id={TENANT_ID}/"
-    f"year=2025/month=03/shard={SHARD}/bundle={BUNDLE_ID}"
-)
-MANIFEST_KEY = f"{OBJECT_PREFIX}/manifest.json"
-
-
-class FakeArchiveStorage:
-    objects: dict[str, bytes]
-    put_objects: dict[str, bytes]
-
-    def __init__(self, objects: dict[str, bytes]) -> None:
-        self.objects = dict(objects)
-        self.put_objects = {}
-
-    def get_object(self, key: str) -> bytes:
-        return self.objects[key]
-
-    def put_object(self, key: str, data: bytes) -> str:
-        self.put_objects[key] = data
-        return hashlib.md5(data).hexdigest()
-
-
-class FakeTaskCache:
-    task: WorkflowRunArchiveDownloadTask | None
-    saved_tasks: list[WorkflowRunArchiveDownloadTask]
-
-    def __init__(self, task: WorkflowRunArchiveDownloadTask | None) -> None:
-        self.task = task
-        self.saved_tasks = []
-
-    def get(self, *, tenant_id: str, download_id: str) -> WorkflowRunArchiveDownloadTask | None:
-        if self.task and self.task.tenant_id == tenant_id and self.task.download_id == download_id:
-            return self.task
-        return None
-
-    def save(self, task: WorkflowRunArchiveDownloadTask) -> None:
-        self.task = task
-        self.saved_tasks.append(task)
-
-
-class FakeSessionContext:
-    bundles: list[WorkflowRunArchiveBundle]
-
-    def __init__(self, bundles: list[WorkflowRunArchiveBundle]) -> None:
-        self.bundles = bundles
-
-    def __enter__(self) -> "FakeSessionContext":
-        return self
-
-    def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
-        return None
-
-    def scalars(self, stmt: object) -> list[WorkflowRunArchiveBundle]:
-        return self.bundles
-
-
-class FakeSessionFactory:
-    bundles: list[WorkflowRunArchiveBundle]
-
-    def __init__(self, bundles: list[WorkflowRunArchiveBundle]) -> None:
-        self.bundles = bundles
-
-    def __call__(self) -> FakeSessionContext:
-        return FakeSessionContext(self.bundles)
-
-
-def _object_prefix(bundle_id: str = BUNDLE_ID) -> str:
-    return (
-        f"{ARCHIVE_BUNDLE_ROOT_PREFIX}tenant_prefix=1/tenant_id={TENANT_ID}/"
-        f"year=2025/month=03/shard={SHARD}/bundle={bundle_id}"
-    )
-
-
-def _bundle(bundle_id: str = BUNDLE_ID) -> WorkflowRunArchiveBundle:
-    return cast(WorkflowRunArchiveBundle, SimpleNamespace(shard=SHARD, bundle_id=bundle_id))
-
-
-def _task(bundle_refs: list[tuple[str, str]] | None = None) -> WorkflowRunArchiveDownloadTask:
-    refs = bundle_refs or [(SHARD, BUNDLE_ID)]
-    return build_pending_archive_download_task(
-        tenant_id=TENANT_ID,
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=[bundle_id for _, bundle_id in refs],
-        bundle_refs=refs,
-        archive_bytes=1024,
-        download_id=build_archive_download_id(
-            tenant_id=TENANT_ID,
-            year=2025,
-            month=3,
-            bundle_refs=refs,
-        ),
-        now=datetime.datetime(2026, 6, 25, 8, 0, tzinfo=datetime.UTC),
-    )
-
-
-def _manifest_bytes(table_payloads: dict[str, bytes], *, bundle_id: str = BUNDLE_ID) -> bytes:
-    object_prefix = _object_prefix(bundle_id)
-    manifest = ArchiveBundleManifest(
-        schema_version=ARCHIVE_BUNDLE_SCHEMA_VERSION,
-        archive_format=ARCHIVE_BUNDLE_FORMAT,
-        tenant_id=TENANT_ID,
-        tenant_prefix="1",
-        year=2025,
-        month=3,
-        shard=SHARD,
-        bundle_id=bundle_id,
-        object_prefix=object_prefix,
-        workflow_run_count=2,
-        workflow_node_execution_count=0,
-        min_created_at="2025-03-01T00:00:00+00:00",
-        max_created_at="2025-03-02T00:00:00+00:00",
-        min_run_id="run-a",
-        max_run_id="run-b",
-        archived_at="2026-06-25T08:00:00+00:00",
-        tables={
-            table_name: {
-                "row_count": 1,
-                "checksum": hashlib.md5(payload).hexdigest(),
-                "size_bytes": len(payload),
-                "object_key": f"{object_prefix}/{table_name}.parquet",
-            }
-            for table_name, payload in table_payloads.items()
-        },
-        run_ids=["run-a", "run-b"],
-    )
-    return json.dumps(manifest).encode("utf-8")
-
-
-def _preparer(
-    *,
-    task: WorkflowRunArchiveDownloadTask,
-    storage: FakeArchiveStorage | None = None,
-    archive_storage: FakeArchiveStorage | None = None,
-    download_storage: FakeArchiveStorage | None = None,
-    cache: FakeTaskCache,
-    bundles: list[WorkflowRunArchiveBundle] | None = None,
-) -> WorkflowRunArchiveDownloadPreparer:
-    source_storage = archive_storage or storage
-    target_storage = download_storage or storage
-    assert source_storage is not None
-    assert target_storage is not None
-    return WorkflowRunArchiveDownloadPreparer(
-        archive_storage=cast(ArchiveStorage, source_storage),
-        download_storage=cast(ArchiveStorage, target_storage),
-        cache=cast(WorkflowRunArchiveDownloadTaskCache, cache),
-        session_factory=cast(sessionmaker[Session], FakeSessionFactory(bundles or [_bundle()])),
-    )
-
-
-def _parquet_bytes(records: list[dict[str, object]]) -> bytes:
-    buffer = io.BytesIO()
-    pq.write_table(pa.Table.from_pylist(records), buffer)
-    return buffer.getvalue()
-
-
-def test_prepare_workflow_run_archive_download_builds_csv_zip_and_marks_ready() -> None:
-    bundle_refs = [(SHARD, "bundle-a"), (SHARD, "bundle-b")]
-    task = _task(bundle_refs)
-    first_bundle_payloads = {
-        "workflow_app_logs": _parquet_bytes([{"id": "log-a", "workflow_run_id": "run-a"}]),
-        "workflow_runs": _parquet_bytes([{"id": "run-a", "status": "succeeded"}]),
-    }
-    second_bundle_payloads = {
-        "workflow_app_logs": _parquet_bytes([{"id": "log-b", "workflow_run_id": "run-b"}]),
-        "workflow_runs": _parquet_bytes([{"id": "run-b", "status": "failed"}]),
-    }
-    archive_storage = FakeArchiveStorage(
-        {
-            f"{_object_prefix('bundle-a')}/manifest.json": _manifest_bytes(
-                first_bundle_payloads,
-                bundle_id="bundle-a",
-            ),
-            **{
-                f"{_object_prefix('bundle-a')}/{table}.parquet": payload
-                for table, payload in first_bundle_payloads.items()
-            },
-            f"{_object_prefix('bundle-b')}/manifest.json": _manifest_bytes(
-                second_bundle_payloads,
-                bundle_id="bundle-b",
-            ),
-            **{
-                f"{_object_prefix('bundle-b')}/{table}.parquet": payload
-                for table, payload in second_bundle_payloads.items()
-            },
-        }
-    )
-    download_storage = FakeArchiveStorage({})
-    cache = FakeTaskCache(task)
-    preparer = _preparer(
-        task=task,
-        archive_storage=archive_storage,
-        download_storage=download_storage,
-        cache=cache,
-        bundles=[_bundle("bundle-a"), _bundle("bundle-b")],
-    )
-
-    result = preparer.prepare(tenant_id=TENANT_ID, download_id=task.download_id)
-
-    assert result is not None
-    assert result.status == WorkflowRunArchiveDownloadStatus.READY
-    assert result.storage_key == build_archive_download_storage_key(task)
-    assert result.file_name == "workflow-run-logs-2025-03.zip"
-    assert cache.saved_tasks[0].status == WorkflowRunArchiveDownloadStatus.PROCESSING
-    assert cache.saved_tasks[-1].status == WorkflowRunArchiveDownloadStatus.READY
-    assert archive_storage.put_objects == {}
-
-    archive_payload = download_storage.put_objects[result.storage_key]
-    with zipfile.ZipFile(io.BytesIO(archive_payload)) as archive:
-        names = set(archive.namelist())
-        assert names == {
-            "workflow-run-logs-2025-03/workflow_app_logs.csv",
-            "workflow-run-logs-2025-03/workflow_runs.csv",
-        }
-        workflow_runs_csv = archive.read("workflow-run-logs-2025-03/workflow_runs.csv").decode("utf-8")
-        assert workflow_runs_csv.count('"id","status"') == 1
-        assert '"run-a","succeeded"' in workflow_runs_csv
-        assert '"run-b","failed"' in workflow_runs_csv
-
-
-def test_prepare_workflow_run_archive_download_marks_failed_on_checksum_mismatch() -> None:
-    task = _task()
-    table_payloads = {"workflow_runs": _parquet_bytes([{"id": "run-a", "status": "succeeded"}])}
-    manifest_data = json.loads(_manifest_bytes(table_payloads).decode("utf-8"))
-    manifest_data["tables"]["workflow_runs"]["checksum"] = "bad-checksum"
-    storage = FakeArchiveStorage(
-        {
-            MANIFEST_KEY: json.dumps(manifest_data).encode("utf-8"),
-            f"{OBJECT_PREFIX}/workflow_runs.parquet": table_payloads["workflow_runs"],
-        }
-    )
-    cache = FakeTaskCache(task)
-    preparer = _preparer(task=task, storage=storage, cache=cache)
-
-    result = preparer.prepare(tenant_id=TENANT_ID, download_id=task.download_id)
-
-    assert result is not None
-    assert result.status == WorkflowRunArchiveDownloadStatus.FAILED
-    assert "checksum mismatch" in (result.error or "")
-    assert storage.put_objects == {}

api/tests/unit_tests/services/retention/workflow_run/test_archive_download_task_cache.py

@@ -1,196 +0,0 @@
-import datetime
-
-import pytest
-
-from services.retention.workflow_run.archive_download_task_cache import (
-    ARCHIVE_DOWNLOAD_FORMAT_VERSION,
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-    build_pending_archive_download_task,
-)
-
-
-class FakeRedis:
-    store: dict[str, tuple[int | datetime.timedelta, str]]
-
-    def __init__(self) -> None:
-        self.store = {}
-
-    def get(self, name: str | bytes) -> bytes | str | None:
-        key = name.decode("utf-8") if isinstance(name, bytes) else name
-        item = self.store.get(key)
-        return item[1] if item else None
-
-    def setex(self, name: str | bytes, time: int | datetime.timedelta, value: str) -> object:
-        key = name.decode("utf-8") if isinstance(name, bytes) else name
-        self.store[key] = (time, value)
-        return True
-
-    def set(
-        self,
-        name: str | bytes,
-        value: str,
-        ex: int | None = None,
-        nx: bool = False,
-    ) -> object:
-        key = name.decode("utf-8") if isinstance(name, bytes) else name
-        if nx and key in self.store:
-            return None
-        self.store[key] = (ex or 0, value)
-        return True
-
-    def delete(self, *names: str | bytes) -> object:
-        deleted = 0
-        for name in names:
-            key = name.decode("utf-8") if isinstance(name, bytes) else name
-            if self.store.pop(key, None) is not None:
-                deleted += 1
-        return deleted
-
-
-def test_build_pending_archive_download_task_sets_ephemeral_payload() -> None:
-    now = datetime.datetime(2026, 6, 25, 8, 0, tzinfo=datetime.UTC)
-
-    task = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a", "bundle-b"],
-        archive_bytes=1024,
-        ttl_seconds=3600,
-        download_id="download-1",
-        now=now,
-    )
-
-    assert task.status == WorkflowRunArchiveDownloadStatus.PENDING
-    assert task.bundle_count == 2
-    assert task.expires_at == now + datetime.timedelta(seconds=3600)
-
-
-def test_build_archive_download_id_is_stable_for_same_bundle_set() -> None:
-    first = build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("01-of-02", "bundle-b"), ("00-of-02", "bundle-a")],
-    )
-    second = build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("00-of-02", "bundle-a"), ("01-of-02", "bundle-b")],
-    )
-
-    assert first == second
-    assert len(first) == 32
-
-
-def test_build_archive_download_id_changes_when_content_or_format_changes() -> None:
-    base = build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("00-of-01", "bundle-a")],
-    )
-    changed_bundle = build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("00-of-01", "bundle-b")],
-    )
-    changed_format = build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("00-of-01", "bundle-a")],
-        download_format_version=f"{ARCHIVE_DOWNLOAD_FORMAT_VERSION}-next",
-    )
-
-    assert base != changed_bundle
-    assert base != changed_format
-
-
-def test_build_archive_download_id_rejects_empty_bundle_refs() -> None:
-    with pytest.raises(ValueError, match="bundle_refs must not be empty"):
-        build_archive_download_id(tenant_id="tenant-1", year=2025, month=3, bundle_refs=[])
-
-
-def test_archive_download_task_cache_round_trips_with_ttl() -> None:
-    redis = FakeRedis()
-    cache = WorkflowRunArchiveDownloadTaskCache(redis)
-    task = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a"],
-        archive_bytes=1024,
-        ttl_seconds=3600,
-        download_id="download-1",
-    )
-
-    cache.save(task)
-    restored = cache.get(tenant_id="tenant-1", download_id="download-1")
-
-    assert restored == task
-    ttl, _ = redis.store["workflow_run_archive_download:tenant-1:download-1"]
-    assert isinstance(ttl, int)
-    assert 0 < ttl <= 3600
-
-
-def test_archive_download_task_cache_create_if_absent_is_idempotent() -> None:
-    redis = FakeRedis()
-    cache = WorkflowRunArchiveDownloadTaskCache(redis)
-    first = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a"],
-        archive_bytes=1024,
-        ttl_seconds=3600,
-        download_id="download-1",
-    )
-    second = first.model_copy(update={"archive_bytes": 2048})
-
-    assert cache.create_if_absent(first) is True
-    assert cache.create_if_absent(second) is False
-
-    restored = cache.get(tenant_id="tenant-1", download_id="download-1")
-    assert restored == first
-
-
-def test_archive_download_task_cache_delete_removes_entry() -> None:
-    redis = FakeRedis()
-    cache = WorkflowRunArchiveDownloadTaskCache(redis)
-    task = WorkflowRunArchiveDownloadTask(
-        download_id="download-1",
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=[],
-        bundle_count=0,
-        archive_bytes=0,
-        status=WorkflowRunArchiveDownloadStatus.FAILED,
-        error="failed",
-        created_at=datetime.datetime.now(datetime.UTC),
-        updated_at=datetime.datetime.now(datetime.UTC),
-        expires_at=datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=3600),
-    )
-    cache.save(task)
-
-    cache.delete(tenant_id="tenant-1", download_id="download-1")
-
-    assert cache.get(tenant_id="tenant-1", download_id="download-1") is None
-
-
-def test_archive_download_task_cache_ignores_malformed_json() -> None:
-    redis = FakeRedis()
-    cache = WorkflowRunArchiveDownloadTaskCache(redis)
-    redis.setex("workflow_run_archive_download:tenant-1:download-1", 3600, "{")
-
-    assert cache.get(tenant_id="tenant-1", download_id="download-1") is None

api/tests/unit_tests/services/retention/workflow_run/test_archive_log_service.py

@@ -1,326 +0,0 @@
-import datetime
-from types import SimpleNamespace
-from typing import cast
-from unittest.mock import MagicMock
-
-import pytest
-
-from models.workflow import WorkflowRunArchiveBundle
-from services.retention.workflow_run.archive_download_task_cache import (
-    WorkflowRunArchiveDownloadStatus,
-    WorkflowRunArchiveDownloadTask,
-    WorkflowRunArchiveDownloadTaskCache,
-    build_archive_download_id,
-    build_pending_archive_download_task,
-)
-from services.retention.workflow_run.archive_log_service import (
-    ArchiveDownloadTaskDispatcher,
-    WorkflowRunArchiveDownloadNotReadyError,
-    WorkflowRunArchiveNotFoundError,
-    create_workflow_run_archive_download_task,
-    get_ready_workflow_run_archive_download_task,
-    list_workflow_run_archives,
-)
-
-
-class FakeTaskCache:
-    created_task: WorkflowRunArchiveDownloadTask | None
-    saved_task: WorkflowRunArchiveDownloadTask | None
-    existing_task: WorkflowRunArchiveDownloadTask | None
-    tasks_by_download_id: dict[str, WorkflowRunArchiveDownloadTask]
-    create_result: bool
-
-    def __init__(
-        self,
-        *,
-        create_result: bool = True,
-        existing_task: WorkflowRunArchiveDownloadTask | None = None,
-        tasks_by_download_id: dict[str, WorkflowRunArchiveDownloadTask] | None = None,
-    ) -> None:
-        self.created_task = None
-        self.saved_task = None
-        self.existing_task = existing_task
-        self.tasks_by_download_id = tasks_by_download_id or {}
-        self.create_result = create_result
-
-    def create_if_absent(self, task: WorkflowRunArchiveDownloadTask) -> bool:
-        self.created_task = task
-        return self.create_result
-
-    def get(self, *, tenant_id: str, download_id: str) -> WorkflowRunArchiveDownloadTask | None:
-        if self.tasks_by_download_id:
-            return self.tasks_by_download_id.get(download_id)
-        return self.existing_task
-
-    def save(self, task: WorkflowRunArchiveDownloadTask) -> None:
-        self.saved_task = task
-
-
-def _bundle(
-    *,
-    shard: str,
-    bundle_id: str,
-    archive_bytes: int,
-    year: int = 2025,
-    month: int = 3,
-    workflow_run_count: int = 1,
-    row_count: int = 9,
-    archived_at: datetime.datetime | None = None,
-) -> WorkflowRunArchiveBundle:
-    return cast(
-        WorkflowRunArchiveBundle,
-        SimpleNamespace(
-            year=year,
-            month=month,
-            shard=shard,
-            bundle_id=bundle_id,
-            workflow_run_count=workflow_run_count,
-            row_count=row_count,
-            archive_bytes=archive_bytes,
-            archived_at=archived_at or datetime.datetime(2026, 6, 25, 8, 0),
-        ),
-    )
-
-
-def _fake_dispatcher(dispatched_tasks: list[WorkflowRunArchiveDownloadTask]) -> ArchiveDownloadTaskDispatcher:
-    def dispatch(
-        task: WorkflowRunArchiveDownloadTask,
-        cache: WorkflowRunArchiveDownloadTaskCache,
-    ) -> WorkflowRunArchiveDownloadTask:
-        dispatched_tasks.append(task)
-        return task.model_copy(update={"celery_task_id": "celery-task-1"})
-
-    return dispatch
-
-
-def test_list_workflow_run_archives_aggregates_month_rows() -> None:
-    latest = datetime.datetime(2026, 6, 25, 8, 0)
-    previous = datetime.datetime(2026, 6, 24, 8, 0)
-    session = MagicMock()
-    march_download_id = build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("00-of-01", "bundle-a"), ("00-of-01", "bundle-b")],
-    )
-    ready_task = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a", "bundle-b"],
-        bundle_refs=[("00-of-01", "bundle-a"), ("00-of-01", "bundle-b")],
-        archive_bytes=4096,
-        download_id=march_download_id,
-    ).model_copy(
-        update={
-            "status": WorkflowRunArchiveDownloadStatus.READY,
-            "file_name": "workflow-run-logs-2025-03.zip",
-            "storage_key": "workflow-run-archive-downloads/tenant-1/2025/03/download.zip",
-            "file_size_bytes": 8192,
-        }
-    )
-    cache = FakeTaskCache(tasks_by_download_id={march_download_id: ready_task})
-    session.scalars.return_value = [
-        _bundle(
-            year=2025,
-            month=3,
-            shard="00-of-01",
-            bundle_id="bundle-a",
-            workflow_run_count=40,
-            row_count=360,
-            archive_bytes=1024,
-            archived_at=previous,
-        ),
-        _bundle(
-            year=2025,
-            month=3,
-            shard="00-of-01",
-            bundle_id="bundle-b",
-            workflow_run_count=60,
-            row_count=540,
-            archive_bytes=3072,
-            archived_at=latest,
-        ),
-        _bundle(
-            year=2025,
-            month=2,
-            shard="00-of-01",
-            bundle_id="bundle-c",
-            workflow_run_count=20,
-            row_count=180,
-            archive_bytes=1024,
-            archived_at=previous,
-        ),
-    ]
-
-    result = list_workflow_run_archives(session, "tenant-1", cache=cast(WorkflowRunArchiveDownloadTaskCache, cache))
-
-    assert result.summary.archived_month_count == 2
-    assert result.summary.workflow_run_count == 120
-    assert result.summary.archive_bytes == 5120
-    assert result.summary.latest_archived_at == latest
-    assert result.months[0].year == 2025
-    assert result.months[0].month == 3
-    assert result.months[0].bundle_count == 2
-    assert result.months[0].workflow_run_count == 100
-    assert result.months[0].row_count == 900
-    assert result.months[0].download_task == ready_task
-    assert result.months[1].download_task is None
-
-
-def test_create_workflow_run_archive_download_task_creates_stable_pending_task() -> None:
-    session = MagicMock()
-    session.scalars.return_value = [
-        _bundle(shard="01-of-02", bundle_id="bundle-b", archive_bytes=2048),
-        _bundle(shard="00-of-02", bundle_id="bundle-a", archive_bytes=1024),
-    ]
-    cache = FakeTaskCache()
-    dispatched_tasks: list[WorkflowRunArchiveDownloadTask] = []
-
-    task = create_workflow_run_archive_download_task(
-        session,
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        cache=cast(WorkflowRunArchiveDownloadTaskCache, cache),
-        dispatcher=_fake_dispatcher(dispatched_tasks),
-    )
-
-    assert task.download_id == build_archive_download_id(
-        tenant_id="tenant-1",
-        year=2025,
-        month=3,
-        bundle_refs=[("01-of-02", "bundle-b"), ("00-of-02", "bundle-a")],
-    )
-    assert task.requested_by == "account-1"
-    assert task.bundle_ids == ["bundle-b", "bundle-a"]
-    assert [(ref.shard, ref.bundle_id) for ref in task.bundle_refs] == [
-        ("01-of-02", "bundle-b"),
-        ("00-of-02", "bundle-a"),
-    ]
-    assert task.archive_bytes == 3072
-    assert cache.created_task == dispatched_tasks[0]
-    assert task.celery_task_id == "celery-task-1"
-
-
-def test_create_workflow_run_archive_download_task_returns_existing_task_when_cache_key_exists() -> None:
-    session = MagicMock()
-    session.scalars.return_value = [_bundle(shard="00-of-01", bundle_id="bundle-a", archive_bytes=1024)]
-    existing_task = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a"],
-        archive_bytes=1024,
-        download_id="existing-download",
-    ).model_copy(update={"celery_task_id": "celery-task-1"})
-    cache = FakeTaskCache(create_result=False, existing_task=existing_task)
-    dispatched_tasks: list[WorkflowRunArchiveDownloadTask] = []
-
-    task = create_workflow_run_archive_download_task(
-        session,
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        cache=cast(WorkflowRunArchiveDownloadTaskCache, cache),
-        dispatcher=_fake_dispatcher(dispatched_tasks),
-    )
-
-    assert task == existing_task
-    assert cache.saved_task is None
-    assert dispatched_tasks == []
-
-
-def test_create_workflow_run_archive_download_task_retries_failed_cached_task() -> None:
-    session = MagicMock()
-    session.scalars.return_value = [_bundle(shard="00-of-01", bundle_id="bundle-a", archive_bytes=1024)]
-    existing_task = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a"],
-        bundle_refs=[("00-of-01", "bundle-a")],
-        archive_bytes=1024,
-        download_id=build_archive_download_id(
-            tenant_id="tenant-1",
-            year=2025,
-            month=3,
-            bundle_refs=[("00-of-01", "bundle-a")],
-        ),
-    ).model_copy(update={"status": WorkflowRunArchiveDownloadStatus.FAILED, "error": "failed"})
-    cache = FakeTaskCache(create_result=False, existing_task=existing_task)
-    dispatched_tasks: list[WorkflowRunArchiveDownloadTask] = []
-
-    task = create_workflow_run_archive_download_task(
-        session,
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        cache=cast(WorkflowRunArchiveDownloadTaskCache, cache),
-        dispatcher=_fake_dispatcher(dispatched_tasks),
-    )
-
-    assert task.status == WorkflowRunArchiveDownloadStatus.PENDING
-    assert task.error is None
-    assert task.celery_task_id == "celery-task-1"
-    assert cache.saved_task == dispatched_tasks[0]
-
-
-def test_create_workflow_run_archive_download_task_rejects_missing_month() -> None:
-    session = MagicMock()
-    session.scalars.return_value = []
-
-    with pytest.raises(WorkflowRunArchiveNotFoundError):
-        create_workflow_run_archive_download_task(
-            session,
-            tenant_id="tenant-1",
-            requested_by="account-1",
-            year=2025,
-            month=3,
-            cache=cast(WorkflowRunArchiveDownloadTaskCache, FakeTaskCache()),
-            dispatcher=_fake_dispatcher([]),
-        )
-
-
-def test_get_ready_workflow_run_archive_download_task_requires_ready_file() -> None:
-    pending_task = build_pending_archive_download_task(
-        tenant_id="tenant-1",
-        requested_by="account-1",
-        year=2025,
-        month=3,
-        bundle_ids=["bundle-a"],
-        archive_bytes=1024,
-        download_id="download-1",
-    )
-    cache = FakeTaskCache(existing_task=pending_task)
-
-    with pytest.raises(WorkflowRunArchiveDownloadNotReadyError):
-        get_ready_workflow_run_archive_download_task(
-            tenant_id="tenant-1",
-            download_id="download-1",
-            cache=cast(WorkflowRunArchiveDownloadTaskCache, cache),
-        )
-
-    ready_task = pending_task.model_copy(
-        update={
-            "status": WorkflowRunArchiveDownloadStatus.READY,
-            "storage_key": "downloads/download-1.zip",
-            "file_name": "workflow-run-logs-2025-03.zip",
-        }
-    )
-    cache = FakeTaskCache(existing_task=ready_task)
-
-    assert (
-        get_ready_workflow_run_archive_download_task(
-            tenant_id="tenant-1",
-            download_id="download-1",
-            cache=cast(WorkflowRunArchiveDownloadTaskCache, cache),
-        )
-        == ready_task
-    )

api/tests/unit_tests/services/test_file_request_service.py

@@ -59,6 +59,31 @@ def test_request_download_url_builds_file_under_bound_scope(
     assert result.download_url == "https://files.example.com/x"
 
 
+def test_request_download_url_supports_internal_download_urls() -> None:
+    fake_file = MagicMock(filename="report.pdf", mime_type="application/pdf", size=123)
+    service = FileRequestService(access_controller=MagicMock())
+
+    with (
+        patch("services.file_request_service.bind_file_access_scope", return_value=nullcontext()),
+        patch.object(service, "_build_file", return_value=fake_file),
+        patch(
+            "services.file_request_service.file_helpers.resolve_file_url",
+            return_value="http://internal-files/report.pdf",
+        ) as resolve_file_url,
+    ):
+        result = service.request_download_url(
+            tenant_id="tenant-1",
+            user_id="user-1",
+            user_from="account",
+            invoke_from="debugger",
+            file_mapping={"transfer_method": "tool_file", "reference": "dify-file-ref:tool-file-1"},
+            for_external=False,
+        )
+
+    resolve_file_url.assert_called_once_with(fake_file, for_external=False)
+    assert result.download_url == "http://internal-files/report.pdf"
+
+
 def test_request_download_url_rejects_unsupported_files() -> None:
     service = FileRequestService(access_controller=MagicMock())
 

api/uv.lock

@@ -322,16 +322,15 @@ wheels = [
 
 [[package]]
 name = "anyio"
-version = "4.11.0"
+version = "4.14.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "idna" },
-    { name = "sniffio" },
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/c6/78/7d432127c41b50bccba979505f272c16cbcadcc33645d5fa3a738110ae75/anyio-4.11.0.tar.gz", hash = "sha256:82a8d0b81e318cc5ce71a5f1f8b5c4e63619620b63141ef8c995fa0db95a57c4", size = 219094, upload-time = "2025-09-23T09:19:12.58Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3b/72/5562aabb8dd7181e8e860622a38bea08d17842b99ecd4c91f84ac95251b0/anyio-4.14.1.tar.gz", hash = "sha256:8d648a3544c1a700e3ff78615cd679e4c5c3f149904287e73687b2596963629e", size = 254831, upload-time = "2026-06-24T20:56:06.017Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/15/b3/9b1a8074496371342ec1e796a96f99c82c945a339cd81a8e73de28b4cf9e/anyio-4.11.0-py3-none-any.whl", hash = "sha256:0287e96f4d26d4149305414d4e3bc32f0dcd0862365a4bddea19d7a1ec38c4fc", size = 109097, upload-time = "2025-09-23T09:19:10.601Z" },
+    { url = "https://files.pythonhosted.org/packages/b0/7b/90df4a0a816d98d6ea26f559d87836d494a2cf1fcf063be67df50a7bcc30/anyio-4.14.1-py3-none-any.whl", hash = "sha256:4e5533c5b8ff0a24f5d7a176cbe6877129cd183893f66b537f8f227d10527d72", size = 124875, upload-time = "2026-06-24T20:56:04.413Z" },
 ]
 
 [[package]]
@@ -1284,7 +1283,9 @@ name = "dify-agent"
 version = "0.1.0"
 source = { editable = "../dify-agent" }
 dependencies = [
+    { name = "anyio" },
     { name = "httpx" },
+    { name = "httpx2" },
     { name = "pydantic" },
     { name = "pydantic-ai-slim" },
     { name = "typer" },
@@ -1293,10 +1294,14 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
+    { name = "aiosqlite", marker = "extra == 'shellctl-server'", specifier = ">=0.21.0,<1.0.0" },
+    { name = "anyio", specifier = ">=4.12.1,<5.0.0" },
     { name = "fastapi", marker = "extra == 'server'", specifier = "==0.136.0" },
+    { name = "fastapi", marker = "extra == 'shellctl-server'", specifier = "==0.136.0" },
     { name = "graphon", marker = "extra == 'server'", specifier = "==0.5.2" },
     { name = "grpclib", extras = ["protobuf"], marker = "extra == 'grpc'", specifier = ">=0.4.9,<0.5.0" },
     { name = "httpx", specifier = "==0.28.1" },
+    { name = "httpx2", specifier = ">=2.5.0,<3.0.0" },
     { name = "jsonschema", marker = "extra == 'server'", specifier = ">=4.23.0,<5.0.0" },
     { name = "jwcrypto", marker = "extra == 'server'", specifier = ">=1.5.6,<2" },
     { name = "logfire", extras = ["fastapi", "httpx", "redis"], marker = "extra == 'server'", specifier = ">=4.37.0,<5.0.0" },
@@ -1306,12 +1311,13 @@ requires-dist = [
     { name = "pydantic-ai-slim", extras = ["anthropic", "google", "openai"], marker = "extra == 'server'", specifier = ">=1.85.1,<2.0.0" },
     { name = "pydantic-settings", marker = "extra == 'server'", specifier = ">=2.12.0,<3.0.0" },
     { name = "redis", marker = "extra == 'server'", specifier = ">=7.4.0,<8.0.0" },
-    { name = "shell-session-manager", marker = "extra == 'server'", specifier = "==2.4.0" },
+    { name = "sqlmodel", marker = "extra == 'shellctl-server'", specifier = ">=0.0.24,<0.1.0" },
     { name = "typer", specifier = ">=0.16.1,<0.17" },
     { name = "typing-extensions", specifier = ">=4.12.2,<5.0.0" },
     { name = "uvicorn", extras = ["standard"], marker = "extra == 'server'", specifier = "==0.46.0" },
+    { name = "uvicorn", extras = ["standard"], marker = "extra == 'shellctl-server'", specifier = "==0.46.0" },
 ]
-provides-extras = ["grpc", "server"]
+provides-extras = ["grpc", "server", "shellctl-server"]
 
 [package.metadata.requires-dev]
 dev = [
@@ -3283,15 +3289,15 @@ wheels = [
 
 [[package]]
 name = "httpcore2"
-version = "2.3.0"
+version = "2.5.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "h11" },
     { name = "truststore" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/e6/34/18f1c596e677962f040284246f393b10a1f8ce440b3a7e69c637d0f1c7ad/httpcore2-2.3.0.tar.gz", hash = "sha256:07327e251560960eea8e969d92d4c6a325feb13cca39e25340731336c3baf924", size = 64300, upload-time = "2026-06-01T13:15:02.998Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/47/06/5c12df521b5322fb1114a83d46911b2fbcb8855ddb3a635f11c01a214af5/httpcore2-2.5.0.tar.gz", hash = "sha256:88aa170137c17328d5ac44234f9fd10706466d5fb347f3edac4d39b91137b09d", size = 64808, upload-time = "2026-06-25T14:16:56.472Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c2/dd/3357218c69360d1cecc196c230c9a1d5c9afd5dba362056e23e60a5e64e5/httpcore2-2.3.0-py3-none-any.whl", hash = "sha256:477e9e334f74e5240dcac002e890580f36a57d40ff0fb14cc9655731d23b8415", size = 80024, upload-time = "2026-06-01T13:15:00.001Z" },
+    { url = "https://files.pythonhosted.org/packages/c9/a1/7564199d1a8728fe737b0a72e5b3f8d92dfe085a74ddf7cdd83bce5f206d/httpcore2-2.5.0-py3-none-any.whl", hash = "sha256:5ce35188de461d31e8d000bfb8ef8bf22c6c16587a211e5571deaa5e9bdf842a", size = 80330, upload-time = "2026-06-25T14:16:53.634Z" },
 ]
 
 [[package]]
@@ -3355,17 +3361,18 @@ wheels = [
 
 [[package]]
 name = "httpx2"
-version = "2.3.0"
+version = "2.5.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "anyio" },
     { name = "httpcore2" },
     { name = "idna" },
     { name = "truststore" },
+    { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/9f/9a/cca0b9145f13d8ae34b885ae28d403a1469a433abc78e0f94f4ce94e650b/httpx2-2.3.0.tar.gz", hash = "sha256:227e7c41d95a76d4077a52640564132777215fc3394e07b66a3116c33d668fa9", size = 81115, upload-time = "2026-06-01T13:15:04.324Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/d0/e2/b5dedc0cf35aa65de5f541ccd30d2bc1fd7f1d43c9ab09f8ed9a7342317b/httpx2-2.5.0.tar.gz", hash = "sha256:e2df9cb4611021527ff8a675b1c320b610a2ec397acc8d6fe6e91df2d9b33c29", size = 83121, upload-time = "2026-06-25T14:16:57.491Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/87/ce/ae2911859847f9ba1d6b23027e53481cbeb50b93234f355a968d300ca2cb/httpx2-2.3.0-py3-none-any.whl", hash = "sha256:6f393663bdf6dbe7fe90118e3eb5b2bd024a675cae0390ac08cec9198812d8b7", size = 74538, upload-time = "2026-06-01T13:15:01.566Z" },
+    { url = "https://files.pythonhosted.org/packages/31/22/859d8252dad9bc9adee34b52e62cde621ece07b042ccb2ab4da1be46695f/httpx2-2.5.0-py3-none-any.whl", hash = "sha256:3d2d4d9cf4b61f1a1f46a95947cfdb47e80cb56a2f91c6256ac8f58e4891df41", size = 76652, upload-time = "2026-06-25T14:16:55.23Z" },
 ]
 
 [[package]]
@@ -3423,11 +3430,11 @@ wheels = [
 
 [[package]]
 name = "idna"
-version = "3.11"
+version = "3.18"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/6f/6d/0703ccc57f3a7233505399edb88de3cbd678da106337b9fcde432b65ed60/idna-3.11.tar.gz", hash = "sha256:795dafcc9c04ed0c1fb032c2aa73654d8e8c5023a7df64a53f39190ada629902", size = 194582, upload-time = "2025-10-12T14:55:20.501Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/cd/63/9496c57188a2ee585e0f1db071d75089a11e98aa86eb99d9d7618fc1edce/idna-3.18.tar.gz", hash = "sha256:ffb385a7e039654cef1ab9ef32c6fafe283c0c0467bba1d9029738ce4a14a848", size = 196711, upload-time = "2026-06-02T14:34:07.794Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/0e/61/66938bbb5fc52dbdf84594873d5b51fb1f7c7794e9c0f5bd885f30bc507b/idna-3.11-py3-none-any.whl", hash = "sha256:771a87f49d9defaf64091e6e6fe9c18d4833f140bd19464795bc32d966ca37ea", size = 71008, upload-time = "2025-10-12T14:55:18.883Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/5e/d4e9f1a599fb8e573b7b87160658329fbf28d19eac2718f51fc3def3aa5a/idna-3.18-py3-none-any.whl", hash = "sha256:7f952cbe720b688055e3f87de14f5c3e5fdaa8bc3928985c4077ca689de849a2", size = 65455, upload-time = "2026-06-02T14:34:06.319Z" },
 ]
 
 [[package]]

dify-agent/.example.env

@@ -20,6 +20,12 @@ DIFY_AGENT_PLUGIN_DAEMON_URL=http://localhost:5002
 # API key sent to the Dify plugin daemon.
 DIFY_AGENT_PLUGIN_DAEMON_API_KEY=
 
+# Dify API inner endpoints
+# Base URL for Dify API inner endpoints used by Agent Stub config/file/drive requests.
+DIFY_AGENT_INNER_API_URL=http://localhost:5001
+# Must match API/worker INNER_API_KEY_FOR_PLUGIN, not the generic INNER_API_KEY.
+DIFY_AGENT_INNER_API_KEY=
+
 # Shell layer
 # Base URL for the shellctl server used by the dify.shell layer. Leave empty to disable shell layer use.
 DIFY_AGENT_SHELLCTL_ENTRYPOINT=
@@ -30,12 +36,12 @@ DIFY_AGENT_SHELLCTL_AUTH_TOKEN=
 # Public Agent Stub URL reachable from shellctl-managed remote machines.
 # Use http(s)://.../agent-stub for HTTP or grpc://host:port for gRPC.
 # Leave empty to avoid injecting DIFY_AGENT_STUB_* into shell.run jobs.
-DIFY_AGENT_STUB_URL=
-# Optional bind override used only when DIFY_AGENT_STUB_URL uses grpc://.
+DIFY_AGENT_STUB_API_BASE_URL=http://localhost:5050/agent-stub
+# Optional bind override used only when DIFY_AGENT_STUB_API_BASE_URL uses grpc://.
 DIFY_AGENT_STUB_GRPC_BIND_ADDRESS=
 # Server-wide root secret used to derive Agent Stub JWE keys.
-# Required when DIFY_AGENT_STUB_URL is set; must be unpadded base64url for 32 bytes.
-DIFY_AGENT_SERVER_SECRET_KEY=
+# Required when DIFY_AGENT_STUB_API_BASE_URL is set; must be unpadded base64url for 32 bytes.
+DIFY_AGENT_SERVER_SECRET_KEY=replace-with-base64url-32-byte-secret
 
 # Shared plugin-daemon HTTP client timeouts and limits.
 # Plugin-daemon HTTP connect timeout in seconds.

dify-agent/Dockerfile

@@ -6,8 +6,8 @@
 #   cd /app/api && .venv/bin/uvicorn dify_agent.server.app:app --host 0.0.0.0 --port 5050
 #
 # Unlike the dify-api image (which only installs the base `dify-agent`
-# dependency), this image installs the `[server]` extra, so jwcrypto,
-# shell-session-manager, fastapi, uvicorn, etc. are present and the server can
+# dependency), this image installs the `[server]` extra, so jwcrypto, fastapi,
+# uvicorn, etc. are present and the server can
 # actually start. dify-api is intentionally left lean.
 
 # base image

dify-agent/docker/local-sandbox/Dockerfile

@@ -12,19 +12,15 @@ FROM python:3.12-slim-bookworm AS base
 ARG NODE_VERSION=22.22.1
 ARG PNPM_VERSION=11.9.0
 ARG UV_VERSION=0.8.9
-ARG DIFY_AGENT_TOOL_SPEC=.[grpc]
-ARG SHELL_SESSION_MANAGER_TOOL_SPEC=shell-session-manager==2.4.0
+ARG DIFY_AGENT_TOOL_SPEC=.[grpc,shellctl-server]
 
 ENV PYTHONDONTWRITEBYTECODE=1 \
     PYTHONUNBUFFERED=1 \
-    PIP_NO_CACHE_DIR=1 \
-    DIFY_AGENT_STUB_DRIVE_BASE=/mnt/drive \
-    UV_TOOL_DIR=/opt/dify-agent-tools/envs \
-    UV_TOOL_BIN_DIR=/opt/dify-agent-tools/bin
-ENV PATH="${UV_TOOL_BIN_DIR}:${PATH}"
+    PIP_NO_CACHE_DIR=1
 
 RUN apt-get update \
     && apt-get install -y --no-install-recommends \
+        bash \
         ca-certificates \
         curl \
         file \
@@ -62,24 +58,25 @@ WORKDIR /opt/dify-agent
 FROM base AS tools
 
 ARG DIFY_AGENT_TOOL_SPEC
-ARG SHELL_SESSION_MANAGER_TOOL_SPEC
 
 COPY pyproject.toml uv.lock README.md ./
 COPY src ./src
 
 RUN uv export --frozen --no-dev --all-extras --no-emit-project --no-hashes \
         > /tmp/dify-agent-constraints.txt \
-    && uv tool install --force --python /usr/local/bin/python --no-python-downloads \
+    && UV_TOOL_DIR=/opt/dify-agent-tools/envs \
+        UV_TOOL_BIN_DIR=/opt/dify-agent-tools/bin \
+        uv tool install --force --python /usr/local/bin/python --no-python-downloads \
         --constraints /tmp/dify-agent-constraints.txt --link-mode=copy "${DIFY_AGENT_TOOL_SPEC}" \
-    && uv tool install --force --python /usr/local/bin/python --no-python-downloads \
-        --constraints /tmp/dify-agent-constraints.txt --link-mode=copy "${SHELL_SESSION_MANAGER_TOOL_SPEC}" \
     && rm -f /tmp/dify-agent-constraints.txt
 
 
 FROM base AS production
 
-COPY --from=tools ${UV_TOOL_DIR} ${UV_TOOL_DIR}
-COPY --from=tools ${UV_TOOL_BIN_DIR} ${UV_TOOL_BIN_DIR}
+ENV PATH="/opt/dify-agent-tools/bin:${PATH}"
+
+COPY --from=tools /opt/dify-agent-tools/envs /opt/dify-agent-tools/envs
+COPY --from=tools /opt/dify-agent-tools/bin /opt/dify-agent-tools/bin
 
 RUN useradd --create-home --shell /bin/sh dify \
     && mkdir -p /mnt/drive \

dify-agent/docs/dify-agent/user-manual/shell-layer/index.md

@@ -230,12 +230,11 @@ The provided `docker/local-sandbox/Dockerfile` installs:
 - `tmux`, required by `shellctl` to manage shell jobs;
 - common shell workspace tools: `git`, `openssh-client`, `jq`, `ripgrep`,
   `unzip`, `zip`, `file`, `procps`, and `less`;
-- `shell-session-manager==2.3.1` as a standalone uv tool, which provides the
-  `shellctl` CLI/server;
+- `dify-agent[grpc,shellctl-server]` as a standalone uv tool, which provides
+  both the Agent Stub client CLI and the built-in `shellctl` CLI/server;
 - `uv`, so uv shebang scripts with PEP 723 metadata can run inside the shell
   workspace and Python CLI tools can be installed with isolated tool
   environments;
 - `node==22.22.1` and `pnpm==11.9.0`, so JavaScript and TypeScript tooling can
   run inside the shell workspace without per-job installation;
-- the `dify-agent[grpc]` Agent Stub client CLI as a standalone uv tool;
 - a non-root default user named `dify`.

dify-agent/proto/dify/agent/stub/v1/agent_stub.proto

@@ -36,6 +36,7 @@ message FileMapping {
 
 message FileDownloadRequest {
   FileMapping file = 1;
+  optional bool for_external = 2;
 }
 
 message FileDownloadResponse {

dify-agent/pyproject.toml

@@ -5,7 +5,9 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.12,<4.0"
 dependencies = [
+    "anyio>=4.12.1,<5.0.0",
     "httpx==0.28.1",
+    "httpx2>=2.5.0,<3.0.0",
     "pydantic>=2.12.5,<2.13",
     "pydantic-ai-slim>=1.102.0,<2.0.0",
     "typer>=0.16.1,<0.17",
@@ -15,6 +17,9 @@ dependencies = [
 [project.scripts]
 dify-agent = "dify_agent.agent_stub.cli.main:main"
 dify-agent-stub-server = "dify_agent.agent_stub.server.cli:main"
+shellctl = "shellctl.cli:main"
+shellctl-sanitize-pty = "shellctl_runtime.sanitize:main"
+shellctl-runner-exit = "shellctl_runtime.runner_exit:main"
 
 [project.optional-dependencies]
 grpc = ["grpclib[protobuf]>=0.4.9,<0.5.0", "protobuf>=6.33.5,<7.0.0"]
@@ -27,13 +32,18 @@ server = [
     "pydantic-ai-slim[anthropic,google,openai]>=1.85.1,<2.0.0",
     "pydantic-settings>=2.12.0,<3.0.0",
     "redis>=7.4.0,<8.0.0",
-    "shell-session-manager==2.4.0",
+    "uvicorn[standard]==0.46.0",
+]
+shellctl-server = [
+    "aiosqlite>=0.21.0,<1.0.0",
+    "fastapi==0.136.0",
+    "sqlmodel>=0.0.24,<0.1.0",
     "uvicorn[standard]==0.46.0",
 ]
 
 [tool.setuptools.packages.find]
 where = ["src"]
-include = ["agenton*", "agenton_collections*", "dify_agent*"]
+include = ["agenton*", "agenton_collections*", "dify_agent*", "shellctl*", "shellctl_runtime*"]
 
 [tool.pyright]
 include = ["src", "examples", "tests"]

dify-agent/src/dify_agent/adapters/llm/model.py

@@ -205,6 +205,7 @@ class DifyStreamedResponse(StreamedResponse):
 
     @override
     async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
+        chunk_sequence = 0
         async for chunk in self.chunks:
             if chunk.delta.usage is not None:
                 self._usage: RequestUsage = _map_usage(chunk.delta.usage)
@@ -216,8 +217,10 @@ class DifyStreamedResponse(StreamedResponse):
                 chunk,
                 self.provider_name_value,
                 self._embedded_thinking_parser,
+                chunk_sequence,
             ):
                 yield event
+            chunk_sequence += 1
 
         for event in self._embedded_thinking_parser.flush(self._parts_manager, self.provider_name_value):
             yield event
@@ -551,11 +554,21 @@ def _normalize_finish_reason(finish_reason: str) -> FinishReason:
     return "error"
 
 
+def _normalize_tool_call_id(tool_call_id: str | None) -> str | None:
+    if tool_call_id is None:
+        return None
+    normalized = tool_call_id.strip()
+    if not normalized or normalized.lower() in {"none", "null"}:
+        return None
+    return normalized
+
+
 def _chunk_to_stream_events(
     parts_manager: ModelResponsePartsManager,
     chunk: LLMResultChunk,
     provider_name: str,
     embedded_thinking_parser: "_EmbeddedThinkingParser",
+    chunk_sequence: int,
 ) -> list[ModelResponseStreamEvent]:
     events: list[ModelResponseStreamEvent] = []
     message = chunk.delta.message
@@ -571,13 +584,14 @@ def _chunk_to_stream_events(
                 events.append(parts_manager.handle_part(vendor_part_id=None, part=part))
 
     for index, tool_call in enumerate(message.tool_calls):
-        vendor_id = tool_call.id or f"chunk-{chunk.delta.index}-tool-{index}"
+        tool_call_id = _normalize_tool_call_id(tool_call.id)
+        vendor_id = tool_call_id or f"chunk-{chunk_sequence}-tool-{index}"
         events.append(
             parts_manager.handle_tool_call_part(
                 vendor_part_id=vendor_id,
                 tool_name=tool_call.function.name,
                 args=tool_call.function.arguments,
-                tool_call_id=tool_call.id,
+                tool_call_id=tool_call_id or vendor_id,
                 provider_name=provider_name,
             )
         )

dify-agent/src/dify_agent/adapters/shell/shellctl.py

@@ -1,6 +1,6 @@
 """Shellctl-backed shell provider adapter for dify-agent.
 
-The shell-session-manager SDK owns the HTTP timeout policy for long-polling
+The built-in shellctl SDK owns the HTTP timeout policy for long-polling
 shellctl requests. This adapter stays narrowly focused on translating SDK and
 transport failures into ``ShellProviderError`` so the shell layer can return
 tool observations instead of aborting the agent loop.
@@ -16,9 +16,9 @@ import time
 from collections.abc import Awaitable
 from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Protocol, TypeVar
+from typing import Protocol, TypeVar, cast
 
-import httpx
+import httpx2 as httpx
 
 from dify_agent.adapters.shell.protocols import (
     ShellCommandProtocol,
@@ -228,8 +228,16 @@ class ShellctlFileTransfer(ShellFileTransferProtocol):
 @dataclass(slots=True)
 class ShellctlResource(ShellResourceProtocol):
     client: ShellctlClientProtocol
-    commands: ShellCommandProtocol
-    files: ShellFileTransferProtocol
+    _commands: ShellCommandProtocol
+    _files: ShellFileTransferProtocol
+
+    @property
+    def commands(self) -> ShellCommandProtocol:
+        return self._commands
+
+    @property
+    def files(self) -> ShellFileTransferProtocol:
+        return self._files
 
     async def close(self) -> None:
         try:
@@ -257,8 +265,8 @@ class ShellctlProvider(ShellProviderProtocol):
         )
         return ShellctlResource(
             client=client,
-            commands=ShellctlCommands(client=client),
-            files=ShellctlFileTransfer(client=client),
+            _commands=ShellctlCommands(client=client),
+            _files=ShellctlFileTransfer(client=client),
         )
 
 
@@ -269,9 +277,15 @@ def create_default_shellctl_client_factory(
     output_limit: int = _SHELLCTL_OUTPUT_LIMIT_BYTES,
 ) -> ShellctlClientFactory:
     def factory() -> ShellctlClientProtocol:
-        from shell_session_manager.shellctl.client import ShellctlClient
+        from shellctl.client import ShellctlClient
 
-        return ShellctlClient(entrypoint, token=token, output_limit=output_limit)
+        return cast(
+            ShellctlClientProtocol,
+            cast(
+                object,
+                ShellctlClient(entrypoint, token=token, output_limit=output_limit),
+            ),
+        )
 
     return factory
 

dify-agent/src/dify_agent/agent_stub/cli/_files.py

@@ -143,6 +143,7 @@ def download_file_from_environment(
         url=environment.url,
         auth_jwe=environment.auth_jwe,
         file=file_mapping,
+        for_external=False,
     )
     if not hasattr(download_request, "filename") or not isinstance(download_request.filename, str):
         raise AgentStubTransferError("signed file download response is missing filename")
@@ -207,8 +208,10 @@ def _request_uploaded_tool_file_download_url(*, url: str, auth_jwe: str, referen
             file=AgentStubFileMapping(transfer_method="tool_file", reference=reference),
         ),
     )
+    if not hasattr(download_request, "download_url") or not isinstance(download_request.download_url, str):
+        raise AgentStubTransferError("signed file download response is missing download_url")
     download_url = download_request.download_url
-    if not isinstance(download_url, str) or not download_url:
+    if not download_url:
         raise AgentStubTransferError("signed file download response is missing download_url")
     return download_url
 

dify-agent/src/dify_agent/agent_stub/cli/main.py

@@ -133,6 +133,26 @@ def config_skills_push(
 
     Pass a directory such as ./skills/researcher that contains SKILL.md. Other files in that directory are
     archived with the skill. Pushing a skill with an existing name replaces that config skill.
+
+    Skill directory requirements:
+
+    - Each PATH must be one skill directory; the directory basename is the config skill name.
+
+    - The directory must contain a top-level SKILL.md.
+
+    - SKILL.md must be non-empty UTF-8 Markdown.
+
+    - SKILL.md must start with YAML frontmatter matching this schema:
+
+    \b
+      ---
+      name: <non-empty string>
+      description: <string>
+      ---
+
+    - Symlinked files are rejected.
+
+    - Dependency/cache folders such as .git, __pycache__, .venv and node_modules should be manually cleared before push.
     """
     _run_config_skills_push(paths=paths)
 

dify-agent/src/dify_agent/agent_stub/client/_agent_stub.py

@@ -97,6 +97,7 @@ def request_agent_stub_file_download_sync(
     url: str,
     auth_jwe: str,
     file: AgentStubFileMapping,
+    for_external: bool = True,
     timeout: float | httpx.Timeout = 30.0,
     sync_http_client: httpx.Client | None = None,
 ):
@@ -109,12 +110,14 @@ def request_agent_stub_file_download_sync(
             url=endpoint.url,
             auth_jwe=auth_jwe,
             file=file,
+            for_external=for_external,
             timeout=timeout,
         )
     return request_agent_stub_file_download_http_sync(
         base_url=endpoint.url,
         auth_jwe=auth_jwe,
         file=file,
+        for_external=for_external,
         timeout=timeout,
         sync_http_client=sync_http_client,
     )

dify-agent/src/dify_agent/agent_stub/client/_agent_stub_grpc.py

@@ -124,6 +124,7 @@ def request_agent_stub_file_download_grpc_sync(
     url: str,
     auth_jwe: str,
     file: AgentStubFileMapping,
+    for_external: bool = True,
     timeout: float | httpx.Timeout = 30.0,
 ):
     """Request one signed download URL through the gRPC Agent Stub endpoint.
@@ -144,7 +145,7 @@ def request_agent_stub_file_download_grpc_sync(
             auth_jwe=auth_jwe,
             method_name="CreateFileDownloadRequest",
             request_factory=lambda runtime: _require_conversions().proto_file_download_request(
-                runtime.agent_stub_pb2, file=file
+                runtime.agent_stub_pb2, file=file, for_external=for_external
             ),
             response_parser=lambda response: _require_conversions().file_download_response_from_proto(response),
             timeout=timeout,

dify-agent/src/dify_agent/agent_stub/client/_agent_stub_http.py

@@ -117,13 +117,14 @@ def request_agent_stub_file_download_http_sync(
     base_url: str,
     auth_jwe: str,
     file: AgentStubFileMapping,
+    for_external: bool = True,
     timeout: float | httpx.Timeout = 30.0,
     sync_http_client: httpx.Client | None = None,
 ) -> AgentStubFileDownloadResponse:
     """Request one signed download URL from the HTTP Agent Stub endpoint."""
 
     try:
-        request_model = AgentStubFileDownloadRequest(file=file)
+        request_model = AgentStubFileDownloadRequest(file=file, for_external=for_external)
     except ValidationError as exc:
         raise AgentStubValidationError("invalid Agent Stub file download request") from exc
     response = _post_agent_stub_json(
@@ -131,7 +132,7 @@ def request_agent_stub_file_download_http_sync(
         auth_jwe=auth_jwe,
         endpoint_name="file download request",
         endpoint_url_factory=agent_stub_file_download_request_url,
-        request_body=request_model.model_dump_json(exclude_none=True),
+        request_body=request_model.model_dump_json(exclude_none=True, exclude_defaults=True),
         timeout=timeout,
         sync_http_client=sync_http_client,
     )

dify-agent/src/dify_agent/agent_stub/grpc/_generated/agent_stub_pb2.py

@@ -1,6 +1,6 @@
-# pyright: reportAttributeAccessIssue=false
 # -*- coding: utf-8 -*-
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
+# NO CHECKED-IN PROTOBUF GENCODE
 # source: dify/agent/stub/v1/agent_stub.proto
 # Protobuf Python Version: 6.33.5
 """Generated protocol buffer code."""
@@ -12,12 +12,7 @@ from google.protobuf import symbol_database as _symbol_database
 from google.protobuf.internal import builder as _builder
 
 _runtime_version.ValidateProtobufRuntimeVersion(
-    _runtime_version.Domain.PUBLIC,
-    6,
-    33,
-    5,
-    "",
-    "dify/agent/stub/v1/agent_stub.proto",
+    _runtime_version.Domain.PUBLIC, 6, 33, 5, "", "dify/agent/stub/v1/agent_stub.proto"
 )
 # @@protoc_insertion_point(imports)
 
@@ -25,12 +20,12 @@ _sym_db = _symbol_database.Default()
 
 
 DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
-    b'\n#dify/agent/stub/v1/agent_stub.proto\x12\x12\x64ify.agent.stub.v1"O\n\x0e\x43onnectRequest\x12\x18\n\x10protocol_version\x18\x01 \x01(\x05\x12\x0c\n\x04\x61rgv\x18\x02 \x03(\t\x12\x15\n\rmetadata_json\x18\x03 \x01(\t"8\n\x0f\x43onnectResponse\x12\x15\n\rconnection_id\x18\x01 \x01(\t\x12\x0e\n\x06status\x18\x02 \x01(\t"7\n\x11\x46ileUploadRequest\x12\x10\n\x08\x66ilename\x18\x01 \x01(\t\x12\x10\n\x08mimetype\x18\x02 \x01(\t"(\n\x12\x46ileUploadResponse\x12\x12\n\nupload_url\x18\x01 \x01(\t"f\n\x0b\x46ileMapping\x12\x17\n\x0ftransfer_method\x18\x01 \x01(\t\x12\x16\n\treference\x18\x02 \x01(\tH\x00\x88\x01\x01\x12\x10\n\x03url\x18\x03 \x01(\tH\x01\x88\x01\x01\x42\x0c\n\n_referenceB\x06\n\x04_url"D\n\x13\x46ileDownloadRequest\x12-\n\x04\x66ile\x18\x01 \x01(\x0b\x32\x1f.dify.agent.stub.v1.FileMapping"r\n\x14\x46ileDownloadResponse\x12\x10\n\x08\x66ilename\x18\x01 \x01(\t\x12\x16\n\tmime_type\x18\x02 \x01(\tH\x00\x88\x01\x01\x12\x0c\n\x04size\x18\x03 \x01(\x03\x12\x14\n\x0c\x64ownload_url\x18\x04 \x01(\tB\x0c\n\n_mime_type2\xc0\x02\n\x10\x41gentStubService\x12R\n\x07\x43onnect\x12".dify.agent.stub.v1.ConnectRequest\x1a#.dify.agent.stub.v1.ConnectResponse\x12h\n\x17\x43reateFileUploadRequest\x12%.dify.agent.stub.v1.FileUploadRequest\x1a&.dify.agent.stub.v1.FileUploadResponse\x12n\n\x19\x43reateFileDownloadRequest\x12\'.dify.agent.stub.v1.FileDownloadRequest\x1a(.dify.agent.stub.v1.FileDownloadResponseb\x06proto3'
+    b'\n#dify/agent/stub/v1/agent_stub.proto\x12\x12\x64ify.agent.stub.v1"O\n\x0e\x43onnectRequest\x12\x18\n\x10protocol_version\x18\x01 \x01(\x05\x12\x0c\n\x04\x61rgv\x18\x02 \x03(\t\x12\x15\n\rmetadata_json\x18\x03 \x01(\t"8\n\x0f\x43onnectResponse\x12\x15\n\rconnection_id\x18\x01 \x01(\t\x12\x0e\n\x06status\x18\x02 \x01(\t"7\n\x11\x46ileUploadRequest\x12\x10\n\x08\x66ilename\x18\x01 \x01(\t\x12\x10\n\x08mimetype\x18\x02 \x01(\t"(\n\x12\x46ileUploadResponse\x12\x12\n\nupload_url\x18\x01 \x01(\t"f\n\x0b\x46ileMapping\x12\x17\n\x0ftransfer_method\x18\x01 \x01(\t\x12\x16\n\treference\x18\x02 \x01(\tH\x00\x88\x01\x01\x12\x10\n\x03url\x18\x03 \x01(\tH\x01\x88\x01\x01\x42\x0c\n\n_referenceB\x06\n\x04_url"p\n\x13\x46ileDownloadRequest\x12-\n\x04\x66ile\x18\x01 \x01(\x0b\x32\x1f.dify.agent.stub.v1.FileMapping\x12\x19\n\x0c\x66or_external\x18\x02 \x01(\x08H\x00\x88\x01\x01\x42\x0f\n\r_for_external"r\n\x14\x46ileDownloadResponse\x12\x10\n\x08\x66ilename\x18\x01 \x01(\t\x12\x16\n\tmime_type\x18\x02 \x01(\tH\x00\x88\x01\x01\x12\x0c\n\x04size\x18\x03 \x01(\x03\x12\x14\n\x0c\x64ownload_url\x18\x04 \x01(\tB\x0c\n\n_mime_type2\xc0\x02\n\x10\x41gentStubService\x12R\n\x07\x43onnect\x12".dify.agent.stub.v1.ConnectRequest\x1a#.dify.agent.stub.v1.ConnectResponse\x12h\n\x17\x43reateFileUploadRequest\x12%.dify.agent.stub.v1.FileUploadRequest\x1a&.dify.agent.stub.v1.FileUploadResponse\x12n\n\x19\x43reateFileDownloadRequest\x12\'.dify.agent.stub.v1.FileDownloadRequest\x1a(.dify.agent.stub.v1.FileDownloadResponseb\x06proto3'
 )
 
 _globals = globals()
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
-_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "dify_agent.agent_stub.grpc._generated.agent_stub_pb2", _globals)
+_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "dify.agent.stub.v1.agent_stub_pb2", _globals)
 if not _descriptor._USE_C_DESCRIPTORS:
     DESCRIPTOR._loaded_options = None
     _globals["_CONNECTREQUEST"]._serialized_start = 59
@@ -44,9 +39,9 @@ if not _descriptor._USE_C_DESCRIPTORS:
     _globals["_FILEMAPPING"]._serialized_start = 297
     _globals["_FILEMAPPING"]._serialized_end = 399
     _globals["_FILEDOWNLOADREQUEST"]._serialized_start = 401
-    _globals["_FILEDOWNLOADREQUEST"]._serialized_end = 469
-    _globals["_FILEDOWNLOADRESPONSE"]._serialized_start = 471
-    _globals["_FILEDOWNLOADRESPONSE"]._serialized_end = 585
-    _globals["_AGENTSTUBSERVICE"]._serialized_start = 588
-    _globals["_AGENTSTUBSERVICE"]._serialized_end = 908
+    _globals["_FILEDOWNLOADREQUEST"]._serialized_end = 513
+    _globals["_FILEDOWNLOADRESPONSE"]._serialized_start = 515
+    _globals["_FILEDOWNLOADRESPONSE"]._serialized_end = 629
+    _globals["_AGENTSTUBSERVICE"]._serialized_start = 632
+    _globals["_AGENTSTUBSERVICE"]._serialized_end = 952
 # @@protoc_insertion_point(module_scope)

dify-agent/src/dify_agent/agent_stub/grpc/_generated/agent_stub_pb2.pyi

@@ -1,71 +1,69 @@
-from __future__ import annotations
+from google.protobuf.internal import containers as _containers
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from collections.abc import Iterable as _Iterable, Mapping as _Mapping
+from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union
 
-from collections.abc import Iterable
+DESCRIPTOR: _descriptor.FileDescriptor
 
-from google.protobuf.message import Message
-
-
-class ConnectRequest(Message):
+class ConnectRequest(_message.Message):
+    __slots__ = ("protocol_version", "argv", "metadata_json")
+    PROTOCOL_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ARGV_FIELD_NUMBER: _ClassVar[int]
+    METADATA_JSON_FIELD_NUMBER: _ClassVar[int]
     protocol_version: int
-    argv: list[str]
+    argv: _containers.RepeatedScalarFieldContainer[str]
     metadata_json: str
+    def __init__(self, protocol_version: _Optional[int] = ..., argv: _Optional[_Iterable[str]] = ..., metadata_json: _Optional[str] = ...) -> None: ...
 
-    def __init__(
-        self,
-        *,
-        protocol_version: int = ...,
-        argv: Iterable[str] = ...,
-        metadata_json: str = ...,
-    ) -> None: ...
-
-
-class ConnectResponse(Message):
+class ConnectResponse(_message.Message):
+    __slots__ = ("connection_id", "status")
+    CONNECTION_ID_FIELD_NUMBER: _ClassVar[int]
+    STATUS_FIELD_NUMBER: _ClassVar[int]
     connection_id: str
     status: str
+    def __init__(self, connection_id: _Optional[str] = ..., status: _Optional[str] = ...) -> None: ...
 
-    def __init__(self, *, connection_id: str = ..., status: str = ...) -> None: ...
-
-
-class FileUploadRequest(Message):
+class FileUploadRequest(_message.Message):
+    __slots__ = ("filename", "mimetype")
+    FILENAME_FIELD_NUMBER: _ClassVar[int]
+    MIMETYPE_FIELD_NUMBER: _ClassVar[int]
     filename: str
     mimetype: str
+    def __init__(self, filename: _Optional[str] = ..., mimetype: _Optional[str] = ...) -> None: ...
 
-    def __init__(self, *, filename: str = ..., mimetype: str = ...) -> None: ...
-
-
-class FileUploadResponse(Message):
+class FileUploadResponse(_message.Message):
+    __slots__ = ("upload_url",)
+    UPLOAD_URL_FIELD_NUMBER: _ClassVar[int]
     upload_url: str
+    def __init__(self, upload_url: _Optional[str] = ...) -> None: ...
 
-    def __init__(self, *, upload_url: str = ...) -> None: ...
-
-
-class FileMapping(Message):
+class FileMapping(_message.Message):
+    __slots__ = ("transfer_method", "reference", "url")
+    TRANSFER_METHOD_FIELD_NUMBER: _ClassVar[int]
+    REFERENCE_FIELD_NUMBER: _ClassVar[int]
+    URL_FIELD_NUMBER: _ClassVar[int]
     transfer_method: str
     reference: str
     url: str
+    def __init__(self, transfer_method: _Optional[str] = ..., reference: _Optional[str] = ..., url: _Optional[str] = ...) -> None: ...
 
-    def __init__(self, *, transfer_method: str = ..., reference: str = ..., url: str = ...) -> None: ...
-    def HasField(self, field_name: str) -> bool: ...
-
-
-class FileDownloadRequest(Message):
+class FileDownloadRequest(_message.Message):
+    __slots__ = ("file", "for_external")
+    FILE_FIELD_NUMBER: _ClassVar[int]
+    FOR_EXTERNAL_FIELD_NUMBER: _ClassVar[int]
     file: FileMapping
+    for_external: bool
+    def __init__(self, file: _Optional[_Union[FileMapping, _Mapping]] = ..., for_external: _Optional[bool] = ...) -> None: ...
 
-    def __init__(self, *, file: FileMapping | None = ...) -> None: ...
-
-
-class FileDownloadResponse(Message):
+class FileDownloadResponse(_message.Message):
+    __slots__ = ("filename", "mime_type", "size", "download_url")
+    FILENAME_FIELD_NUMBER: _ClassVar[int]
+    MIME_TYPE_FIELD_NUMBER: _ClassVar[int]
+    SIZE_FIELD_NUMBER: _ClassVar[int]
+    DOWNLOAD_URL_FIELD_NUMBER: _ClassVar[int]
     filename: str
     mime_type: str
     size: int
     download_url: str
-
-    def __init__(
-        self,
-        *,
-        filename: str = ...,
-        mime_type: str = ...,
-        size: int = ...,
-        download_url: str = ...,
-    ) -> None: ...
-    def HasField(self, field_name: str) -> bool: ...
+    def __init__(self, filename: _Optional[str] = ..., mime_type: _Optional[str] = ..., size: _Optional[int] = ..., download_url: _Optional[str] = ...) -> None: ...

dify-agent/src/dify_agent/agent_stub/grpc/conversions.py

@@ -98,13 +98,19 @@ def file_download_request_from_proto(message: agent_stub_pb2.FileDownloadRequest
         "reference": message.file.reference if message.file.HasField("reference") else None,
         "url": message.file.url if message.file.HasField("url") else None,
     }
-    return AgentStubFileDownloadRequest.model_validate({"file": file_mapping_kwargs})
+    return AgentStubFileDownloadRequest.model_validate(
+        {
+            "file": file_mapping_kwargs,
+            "for_external": message.for_external if message.HasField("for_external") else True,
+        }
+    )
 
 
 def proto_file_download_request(
     pb2_module,
     *,
     file: AgentStubFileMapping,
+    for_external: bool = True,
 ) -> agent_stub_pb2.FileDownloadRequest:
     """Build one protobuf file-download request from the public DTO."""
     mapping = pb2_module.FileMapping(transfer_method=file.transfer_method)
@@ -112,7 +118,9 @@ def proto_file_download_request(
         mapping.reference = file.reference
     if file.url is not None:
         mapping.url = file.url
-    return pb2_module.FileDownloadRequest(file=mapping)
+    request = pb2_module.FileDownloadRequest(file=mapping)
+    request.for_external = for_external
+    return request
 
 
 def file_download_response_from_proto(message: agent_stub_pb2.FileDownloadResponse) -> AgentStubFileDownloadResponse:

dify-agent/src/dify_agent/agent_stub/protocol/agent_stub.py

@@ -249,6 +249,7 @@ class AgentStubFileDownloadRequest(BaseModel):
     """Request body for one signed download URL allocation."""
 
     file: AgentStubFileMapping
+    for_external: bool = True
 
     model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
 

dify-agent/src/dify_agent/agent_stub/server/agent_stub_files.py

@@ -161,6 +161,8 @@ class DifyApiAgentStubFileRequestHandler:
             "invoke_from": execution_context.invoke_from,
             "file": request.file.model_dump(mode="json", exclude_none=True),
         }
+        if request.for_external is False:
+            payload["for_external"] = False
         data = await self._post_inner_api("/inner/api/download/file/request", payload)
         try:
             return AgentStubFileDownloadResponse.model_validate(data)

dify-agent/src/dify_agent/layers/shell/layer.py

@@ -141,7 +141,18 @@ shell_run script rules:
 
 Tips:
 
-- When using Python, prefer a uv script with a PEP 723 dependency header.
+- Python 3.12, uv, pip, Node.js, pnpm, and pnx are preinstalled in the local sandbox.
+- For one-off Python dependencies, prefer a uv script with a PEP 723 dependency header or:
+  `uv run --with <package> python <script-or--c>`.
+- For reusable Python CLI tools, use `uv tool install <tool>`; installed commands land in `$HOME/.local/bin`.
+  Run them by full path or add `$HOME/.local/bin` to PATH in the command that needs them.
+- `python3 -m pip install --user <package>` also installs into `$HOME/.local`; add `$HOME/.local/bin` to PATH
+  when you need console scripts.
+- For reusable Node.js CLIs, use user-level global installs:
+  `PNPM_HOME=$HOME/.local/share/pnpm PATH=$HOME/.local/share/pnpm/bin:$PATH pnpm add -g <package>`.
+  Installed commands land in `$PNPM_HOME/bin`; run them by full path or with the same PATH prefix.
+- For one-off Node.js CLIs, prefer `pnx <command> [args]`.
+- Do not install new packages into system or image tool paths such as `/usr/local`, `/usr`, or `/opt/dify-agent-tools`.
 - If you need MCP, install the MCP server in the shell environment and start that server when you use it.
 
 Example shell_run script:

dify-agent/src/dify_agent/runtime/runner.py

@@ -31,13 +31,14 @@ both the JSON-safe final output or deferred tool call and the session snapshot;
 there are no separate output or snapshot events to correlate.
 """
 
-from collections.abc import AsyncIterable, Callable
+from collections.abc import AsyncIterable, Callable, Mapping
 from collections import Counter
 from dataclasses import dataclass
 from typing import Any, Literal, Protocol, cast, runtime_checkable
 
 import httpx
 from pydantic import JsonValue, TypeAdapter
+from pydantic_ai.exceptions import ModelHTTPError
 from pydantic_ai.messages import AgentStreamEvent, PartDeltaEvent, PartStartEvent, TextPart, TextPartDelta
 from pydantic_ai.output import OutputSpec
 from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults
@@ -104,6 +105,28 @@ class AgentRunValidationError(ValueError):
     """Raised when a run request is valid JSON but cannot execute."""
 
 
+def _run_failed_error_payload(exc: Exception) -> tuple[str, str | None]:
+    """Return the public failed-run error text and structured reason."""
+    message = str(exc) or type(exc).__name__
+    reason: str | None = None
+
+    if isinstance(exc, ModelHTTPError):
+        body = exc.body
+        if isinstance(body, Mapping):
+            body_message = body.get("message")
+            if isinstance(body_message, str) and body_message:
+                message = body_message
+
+            error_type = body.get("error_type")
+            if isinstance(error_type, str) and error_type:
+                reason = error_type
+
+        if reason is None and exc.status_code == 429:
+            reason = "InvokeRateLimitError"
+
+    return message, reason
+
+
 def _has_model_layer(request: CreateRunRequest) -> bool:
     """Return whether the public composition includes the reserved model layer."""
     return any(layer.name == DIFY_AGENT_MODEL_LAYER_ID for layer in request.composition.layers)
@@ -165,8 +188,8 @@ class AgentRunRunner:
         try:
             outcome = await self._run_agent()
         except Exception as exc:
-            message = str(exc) or type(exc).__name__
-            _ = await emit_run_failed(self.sink, run_id=self.run_id, error=message)
+            message, reason = _run_failed_error_payload(exc)
+            _ = await emit_run_failed(self.sink, run_id=self.run_id, error=message, reason=reason)
             await self.sink.update_status(self.run_id, "failed", message)
             raise
 

dify-agent/src/shellctl/__init__.py

@@ -0,0 +1,124 @@
+"""Public shellctl package exports.
+
+This package stays lazy on purpose. Hot-path runtime helpers live outside the
+`shellctl` package, and importing this package root should
+not pull the full client/server/public DTO surface unless a caller explicitly
+asks for those exports.
+"""
+
+from __future__ import annotations
+
+from importlib import import_module
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from shellctl.client import (
+        ShellctlClient,
+        ShellctlClientError,
+    )
+    from shellctl.shared import (
+        DEFAULT_AUTH_TOKEN_ENV,
+        DEFAULT_BASE_URL,
+        DEFAULT_BASE_URL_ENV,
+        DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+        DEFAULT_GC_INTERVAL_SECONDS,
+        DEFAULT_IDLE_FLUSH_SECONDS,
+        DEFAULT_LIST_LIMIT,
+        DEFAULT_OUTPUT_LIMIT_BYTES,
+        DEFAULT_TERMINAL_COLS,
+        DEFAULT_TERMINAL_ROWS,
+        DEFAULT_TERMINATE_GRACE_SECONDS,
+        DEFAULT_TIMEOUT_SECONDS,
+        DeleteJobResponse,
+        HealthResponse,
+        InputJobRequest,
+        JobInfo,
+        JobResult,
+        JobStatusName,
+        JobStatusView,
+        ListJobsResponse,
+        RunJobRequest,
+        TerminalSize,
+        TerminateJobRequest,
+        WaitJobRequest,
+        generate_job_id,
+        read_output_window,
+        tail_output_window,
+    )
+
+__all__ = [
+    "DEFAULT_AUTH_TOKEN_ENV",
+    "DEFAULT_BASE_URL",
+    "DEFAULT_BASE_URL_ENV",
+    "DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS",
+    "DEFAULT_GC_INTERVAL_SECONDS",
+    "DEFAULT_IDLE_FLUSH_SECONDS",
+    "DEFAULT_LIST_LIMIT",
+    "DEFAULT_OUTPUT_LIMIT_BYTES",
+    "DEFAULT_TERMINAL_COLS",
+    "DEFAULT_TERMINAL_ROWS",
+    "DEFAULT_TERMINATE_GRACE_SECONDS",
+    "DEFAULT_TIMEOUT_SECONDS",
+    "DeleteJobResponse",
+    "HealthResponse",
+    "InputJobRequest",
+    "JobInfo",
+    "JobResult",
+    "JobStatusName",
+    "JobStatusView",
+    "ListJobsResponse",
+    "RunJobRequest",
+    "ShellctlClient",
+    "ShellctlClientError",
+    "TerminalSize",
+    "TerminateJobRequest",
+    "WaitJobRequest",
+    "generate_job_id",
+    "read_output_window",
+    "tail_output_window",
+]
+
+_EXPORTS = {
+    "ShellctlClient": "shellctl.client",
+    "ShellctlClientError": "shellctl.client",
+    "DEFAULT_AUTH_TOKEN_ENV": "shellctl.shared",
+    "DEFAULT_BASE_URL": "shellctl.shared",
+    "DEFAULT_BASE_URL_ENV": "shellctl.shared",
+    "DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS": "shellctl.shared",
+    "DEFAULT_GC_INTERVAL_SECONDS": "shellctl.shared",
+    "DEFAULT_IDLE_FLUSH_SECONDS": "shellctl.shared",
+    "DEFAULT_LIST_LIMIT": "shellctl.shared",
+    "DEFAULT_OUTPUT_LIMIT_BYTES": "shellctl.shared",
+    "DEFAULT_TERMINAL_COLS": "shellctl.shared",
+    "DEFAULT_TERMINAL_ROWS": "shellctl.shared",
+    "DEFAULT_TERMINATE_GRACE_SECONDS": "shellctl.shared",
+    "DEFAULT_TIMEOUT_SECONDS": "shellctl.shared",
+    "DeleteJobResponse": "shellctl.shared",
+    "HealthResponse": "shellctl.shared",
+    "InputJobRequest": "shellctl.shared",
+    "JobInfo": "shellctl.shared",
+    "JobResult": "shellctl.shared",
+    "JobStatusName": "shellctl.shared",
+    "JobStatusView": "shellctl.shared",
+    "ListJobsResponse": "shellctl.shared",
+    "RunJobRequest": "shellctl.shared",
+    "TerminalSize": "shellctl.shared",
+    "TerminateJobRequest": "shellctl.shared",
+    "WaitJobRequest": "shellctl.shared",
+    "generate_job_id": "shellctl.shared",
+    "read_output_window": "shellctl.shared",
+    "tail_output_window": "shellctl.shared",
+}
+
+
+def __getattr__(name: str) -> Any:
+    if name not in _EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    module = import_module(_EXPORTS[name])
+    value = getattr(module, name)  # noqa: no-new-getattr lazy export proxy
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__))

dify-agent/src/shellctl/cli.py

@@ -0,0 +1,587 @@
+"""Typer CLI for network-backed shellctl commands.
+
+Job-management commands in this module intentionally stay on the SDK side of
+the boundary: they parse CLI options, call `ShellctlClient`, and render compact
+JSON. That keeps `shellctl --help` and `shellctl run --help` free of FastAPI,
+SQLAlchemy, tmux, and local runtime bootstrap imports.
+
+Only `serve` lazily imports server-side modules when that subcommand is
+actually invoked.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Awaitable, Callable
+from pathlib import Path
+from typing import NoReturn
+
+import anyio
+import httpx2 as httpx
+import typer
+from pydantic import BaseModel, ValidationError
+
+from shellctl.client import ShellctlClient, ShellctlClientError
+from shellctl.shared.constants import (
+    DEFAULT_AUTH_TOKEN_ENV,
+    DEFAULT_BASE_URL,
+    DEFAULT_BASE_URL_ENV,
+    DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+    DEFAULT_GC_INTERVAL_SECONDS,
+    DEFAULT_IDLE_FLUSH_SECONDS,
+    DEFAULT_LIST_LIMIT,
+    DEFAULT_OUTPUT_LIMIT_BYTES,
+    DEFAULT_TERMINAL_COLS,
+    DEFAULT_TERMINAL_ROWS,
+    DEFAULT_TERMINATE_GRACE_SECONDS,
+    DEFAULT_TIMEOUT_SECONDS,
+    MAX_LIST_LIMIT,
+    MAX_OUTPUT_LIMIT_BYTES,
+)
+from shellctl.shared.schemas import (
+    DeleteJobResponse,
+    HealthResponse,
+    JobInfo,
+    JobResult,
+    JobStatusName,
+    JobStatusView,
+    RunJobRequest,
+    TerminalSize,
+)
+
+cli = typer.Typer(
+    no_args_is_help=True,
+    pretty_exceptions_enable=False,
+    rich_markup_mode=None,
+)
+
+
+@cli.command("health")
+def health_command(
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help="Accepted for CLI consistency but ignored because /healthz is public.",
+    ),
+) -> None:
+    """Call the public health endpoint and report JSON."""
+
+    del auth_token
+
+    async def action(client: ShellctlClient) -> HealthResponse:
+        return await client.health()
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=None,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("run")
+def run_command(
+    script: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    cwd: Path | None = typer.Option(None, "--cwd"),
+    env: list[str] | None = typer.Option(None, "--env"),
+    timeout: float = typer.Option(DEFAULT_TIMEOUT_SECONDS, "--timeout"),
+    output_limit: int = typer.Option(DEFAULT_OUTPUT_LIMIT_BYTES, "--output-limit"),
+    idle_flush_seconds: float = typer.Option(
+        DEFAULT_IDLE_FLUSH_SECONDS,
+        "--idle-flush-seconds",
+    ),
+    cols: int | None = typer.Option(None, "--cols"),
+    rows: int | None = typer.Option(None, "--rows"),
+) -> None:
+    """Create a job through the running shellctl server."""
+
+    request = _build_model(
+        RunJobRequest,
+        script=script,
+        cwd=str(cwd) if cwd is not None else None,
+        env=_parse_env(env),
+        terminal=_terminal_size(cols=cols, rows=rows),
+        timeout=timeout,
+        output_limit=output_limit,
+        idle_flush_seconds=idle_flush_seconds,
+    )
+
+    async def action(client: ShellctlClient) -> JobResult:
+        return await client.run(
+            request.script,
+            cwd=request.cwd,
+            env=request.env,
+            timeout=request.timeout,
+            terminal=request.terminal,
+        )
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        output_limit=output_limit,
+        idle_flush_seconds=idle_flush_seconds,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("wait")
+def wait_command(
+    job_id: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    offset: int = typer.Option(..., "--offset"),
+    timeout: float = typer.Option(DEFAULT_TIMEOUT_SECONDS, "--timeout"),
+    output_limit: int = typer.Option(DEFAULT_OUTPUT_LIMIT_BYTES, "--output-limit"),
+    idle_flush_seconds: float = typer.Option(
+        DEFAULT_IDLE_FLUSH_SECONDS,
+        "--idle-flush-seconds",
+    ),
+) -> None:
+    """Wait for incremental output, completion, truncation, or timeout."""
+
+    async def action(client: ShellctlClient) -> JobResult:
+        return await client.wait(job_id, offset=offset, timeout=timeout)
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        output_limit=output_limit,
+        idle_flush_seconds=idle_flush_seconds,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("status")
+def status_command(
+    job_id: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+) -> None:
+    """Materialize the current status view for one job."""
+
+    async def action(client: ShellctlClient) -> JobStatusView:
+        return await client.status(job_id)
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("list")
+def list_command(
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    status: JobStatusName | None = typer.Option(None, "--status"),
+    limit: int = typer.Option(DEFAULT_LIST_LIMIT, "--limit", min=1, max=MAX_LIST_LIMIT),
+) -> None:
+    """List recent jobs, optionally filtered by lifecycle status."""
+
+    async def action(client: ShellctlClient) -> list[JobInfo]:
+        return await client.list_jobs(status=status, limit=limit)
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        action=action,
+        emit=_emit_job_list,
+    )
+
+
+@cli.command("input")
+def input_command(
+    job_id: str = typer.Argument(...),
+    text: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    offset: int = typer.Option(..., "--offset"),
+    timeout: float = typer.Option(DEFAULT_TIMEOUT_SECONDS, "--timeout"),
+    output_limit: int = typer.Option(DEFAULT_OUTPUT_LIMIT_BYTES, "--output-limit"),
+    idle_flush_seconds: float = typer.Option(
+        DEFAULT_IDLE_FLUSH_SECONDS,
+        "--idle-flush-seconds",
+    ),
+) -> None:
+    """Send text input to a running job and wait for the next result window."""
+
+    async def action(client: ShellctlClient) -> JobResult:
+        return await client.input(job_id, text, offset=offset, timeout=timeout)
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        output_limit=output_limit,
+        idle_flush_seconds=idle_flush_seconds,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("tail")
+def tail_command(
+    job_id: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    output_limit: int = typer.Option(
+        DEFAULT_OUTPUT_LIMIT_BYTES,
+        "--output-limit",
+        min=1,
+        max=MAX_OUTPUT_LIMIT_BYTES,
+    ),
+) -> None:
+    """Read a UTF-8-safe output tail for one job."""
+
+    async def action(client: ShellctlClient) -> JobResult:
+        return await client.tail(job_id)
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        output_limit=output_limit,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("terminate")
+def terminate_command(
+    job_id: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    grace_seconds: float = typer.Option(
+        DEFAULT_TERMINATE_GRACE_SECONDS,
+        "--grace-seconds",
+    ),
+) -> None:
+    """Terminate a job and return its materialized status."""
+
+    async def action(client: ShellctlClient) -> JobStatusView:
+        return await client.terminate(job_id, grace_seconds=grace_seconds)
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("delete")
+def delete_command(
+    job_id: str = typer.Argument(...),
+    base_url: str = typer.Option(
+        DEFAULT_BASE_URL,
+        "--base-url",
+        envvar=DEFAULT_BASE_URL_ENV,
+        help="shellctl server base URL. You can also set SHELLCTL_BASE_URL.",
+    ),
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty when the server does not require auth."
+        ),
+    ),
+    force: bool = typer.Option(False, "--force"),
+    grace_seconds: float = typer.Option(
+        DEFAULT_TERMINATE_GRACE_SECONDS,
+        "--grace-seconds",
+    ),
+) -> None:
+    """Delete a job row and artifacts, optionally terminating first."""
+
+    async def action(client: ShellctlClient) -> DeleteJobResponse:
+        return await client.delete(
+            job_id,
+            force=force,
+            grace_seconds=grace_seconds,
+        )
+
+    _run_client_action(
+        base_url=base_url,
+        auth_token=auth_token,
+        action=action,
+        emit=_emit_model,
+    )
+
+
+@cli.command("serve")
+def serve_command(
+    listen: str = "127.0.0.1:8765",
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty to disable HTTP bearer auth."
+        ),
+    ),
+    state_dir: Path | None = None,
+    runtime_dir: Path | None = None,
+    gc_interval_seconds: float = typer.Option(
+        DEFAULT_GC_INTERVAL_SECONDS,
+        "--gc-interval-seconds",
+    ),
+    gc_finished_job_retention_seconds: float = typer.Option(
+        DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+        "--gc-finished-job-retention-seconds",
+    ),
+) -> None:
+    """Run the local shellctl FastAPI server via uvicorn."""
+
+    from shellctl.server.serve import (
+        serve_command as server_serve_command,
+    )
+
+    server_serve_command(
+        listen=listen,
+        auth_token=auth_token,
+        state_dir=state_dir,
+        runtime_dir=runtime_dir,
+        gc_interval_seconds=gc_interval_seconds,
+        gc_finished_job_retention_seconds=gc_finished_job_retention_seconds,
+    )
+
+
+def main() -> None:
+    """CLI entrypoint used by the console script and `python -m` invocations."""
+
+    cli()
+
+
+def _parse_env(values: list[str] | None) -> dict[str, str] | None:
+    if not values:
+        return None
+
+    parsed: dict[str, str] = {}
+    for value in values:
+        if "=" not in value:
+            raise typer.BadParameter(
+                "env entries must use NAME=VALUE format",
+                param_hint="--env",
+            )
+        name, env_value = value.split("=", 1)
+        if not name:
+            raise typer.BadParameter(
+                "env names must be non-empty",
+                param_hint="--env",
+            )
+        parsed[name] = env_value
+    return parsed
+
+
+def _terminal_size(*, cols: int | None, rows: int | None) -> TerminalSize | None:
+    if cols is None and rows is None:
+        return None
+    return _build_model(
+        TerminalSize,
+        cols=cols if cols is not None else DEFAULT_TERMINAL_COLS,
+        rows=rows if rows is not None else DEFAULT_TERMINAL_ROWS,
+    )
+
+
+def _build_model[ModelT: BaseModel](model_type: type[ModelT], /, **data: object) -> ModelT:
+    try:
+        return model_type(**data)
+    except ValidationError as exc:
+        raise typer.BadParameter(_validation_error_message(exc)) from exc
+
+
+async def _with_client[ResponseT](
+    base_url: str,
+    auth_token: str | None,
+    output_limit: int,
+    idle_flush_seconds: float,
+    action: Callable[[ShellctlClient], Awaitable[ResponseT]],
+) -> ResponseT:
+    async with ShellctlClient(
+        base_url,
+        output_limit=output_limit,
+        idle_flush_seconds=idle_flush_seconds,
+        token=auth_token,
+    ) as client:
+        return await action(client)
+
+
+def _run_client_action[ResponseT](
+    *,
+    base_url: str,
+    auth_token: str | None,
+    action: Callable[[ShellctlClient], Awaitable[ResponseT]],
+    emit: Callable[[ResponseT], None],
+    output_limit: int = DEFAULT_OUTPUT_LIMIT_BYTES,
+    idle_flush_seconds: float = DEFAULT_IDLE_FLUSH_SECONDS,
+) -> None:
+    try:
+        payload = anyio.run(
+            _with_client,
+            base_url,
+            auth_token,
+            output_limit,
+            idle_flush_seconds,
+            action,
+        )
+    except ShellctlClientError as exc:
+        _emit_error_and_exit(exc.code, exc.message)
+    except httpx.TimeoutException:
+        _emit_error_and_exit("request_timeout", "request timed out")
+    except httpx.TransportError as exc:
+        _emit_error_and_exit("connection_error", str(exc))
+
+    emit(payload)
+
+
+def _emit_model(model: BaseModel) -> None:
+    typer.echo(model.model_dump_json(exclude_none=True), color=False)
+
+
+def _emit_job_list(jobs: list[JobInfo]) -> None:
+    typer.echo(
+        json.dumps(
+            [item.model_dump(mode="json", exclude_none=True) for item in jobs],
+            separators=(",", ":"),
+        ),
+        color=False,
+    )
+
+
+def _emit_error_and_exit(code: str, message: str) -> NoReturn:
+    typer.echo(
+        json.dumps(
+            {"error": {"code": code, "message": message}},
+            separators=(",", ":"),
+        ),
+        err=True,
+        color=False,
+    )
+    raise typer.Exit(code=1)
+
+
+def _validation_error_message(exc: ValidationError) -> str:
+    detail = exc.errors(include_url=False)[0]
+    location = ".".join(str(part) for part in detail.get("loc", ()))
+    message = detail["msg"]
+    return f"{location}: {message}" if location else str(message)
+
+
+__all__ = [
+    "cli",
+    "delete_command",
+    "health_command",
+    "input_command",
+    "list_command",
+    "main",
+    "run_command",
+    "serve_command",
+    "status_command",
+    "tail_command",
+    "terminate_command",
+    "wait_command",
+]

dify-agent/src/shellctl/client/__init__.py

@@ -0,0 +1,12 @@
+"""Async HTTP client package for shellctl.
+
+`shellctl.client` remains importable as before, but it is
+ now a package so future client helpers can live beside the main SDK class.
+"""
+
+from shellctl.client.sdk import (
+    ShellctlClient,
+    ShellctlClientError,
+)
+
+__all__ = ["ShellctlClient", "ShellctlClientError"]

dify-agent/src/shellctl/client/sdk.py

@@ -0,0 +1,319 @@
+"""Async HTTP client for the shellctl server API.
+
+The SDK keeps transport-level knobs (`output_limit`, `idle_flush_seconds`, base
+URL selection, and bearer-token handling) on the client instance so individual
+method calls stay close to the network CLI's high-level workflow. Blocking shell
+operations reuse the shared client, but they override the HTTP read timeout per
+request so the transport does not fail before the server-side shell wait timeout
+or terminate-grace budget does.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx2 as httpx
+
+from shellctl.shared.constants import (
+    DEFAULT_AUTH_TOKEN_ENV,
+    DEFAULT_BASE_URL,
+    DEFAULT_IDLE_FLUSH_SECONDS,
+    DEFAULT_LIST_LIMIT,
+    DEFAULT_OUTPUT_LIMIT_BYTES,
+    DEFAULT_TERMINATE_GRACE_SECONDS,
+    DEFAULT_TIMEOUT_SECONDS,
+)
+from shellctl.shared.schemas import (
+    DeleteJobResponse,
+    HealthResponse,
+    JobInfo,
+    JobResult,
+    JobStatusView,
+    ListJobsResponse,
+    RunJobRequest,
+    TerminalSize,
+)
+
+
+class ShellctlClientError(RuntimeError):
+    """Raised for API-declared failures and response decode/shape problems.
+
+    `ShellctlClient` raises this error when the server returns a structured
+    error payload, and also when an otherwise successful HTTP response contains
+    invalid JSON or a top-level payload shape that does not match the SDK
+    contract. Transport and timeout failures remain raw `httpx2` exceptions so
+    library callers can decide how to handle network-layer failures.
+    """
+
+    def __init__(self, status_code: int, code: str, message: str) -> None:
+        super().__init__(f"{code} ({status_code}): {message}")
+        self.status_code = status_code
+        self.code = code
+        self.message = message
+
+
+class ShellctlClient:
+    """Thin async SDK for the shellctl HTTP API.
+
+    The client owns a reusable `httpx.AsyncClient` unless one is injected via the
+    `client` argument. Callers can therefore either keep one instance for a full
+    workflow or treat it as an async context manager. Injected clients keep their
+    original lifecycle; `close()` only closes clients that this SDK created.
+    """
+
+    def __init__(
+        self,
+        base_url: str = DEFAULT_BASE_URL,
+        *,
+        output_limit: int = DEFAULT_OUTPUT_LIMIT_BYTES,
+        idle_flush_seconds: float = DEFAULT_IDLE_FLUSH_SECONDS,
+        token: str | None = None,
+        client: httpx.AsyncClient | None = None,
+        transport: httpx.AsyncBaseTransport | None = None,
+        request_timeout_grace_seconds: float = 10.0,
+    ) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.output_limit = output_limit
+        self.idle_flush_seconds = idle_flush_seconds
+        self.request_timeout_grace_seconds = request_timeout_grace_seconds
+        self.token = token if token is not None else os.environ.get(DEFAULT_AUTH_TOKEN_ENV)
+        self._owns_client = client is None
+        self._client = client or httpx.AsyncClient(
+            base_url=self.base_url,
+            follow_redirects=True,
+            timeout=httpx.Timeout(DEFAULT_TIMEOUT_SECONDS, connect=DEFAULT_TIMEOUT_SECONDS),
+            transport=transport,
+        )
+
+    async def __aenter__(self) -> ShellctlClient:
+        return self
+
+    async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
+        await self.close()
+
+    async def close(self) -> None:
+        """Close the underlying HTTP client if this SDK instance owns it."""
+
+        if self._owns_client:
+            await self._client.aclose()
+
+    def _wait_request_timeout(self, timeout: float) -> httpx.Timeout:
+        """Return a request timeout for blocking shell calls.
+
+        The shellctl server enforces the payload timeout, while the SDK keeps a
+        small HTTP read-timeout grace so the transport can wait slightly longer
+        for that response without loosening connect/write/pool timeouts.
+        """
+
+        return httpx.Timeout(
+            connect=DEFAULT_TIMEOUT_SECONDS,
+            read=timeout + self.request_timeout_grace_seconds,
+            write=DEFAULT_TIMEOUT_SECONDS,
+            pool=DEFAULT_TIMEOUT_SECONDS,
+        )
+
+    def _terminate_request_timeout(self, grace_seconds: float | None = None) -> httpx.Timeout:
+        """Return a request timeout for terminate-style calls.
+
+        `terminate()` and forced `delete()` block until the server finishes the
+        terminate grace window, so their HTTP read timeout must cover that
+        business wait budget even when the request relies on the API default.
+        """
+
+        effective_grace_seconds = DEFAULT_TERMINATE_GRACE_SECONDS if grace_seconds is None else grace_seconds
+        return self._wait_request_timeout(effective_grace_seconds)
+
+    async def health(self) -> HealthResponse:
+        """Call the public health endpoint and decode it as `HealthResponse`."""
+
+        return HealthResponse.model_validate(await self.healthz())
+
+    async def healthz(self) -> dict[str, Any]:
+        """Call the public health endpoint without requiring auth."""
+
+        response = await self._client.get("/healthz")
+        return self._decode_response(response)
+
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        timeout: float = DEFAULT_TIMEOUT_SECONDS,
+        terminal: TerminalSize | None = None,
+    ) -> JobResult:
+        """Create a new job and wait for initial output or completion.
+
+        `cwd` and `env` preset the script's working directory and environment
+        overlay on the server side.
+        """
+
+        payload = RunJobRequest(
+            script=script,
+            cwd=cwd,
+            env=env,
+            terminal=terminal,
+            timeout=timeout,
+            output_limit=self.output_limit,
+            idle_flush_seconds=self.idle_flush_seconds,
+        )
+        response = await self._client.post(
+            "/v1/jobs/run",
+            json=payload.model_dump(mode="json", exclude_none=True),
+            headers=self._auth_headers(),
+            timeout=self._wait_request_timeout(timeout),
+        )
+        return JobResult.model_validate(self._decode_response(response))
+
+    async def wait(
+        self,
+        job_id: str,
+        *,
+        offset: int,
+        timeout: float = DEFAULT_TIMEOUT_SECONDS,
+    ) -> JobResult:
+        """Wait for incremental output, completion, truncation, or timeout."""
+
+        response = await self._client.post(
+            f"/v1/jobs/{job_id}/wait",
+            json={
+                "offset": offset,
+                "timeout": timeout,
+                "output_limit": self.output_limit,
+                "idle_flush_seconds": self.idle_flush_seconds,
+            },
+            headers=self._auth_headers(),
+            timeout=self._wait_request_timeout(timeout),
+        )
+        return JobResult.model_validate(self._decode_response(response))
+
+    async def status(self, job_id: str) -> JobStatusView:
+        """Fetch the materialized status view for one job."""
+
+        response = await self._client.get(
+            f"/v1/jobs/{job_id}",
+            headers=self._auth_headers(),
+        )
+        return JobStatusView.model_validate(self._decode_response(response))
+
+    async def list_jobs(
+        self,
+        *,
+        status: str | None = None,
+        limit: int = DEFAULT_LIST_LIMIT,
+    ) -> list[JobInfo]:
+        """List recent jobs, optionally filtered by lifecycle status."""
+
+        params: dict[str, Any] = {"limit": limit}
+        if status is not None:
+            params["status"] = status
+        response = await self._client.get(
+            "/v1/jobs",
+            params=params,
+            headers=self._auth_headers(),
+        )
+        payload = ListJobsResponse.model_validate(self._decode_response(response))
+        return payload.jobs
+
+    async def input(
+        self,
+        job_id: str,
+        text: str,
+        *,
+        offset: int,
+        timeout: float = DEFAULT_TIMEOUT_SECONDS,
+    ) -> JobResult:
+        """Send text input to a running job and then wait like `wait()`."""
+
+        response = await self._client.post(
+            f"/v1/jobs/{job_id}/input",
+            json={
+                "text": text,
+                "offset": offset,
+                "timeout": timeout,
+                "output_limit": self.output_limit,
+                "idle_flush_seconds": self.idle_flush_seconds,
+            },
+            headers=self._auth_headers(),
+            timeout=self._wait_request_timeout(timeout),
+        )
+        return JobResult.model_validate(self._decode_response(response))
+
+    async def tail(self, job_id: str) -> JobResult:
+        """Fetch an immediate UTF-8-safe tail snapshot for a job."""
+
+        response = await self._client.get(
+            f"/v1/jobs/{job_id}/log/tail",
+            params={"output_limit": self.output_limit},
+            headers=self._auth_headers(),
+        )
+        return JobResult.model_validate(self._decode_response(response))
+
+    async def terminate(
+        self,
+        job_id: str,
+        grace_seconds: float = DEFAULT_TERMINATE_GRACE_SECONDS,
+    ) -> JobStatusView:
+        """Terminate a job, waiting long enough for the grace window to finish."""
+
+        response = await self._client.post(
+            f"/v1/jobs/{job_id}/terminate",
+            json={"grace_seconds": grace_seconds},
+            headers=self._auth_headers(),
+            timeout=self._terminate_request_timeout(grace_seconds),
+        )
+        return JobStatusView.model_validate(self._decode_response(response))
+
+    async def delete(
+        self,
+        job_id: str,
+        *,
+        force: bool = False,
+        grace_seconds: float | None = None,
+    ) -> DeleteJobResponse:
+        """Delete job artifacts, optionally waiting for forced termination first."""
+
+        params: dict[str, Any] = {"force": str(force).lower()}
+        if grace_seconds is not None:
+            params["grace_seconds"] = grace_seconds
+        request_kwargs: dict[str, Any] = {
+            "params": params,
+            "headers": self._auth_headers(),
+        }
+        if force:
+            request_kwargs["timeout"] = self._terminate_request_timeout(grace_seconds)
+        response = await self._client.delete(
+            f"/v1/jobs/{job_id}",
+            **request_kwargs,
+        )
+        return DeleteJobResponse.model_validate(self._decode_response(response))
+
+    def _auth_headers(self) -> dict[str, str]:
+        if not self.token:
+            return {}
+        return {"Authorization": f"Bearer {self.token}"}
+
+    def _decode_response(self, response: httpx.Response) -> dict[str, Any]:
+        try:
+            payload = response.json()
+        except ValueError as exc:  # pragma: no cover - network/proxy corruption
+            raise ShellctlClientError(response.status_code, "invalid_json", response.text) from exc
+
+        if response.is_error:
+            error = payload.get("error") if isinstance(payload, dict) else None
+            if isinstance(error, dict):
+                code = str(error.get("code", "request_failed"))
+                message = str(error.get("message", response.text))
+            else:
+                code = "request_failed"
+                message = response.text
+            raise ShellctlClientError(response.status_code, code, message)
+
+        if not isinstance(payload, dict):
+            raise ShellctlClientError(response.status_code, "invalid_payload", response.text)
+        return payload
+
+
+__all__ = ["ShellctlClient", "ShellctlClientError"]

dify-agent/src/shellctl/py.typed

@@ -0,0 +1 @@
+

dify-agent/src/shellctl/server/__init__.py

@@ -0,0 +1,58 @@
+"""shellctl server package.
+
+The server stack sits behind lazy exports so importing the network CLI does not
+pull in FastAPI, SQLAlchemy, tmux, or the local service runtime unless a
+server-side symbol is actually used.
+"""
+
+from __future__ import annotations
+
+from importlib import import_module
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from shellctl.server.api import create_app
+    from shellctl.server.cli import (
+        cli,
+        main,
+    )
+    from shellctl.server.config import ShellctlConfig
+    from shellctl.server.db import JobRow
+    from shellctl.server.errors import ShellctlServerError
+    from shellctl.server.serve import serve_command
+    from shellctl.server.service import ShellctlService
+
+__all__ = [
+    "JobRow",
+    "ShellctlConfig",
+    "ShellctlServerError",
+    "ShellctlService",
+    "cli",
+    "create_app",
+    "main",
+    "serve_command",
+]
+
+_EXPORTS = {
+    "JobRow": "shellctl.server.db",
+    "ShellctlConfig": "shellctl.server.config",
+    "ShellctlServerError": "shellctl.server.errors",
+    "ShellctlService": "shellctl.server.service",
+    "cli": "shellctl.server.cli",
+    "create_app": "shellctl.server.api",
+    "main": "shellctl.server.cli",
+    "serve_command": "shellctl.server.serve",
+}
+
+
+def __getattr__(name: str) -> Any:
+    if name not in _EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    module = import_module(_EXPORTS[name])
+    value = getattr(module, name)  # noqa: no-new-getattr lazy export proxy
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__))

dify-agent/src/shellctl/server/__main__.py

@@ -0,0 +1,6 @@
+"""Support `python -m shellctl.server`."""
+
+from shellctl.cli import main
+
+if __name__ == "__main__":
+    main()

dify-agent/src/shellctl/server/api.py

@@ -0,0 +1,198 @@
+"""FastAPI wiring for shellctl server endpoints."""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Annotated, cast
+
+from fastapi import Depends, FastAPI, Header, Query, Request
+from fastapi.responses import JSONResponse
+
+from shellctl.server.config import ShellctlConfig
+from shellctl.server.errors import ShellctlServerError
+from shellctl.server.service import ShellctlService
+from shellctl.shared.constants import (
+    DEFAULT_HEALTH_STATUS,
+    DEFAULT_LIST_LIMIT,
+    DEFAULT_OUTPUT_LIMIT_BYTES,
+    DEFAULT_TERMINATE_GRACE_SECONDS,
+    MAX_LIST_LIMIT,
+    MAX_OUTPUT_LIMIT_BYTES,
+)
+from shellctl.shared.schemas import (
+    DeleteJobResponse,
+    ErrorDetail,
+    ErrorResponse,
+    HealthResponse,
+    InputJobRequest,
+    JobResult,
+    JobStatusName,
+    JobStatusView,
+    ListJobsResponse,
+    RunJobRequest,
+    TerminateJobRequest,
+    WaitJobRequest,
+)
+
+
+def create_app(
+    config: ShellctlConfig | None = None,
+    *,
+    service: ShellctlService | None = None,
+) -> FastAPI:
+    """Create the FastAPI application used by `shellctl serve`."""
+
+    resolved_config = config or ShellctlConfig()
+    resolved_service = service or ShellctlService(resolved_config)
+
+    @asynccontextmanager
+    async def lifespan(_app: FastAPI):
+        await resolved_service.initialize()
+        resolved_service.start_background_gc()
+        resolved_service.start_background_pipe_monitor()
+        try:
+            yield
+        finally:
+            await resolved_service.shutdown()
+
+    app = FastAPI(title="shellctl", version="0.1.0", lifespan=lifespan)
+    app.state.shellctl_service = resolved_service
+
+    @app.exception_handler(ShellctlServerError)
+    async def handle_shellctl_error(
+        _request: Request,
+        exc: ShellctlServerError,
+    ) -> JSONResponse:
+        return JSONResponse(
+            status_code=exc.status_code,
+            content=ErrorResponse(error=ErrorDetail(code=exc.code, message=exc.message)).model_dump(mode="json"),
+        )
+
+    @app.exception_handler(RuntimeError)
+    async def handle_runtime_error(_request: Request, exc: RuntimeError) -> JSONResponse:
+        return JSONResponse(
+            status_code=500,
+            content=ErrorResponse(
+                error=ErrorDetail(
+                    code="internal_error",
+                    message=str(exc) or "internal server error",
+                )
+            ).model_dump(mode="json"),
+        )
+
+    def get_service() -> ShellctlService:
+        return cast(ShellctlService, app.state.shellctl_service)
+
+    def verify_auth(
+        authorization: Annotated[str | None, Header()] = None,
+    ) -> None:
+        token = resolved_config.auth_token
+        if token is None:
+            return
+        expected = f"Bearer {token}"
+        if authorization != expected:
+            raise ShellctlServerError(401, "unauthorized", "Missing or invalid bearer token")
+
+    @app.get("/healthz", response_model=HealthResponse)
+    async def healthz() -> HealthResponse:
+        return HealthResponse(status=DEFAULT_HEALTH_STATUS)
+
+    @app.post(
+        "/v1/jobs/run",
+        response_model=JobResult,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def run_job(
+        payload: RunJobRequest,
+        svc: ShellctlService = Depends(get_service),
+    ) -> JobResult:
+        return await svc.run_job(payload)
+
+    @app.post(
+        "/v1/jobs/{job_id}/wait",
+        response_model=JobResult,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def wait_job(
+        job_id: str,
+        payload: WaitJobRequest,
+        svc: ShellctlService = Depends(get_service),
+    ) -> JobResult:
+        return await svc.wait_job(job_id, payload)
+
+    @app.get(
+        "/v1/jobs/{job_id}/log/tail",
+        response_model=JobResult,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def tail_job(
+        job_id: str,
+        output_limit: Annotated[int, Query(ge=1, le=MAX_OUTPUT_LIMIT_BYTES)] = DEFAULT_OUTPUT_LIMIT_BYTES,
+        svc: ShellctlService = Depends(get_service),
+    ) -> JobResult:
+        return await svc.tail_job(job_id, output_limit=output_limit)
+
+    @app.get(
+        "/v1/jobs/{job_id}",
+        response_model=JobStatusView,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def job_status(
+        job_id: str,
+        svc: ShellctlService = Depends(get_service),
+    ) -> JobStatusView:
+        return await svc.get_job_status(job_id)
+
+    @app.get(
+        "/v1/jobs",
+        response_model=ListJobsResponse,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def list_jobs(
+        status: Annotated[JobStatusName | None, Query()] = None,
+        limit: Annotated[int, Query(ge=1, le=MAX_LIST_LIMIT)] = DEFAULT_LIST_LIMIT,
+        svc: ShellctlService = Depends(get_service),
+    ) -> ListJobsResponse:
+        return await svc.list_jobs(status=status, limit=limit)
+
+    @app.post(
+        "/v1/jobs/{job_id}/input",
+        response_model=JobResult,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def input_job(
+        job_id: str,
+        payload: InputJobRequest,
+        svc: ShellctlService = Depends(get_service),
+    ) -> JobResult:
+        return await svc.send_input(job_id, payload)
+
+    @app.post(
+        "/v1/jobs/{job_id}/terminate",
+        response_model=JobStatusView,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def terminate_job(
+        job_id: str,
+        payload: TerminateJobRequest,
+        svc: ShellctlService = Depends(get_service),
+    ) -> JobStatusView:
+        return await svc.terminate_job(job_id, payload)
+
+    @app.delete(
+        "/v1/jobs/{job_id}",
+        response_model=DeleteJobResponse,
+        dependencies=[Depends(verify_auth)],
+    )
+    async def delete_job(
+        job_id: str,
+        force: bool = False,
+        grace_seconds: float = DEFAULT_TERMINATE_GRACE_SECONDS,
+        svc: ShellctlService = Depends(get_service),
+    ) -> DeleteJobResponse:
+        return await svc.delete_job(job_id, force=force, grace_seconds=grace_seconds)
+
+    return app
+
+
+__all__ = ["create_app"]

dify-agent/src/shellctl/server/artifacts.py

@@ -0,0 +1,62 @@
+"""Per-job artifact names used by shellctl server/runtime code.
+
+Normal job completion is coordinated through small marker files inside each
+`jobs/<job_id>/` directory so the tmux output-pipe finalizer can publish the
+SQLite `exited(exit_code, ended_at)` state only after PTY output is fully
+drained into `output.log`. The same artifact directory also stores the request's
+environment overlay so the runner can merge arbitrary key/value pairs without
+shell-escaping them into the generated script. Separate failure markers and a
+dedicated `pipe-error.log` stderr capture keep startup diagnostics available
+when the sanitizer never reaches its ready-file handshake.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+RUNNER_EXIT_CODE_FILENAME = ".runner-exit-code"
+RUNNER_ENDED_AT_FILENAME = ".runner-ended-at"
+JOB_ENV_FILENAME = ".job-env.json"
+PIPE_DRAINED_FILENAME = ".pipe-drained"
+PIPE_FAILED_FILENAME = ".pipe-failed"
+PIPE_ERROR_LOG_FILENAME = "pipe-error.log"
+
+
+def runner_exit_code_path(job_dir: Path) -> Path:
+    return job_dir / RUNNER_EXIT_CODE_FILENAME
+
+
+def runner_ended_at_path(job_dir: Path) -> Path:
+    return job_dir / RUNNER_ENDED_AT_FILENAME
+
+
+def job_env_path(job_dir: Path) -> Path:
+    return job_dir / JOB_ENV_FILENAME
+
+
+def pipe_drained_path(job_dir: Path) -> Path:
+    return job_dir / PIPE_DRAINED_FILENAME
+
+
+def pipe_failed_path(job_dir: Path) -> Path:
+    return job_dir / PIPE_FAILED_FILENAME
+
+
+def pipe_error_log_path(job_dir: Path) -> Path:
+    return job_dir / PIPE_ERROR_LOG_FILENAME
+
+
+__all__ = [
+    "JOB_ENV_FILENAME",
+    "PIPE_DRAINED_FILENAME",
+    "PIPE_ERROR_LOG_FILENAME",
+    "PIPE_FAILED_FILENAME",
+    "RUNNER_ENDED_AT_FILENAME",
+    "RUNNER_EXIT_CODE_FILENAME",
+    "job_env_path",
+    "pipe_drained_path",
+    "pipe_error_log_path",
+    "pipe_failed_path",
+    "runner_ended_at_path",
+    "runner_exit_code_path",
+]

dify-agent/src/shellctl/server/cli.py

@@ -0,0 +1,17 @@
+"""Compatibility shim for historical `shellctl.server.cli` imports.
+
+Job-management commands now live in `shellctl.cli`. This
+module is intentionally a thin legacy re-export for callers that still import
+CLI symbols from `shellctl.server.cli`.
+"""
+
+from __future__ import annotations
+
+from shellctl.cli import cli, main
+from shellctl.server.serve import serve_command
+
+__all__ = [
+    "cli",
+    "main",
+    "serve_command",
+]

dify-agent/src/shellctl/server/config.py

@@ -0,0 +1,105 @@
+"""Configuration objects for shellctl server/runtime modules."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import cast
+
+from shellctl.shared.constants import (
+    DEFAULT_AUTH_TOKEN_ENV,
+    DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+    DEFAULT_GC_INTERVAL_SECONDS,
+    DEFAULT_IDLE_FLUSH_SECONDS,
+    DEFAULT_LIST_LIMIT,
+    DEFAULT_OUTPUT_LIMIT_BYTES,
+    DEFAULT_TERMINAL_COLS,
+    DEFAULT_TERMINAL_ROWS,
+    DEFAULT_TERMINATE_GRACE_SECONDS,
+    DEFAULT_TIMEOUT_SECONDS,
+    MAX_LIST_LIMIT,
+    MAX_OUTPUT_LIMIT_BYTES,
+    MAX_WAIT_TIMEOUT_SECONDS,
+)
+from shellctl.shared.runtime import (
+    default_runtime_dir,
+    default_state_dir,
+)
+from shellctl_runtime.paths import DEFAULT_SQLITE_BUSY_TIMEOUT_MS
+
+
+@dataclass(slots=True, frozen=True)
+class ShellctlConfig:
+    """Runtime configuration for the shellctl service and CLI.
+
+    The tmux subprocess hooks use dedicated console scripts instead of
+    `python -m shellctl...` entrypoints so every job does not pay
+    the shellctl client/server import cost just to sanitize PTY bytes or record
+    an exit row. `sqlite_busy_timeout_ms` still applies to the out-of-process
+    `runner-exit` callback, so non-default deployments keep one SQLite timeout
+    policy across the service and tmux finalizer.
+
+    Bearer auth is opt-in: if the explicit `auth_token` and the fallback
+    `SHELLCTL_AUTH_TOKEN` environment variable are both missing or empty,
+    `shellctl serve` accepts requests without checking an Authorization header.
+    """
+
+    listen: str = "127.0.0.1:8765"
+    auth_token: str | None = None
+    state_dir: Path = field(default_factory=default_state_dir)
+    runtime_dir: Path | None = None
+    gc_interval_seconds: float = DEFAULT_GC_INTERVAL_SECONDS
+    gc_finished_job_retention_seconds: float = DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS
+    default_timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS
+    max_wait_timeout_seconds: float = MAX_WAIT_TIMEOUT_SECONDS
+    idle_flush_seconds: float = DEFAULT_IDLE_FLUSH_SECONDS
+    default_cwd: Path = field(default_factory=Path.home)
+    default_terminal_cols: int = DEFAULT_TERMINAL_COLS
+    default_terminal_rows: int = DEFAULT_TERMINAL_ROWS
+    default_list_limit: int = DEFAULT_LIST_LIMIT
+    max_list_limit: int = MAX_LIST_LIMIT
+    default_output_limit_bytes: int = DEFAULT_OUTPUT_LIMIT_BYTES
+    max_output_limit_bytes: int = MAX_OUTPUT_LIMIT_BYTES
+    default_terminate_grace_seconds: float = DEFAULT_TERMINATE_GRACE_SECONDS
+    poll_interval_seconds: float = 0.05
+    pipe_monitor_interval_seconds: float = 1.0
+    pipe_ready_timeout_seconds: float = 10.0
+    sqlite_busy_timeout_ms: int = DEFAULT_SQLITE_BUSY_TIMEOUT_MS
+    sanitize_pty_command: tuple[str, ...] = ("shellctl-sanitize-pty",)
+    runner_exit_command: tuple[str, ...] = ("shellctl-runner-exit",)
+
+    def __post_init__(self) -> None:
+        if self.runtime_dir is None:
+            object.__setattr__(self, "runtime_dir", default_runtime_dir(self.state_dir))
+        token = self.auth_token
+        if token is None:
+            token = os.environ.get(DEFAULT_AUTH_TOKEN_ENV)
+        if not token:
+            token = None
+        object.__setattr__(self, "auth_token", token)
+
+    @property
+    def jobs_dir(self) -> Path:
+        return self.state_dir / "jobs"
+
+    @property
+    def db_path(self) -> Path:
+        return self.state_dir / "shellctl.db"
+
+    @property
+    def database_url(self) -> str:
+        return f"sqlite+aiosqlite:///{self.db_path}"
+
+    @property
+    def tmux_socket(self) -> Path:
+        runtime_dir = cast(Path, self.runtime_dir)
+        return runtime_dir / "tmux.sock"
+
+    @property
+    def runner_path(self) -> Path:
+        runtime_dir = cast(Path, self.runtime_dir)
+        return runtime_dir / "bin" / "shellctl-runner"
+
+
+__all__ = ["ShellctlConfig"]

dify-agent/src/shellctl/server/db.py

@@ -0,0 +1,52 @@
+"""SQLite models and engine helpers for shellctl."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+from sqlalchemy import event
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlmodel import Field, SQLModel
+
+
+class JobRow(SQLModel, table=True):
+    """SQLite source-of-truth row for shellctl jobs.
+
+    The table intentionally combines immutable metadata, mutable lifecycle state,
+    and exit facts so state transitions can be expressed as single conditional
+    `UPDATE` statements without synchronizing separate records.
+    """
+
+    __tablename__ = cast(Any, "jobs")
+
+    job_id: str = Field(primary_key=True)
+    script_path: str
+    output_path: str
+    cwd: str
+    terminal_cols: int
+    terminal_rows: int
+    status: str = Field(index=True)
+    session_name: str
+    pane_target: str
+    exit_code: int | None = Field(default=None, nullable=True)
+    reason: str | None = Field(default=None, nullable=True)
+    message: str | None = Field(default=None, nullable=True)
+    created_at: str = Field(index=True)
+    started_at: str | None = Field(default=None, nullable=True)
+    ended_at: str | None = Field(default=None, nullable=True, index=True)
+    updated_at: str
+
+
+def configure_sqlite_engine(engine: AsyncEngine, *, busy_timeout_ms: int) -> None:
+    """Install SQLite pragmas required by the proposal's concurrency model."""
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def _set_pragmas(dbapi_connection: Any, _connection_record: Any) -> None:
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA journal_mode=WAL")
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.execute(f"PRAGMA busy_timeout={busy_timeout_ms}")
+        cursor.close()
+
+
+__all__ = ["JobRow", "configure_sqlite_engine"]

dify-agent/src/shellctl/server/errors.py

@@ -0,0 +1,14 @@
+"""Shared exception types for shellctl server-side modules."""
+
+
+class ShellctlServerError(RuntimeError):
+    """Structured server-side error that maps directly to API responses."""
+
+    def __init__(self, status_code: int, code: str, message: str) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.code = code
+        self.message = message
+
+
+__all__ = ["ShellctlServerError"]

dify-agent/src/shellctl/server/serve.py

@@ -0,0 +1,103 @@
+"""Server-side `shellctl serve` implementation.
+
+This module stays separate from the top-level CLI so ordinary client commands
+do not import the FastAPI or uvicorn stack. `serve_command()` performs those
+imports lazily because only the long-running server path needs them.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import typer
+
+from shellctl.shared.constants import (
+    DEFAULT_AUTH_TOKEN_ENV,
+    DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+    DEFAULT_GC_INTERVAL_SECONDS,
+)
+from shellctl.shared.runtime import (
+    default_state_dir,
+)
+
+if TYPE_CHECKING:
+    from shellctl.server.config import ShellctlConfig
+
+
+def serve_command(
+    listen: str = "127.0.0.1:8765",
+    auth_token: str | None = typer.Option(
+        None,
+        "--auth-token",
+        envvar=DEFAULT_AUTH_TOKEN_ENV,
+        help=(
+            "Bearer token value. You can also set SHELLCTL_AUTH_TOKEN. "
+            "Leave it unset or empty to disable HTTP bearer auth."
+        ),
+    ),
+    state_dir: Path | None = None,
+    runtime_dir: Path | None = None,
+    gc_interval_seconds: float = DEFAULT_GC_INTERVAL_SECONDS,
+    gc_finished_job_retention_seconds: float = DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+) -> None:
+    """Build `ShellctlConfig` from CLI inputs and run the local HTTP server.
+
+    Args:
+        listen: Host/port pair for the uvicorn listener in `host:port` form.
+        auth_token: Optional bearer token value. An explicit empty string
+            disables HTTP auth, an explicit non-empty token enables it, and an
+            omitted/`None` value may still resolve from `SHELLCTL_AUTH_TOKEN`
+            through the Typer env var binding or `ShellctlConfig` fallback.
+        state_dir: Persistent shellctl state directory; defaults to the shared
+            XDG-style state path when omitted.
+        runtime_dir: Optional runtime directory override for tmux/runtime
+            artifacts.
+        gc_interval_seconds: Background GC wake-up cadence for finished jobs.
+        gc_finished_job_retention_seconds: Retention window before finished jobs
+            are eligible for GC.
+
+    This entrypoint is the only CLI path that should pull in the FastAPI and
+    uvicorn stack. It parses the listener, constructs `ShellctlConfig`, and
+    then hands the configured app to uvicorn.
+    """
+
+    from shellctl.server.config import ShellctlConfig
+
+    host, port = _parse_listen(listen)
+    config = ShellctlConfig(
+        listen=listen,
+        auth_token=auth_token,
+        state_dir=state_dir or default_state_dir(),
+        runtime_dir=runtime_dir,
+        gc_interval_seconds=gc_interval_seconds,
+        gc_finished_job_retention_seconds=gc_finished_job_retention_seconds,
+    )
+    _uvicorn_run(_create_app(config), host=host, port=port, log_level="info")
+
+
+def _parse_listen(value: str) -> tuple[str, int]:
+    if ":" not in value:
+        raise typer.BadParameter("listen must use host:port format")
+    host, raw_port = value.rsplit(":", 1)
+    host = host.strip("[]")
+    try:
+        port = int(raw_port)
+    except ValueError as exc:
+        raise typer.BadParameter(f"invalid port: {raw_port}") from exc
+    return host, port
+
+
+def _create_app(config: ShellctlConfig) -> Any:
+    from shellctl.server.api import create_app
+
+    return create_app(config)
+
+
+def _uvicorn_run(app: Any, *, host: str, port: int, log_level: str) -> None:
+    import uvicorn
+
+    uvicorn.run(app, host=host, port=port, log_level=log_level)
+
+
+__all__ = ["serve_command"]

dify-agent/src/shellctl/server/tmux.py

@@ -0,0 +1,343 @@
+"""tmux control layer for shellctl jobs.
+
+The rest of shellctl treats this module as the only place that knows tmux CLI
+ command shapes. Tests can replace `TmuxControllerProtocol` with a fake to
+ exercise SQLite/output semantics without depending on a local tmux daemon.
+"""
+
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Protocol, cast
+
+import anyio
+
+from shellctl.server.artifacts import (
+    pipe_drained_path,
+    pipe_error_log_path,
+    pipe_failed_path,
+    runner_ended_at_path,
+    runner_exit_code_path,
+)
+from shellctl.server.config import ShellctlConfig
+from shellctl.server.errors import ShellctlServerError
+from shellctl.shared.runtime import (
+    job_pane_target,
+    job_session_name,
+)
+from shellctl.shared.schemas import TerminalSize
+
+
+class TmuxControllerProtocol(Protocol):
+    """Protocol used by `ShellctlService` for tmux interactions."""
+
+    async def start_server(self) -> None: ...
+
+    async def list_sessions(self) -> set[str]: ...
+
+    async def session_exists(self, session_name: str) -> bool: ...
+
+    async def is_output_pipe_active(self, *, job_id: str) -> bool | None: ...
+
+    async def create_job_session(
+        self,
+        *,
+        job_id: str,
+        job_dir: Path,
+        cwd: Path,
+        terminal: TerminalSize,
+    ) -> None: ...
+
+    async def enable_output_pipe(self, *, job_id: str, job_dir: Path, ready_file: Path) -> None: ...
+
+    async def send_input(self, *, job_id: str, text: str) -> None: ...
+
+    async def send_interrupt(self, *, job_id: str) -> None: ...
+
+    async def cleanup_session(self, *, job_id: str) -> None: ...
+
+
+class TmuxController:
+    """Best-effort wrapper around a dedicated tmux socket.
+
+    The controller always clears `TMUX` from the child environment and always
+    passes `-S <socket>` so shellctl sessions stay isolated from the user's
+    default tmux server.
+    """
+
+    def __init__(self, config: ShellctlConfig) -> None:
+        self._config = config
+
+    async def start_server(self) -> None:
+        await self._run_tmux("start-server")
+
+    async def list_sessions(self) -> set[str]:
+        result = await self._run_tmux("list-sessions", "-F", "#{session_name}", check=False)
+        if result.returncode != 0:
+            stderr = result.stderr.decode("utf-8", errors="replace")
+            if _tmux_target_missing(stderr):
+                return set()
+            raise ShellctlServerError(500, "tmux_error", stderr.strip() or "tmux list-sessions failed")
+        output = result.stdout.decode("utf-8", errors="replace")
+        return {line.strip() for line in output.splitlines() if line.strip()}
+
+    async def session_exists(self, session_name: str) -> bool:
+        return session_name in await self.list_sessions()
+
+    async def is_output_pipe_active(self, *, job_id: str) -> bool | None:
+        result = await self._run_tmux(
+            "display-message",
+            "-p",
+            "-t",
+            job_pane_target(job_id),
+            "#{pane_pipe}",
+            check=False,
+        )
+        if result.returncode != 0:
+            stderr = result.stderr.decode("utf-8", errors="replace")
+            if _tmux_target_missing(stderr):
+                return None
+            raise ShellctlServerError(
+                500,
+                "tmux_error",
+                stderr.strip() or f"Failed to inspect output pipe for {job_id}",
+            )
+        return result.stdout.decode("utf-8", errors="replace").strip() == "1"
+
+    async def create_job_session(
+        self,
+        *,
+        job_id: str,
+        job_dir: Path,
+        cwd: Path,
+        terminal: TerminalSize,
+    ) -> None:
+        runner_command = self._shell_join(
+            [
+                str(self._config.runner_path),
+                str(job_dir),
+                job_id,
+                str(cwd),
+            ]
+        )
+        result = await self._run_tmux(
+            "-f",
+            "/dev/null",
+            "new-session",
+            "-d",
+            "-s",
+            job_session_name(job_id),
+            "-x",
+            str(terminal.cols),
+            "-y",
+            str(terminal.rows),
+            runner_command,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise ShellctlServerError(
+                500,
+                "tmux_new_session_failed",
+                result.stderr.decode("utf-8", errors="replace").strip()
+                or f"Failed to create tmux session for {job_id}",
+            )
+
+    async def enable_output_pipe(self, *, job_id: str, job_dir: Path, ready_file: Path) -> None:
+        output_command = self._pipe_command_source(
+            job_id=job_id,
+            job_dir=job_dir,
+            ready_file=ready_file,
+        )
+        result = await self._run_tmux(
+            "pipe-pane",
+            "-o",
+            "-t",
+            job_pane_target(job_id),
+            output_command,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise ShellctlServerError(
+                500,
+                "pipe_failed",
+                result.stderr.decode("utf-8", errors="replace").strip() or f"Failed to attach output pipe for {job_id}",
+            )
+
+    def _pipe_command_source(self, *, job_id: str, job_dir: Path, ready_file: Path) -> str:
+        """Build the tmux `pipe-pane` command that drains and finalizes output.
+
+        For normal exits, the runner now records completion metadata into job
+        artifacts and the pipe finalizer commits `runner-exit` only after
+        the lightweight sanitizer reaches EOF and flushes `output.log`
+        successfully. Sanitizer stderr is captured into `pipe-error.log` so
+        startup timeouts can distinguish slow imports from subprocess crashes.
+        If the follow-up `runner-exit` write fails, the drain marker remains in
+        place and stderr is appended to the same log with an explicit status
+        line. The pipe still exits with the sanitizer status so a drained job is
+        not misclassified as `pipe_failed` before reconciliation can recover the
+        SQLite write from the drained artifacts.
+        """
+
+        sanitize_command = self._shell_join(
+            (
+                *self._config.sanitize_pty_command,
+                "--ready-file",
+                str(ready_file),
+            )
+        )
+        runner_exit_command = self._shell_join(
+            (
+                *self._config.runner_exit_command,
+                "--state-dir",
+                str(self._config.state_dir),
+                "--job-id",
+                job_id,
+                "--sqlite-busy-timeout-ms",
+                str(self._config.sqlite_busy_timeout_ms),
+            )
+        )
+        output_path = shlex.quote(str(job_dir / "output.log"))
+        drained_path = shlex.quote(str(pipe_drained_path(job_dir)))
+        error_log_path = shlex.quote(str(pipe_error_log_path(job_dir)))
+        failed_path = shlex.quote(str(pipe_failed_path(job_dir)))
+        exit_code_path = shlex.quote(str(runner_exit_code_path(job_dir)))
+        ended_at_path = shlex.quote(str(runner_ended_at_path(job_dir)))
+        return " ; ".join(
+            [
+                f"{sanitize_command} >> {output_path} 2> {error_log_path}",
+                "sanitize_status=$?",
+                "runner_exit_status=0",
+                (
+                    'if [ "$sanitize_status" -eq 0 ]; then '
+                    f": > {drained_path}; "
+                    f"if [ -s {exit_code_path} ] && [ -s {ended_at_path} ]; then "
+                    f'{runner_exit_command} --exit-code "$(cat {exit_code_path})" '
+                    f'--ended-at "$(cat {ended_at_path})" 2>> {error_log_path}; '
+                    "runner_exit_status=$?; "
+                    'if [ "$runner_exit_status" -ne 0 ]; then '
+                    f"printf 'runner-exit failed with status %s\\n' \"$runner_exit_status\" >> {error_log_path}; "
+                    "fi; fi; "
+                    f"else : > {failed_path}; fi"
+                ),
+                'if [ "$sanitize_status" -ne 0 ]; then exit "$sanitize_status"; fi',
+                'exit "$sanitize_status"',
+            ]
+        )
+
+    async def send_input(self, *, job_id: str, text: str) -> None:
+        buffer_name = f"shellctl-in-{job_id}"
+        runtime_dir = cast(Path, self._config.runtime_dir)
+        runtime_dir.mkdir(parents=True, exist_ok=True)
+        fd, tmp_name = tempfile.mkstemp(prefix=f"shellctl-input-{job_id}-", dir=runtime_dir)
+        tmp_path = Path(tmp_name)
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                handle.write(text)
+            load_result = await self._run_tmux(
+                "load-buffer",
+                "-b",
+                buffer_name,
+                str(tmp_path),
+                check=False,
+            )
+            if load_result.returncode != 0:
+                stderr = load_result.stderr.decode("utf-8", errors="replace").strip()
+                if _tmux_target_missing(stderr):
+                    raise ShellctlServerError(
+                        409,
+                        "tmux_target_missing",
+                        stderr or f"The tmux pane for {job_id} is no longer available",
+                    )
+                raise ShellctlServerError(
+                    500,
+                    "tmux_input_failed",
+                    stderr or f"Failed to load input buffer for {job_id}",
+                )
+            paste_result = await self._run_tmux(
+                "paste-buffer",
+                "-t",
+                job_pane_target(job_id),
+                "-b",
+                buffer_name,
+                check=False,
+            )
+            if paste_result.returncode != 0:
+                stderr = paste_result.stderr.decode("utf-8", errors="replace").strip()
+                if _tmux_target_missing(stderr):
+                    raise ShellctlServerError(
+                        409,
+                        "tmux_target_missing",
+                        stderr or f"The tmux pane for {job_id} is no longer available",
+                    )
+                raise ShellctlServerError(
+                    500,
+                    "tmux_input_failed",
+                    stderr or f"Failed to paste input buffer for {job_id}",
+                )
+        finally:
+            await self._run_tmux("delete-buffer", "-b", buffer_name, check=False)
+            tmp_path.unlink(missing_ok=True)
+
+    async def send_interrupt(self, *, job_id: str) -> None:
+        await self._run_tmux(
+            "send-keys",
+            "-t",
+            job_pane_target(job_id),
+            "C-c",
+            check=False,
+        )
+
+    async def cleanup_session(self, *, job_id: str) -> None:
+        await self._run_tmux(
+            "kill-session",
+            "-t",
+            job_session_name(job_id),
+            check=False,
+        )
+
+    async def _run_tmux(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[bytes]:
+        env = dict(os.environ)
+        env.pop("TMUX", None)
+        try:
+            result = await anyio.run_process(
+                ["tmux", "-S", str(self._config.tmux_socket), *args],
+                env=env,
+                check=False,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+        except FileNotFoundError as exc:
+            raise ShellctlServerError(500, "tmux_not_installed", "tmux executable was not found") from exc
+        if check and result.returncode != 0:
+            raise ShellctlServerError(
+                500,
+                "tmux_error",
+                result.stderr.decode("utf-8", errors="replace").strip() or "tmux command failed",
+            )
+        return result
+
+    @staticmethod
+    def _shell_join(parts: tuple[str, ...] | list[str]) -> str:
+        return " ".join(shlex.quote(part) for part in parts)
+
+
+def _tmux_target_missing(stderr: str) -> bool:
+    normalized = stderr.lower()
+    return (
+        "can't find pane" in normalized
+        or "can't find session" in normalized
+        or "no server running" in normalized
+        or "failed to connect" in normalized
+        or "server exited unexpectedly" in normalized
+    )
+
+
+__all__ = [
+    "TmuxController",
+    "TmuxControllerProtocol",
+    "_tmux_target_missing",
+]

dify-agent/src/shellctl/shared/__init__.py

@@ -0,0 +1,182 @@
+"""Shared shellctl transport/runtime helpers.
+
+This package preserves the historical import surface while keeping the package
+root lazy. Lightweight callers can import concrete submodules such as
+`shared.runtime` without eagerly importing the pydantic schema layer, output
+or helpers outside the shared compatibility surface.
+"""
+
+from __future__ import annotations
+
+from importlib import import_module
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from shellctl.shared.constants import (
+        DEFAULT_AUTH_TOKEN_ENV,
+        DEFAULT_BASE_URL,
+        DEFAULT_BASE_URL_ENV,
+        DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS,
+        DEFAULT_GC_INTERVAL_SECONDS,
+        DEFAULT_HEALTH_STATUS,
+        DEFAULT_IDLE_FLUSH_SECONDS,
+        DEFAULT_LIST_LIMIT,
+        DEFAULT_OUTPUT_LIMIT_BYTES,
+        DEFAULT_TERMINAL_COLS,
+        DEFAULT_TERMINAL_ROWS,
+        DEFAULT_TERMINATE_GRACE_SECONDS,
+        DEFAULT_TIMEOUT_SECONDS,
+        JOB_ID_ALPHABET,
+        JOB_ID_RANDOM_SUFFIX_LENGTH,
+        MAX_LIST_LIMIT,
+        MAX_OUTPUT_LIMIT_BYTES,
+        MAX_WAIT_TIMEOUT_SECONDS,
+        SESSION_NAME_PREFIX,
+    )
+    from shellctl.shared.output import (
+        OutputWindow,
+        read_output_window,
+        tail_output_window,
+    )
+    from shellctl.shared.runtime import (
+        default_runtime_dir,
+        default_state_dir,
+        format_timestamp,
+        generate_job_id,
+        is_terminal_status,
+        job_pane_target,
+        job_session_name,
+        parse_timestamp,
+        utc_now,
+    )
+    from shellctl.shared.schemas import (
+        TERMINAL_JOB_STATUSES,
+        DeleteJobResponse,
+        ErrorDetail,
+        ErrorResponse,
+        HealthResponse,
+        InputJobRequest,
+        JobInfo,
+        JobResult,
+        JobStatusName,
+        JobStatusView,
+        ListJobsResponse,
+        RunJobRequest,
+        ShellctlModel,
+        TerminalSize,
+        TerminateJobRequest,
+        WaitJobRequest,
+    )
+
+__all__ = [
+    "DEFAULT_AUTH_TOKEN_ENV",
+    "DEFAULT_BASE_URL",
+    "DEFAULT_BASE_URL_ENV",
+    "DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS",
+    "DEFAULT_GC_INTERVAL_SECONDS",
+    "DEFAULT_HEALTH_STATUS",
+    "DEFAULT_IDLE_FLUSH_SECONDS",
+    "DEFAULT_LIST_LIMIT",
+    "DEFAULT_OUTPUT_LIMIT_BYTES",
+    "DEFAULT_TERMINAL_COLS",
+    "DEFAULT_TERMINAL_ROWS",
+    "DEFAULT_TERMINATE_GRACE_SECONDS",
+    "DEFAULT_TIMEOUT_SECONDS",
+    "JOB_ID_ALPHABET",
+    "JOB_ID_RANDOM_SUFFIX_LENGTH",
+    "MAX_LIST_LIMIT",
+    "MAX_OUTPUT_LIMIT_BYTES",
+    "MAX_WAIT_TIMEOUT_SECONDS",
+    "SESSION_NAME_PREFIX",
+    "TERMINAL_JOB_STATUSES",
+    "DeleteJobResponse",
+    "ErrorDetail",
+    "ErrorResponse",
+    "HealthResponse",
+    "InputJobRequest",
+    "JobInfo",
+    "JobResult",
+    "JobStatusName",
+    "JobStatusView",
+    "ListJobsResponse",
+    "OutputWindow",
+    "RunJobRequest",
+    "ShellctlModel",
+    "TerminalSize",
+    "TerminateJobRequest",
+    "WaitJobRequest",
+    "default_runtime_dir",
+    "default_state_dir",
+    "format_timestamp",
+    "generate_job_id",
+    "is_terminal_status",
+    "job_pane_target",
+    "job_session_name",
+    "parse_timestamp",
+    "read_output_window",
+    "tail_output_window",
+    "utc_now",
+]
+
+_EXPORTS = {
+    "DEFAULT_AUTH_TOKEN_ENV": "shellctl.shared.constants",
+    "DEFAULT_BASE_URL": "shellctl.shared.constants",
+    "DEFAULT_BASE_URL_ENV": "shellctl.shared.constants",
+    "DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS": "shellctl.shared.constants",
+    "DEFAULT_GC_INTERVAL_SECONDS": "shellctl.shared.constants",
+    "DEFAULT_HEALTH_STATUS": "shellctl.shared.constants",
+    "DEFAULT_IDLE_FLUSH_SECONDS": "shellctl.shared.constants",
+    "DEFAULT_LIST_LIMIT": "shellctl.shared.constants",
+    "DEFAULT_OUTPUT_LIMIT_BYTES": "shellctl.shared.constants",
+    "DEFAULT_TERMINAL_COLS": "shellctl.shared.constants",
+    "DEFAULT_TERMINAL_ROWS": "shellctl.shared.constants",
+    "DEFAULT_TERMINATE_GRACE_SECONDS": "shellctl.shared.constants",
+    "DEFAULT_TIMEOUT_SECONDS": "shellctl.shared.constants",
+    "JOB_ID_ALPHABET": "shellctl.shared.constants",
+    "JOB_ID_RANDOM_SUFFIX_LENGTH": "shellctl.shared.constants",
+    "MAX_LIST_LIMIT": "shellctl.shared.constants",
+    "MAX_OUTPUT_LIMIT_BYTES": "shellctl.shared.constants",
+    "MAX_WAIT_TIMEOUT_SECONDS": "shellctl.shared.constants",
+    "SESSION_NAME_PREFIX": "shellctl.shared.constants",
+    "OutputWindow": "shellctl.shared.output",
+    "read_output_window": "shellctl.shared.output",
+    "tail_output_window": "shellctl.shared.output",
+    "default_runtime_dir": "shellctl.shared.runtime",
+    "default_state_dir": "shellctl.shared.runtime",
+    "format_timestamp": "shellctl.shared.runtime",
+    "generate_job_id": "shellctl.shared.runtime",
+    "is_terminal_status": "shellctl.shared.runtime",
+    "job_pane_target": "shellctl.shared.runtime",
+    "job_session_name": "shellctl.shared.runtime",
+    "parse_timestamp": "shellctl.shared.runtime",
+    "utc_now": "shellctl.shared.runtime",
+    "TERMINAL_JOB_STATUSES": "shellctl.shared.schemas",
+    "DeleteJobResponse": "shellctl.shared.schemas",
+    "ErrorDetail": "shellctl.shared.schemas",
+    "ErrorResponse": "shellctl.shared.schemas",
+    "HealthResponse": "shellctl.shared.schemas",
+    "InputJobRequest": "shellctl.shared.schemas",
+    "JobInfo": "shellctl.shared.schemas",
+    "JobResult": "shellctl.shared.schemas",
+    "JobStatusName": "shellctl.shared.schemas",
+    "JobStatusView": "shellctl.shared.schemas",
+    "ListJobsResponse": "shellctl.shared.schemas",
+    "RunJobRequest": "shellctl.shared.schemas",
+    "ShellctlModel": "shellctl.shared.schemas",
+    "TerminalSize": "shellctl.shared.schemas",
+    "TerminateJobRequest": "shellctl.shared.schemas",
+    "WaitJobRequest": "shellctl.shared.schemas",
+}
+
+
+def __getattr__(name: str) -> Any:
+    if name not in _EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    module = import_module(_EXPORTS[name])
+    value = getattr(module, name)  # noqa: no-new-getattr lazy export proxy
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__))

dify-agent/src/shellctl/shared/constants.py

@@ -0,0 +1,49 @@
+"""Shared shellctl constants.
+
+This module intentionally contains only literal defaults and bounds so callers
+can import stable configuration values without pulling in the heavier DTO,
+sanitize, or server packages. The network CLI depends on this file for its
+base URL and auth-token environment contract, so keep it import-light.
+"""
+
+DEFAULT_AUTH_TOKEN_ENV = "SHELLCTL_AUTH_TOKEN"
+DEFAULT_BASE_URL_ENV = "SHELLCTL_BASE_URL"
+DEFAULT_BASE_URL = "http://127.0.0.1:8765"
+DEFAULT_OUTPUT_LIMIT_BYTES = 1024 * 8
+MAX_OUTPUT_LIMIT_BYTES = 1024 * 1024
+DEFAULT_TIMEOUT_SECONDS = 30.0
+MAX_WAIT_TIMEOUT_SECONDS = 5.0 * 60.0
+DEFAULT_IDLE_FLUSH_SECONDS = 0.5
+DEFAULT_TERMINATE_GRACE_SECONDS = 5.0
+DEFAULT_TERMINAL_COLS = 120
+DEFAULT_TERMINAL_ROWS = 80
+DEFAULT_LIST_LIMIT = 100
+MAX_LIST_LIMIT = 1000
+DEFAULT_GC_INTERVAL_SECONDS = 10.0 * 60.0
+DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS = 24.0 * 60.0 * 60.0
+JOB_ID_ALPHABET = "0123456789abcdefghjkmnpqrstvwxyz"
+JOB_ID_RANDOM_SUFFIX_LENGTH = 3
+SESSION_NAME_PREFIX = "shellctl-job-"
+DEFAULT_HEALTH_STATUS = "ok"
+
+__all__ = [
+    "DEFAULT_AUTH_TOKEN_ENV",
+    "DEFAULT_BASE_URL",
+    "DEFAULT_BASE_URL_ENV",
+    "DEFAULT_GC_FINISHED_JOB_RETENTION_SECONDS",
+    "DEFAULT_GC_INTERVAL_SECONDS",
+    "DEFAULT_HEALTH_STATUS",
+    "DEFAULT_IDLE_FLUSH_SECONDS",
+    "DEFAULT_LIST_LIMIT",
+    "DEFAULT_OUTPUT_LIMIT_BYTES",
+    "DEFAULT_TERMINAL_COLS",
+    "DEFAULT_TERMINAL_ROWS",
+    "DEFAULT_TERMINATE_GRACE_SECONDS",
+    "DEFAULT_TIMEOUT_SECONDS",
+    "JOB_ID_ALPHABET",
+    "JOB_ID_RANDOM_SUFFIX_LENGTH",
+    "MAX_LIST_LIMIT",
+    "MAX_OUTPUT_LIMIT_BYTES",
+    "MAX_WAIT_TIMEOUT_SECONDS",
+    "SESSION_NAME_PREFIX",
+]