diff --git a/api/controllers/openapi/_catalog.py b/api/controllers/openapi/_catalog.py index de31dedb8ff..49ad104b61a 100644 --- a/api/controllers/openapi/_catalog.py +++ b/api/controllers/openapi/_catalog.py @@ -27,7 +27,7 @@ from werkzeug.routing import Rule from configs import dify_config from constants import HEADER_NAME_CATALOG from controllers.openapi._upload import has_binary_leaf -from controllers.openapi.auth.spec import spec_of +from controllers.openapi.auth.spec import Effect, spec_of CATALOG_HEADER: Final = HEADER_NAME_CATALOG _SURFACE: Final = "/openapi/v1" @@ -137,6 +137,7 @@ def build_catalog(app: Flask) -> dict[str, Any]: "method": verb, "path": _catalog_path(rule.rule), "kind": meta.kind.value, + "effect": (meta.effect or Effect.for_method(verb)).value, "input": schema, "bind": derive_bind(method=verb, path_params=path_params, schema=schema), "internal": meta.internal, diff --git a/api/controllers/openapi/_contract.py b/api/controllers/openapi/_contract.py index e5b4b0a290c..6585e2b2a0f 100644 --- a/api/controllers/openapi/_contract.py +++ b/api/controllers/openapi/_contract.py @@ -32,10 +32,10 @@ from controllers.openapi._multipart import body_from_request from controllers.openapi._upload import file_fields from controllers.openapi.auth.requirements import Requirement from controllers.openapi.auth.router import subject_router -from controllers.openapi.auth.spec import AdmissionContext, CatalogMeta, EndpointSpec, Example, Kind +from controllers.openapi.auth.spec import AdmissionContext, CatalogMeta, Effect, EndpointSpec, Example, Kind from enums import DeploymentEdition -__all__ = ["Example", "Kind", "accepts", "endpoint", "op_of", "paginated", "returns"] +__all__ = ["Effect", "Example", "Kind", "accepts", "endpoint", "op_of", "paginated", "returns"] _INJECTED_KWARGS: Final = frozenset({"ctx", "query", "body"}) @@ -159,6 +159,7 @@ def endpoint( context: AdmissionContext | None = "orm", internal: bool = False, deprecated: bool = False, + effect: Effect | None = None, examples: Sequence[Example] = (), ) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """The one seam a route attaches to for auth, request validation and response @@ -185,7 +186,8 @@ def endpoint( ``op``/``kind``/``summary`` feed the catalog (see ``_catalog.py``); ``internal`` hides the op from the CLI's default listing; ``deprecated`` marks it as kept - for compatibility only; ``examples`` are complete inputs the catalog shows + for compatibility only; ``effect`` overrides the effect the HTTP method implies; + ``examples`` are complete inputs the catalog shows next to the schema. """ requirements = tuple(requirements) @@ -204,6 +206,7 @@ def endpoint( body=body, internal=internal, deprecated=deprecated, + effect=effect, examples=tuple(examples), ), ) diff --git a/api/controllers/openapi/_errors.py b/api/controllers/openapi/_errors.py index d5d5a86b516..43b5737f836 100644 --- a/api/controllers/openapi/_errors.py +++ b/api/controllers/openapi/_errors.py @@ -81,6 +81,7 @@ class OpenApiErrorCode(StrEnum): CATALOG_STALE = "catalog_stale" TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE = "trigger_workflow_service_mode_unavailable" RUN_NOT_FOUND = "run_not_found" + RUN_NOT_STOPPABLE = "run_not_stoppable" VERSION_NOT_FOUND = "version_not_found" VERSION_NOT_RESTORABLE = "version_not_restorable" DRAFT_NOT_FOUND = "draft_not_found" @@ -367,6 +368,14 @@ class RunNotFound(OpenApiError): # noqa: N818 description = "No run with this id exists for this app." +class RunNotStoppable(OpenApiError): # noqa: N818 + code = 409 + error_code = OpenApiErrorCode.RUN_NOT_STOPPABLE + description = ( + "This run has already finished, or it started before runs recorded a stop id; stop it from the console." + ) + + class VersionNotFound(OpenApiError): # noqa: N818 code = 404 error_code = OpenApiErrorCode.VERSION_NOT_FOUND diff --git a/api/controllers/openapi/_models.py b/api/controllers/openapi/_models.py index 6c080ad0e9d..b2ea11a5fa4 100644 --- a/api/controllers/openapi/_models.py +++ b/api/controllers/openapi/_models.py @@ -588,8 +588,8 @@ class CheckDependenciesResponse(CheckDependenciesResult, Hinted): class TaskStopResponse(BaseModel): - """200 body for POST /apps//tasks/:stop. The handler always returns - {"result": "success"}, so `result` is required (no default) — the generated contract + """200 body for POST /apps//runs/:stop and POST /apps//tasks/:stop. + The handlers always return {"result": "success"}, so `result` is required (no default) — the generated contract types it as a required `'success'` rather than an optional field.""" result: Literal["success"] diff --git a/api/controllers/openapi/app_dsl.py b/api/controllers/openapi/app_dsl.py index 91e9d030106..110a341c386 100644 --- a/api/controllers/openapi/app_dsl.py +++ b/api/controllers/openapi/app_dsl.py @@ -11,7 +11,7 @@ from constants.oauth_bearer import Scope from controllers.common.errors import ForbiddenError, NotFoundError from controllers.common.rbac import RBACPermission from controllers.openapi import openapi_ns -from controllers.openapi._contract import Example, Kind, endpoint, op_of +from controllers.openapi._contract import Effect, Example, Kind, endpoint, op_of from controllers.openapi._errors import DraftChanged, DraftNotFound, DslInvalid, ErrorDetail, VersionNotFound from controllers.openapi._models import ( AppDslExportQuery, @@ -114,6 +114,7 @@ class AppDslImportApi(Resource): context="workspace", op="import.console_app.dsl", kind=Kind.OBJECT, + effect=Effect.DESTRUCTIVE, summary="Import an app from DSL text or URL", examples=( Example( @@ -175,6 +176,7 @@ class AppDslCheckApi(Resource): context="workspace", op="check.console_app.dsl", kind=Kind.OBJECT, + effect=Effect.READ, summary="Check a DSL without importing it: every problem with its node and field", examples=(Example(title="Check a DSL (pipe the YAML)", input={"yaml_content": ""}),), requirements=_DSL_WRITE_GUARDS, @@ -212,6 +214,7 @@ class AppDslImportConfirmApi(Resource): context="workspace", op="confirm.console_app.dsl_import", kind=Kind.OBJECT, + effect=Effect.DESTRUCTIVE, summary="Confirm a pending DSL import", examples=(Example(title="Confirm an import that is pending confirmation", input={"import_id": ""}),), requirements=_DSL_WRITE_GUARDS, diff --git a/api/controllers/openapi/app_run.py b/api/controllers/openapi/app_run.py index 7ef165744f3..8182399cc96 100644 --- a/api/controllers/openapi/app_run.py +++ b/api/controllers/openapi/app_run.py @@ -24,8 +24,8 @@ from controllers.common.fields import EventStreamResponse from controllers.common.rbac import PlainApp, RBACCheck, RBACPermission from controllers.openapi import openapi_ns from controllers.openapi._audit import emit_app_run -from controllers.openapi._contract import Example, Kind, endpoint -from controllers.openapi._errors import AppModeMismatch +from controllers.openapi._contract import Effect, Example, Kind, endpoint +from controllers.openapi._errors import AppModeMismatch, RunNotFound, RunNotStoppable from controllers.openapi._files import end_read_transaction, materialize, merge_files from controllers.openapi._hints import attach_stream_hints from controllers.openapi._models import ( @@ -38,6 +38,7 @@ from controllers.openapi._models import ( TaskStopResponse, WorkflowRunPayload, ) +from controllers.openapi.app_workflow import require_uuid from controllers.openapi.auth.context import Context from controllers.openapi.auth.requirements import ( EDITOR_ROLES, @@ -75,6 +76,7 @@ from graphon.model_runtime.errors.invoke import InvokeError from libs import helper from models.model import App, AppMode from services.app_generate_service import AppGenerateService +from services.app_task_service import WorkflowStopOutcome from services.errors.app import ( IsDraftWorkflowError, WorkflowIdFormatError, @@ -84,6 +86,7 @@ from services.errors.app import ( TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, ) from services.errors.llm import InvokeRateLimitError +from services.workflow.graph_check import GRAPH_MODES, MESSAGE_MODES logger = logging.getLogger(__name__) @@ -440,24 +443,52 @@ def _run_api(route: _RunRoute) -> type[Resource]: ) = (_run_api(route) for route in _RUN_ROUTES) -@openapi_ns.route("/apps//tasks/:stop") -class AppRunTaskStopApi(Resource): +@openapi_ns.route("/apps//runs/:stop") +class AppRunStopApi(Resource): @endpoint( op="stop.run", kind=Kind.OBJECT, - summary="Stop a running task", - examples=(Example(title="Stop a running task", input={"app_id": "", "task_id": ""}),), + effect=Effect.DESTRUCTIVE, + summary="Stop a running or paused run of a workflow or advanced-chat app", + examples=(Example(title="Stop a run", input={"app_id": "", "run_id": ""}),), + requirements=_RUN_GUARDS, + returns=(200, TaskStopResponse, "Run stopped"), + ) + def post(self, ctx: Context, app_id: str, run_id: str): + _require_mode(ctx.app, *GRAPH_MODES) + outcome = application_services().app_tasks.stop_workflow_run( + tenant_id=ctx.app.tenant_id, + app_id=ctx.app.id, + run_id=require_uuid(run_id, not_found=RunNotFound), + app_mode=AppMode.value_of(ctx.app.mode), + caller_role=ctx.subject.caller_role, + caller_id=ctx.caller.id, + ) + if outcome == WorkflowStopOutcome.NOT_FOUND: + raise RunNotFound() + if outcome == WorkflowStopOutcome.NOT_STOPPABLE: + raise RunNotStoppable() + return TaskStopResponse(result="success") + + +@openapi_ns.route("/apps//tasks/:stop") +class AppTaskStopApi(Resource): + @endpoint( + op="stop.task", + kind=Kind.OBJECT, + effect=Effect.DESTRUCTIVE, + summary=( + "Stop a streaming reply of a chat, agent-chat or completion app; " + "`task_id` is the stream's task_id. An unknown task id is accepted" + ), + examples=(Example(title="Stop a streaming reply", input={"app_id": "", "task_id": ""}),), requirements=_RUN_GUARDS, returns=(200, TaskStopResponse, "Task stopped"), ) def post(self, ctx: Context, app_id: str, task_id: str): - if not application_services().app_tasks.stop_app_task( - tenant_id=ctx.app.tenant_id, - app_id=ctx.app.id, - task_id=task_id, - app_mode=AppMode.value_of(ctx.app.mode), - caller_role=ctx.subject.caller_role, - caller_id=ctx.caller.id, + _require_mode(ctx.app, *MESSAGE_MODES) + if not application_services().app_tasks.stop_message_task( + task_id=task_id, caller_role=ctx.subject.caller_role, caller_id=ctx.caller.id ): raise NotFound("Task not found") return TaskStopResponse(result="success") diff --git a/api/controllers/openapi/app_webapp.py b/api/controllers/openapi/app_webapp.py index 597fa034a05..d73945050ca 100644 --- a/api/controllers/openapi/app_webapp.py +++ b/api/controllers/openapi/app_webapp.py @@ -11,7 +11,7 @@ from constants.oauth_bearer import Scope from controllers.common.rbac import AgentBehindApp, PlainApp, RBACCheck, RBACPermission from controllers.openapi import _app_settings as settings from controllers.openapi import openapi_ns -from controllers.openapi._contract import Example, Kind, endpoint +from controllers.openapi._contract import Effect, Example, Kind, endpoint from controllers.openapi._models import ( AdvancedChatWebApp, AdvancedChatWebAppPatch, @@ -234,6 +234,7 @@ class AgentWebAppResetApi(Resource): @endpoint( op="reset.webapp.agent", kind=Kind.OBJECT, + effect=Effect.DESTRUCTIVE, summary="Rotate an agent app's web app URL; the old URL stops working", examples=(_RESET_EXAMPLE,), requirements=account_settings_guards( @@ -284,6 +285,7 @@ class WebAppResetApi(Resource): @endpoint( op="reset.webapp", kind=Kind.OBJECT, + effect=Effect.DESTRUCTIVE, summary="Rotate an app's web app URL; the old URL stops working (agent apps: reset.webapp.agent)", examples=(_RESET_EXAMPLE,), requirements=account_settings_guards( diff --git a/api/controllers/openapi/app_workflow.py b/api/controllers/openapi/app_workflow.py index 7a3179c7eb6..2fbf8fce4f4 100644 --- a/api/controllers/openapi/app_workflow.py +++ b/api/controllers/openapi/app_workflow.py @@ -16,7 +16,7 @@ from constants.oauth_bearer import Scope from controllers.common.fields import SimpleResultResponse from controllers.common.rbac import RBACPermission from controllers.openapi import openapi_ns -from controllers.openapi._contract import Example, Kind, endpoint, op_of +from controllers.openapi._contract import Effect, Example, Kind, endpoint, op_of from controllers.openapi._errors import ( DraftNotFound, EnvVariableNotFound, @@ -89,7 +89,7 @@ _ENV_WRITE_GUARDS: Final = account_app_guards( ) -def _require_uuid(value: str, *, not_found: type[OpenApiError]) -> str: +def require_uuid(value: str, *, not_found: type[OpenApiError]) -> str: """Parse a path id as a UUID or raise 404; an unparsed id would otherwise reach a UUID-typed database column and fail as a 500.""" try: @@ -100,7 +100,7 @@ def _require_uuid(value: str, *, not_found: type[OpenApiError]) -> str: def _require_run(ctx: Context, run_id: str) -> WorkflowRun: run = application_services().workflow_runs.get_workflow_run( - ctx.request_context, app_id=ctx.app.id, run_id=_require_uuid(run_id, not_found=RunNotFound) + ctx.request_context, app_id=ctx.app.id, run_id=require_uuid(run_id, not_found=RunNotFound) ) if run is None: raise RunNotFound() @@ -269,6 +269,7 @@ class AppVersionRestoreApi(Resource): @endpoint( op="restore.console_app.version", kind=Kind.OBJECT, + effect=Effect.DESTRUCTIVE, summary="Copy a published version into the draft; publish again to make it live", examples=( Example(title="Restore an older version", input={"app_id": "", "version_id": ""}), @@ -277,7 +278,7 @@ class AppVersionRestoreApi(Resource): returns=(HTTPStatus.OK, RestoreResponse, "Restored into the draft"), ) def post(self, ctx: Context, app_id: str, version_id: str): - version_id = _require_uuid(version_id, not_found=VersionNotFound) + version_id = require_uuid(version_id, not_found=VersionNotFound) try: draft = WorkflowService().restore_published_workflow_to_draft( app_model=ctx.app, workflow_id=version_id, account=ctx.account, session=ctx.session diff --git a/api/controllers/openapi/auth/spec.py b/api/controllers/openapi/auth/spec.py index d722aba9dcf..243275ef175 100644 --- a/api/controllers/openapi/auth/spec.py +++ b/api/controllers/openapi/auth/spec.py @@ -31,6 +31,33 @@ class Kind(StrEnum): FILE = "file" +class Effect(StrEnum): + """What running an op does to the server. Frozen protocol names, published in the catalog. + + Each member owns the HTTP methods that imply it; an op overrides that default + through `endpoint(effect=...)` when its method understates what it does. + """ + + methods: frozenset[str] + + READ = ("read", frozenset({"GET"})) + WRITE = ("write", frozenset({"POST", "PUT", "PATCH"})) + DESTRUCTIVE = ("destructive", frozenset({"DELETE"})) + + def __new__(cls, value: str, methods: frozenset[str]) -> Effect: + obj = str.__new__(cls, value) + obj._value_ = value + obj.methods = methods + return obj + + @classmethod + def for_method(cls, method: str) -> Effect: + for effect in cls: + if method.upper() in effect.methods: + return effect + raise ValueError(f"no effect for HTTP method {method!r}") + + @dataclass(frozen=True, slots=True, kw_only=True) class Example: """One complete input for an op, shown by the CLI next to the schema.""" @@ -44,8 +71,10 @@ class CatalogMeta: """What the catalog (`_catalog.py`) says about a route; the router never reads it. `internal` hides the op from the CLI's default listing; `deprecated` marks it - as kept for compatibility only. `examples` ride into the catalog as complete - inputs; a catalog test checks each one against the op's input schema. + as kept for compatibility only. `effect` overrides the effect the op's HTTP + method implies; leave it unset when the method already says it. `examples` + ride into the catalog as complete inputs; a catalog test checks each one against + the op's input schema. """ op: str @@ -55,6 +84,7 @@ class CatalogMeta: body: type[BaseModel] | None = None internal: bool = False deprecated: bool = False + effect: Effect | None = None examples: tuple[Example, ...] = () diff --git a/api/controllers/openapi/human_input_form.py b/api/controllers/openapi/human_input_form.py index 405958e99e7..4221b8b3c79 100644 --- a/api/controllers/openapi/human_input_form.py +++ b/api/controllers/openapi/human_input_form.py @@ -207,7 +207,7 @@ class OpenApiWorkflowHumanInputFormSubmitApi(Resource): op=op_of(OpenApiWorkflowEventsApi.get), input={ "app_id": ctx.app.id, - "task_id": form.workflow_run_id, + "run_id": form.workflow_run_id, "include_state_snapshot": True, "continue_on_pause": False, }, diff --git a/api/controllers/openapi/workflow_events.py b/api/controllers/openapi/workflow_events.py index f986744b759..d7e6e7e5a7b 100644 --- a/api/controllers/openapi/workflow_events.py +++ b/api/controllers/openapi/workflow_events.py @@ -1,9 +1,8 @@ """ OpenAPI bearer-authed workflow reconnect event stream endpoint. -GET /apps//tasks//events +GET /apps//runs//events — reconnect to the SSE stream for a paused/running workflow run. - `task_id` is treated as `workflow_run_id`. """ from __future__ import annotations @@ -21,6 +20,8 @@ from controllers.common.fields import EventStreamResponse from controllers.common.rbac import PlainApp, RBACCheck, RBACPermission from controllers.openapi import openapi_ns from controllers.openapi._contract import Example, Kind, endpoint +from controllers.openapi._errors import RunNotFound +from controllers.openapi.app_workflow import require_uuid from controllers.openapi.auth.context import Context from controllers.openapi.auth.requirements import ( CheckAppAccess, @@ -50,20 +51,20 @@ class WorkflowEventsQuery(BaseModel): continue_on_pause: bool = Field(default=False, description="Whether to keep the event stream open on pause") -@openapi_ns.route("/apps//tasks//events") +@openapi_ns.route("/apps//runs//events") class OpenApiWorkflowEventsApi(Resource): @endpoint( op="get.run.event", kind=Kind.SSE, - summary="Stream the events of a workflow run; `task_id` takes the workflow run id", + summary="Stream the events of a workflow run", examples=( Example( title="Follow the events of a workflow run by its run id", - input={"app_id": "", "task_id": ""}, + input={"app_id": "", "run_id": ""}, ), Example( title="Keep the stream open across a human-input pause", - input={"app_id": "", "task_id": "", "continue_on_pause": True}, + input={"app_id": "", "run_id": "", "continue_on_pause": True}, ), ), requirements=( @@ -77,7 +78,7 @@ class OpenApiWorkflowEventsApi(Resource): query=WorkflowEventsQuery, returns=(200, EventStreamResponse, "SSE event stream"), ) - def get(self, ctx: Context, app_id: str, task_id: str, *, query: WorkflowEventsQuery): + def get(self, ctx: Context, app_id: str, run_id: str, *, query: WorkflowEventsQuery): # The router's session closes as soon as this returns, so everything the SSE # body needs is read off `ctx` here and the generators below close over plain # values only. @@ -94,7 +95,7 @@ class OpenApiWorkflowEventsApi(Resource): repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker) workflow_run = repo.get_workflow_run_by_id_and_tenant_id( tenant_id=app_model.tenant_id, - run_id=task_id, + run_id=require_uuid(run_id, not_found=RunNotFound), ) if workflow_run is None: diff --git a/api/openapi/markdown/openapi-openapi.md b/api/openapi/markdown/openapi-openapi.md index b2d276334d3..73cd80c103c 100644 --- a/api/openapi/markdown/openapi-openapi.md +++ b/api/openapi/markdown/openapi-openapi.md @@ -672,6 +672,24 @@ Upload a file to use as an input variable when running the app | 200 | Run detail | **application/json**: [WorkflowRunDetailResponse](#workflowrundetailresponse)
| | default | Error | **application/json**: [ErrorBody](#errorbody)
| +### [GET] /apps/{app_id}/runs/{run_id}/events +#### Parameters + +| Name | Located in | Description | Required | Schema | +| ---- | ---------- | ----------- | -------- | ------ | +| continue_on_pause | query | Whether to keep the event stream open on pause | No | boolean | +| include_state_snapshot | query | Whether to include workflow state snapshots | No | boolean | +| app_id | path | | Yes | string | +| run_id | path | | Yes | string | + +#### Responses + +| Code | Description | Schema | +| ---- | ----------- | ------ | +| 200 | SSE event stream | **application/json**: [EventStreamResponse](#eventstreamresponse)
| +| 422 | Validation error | **application/json**: [ErrorBody](#errorbody)
| +| default | Error | **application/json**: [ErrorBody](#errorbody)
| + ### [GET] /apps/{app_id}/runs/{run_id}/nodes #### Parameters @@ -687,6 +705,21 @@ Upload a file to use as an input variable when running the app | 200 | Node steps | **application/json**: [WorkflowRunNodeExecutionListResponse](#workflowrunnodeexecutionlistresponse)
| | default | Error | **application/json**: [ErrorBody](#errorbody)
| +### [POST] /apps/{app_id}/runs/{run_id}:stop +#### Parameters + +| Name | Located in | Description | Required | Schema | +| ---- | ---------- | ----------- | -------- | ------ | +| app_id | path | | Yes | string | +| run_id | path | | Yes | string | + +#### Responses + +| Code | Description | Schema | +| ---- | ----------- | ------ | +| 200 | Run stopped | **application/json**: [TaskStopResponse](#taskstopresponse)
| +| default | Error | **application/json**: [ErrorBody](#errorbody)
| + ### [GET] /apps/{app_id}/service-api #### Parameters @@ -757,24 +790,6 @@ Upload a file to use as an input variable when running the app | 422 | Validation error | **application/json**: [ErrorBody](#errorbody)
| | default | Error | **application/json**: [ErrorBody](#errorbody)
| -### [GET] /apps/{app_id}/tasks/{task_id}/events -#### Parameters - -| Name | Located in | Description | Required | Schema | -| ---- | ---------- | ----------- | -------- | ------ | -| continue_on_pause | query | Whether to keep the event stream open on pause | No | boolean | -| include_state_snapshot | query | Whether to include workflow state snapshots | No | boolean | -| app_id | path | | Yes | string | -| task_id | path | | Yes | string | - -#### Responses - -| Code | Description | Schema | -| ---- | ----------- | ------ | -| 200 | SSE event stream | **application/json**: [EventStreamResponse](#eventstreamresponse)
| -| 422 | Validation error | **application/json**: [ErrorBody](#errorbody)
| -| default | Error | **application/json**: [ErrorBody](#errorbody)
| - ### [POST] /apps/{app_id}/tasks/{task_id}:stop #### Parameters @@ -3660,8 +3675,8 @@ generated CLI whitelist all derive from it. #### TaskStopResponse -200 body for POST /apps//tasks/:stop. The handler always returns -{"result": "success"}, so `result` is required (no default) — the generated contract +200 body for POST /apps//runs/:stop and POST /apps//tasks/:stop. +The handlers always return {"result": "success"}, so `result` is required (no default) — the generated contract types it as a required `'success'` rather than an optional field. | Name | Type | Description | Required | diff --git a/api/services/app_task_service.py b/api/services/app_task_service.py index a66b77ccd08..0d14d721e63 100644 --- a/api/services/app_task_service.py +++ b/api/services/app_task_service.py @@ -7,6 +7,7 @@ new Engine command channel mechanism. import json import logging +from enum import StrEnum from uuid import UUID from sqlalchemy import select @@ -39,6 +40,12 @@ from repositories.sqlalchemy_api_workflow_run_repository import ( logger = logging.getLogger(__name__) +class WorkflowStopOutcome(StrEnum): + STOPPED = "stopped" + NOT_FOUND = "not_found" + NOT_STOPPABLE = "not_stoppable" + + class AppTaskControlService: """Service for managing application task operations.""" @@ -63,27 +70,33 @@ class AppTaskControlService: """ return AppQueueManager.set_stop_flag(task_id, invoke_from, user_id, redis=self._redis_client) - def stop_app_task( + def stop_message_task(self, *, task_id: str, caller_role: CreatorUserRole, caller_id: str) -> bool: + """Stop a chat, agent-chat or completion task, restricting end users to their own tasks.""" + if caller_role == CreatorUserRole.END_USER: + return self.stop_task(task_id, InvokeFrom.OPENAPI, caller_id) + set_app_task_stop_flag(task_id, redis=self._redis_client) + return True + + def stop_workflow_run( self, *, tenant_id: str, app_id: str, - task_id: str, + run_id: str, app_mode: AppMode, caller_role: CreatorUserRole, caller_id: str, - ) -> bool: - """Stop after app admission, restricting all end users to their own tasks.""" + ) -> WorkflowStopOutcome: + """Stop a workflow run by its run ID, restricting end users to their own runs.""" owner = (caller_role, caller_id) if caller_role == CreatorUserRole.END_USER else None - if app_mode in (AppMode.WORKFLOW, AppMode.ADVANCED_CHAT): - admitted = self.stop_workflow_task( - tenant_id=tenant_id, app_id=app_id, task_id=task_id, app_mode=app_mode, owner=owner - ) - return admitted or owner is None - if owner is not None: - return self.stop_task(task_id, InvokeFrom.OPENAPI, caller_id) - set_app_task_stop_flag(task_id, redis=self._redis_client) - return True + return AppTaskService.stop_workflow_run( + tenant_id=tenant_id, + app_id=app_id, + run_id=run_id, + app_mode=app_mode, + owner=owner, + redis=self._redis_client, + ) def stop_workflow_task_no_user_check(self, *, task_id: str) -> None: """Stop a workflow after app admission, without consulting the user ownership cache. @@ -185,6 +198,47 @@ class AppTaskService: return after_id = pauses[-1][1].id + @staticmethod + def stop_workflow_run( + *, + tenant_id: str, + app_id: str, + run_id: str, + app_mode: AppMode, + owner: tuple[CreatorUserRole, str] | None, + redis: RedisClientWrapper, + ) -> WorkflowStopOutcome: + repository = DifyAPISQLAlchemyWorkflowRunRepository(sessionmaker(bind=db.engine)) + run = repository.get_workflow_run_by_id_and_tenant_id(tenant_id, run_id) + if run is None: + return WorkflowStopOutcome.NOT_FOUND + if run.app_id != app_id: + return WorkflowStopOutcome.NOT_FOUND + if owner is not None and (run.created_by_role, run.created_by) != owner: + return WorkflowStopOutcome.NOT_FOUND + if run.task_id is None: + return WorkflowStopOutcome.NOT_STOPPABLE + task_id = AppTaskService._parse_task_id(run.task_id) + try: + paused_run = repository.stop_paused_workflow_task( + tenant_id=tenant_id, app_id=app_id, task_id=task_id, owner=owner + ) + except WorkflowTaskOwnerMismatchError: + return WorkflowStopOutcome.NOT_STOPPABLE + stopped = AppTaskService._finish_workflow_stop( + run=paused_run, tenant_id=tenant_id, app_id=app_id, task_id=task_id, app_mode=app_mode, redis=redis + ) + if not stopped: + return WorkflowStopOutcome.NOT_STOPPABLE + return WorkflowStopOutcome.STOPPED + + @staticmethod + def _parse_task_id(task_id: str) -> str: + try: + return str(UUID(task_id)) + except ValueError as exc: + raise ValueError("Invalid task ID: expected a UUID") from exc + @staticmethod def stop_workflow_task( *, @@ -201,10 +255,7 @@ class AppTaskService: transition and form invalidation must complete before acknowledging stop. Omitted Redis uses the compatibility transport; injected callers supply theirs. """ - try: - task_id = str(UUID(task_id)) - except ValueError as exc: - raise ValueError("Invalid task ID: expected a UUID") from exc + task_id = AppTaskService._parse_task_id(task_id) repository = DifyAPISQLAlchemyWorkflowRunRepository(sessionmaker(bind=db.engine)) try: run = repository.stop_paused_workflow_task(tenant_id=tenant_id, app_id=app_id, task_id=task_id, owner=owner) @@ -217,6 +268,20 @@ class AppTaskService: ) except WorkflowTaskOwnerMismatchError: return False + return AppTaskService._finish_workflow_stop( + run=run, tenant_id=tenant_id, app_id=app_id, task_id=task_id, app_mode=app_mode, redis=redis + ) + + @staticmethod + def _finish_workflow_stop( + *, + run: WorkflowRun | None, + tenant_id: str, + app_id: str, + task_id: str, + app_mode: AppMode, + redis: RedisClientWrapper | None, + ) -> bool: if run is not None and run.status == WorkflowExecutionStatus.RUNNING: # Resume has claimed the pause, but may not have opened its queue. # The repository has already checked the complete app/user owner. diff --git a/api/services/workflow/graph_check.py b/api/services/workflow/graph_check.py index 18d6e1569ad..03088b51e7e 100644 --- a/api/services/workflow/graph_check.py +++ b/api/services/workflow/graph_check.py @@ -27,6 +27,7 @@ from services.workflow_variable_reference_validator import SOURCE_HANDLE, branch logger = logging.getLogger(__name__) GRAPH_MODES: Final = frozenset({AppMode.WORKFLOW, AppMode.ADVANCED_CHAT}) +MESSAGE_MODES: Final = frozenset({AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.COMPLETION}) CONTAINER_NODE_TYPES: Final = frozenset({BuiltinNodeTypes.ITERATION, BuiltinNodeTypes.LOOP}) _CANVAS_NOTE_TYPE: Final = "custom-note" diff --git a/api/tests/e2e_tests/containers/controllers/openapi/test_app_run.py b/api/tests/e2e_tests/containers/controllers/openapi/test_app_run.py index ef6077983fb..2bdbb3ec9be 100644 --- a/api/tests/e2e_tests/containers/controllers/openapi/test_app_run.py +++ b/api/tests/e2e_tests/containers/controllers/openapi/test_app_run.py @@ -3,10 +3,12 @@ from __future__ import annotations from collections.abc import Callable from uuid import uuid4 +import pytest from flask import Flask from sqlalchemy.orm import Session -from controllers.openapi.app_run import AppRunTaskStopApi +from controllers.openapi._errors import RunNotFound +from controllers.openapi.app_run import AppRunStopApi from models import Account, App from services.app_service import AppService, CreateAppParams from tests.e2e_tests.containers.controllers.openapi.conftest import context_for @@ -28,25 +30,26 @@ def _create_app(db_session: Session, account: Account, *, name: str = "Runner") return app_model -class TestAppRunTaskStop: - def test_task_stop_returns_success( +class TestAppRunStop: + def test_run_stop_rejects_an_unknown_run( self, app: Flask, db_session_with_containers: Session, make_account: Callable[..., Account] ) -> None: account = make_account() app_model = _create_app(db_session_with_containers, account) - task_id = str(uuid4()) + run_id = str(uuid4()) - api = AppRunTaskStopApi() - with app.test_request_context(f"/openapi/v1/apps/{app_model.id}/tasks/{task_id}:stop", method="POST"): - result = api.post.__handler__( + api = AppRunStopApi() + with ( + app.test_request_context(f"/openapi/v1/apps/{app_model.id}/runs/{run_id}:stop", method="POST"), + pytest.raises(RunNotFound), + ): + api.post.__handler__( api, context_for( account, session=db_session_with_containers, - view_args={"app_id": app_model.id, "task_id": task_id}, + view_args={"app_id": app_model.id, "run_id": run_id}, ), app_model.id, - task_id, + run_id, ) - - assert result.result == "success" diff --git a/api/tests/unit_tests/controllers/openapi/auth/test_requirements.py b/api/tests/unit_tests/controllers/openapi/auth/test_requirements.py index be751f9abff..aab671f8276 100644 --- a/api/tests/unit_tests/controllers/openapi/auth/test_requirements.py +++ b/api/tests/unit_tests/controllers/openapi/auth/test_requirements.py @@ -994,6 +994,12 @@ ROUTES: tuple[Route, ...] = ( Route( "app_run.stop", "POST", + "/apps/{app_id}/runs/{run_id}:stop", + frozenset({Trait.ACCOUNT_PRIMARY, Trait.APP_SCOPED, Trait.EXTERNAL_REACHABLE}), + ), + Route( + "app_run.stop.task", + "POST", "/apps/{app_id}/tasks/{task_id}:stop", frozenset({Trait.ACCOUNT_PRIMARY, Trait.APP_SCOPED, Trait.EXTERNAL_REACHABLE}), ), @@ -1018,7 +1024,7 @@ ROUTES: tuple[Route, ...] = ( Route( "workflow_events.stream", "GET", - "/apps/{app_id}/tasks/{task_id}/events", + "/apps/{app_id}/runs/{run_id}/events", frozenset({Trait.ACCOUNT_PRIMARY, Trait.APP_SCOPED, Trait.EXTERNAL_REACHABLE}), ), Route( @@ -1780,6 +1786,7 @@ DECLARED: dict[str, tuple[Requirement, ...]] = { CheckWebAppAuthEnterprise(), ), "app_run.stop": _RESOURCE_RUN, + "app_run.stop.task": _RESOURCE_RUN, "files.upload": _REQ_FILES, "human_input_form.get": _REQ_RUN_FORM, "human_input_form.submit": _REQ_RUN_FORM, @@ -2166,7 +2173,8 @@ EXPECTED_RESPONSE_CODES: dict[tuple[str, str], frozenset[str]] = { ("post", "/apps/{app_id}/files"): frozenset({"201", "400", "401", "413", "415", "422", "default"}), ("get", "/apps/{app_id}/human-input-forms/{form_token}"): frozenset({"200", "default"}), ("post", "/apps/{app_id}/human-input-forms/{form_token}:submit"): frozenset({"200", "422", "default"}), - ("get", "/apps/{app_id}/tasks/{task_id}/events"): frozenset({"200", "422", "default"}), + ("get", "/apps/{app_id}/runs/{run_id}/events"): frozenset({"200", "422", "default"}), + ("post", "/apps/{app_id}/runs/{run_id}:stop"): frozenset({"200", "default"}), ("post", "/apps/{app_id}/tasks/{task_id}:stop"): frozenset({"200", "default"}), ("post", "/apps/{app_id}/workflow:run"): frozenset({"200", "422", "default"}), ("post", "/apps/{app_id}/chat:run"): frozenset({"200", "422", "default"}), diff --git a/api/tests/unit_tests/controllers/openapi/auth/test_resource_access.py b/api/tests/unit_tests/controllers/openapi/auth/test_resource_access.py index 0e98ac53a28..8bbc876a89e 100644 --- a/api/tests/unit_tests/controllers/openapi/auth/test_resource_access.py +++ b/api/tests/unit_tests/controllers/openapi/auth/test_resource_access.py @@ -261,7 +261,7 @@ def test_stop_task_rejects_other_token_owner( from werkzeug.exceptions import NotFound - from controllers.openapi.app_run import AppRunTaskStopApi + from controllers.openapi.app_run import AppTaskStopApi flask_app, app_id, _, _, end_user = resource_fixture data = call(flask_app, app_id) @@ -275,10 +275,10 @@ def test_stop_task_rejects_other_token_owner( return_value=replace(account_application_services, app_tasks=AppTaskControlService(redis_client=redis)), ): with pytest.raises(NotFound): - inspect.unwrap(AppRunTaskStopApi.post)(AppRunTaskStopApi(), data, app_id, "task") + inspect.unwrap(AppTaskStopApi.post)(AppTaskStopApi(), data, app_id, "task") assert redis.operations == [] redis.values["generate_task_belong:task"] = b"end-user-machine-user" - inspect.unwrap(AppRunTaskStopApi.post)(AppRunTaskStopApi(), data, app_id, "task") + inspect.unwrap(AppTaskStopApi.post)(AppTaskStopApi(), data, app_id, "task") assert redis.values["generate_task_stopped:task"] == b"1" diff --git a/api/tests/unit_tests/controllers/openapi/test__catalog.py b/api/tests/unit_tests/controllers/openapi/test__catalog.py index fa2ed38db13..097dd0ef424 100644 --- a/api/tests/unit_tests/controllers/openapi/test__catalog.py +++ b/api/tests/unit_tests/controllers/openapi/test__catalog.py @@ -22,7 +22,7 @@ from controllers.openapi._catalog import ( ) from controllers.openapi._errors import OpenApiErrorCode from controllers.openapi._models import Hinted -from controllers.openapi.auth.spec import EndpointSpec, Kind, spec_of +from controllers.openapi.auth.spec import Effect, EndpointSpec, Kind, spec_of from tests.unit_tests.controllers.openapi.conftest import AdmittedWorld OP_ID_RE = re.compile(r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)+$") @@ -97,9 +97,40 @@ def test_every_guarded_route_declares_catalog_meta_once(app: Flask, ops: dict[st assert streaming == per_mode | draft | {"get.run.event"} +# Every op whose effect differs from its HTTP method, plus one default per method. +@pytest.mark.parametrize( + ("op", "effect"), + [ + ("reset.webapp", Effect.DESTRUCTIVE), + ("reset.webapp.agent", Effect.DESTRUCTIVE), + ("restore.console_app.version", Effect.DESTRUCTIVE), + ("import.console_app.dsl", Effect.DESTRUCTIVE), + ("confirm.console_app.dsl_import", Effect.DESTRUCTIVE), + ("stop.run", Effect.DESTRUCTIVE), + ("stop.task", Effect.DESTRUCTIVE), + ("check.console_app.dsl", Effect.READ), + ("run.console_app.chat", Effect.WRITE), + ("get.run.event", Effect.READ), + ], +) +def test_catalog_rows_carry_the_op_effect(ops: dict[str, CatalogOp], op: str, effect: Effect) -> None: + assert ops[op]["effect"] == effect.value + + def test_run_entries_carry_path_bind_kind_and_flags(ops: dict[str, CatalogOp]) -> None: chat = ops["run.console_app.chat"] - assert set(chat) == {"summary", "method", "path", "kind", "input", "bind", "internal", "deprecated", "examples"} + assert set(chat) == { + "summary", + "method", + "path", + "kind", + "effect", + "input", + "bind", + "internal", + "deprecated", + "examples", + } assert (chat["method"], chat["path"], chat["kind"]) == ( "POST", "/openapi/v1/apps/{app_id}/chat:run", diff --git a/api/tests/unit_tests/controllers/openapi/test_app_run.py b/api/tests/unit_tests/controllers/openapi/test_app_run.py index aadf1af3ced..7d3dcf069bb 100644 --- a/api/tests/unit_tests/controllers/openapi/test_app_run.py +++ b/api/tests/unit_tests/controllers/openapi/test_app_run.py @@ -28,7 +28,7 @@ from controllers.openapi._models import ( ) from controllers.openapi.app_run import ( AdvancedChatRunApi, - AppRunTaskStopApi, + AppTaskStopApi, ChatRunApi, CompletionRunApi, _translate_service_errors, @@ -39,13 +39,11 @@ from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpErr from core.app.entities.app_invoke_entities import InvokeFrom from core.errors.error import AppInvokeQuotaExceededError from extensions.ext_application_services import ApplicationServices -from graphon.enums import WorkflowExecutionStatus, WorkflowType from graphon.file import FileType from libs.oauth_bearer import AuthContext from models import Account -from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom +from models.enums import CreatorUserRole from models.model import App, AppMode, EndUser, UploadFile -from models.workflow import WorkflowRun from services.app_task_service import AppTaskControlService from services.errors.app import ( TriggerWorkflowServiceModeUnavailableError as TriggerWorkflowServiceModeUnavailableServiceError, @@ -158,7 +156,7 @@ def test_chat_payload_normalizes_conversation_id(): @pytest.mark.parametrize( "token_type", [TokenType.OAUTH_ACCOUNT, TokenType.OAUTH_EXTERNAL_SSO, TokenType.RESOURCE_ACCESS] ) -@pytest.mark.parametrize("mode", [AppMode.WORKFLOW, AppMode.ADVANCED_CHAT, AppMode.CHAT, AppMode.COMPLETION]) +@pytest.mark.parametrize("mode", [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.COMPLETION]) @pytest.mark.parametrize("owns_task", [False, True]) def test_stop_task_enforces_resolved_caller_ownership( app: Flask, @@ -176,23 +174,6 @@ def test_stop_task_enforces_resolved_caller_ownership( owner_id = caller.id if owns_task else "another-user" task_id = str(uuid.uuid4()) redis = _StopRedis(values={f"generate_task_belong:{task_id}": f"end-user-{owner_id}".encode()}) - if mode in (AppMode.WORKFLOW, AppMode.ADVANCED_CHAT): - sqlite_session.add( - WorkflowRun( - id="run-1", - tenant_id="tenant-1", - app_id="app-1", - workflow_id="workflow-1", - task_id=task_id, - type=WorkflowType.CHAT if mode == AppMode.ADVANCED_CHAT else WorkflowType.WORKFLOW, - version="1", - triggered_from=WorkflowRunTriggeredFrom.APP_RUN, - status=WorkflowExecutionStatus.RUNNING, - created_by_role=CreatorUserRole.END_USER, - created_by=owner_id, - ) - ) - sqlite_session.commit() monkeypatch.setattr(type(task_module.db), "engine", property(lambda _: sqlite_session.get_bind())) monkeypatch.setattr( sys.modules["controllers.openapi.app_run"], @@ -204,7 +185,7 @@ def test_stop_task_enforces_resolved_caller_ownership( caller=caller, app=App(id="app-1", tenant_id="tenant-1", mode=mode), ) - api = AppRunTaskStopApi() + api = AppTaskStopApi() allowed = owns_task or subject.caller_role == CreatorUserRole.ACCOUNT with app.test_request_context(f"/openapi/v1/apps/app-1/tasks/{task_id}:stop", method="POST"): if allowed: @@ -214,10 +195,9 @@ def test_stop_task_enforces_resolved_caller_ownership( api.post.__handler__(api, ctx, app_id="app-1", task_id=task_id) if allowed: assert redis.values[f"generate_task_stopped:{task_id}"] == b"1" - assert bool(redis.commands) == (mode in (AppMode.WORKFLOW, AppMode.ADVANCED_CHAT)) else: assert redis.operations == [] - assert redis.commands == {} + assert redis.commands == {} class _SealableContext: diff --git a/api/tests/unit_tests/controllers/openapi/test_human_input_form.py b/api/tests/unit_tests/controllers/openapi/test_human_input_form.py index a81cc621809..5bf05a2609f 100644 --- a/api/tests/unit_tests/controllers/openapi/test_human_input_form.py +++ b/api/tests/unit_tests/controllers/openapi/test_human_input_form.py @@ -241,7 +241,7 @@ class TestSubmitFollowRunHint: assert hint.op == op_of(OpenApiWorkflowEventsApi.get) == "get.run.event" assert hint.input == { "app_id": "app-1", - "task_id": "run-9", + "run_id": "run-9", "include_state_snapshot": True, "continue_on_pause": False, } diff --git a/api/tests/unit_tests/controllers/openapi/test_workflow_events.py b/api/tests/unit_tests/controllers/openapi/test_workflow_events.py index 64026d07fff..36c4d7cfe57 100644 --- a/api/tests/unit_tests/controllers/openapi/test_workflow_events.py +++ b/api/tests/unit_tests/controllers/openapi/test_workflow_events.py @@ -26,6 +26,8 @@ from models.enums import CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom from models.model import App, AppMode, EndUser from models.workflow import WorkflowRun, WorkflowType +_RUN_ID = "11111111-1111-1111-1111-111111111111" + class _SealableContext: """A `Context` stand-in that refuses reads once `seal()` is called. @@ -64,7 +66,7 @@ def _make_workflow_run( finished_at=None, ) -> WorkflowRun: return WorkflowRun( - id="wf-run-1", + id=_RUN_ID, app_id=app_id, tenant_id=tenant_id, workflow_id="workflow-1", @@ -127,13 +129,13 @@ class TestOpenApiWorkflowEventsApi: factory_mock = self._bind_repo(monkeypatch, None) api = OpenApiWorkflowEventsApi() - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): with pytest.raises(NotFound): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), app_id="app-1", - task_id="wf-run-1", + run_id=_RUN_ID, query=WorkflowEventsQuery(), ) @@ -146,13 +148,13 @@ class TestOpenApiWorkflowEventsApi: self._bind_repo(monkeypatch, _make_workflow_run(app_id="other-app")) api = OpenApiWorkflowEventsApi() - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): with pytest.raises(NotFound): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), app_id="app-1", - task_id="wf-run-1", + run_id=_RUN_ID, query=WorkflowEventsQuery(), ) @@ -160,13 +162,13 @@ class TestOpenApiWorkflowEventsApi: self._bind_repo(monkeypatch, _make_workflow_run(created_by_role=CreatorUserRole.END_USER, created_by="eu-1")) api = OpenApiWorkflowEventsApi() - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): with pytest.raises(NotFound): api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), app_id="app-1", - task_id="wf-run-1", + run_id=_RUN_ID, query=WorkflowEventsQuery(), ) @@ -176,12 +178,12 @@ class TestOpenApiWorkflowEventsApi: self._bind_generators(monkeypatch) api = OpenApiWorkflowEventsApi() - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): resp = api.get.__handler__( api, _context(_make_end_user(), CreatorUserRole.END_USER), app_id="app-1", - task_id="wf-run-1", + run_id=_RUN_ID, query=WorkflowEventsQuery(), ) assert resp.mimetype == "text/event-stream" @@ -218,8 +220,8 @@ class TestOpenApiWorkflowEventsApi: ctx = _context(_make_account(), CreatorUserRole.ACCOUNT) api = OpenApiWorkflowEventsApi() - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): - resp = api.get.__handler__(api, ctx, app_id="app-1", task_id="wf-run-1", query=WorkflowEventsQuery()) + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): + resp = api.get.__handler__(api, ctx, app_id="app-1", run_id=_RUN_ID, query=WorkflowEventsQuery()) ctx.seal() body = "".join(resp.response) @@ -248,8 +250,8 @@ class TestOpenApiWorkflowEventsApi: ctx = _context(_make_account(), CreatorUserRole.ACCOUNT) api = OpenApiWorkflowEventsApi() query = WorkflowEventsQuery(include_state_snapshot=include_state_snapshot) - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): - resp = api.get.__handler__(api, ctx, app_id="app-1", task_id="wf-run-1", query=query) + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): + resp = api.get.__handler__(api, ctx, app_id="app-1", run_id=_RUN_ID, query=query) ctx.seal() body = "".join(resp.response) @@ -267,12 +269,12 @@ class TestOpenApiWorkflowEventsApi: monkeypatch.setattr(module, "MessageGenerator", lambda: msg_gen_mock) api = OpenApiWorkflowEventsApi() - with app.test_request_context("/openapi/v1/apps/app-1/tasks/wf-run-1/events"): + with app.test_request_context(f"/openapi/v1/apps/app-1/runs/{_RUN_ID}/events"): resp = api.get.__handler__( api, _context(_make_account(), CreatorUserRole.ACCOUNT), app_id="app-1", - task_id="wf-run-1", + run_id=_RUN_ID, query=WorkflowEventsQuery(continue_on_pause=continue_on_pause), ) "".join(resp.response) diff --git a/api/tests/unit_tests/services/test_app_task_service.py b/api/tests/unit_tests/services/test_app_task_service.py index c741d445621..ef347d47861 100644 --- a/api/tests/unit_tests/services/test_app_task_service.py +++ b/api/tests/unit_tests/services/test_app_task_service.py @@ -44,7 +44,7 @@ from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionTri from models.workflow import WorkflowType as ModelWorkflowType from models.workflow import WorkflowType as _legacy_workflow_task_recovery_WorkflowType from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository -from services.app_task_service import AppTaskControlService, AppTaskService +from services.app_task_service import AppTaskControlService, AppTaskService, WorkflowStopOutcome """Injected workflow cancellation preserves both transport mechanisms.""" @@ -249,7 +249,6 @@ class TestAppTaskServiceCases: assert redis.commands == {} @staticmethod - @pytest.mark.parametrize("app_mode", [AppMode.CHAT, AppMode.COMPLETION]) @pytest.mark.parametrize( ("caller_role", "cached_owner", "admitted"), [ @@ -261,18 +260,13 @@ class TestAppTaskServiceCases: (CreatorUserRole.ACCOUNT, None, True), ], ) - def test_stop_app_task_applies_legacy_caller_ownership( - app_mode: AppMode, caller_role: CreatorUserRole, cached_owner: bytes | None, admitted: bool + def test_stop_message_task_applies_legacy_caller_ownership( + caller_role: CreatorUserRole, cached_owner: bytes | None, admitted: bool ) -> None: redis = _AppTaskServiceStopredis(values={_OWNER_KEY: cached_owner} if cached_owner is not None else {}) - result = AppTaskControlService(redis_client=redis).stop_app_task( - tenant_id="tenant-1", - app_id="app-1", - task_id=_TASK_ID, - app_mode=app_mode, - caller_role=caller_role, - caller_id=_USER_ID, + result = AppTaskControlService(redis_client=redis).stop_message_task( + task_id=_TASK_ID, caller_role=caller_role, caller_id=_USER_ID ) assert result is admitted @@ -314,7 +308,7 @@ class TestAppTaskServiceCases: (CreatorUserRole.ACCOUNT, CreatorUserRole.ACCOUNT, "other-account"), ], ) - def test_stop_app_task_signals_admitted_running_workflow( + def test_stop_workflow_run_signals_a_running_run( sqlite_session: Session, app_task_workflow_run: WorkflowRun, app_mode: AppMode, @@ -332,16 +326,16 @@ class TestAppTaskServiceCases: read_error=AssertionError("Durable workflow ownership must not depend on Redis") ) - admitted = AppTaskControlService(redis_client=redis).stop_app_task( + outcome = AppTaskControlService(redis_client=redis).stop_workflow_run( tenant_id=run.tenant_id, app_id=run.app_id, - task_id=task_id, + run_id=run.id, app_mode=app_mode, caller_role=caller_role, caller_id=_USER_ID, ) - assert admitted is True + assert outcome is WorkflowStopOutcome.STOPPED sqlite_session.refresh(run) assert run.status == WorkflowExecutionStatus.RUNNING assert run.stop_requested_at is not None @@ -368,7 +362,7 @@ class TestAppTaskServiceCases: (CreatorUserRole.ACCOUNT, "missing"), ], ) - def test_stop_app_task_preserves_unadmitted_workflow_and_pause( + def test_stop_workflow_run_preserves_unadmitted_run_and_pause( sqlite_session: Session, app_task_workflow_run: WorkflowRun, app_mode: AppMode, @@ -398,22 +392,20 @@ class TestAppTaskServiceCases: ) sqlite_session.add_all([pause, form]) sqlite_session.commit() - task_id = "22222222-2222-2222-2222-222222222222" if mismatch == "missing" else run.task_id - assert task_id is not None redis = _AppTaskServiceStopredis( read_error=AssertionError("Durable workflow ownership must not depend on Redis") ) - admitted = AppTaskControlService(redis_client=redis).stop_app_task( + outcome = AppTaskControlService(redis_client=redis).stop_workflow_run( tenant_id="other-tenant" if mismatch == "tenant" else run.tenant_id, app_id="other-app" if mismatch == "app" else run.app_id, - task_id=task_id, + run_id="22222222-2222-2222-2222-222222222222" if mismatch == "missing" else run.id, app_mode=app_mode, caller_role=caller_role, caller_id=_USER_ID, ) - assert admitted is (caller_role is CreatorUserRole.ACCOUNT) + assert outcome is WorkflowStopOutcome.NOT_FOUND assert redis.reads == [] assert redis.operations == [] assert redis.values == {} @@ -429,6 +421,31 @@ class TestAppTaskServiceCases: assert form.status == HumanInputFormStatus.WAITING assert form.submitted_at is None + @staticmethod + @pytest.mark.parametrize( + "change", [{"status": WorkflowExecutionStatus.SUCCEEDED}, {"task_id": None}], ids=["finished", "no-task-id"] + ) + def test_stop_workflow_run_refuses_a_run_it_cannot_stop( + sqlite_session: Session, app_task_workflow_run: WorkflowRun, change: dict[str, object] + ) -> None: + run = app_task_workflow_run + for name, value in change.items(): + setattr(run, name, value) + sqlite_session.commit() + redis = _AppTaskServiceStopredis() + + outcome = AppTaskControlService(redis_client=redis).stop_workflow_run( + tenant_id=run.tenant_id, + app_id=run.app_id, + run_id=run.id, + app_mode=AppMode.WORKFLOW, + caller_role=CreatorUserRole.ACCOUNT, + caller_id=_USER_ID, + ) + + assert outcome is WorkflowStopOutcome.NOT_STOPPABLE + assert redis.operations == [] + @staticmethod @pytest.mark.parametrize("owner", [None, b"account-another-user"]) def test_unchecked_workflow_stop_skips_ownership_read(owner: bytes | None) -> None: diff --git a/cli/src/call/execute.test.ts b/cli/src/call/execute.test.ts index 889fd6fcff4..637f90eb510 100644 --- a/cli/src/call/execute.test.ts +++ b/cli/src/call/execute.test.ts @@ -165,7 +165,7 @@ it('a form submit prints the hint to follow the run', async () => { expect(JSON.parse(w.io.outBuf()).hints).toEqual([ expect.objectContaining({ op: 'get.run.event', - input: { app_id: 'app-2', task_id: 'wf-run-hitl-1', continue_on_pause: false }, + input: { app_id: 'app-2', run_id: 'wf-run-hitl-1', continue_on_pause: false }, }), ]) }) diff --git a/cli/src/plugins/catalog/types.ts b/cli/src/plugins/catalog/types.ts index e154372415f..97baa0b06d0 100644 --- a/cli/src/plugins/catalog/types.ts +++ b/cli/src/plugins/catalog/types.ts @@ -16,13 +16,14 @@ export type CatalogOp = Readonly<{ bind: Readonly> internal: boolean deprecated: boolean + effect?: string examples: readonly Example[] }> export type CatalogDoc = Readonly<{ ops: Readonly> }> const EXAMPLE_SCHEMA = z.object({ title: z.string(), input: z.record(z.string(), z.unknown()) }) -// kind and bind values are unrestricted here: unknown kinds/binds are a forward-compat +// kind, effect and bind values are unrestricted here: unknown kinds, binds and effects are a forward-compat // contract, not a parse error. const CATALOG_OP_SCHEMA = z .object({ @@ -34,6 +35,7 @@ const CATALOG_OP_SCHEMA = z bind: z.record(z.string(), z.string()), internal: z.boolean(), deprecated: z.boolean(), + effect: z.string().optional(), examples: z.array(EXAMPLE_SCHEMA).default([]), }) .catchall(z.unknown()) diff --git a/cli/src/plugins/commands/command.ts b/cli/src/plugins/commands/command.ts index 64deb027bc5..c4147270879 100644 --- a/cli/src/plugins/commands/command.ts +++ b/cli/src/plugins/commands/command.ts @@ -15,7 +15,12 @@ import { commandRow } from './describe' export type { Descriptor, OpFacets } from './describe' -export type CommandEffect = 'read' | 'write' | 'destructive' +export const COMMAND_EFFECTS = ['read', 'write', 'destructive'] as const +export type CommandEffect = (typeof COMMAND_EFFECTS)[number] + +export function isCommandEffect(value: string): value is CommandEffect { + return (COMMAND_EFFECTS as readonly string[]).includes(value) +} export const HELP_WORD = 'help' diff --git a/cli/src/plugins/ops/command.test.ts b/cli/src/plugins/ops/command.test.ts index dc03a638f66..7cdc11105a3 100644 --- a/cli/src/plugins/ops/command.test.ts +++ b/cli/src/plugins/ops/command.test.ts @@ -9,7 +9,7 @@ import { callOptionsSchema } from '@/call/flags' import { parseArgv } from '@/plugins/argv/parse' import { parseCatalog } from '@/plugins/catalog' import { propertiesOf } from '@/protocol/shape' -import { effectOf, opCommand } from './command' +import { effectOf, opCommand, opEffect } from './command' const OPS = parseCatalog( readFileSync(join(__dirname, '../../../test/fixtures/catalog-v2.json')), @@ -99,6 +99,14 @@ it.each(EFFECTS)('effect of %s is %s', (method, effect) => { expect(effectOf(method)).toBe(effect) }) +it.each([ + ['destructive', 'destructive'], + [undefined, 'write'], + ['irreversible', undefined], +] as const)('a POST op declaring effect %s shows %s', (declared, shown) => { + expect(opEffect({ ...fixtureOp('stop.run'), effect: declared })).toBe(shown) +}) + it('help shows the business schema, the op facets, the call options and the pin', async () => { const w = await world() const row = await opCommand('get.console_app', list).help({ diff --git a/cli/src/plugins/ops/command.ts b/cli/src/plugins/ops/command.ts index ef7e88e2405..2be0f6b7f9b 100644 --- a/cli/src/plugins/ops/command.ts +++ b/cli/src/plugins/ops/command.ts @@ -9,7 +9,7 @@ import type { } from '@/plugins/commands/command' import { executeCall } from '@/call/execute' import { CALL_FLAG, CALL_FLAG_SCHEMA, CALL_FLAGS, callOptionsSchema } from '@/call/flags' -import { Command, HELP_WORD } from '@/plugins/commands/command' +import { Command, HELP_WORD, isCommandEffect } from '@/plugins/commands/command' import { GLOBAL_INPUT } from '@/plugins/global-flags' import { propertiesOf } from '@/protocol/shape' import { pinsFor } from './pins' @@ -39,6 +39,13 @@ export function effectOf(method: string): CommandEffect | undefined { return METHOD_EFFECT[method.toUpperCase()] } +// An absent effect is an older server: the method decides. A value this CLI does not +// know shows no label rather than a guess. +export function opEffect(op: CatalogOp): CommandEffect | undefined { + if (op.effect === undefined) return effectOf(op.method) + return isCommandEffect(op.effect) ? op.effect : undefined +} + // Parse-only: the op's own fields minus the reserved names, plus the call options its // kind accepts. `required` is left out on purpose — executeCall validates the assembled // body against the op's real schema, which the fields alone do not have to satisfy. @@ -57,7 +64,7 @@ export function opCommand(id: string, op: CatalogOp): CommandConstructor { const parseSchema = flagSchema(op, options) return class OpCommand extends Command { static override summary = op.summary - static override effect = effectOf(op.method) + static override effect = opEffect(op) static override examples = op.examples static override hidden = op.internal diff --git a/cli/test/fixtures/catalog-v2.json b/cli/test/fixtures/catalog-v2.json index 613cc038677..4341995a492 100644 --- a/cli/test/fixtures/catalog-v2.json +++ b/cli/test/fixtures/catalog-v2.json @@ -1068,7 +1068,7 @@ "app_id": "path", "continue_on_pause": "query", "include_state_snapshot": "query", - "task_id": "path" + "run_id": "path" }, "deprecated": false, "input": { @@ -1088,17 +1088,17 @@ "title": "Include State Snapshot", "type": "boolean" }, - "task_id": { + "run_id": { "type": "string" } }, - "required": ["app_id", "task_id"], + "required": ["app_id", "run_id"], "type": "object" }, "internal": false, "kind": "sse", "method": "GET", - "path": "/openapi/v1/apps/{app_id}/tasks/{task_id}/events", + "path": "/openapi/v1/apps/{app_id}/runs/{run_id}/events", "summary": "Stream the events of a workflow run" }, "describe.run.form": { @@ -1216,7 +1216,7 @@ "stop.run": { "bind": { "app_id": "path", - "task_id": "path" + "run_id": "path" }, "deprecated": false, "input": { @@ -1224,17 +1224,17 @@ "app_id": { "type": "string" }, - "task_id": { + "run_id": { "type": "string" } }, - "required": ["app_id", "task_id"], + "required": ["app_id", "run_id"], "type": "object" }, "internal": false, "kind": "object", "method": "POST", - "path": "/openapi/v1/apps/{app_id}/tasks/{task_id}:stop", + "path": "/openapi/v1/apps/{app_id}/runs/{run_id}:stop", "summary": "Stop a running task" }, "describe.workspace": { diff --git a/cli/test/fixtures/catalog.json b/cli/test/fixtures/catalog.json index 8677f45da3c..85e3badda2c 100644 --- a/cli/test/fixtures/catalog.json +++ b/cli/test/fixtures/catalog.json @@ -727,7 +727,7 @@ "app_id": "path", "continue_on_pause": "query", "include_state_snapshot": "query", - "task_id": "path" + "run_id": "path" }, "deprecated": false, "input": { @@ -745,15 +745,15 @@ "title": "Include State Snapshot", "type": "boolean" }, - "task_id": { "type": "string" } + "run_id": { "type": "string" } }, - "required": ["app_id", "task_id"], + "required": ["app_id", "run_id"], "type": "object" }, "internal": false, "kind": "sse", "method": "GET", - "path": "/openapi/v1/apps/{app_id}/tasks/{task_id}/events", + "path": "/openapi/v1/apps/{app_id}/runs/{run_id}/events", "summary": "Stream the events of a workflow run" }, "describe.run.form": { @@ -844,6 +844,20 @@ "summary": "Submit a human-input form" }, "stop.run": { + "bind": { "app_id": "path", "run_id": "path" }, + "deprecated": false, + "input": { + "properties": { "app_id": { "type": "string" }, "run_id": { "type": "string" } }, + "required": ["app_id", "run_id"], + "type": "object" + }, + "internal": false, + "kind": "object", + "method": "POST", + "path": "/openapi/v1/apps/{app_id}/runs/{run_id}:stop", + "summary": "Stop a running or paused run of a workflow or advanced-chat app" + }, + "stop.task": { "bind": { "app_id": "path", "task_id": "path" }, "deprecated": false, "input": { @@ -855,7 +869,7 @@ "kind": "object", "method": "POST", "path": "/openapi/v1/apps/{app_id}/tasks/{task_id}:stop", - "summary": "Stop a running task" + "summary": "Stop a streaming reply of a chat, agent-chat or completion app" }, "describe.workspace": { "bind": { "workspace_id": "path" }, diff --git a/cli/test/fixtures/dify-mock/server.ts b/cli/test/fixtures/dify-mock/server.ts index 4dc78c65ae6..1f006197142 100644 --- a/cli/test/fixtures/dify-mock/server.ts +++ b/cli/test/fixtures/dify-mock/server.ts @@ -761,7 +761,7 @@ export function buildApp(getScenario: () => Scenario, state?: MockState): Hono { ) }) - app.post('/openapi/v1/apps/:id/tasks/:taskId:stop', (c) => { + app.post('/openapi/v1/apps/:id/runs/:runId:stop', (c) => { return c.json({ result: 'success' }) }) @@ -772,13 +772,13 @@ export function buildApp(getScenario: () => Scenario, state?: MockState): Hono { { summary: 'Follow the run', op: OpId.RunEvents, - input: { app_id: c.req.param('id'), task_id: 'wf-run-hitl-1', continue_on_pause: false }, + input: { app_id: c.req.param('id'), run_id: 'wf-run-hitl-1', continue_on_pause: false }, }, ], }), ) - app.get('/openapi/v1/apps/:id/tasks/:task_id/events', (_c) => sseResponse(hitlResumedResponse())) + app.get('/openapi/v1/apps/:id/runs/:run_id/events', (_c) => sseResponse(hitlResumedResponse())) app.post('/openapi/v1/oauth/device/code', (c) => { const scenario = getScenario() diff --git a/packages/contracts/generated/api/openapi/orpc.gen.ts b/packages/contracts/generated/api/openapi/orpc.gen.ts index fb4af2cdf7f..b6f5b92fc30 100644 --- a/packages/contracts/generated/api/openapi/orpc.gen.ts +++ b/packages/contracts/generated/api/openapi/orpc.gen.ts @@ -41,6 +41,9 @@ import { zGetAppsByAppIdReleaseCheckPath, zGetAppsByAppIdReleaseCheckResponse, zGetAppsByAppIdResponse, + zGetAppsByAppIdRunsByRunIdEventsPath, + zGetAppsByAppIdRunsByRunIdEventsQuery, + zGetAppsByAppIdRunsByRunIdEventsResponse, zGetAppsByAppIdRunsByRunIdNodesPath, zGetAppsByAppIdRunsByRunIdNodesResponse, zGetAppsByAppIdRunsByRunIdPath, @@ -52,9 +55,6 @@ import { zGetAppsByAppIdServiceApiAgentResponse, zGetAppsByAppIdServiceApiPath, zGetAppsByAppIdServiceApiResponse, - zGetAppsByAppIdTasksByTaskIdEventsPath, - zGetAppsByAppIdTasksByTaskIdEventsQuery, - zGetAppsByAppIdTasksByTaskIdEventsResponse, zGetAppsByAppIdVersionsPath, zGetAppsByAppIdVersionsQuery, zGetAppsByAppIdVersionsResponse, @@ -219,6 +219,8 @@ import { zPostAppsByAppIdPublishBody, zPostAppsByAppIdPublishPath, zPostAppsByAppIdPublishResponse, + zPostAppsByAppIdRunsByRunIdStopPath, + zPostAppsByAppIdRunsByRunIdStopResponse, zPostAppsByAppIdTasksByTaskIdStopPath, zPostAppsByAppIdTasksByTaskIdStopResponse, zPostAppsByAppIdVersionsByVersionIdRestorePath, @@ -921,6 +923,26 @@ export const release = { } export const get17 = oc + .route({ + inputStructure: 'detailed', + method: 'GET', + operationId: 'getAppsByAppIdRunsByRunIdEvents', + path: '/apps/{app_id}/runs/{run_id}/events', + tags: ['openapi'], + }) + .input( + z.object({ + params: zGetAppsByAppIdRunsByRunIdEventsPath, + query: zGetAppsByAppIdRunsByRunIdEventsQuery.optional(), + }), + ) + .output(zGetAppsByAppIdRunsByRunIdEventsResponse) + +export const events = { + get: get17, +} + +export const get18 = oc .route({ inputStructure: 'detailed', method: 'GET', @@ -932,10 +954,25 @@ export const get17 = oc .output(zGetAppsByAppIdRunsByRunIdNodesResponse) export const nodes3 = { - get: get17, + get: get18, } -export const get18 = oc +export const post10 = oc + .route({ + inputStructure: 'detailed', + method: 'POST', + operationId: 'postAppsByAppIdRunsByRunIdStop', + path: '/apps/{app_id}/runs/{run_id}:stop', + tags: ['openapi'], + }) + .input(z.object({ params: zPostAppsByAppIdRunsByRunIdStopPath })) + .output(zPostAppsByAppIdRunsByRunIdStopResponse) + +export const stop = { + post: post10, +} + +export const get19 = oc .route({ inputStructure: 'detailed', method: 'GET', @@ -947,11 +984,13 @@ export const get18 = oc .output(zGetAppsByAppIdRunsByRunIdResponse) export const byRunId = { - get: get18, + get: get19, + events, nodes: nodes3, + stop, } -export const get19 = oc +export const get20 = oc .route({ inputStructure: 'detailed', method: 'GET', @@ -963,11 +1002,11 @@ export const get19 = oc .output(zGetAppsByAppIdRunsResponse) export const runs = { - get: get19, + get: get20, byRunId, } -export const get20 = oc +export const get21 = oc .route({ inputStructure: 'detailed', method: 'GET', @@ -995,11 +1034,11 @@ export const patch7 = oc .output(zPatchAppsByAppIdServiceApiAgentResponse) export const agent2 = { - get: get20, + get: get21, patch: patch7, } -export const get21 = oc +export const get22 = oc .route({ inputStructure: 'detailed', method: 'GET', @@ -1024,32 +1063,12 @@ export const patch8 = oc .output(zPatchAppsByAppIdServiceApiResponse) export const serviceApi = { - get: get21, + get: get22, patch: patch8, agent: agent2, } -export const get22 = oc - .route({ - inputStructure: 'detailed', - method: 'GET', - operationId: 'getAppsByAppIdTasksByTaskIdEvents', - path: '/apps/{app_id}/tasks/{task_id}/events', - tags: ['openapi'], - }) - .input( - z.object({ - params: zGetAppsByAppIdTasksByTaskIdEventsPath, - query: zGetAppsByAppIdTasksByTaskIdEventsQuery.optional(), - }), - ) - .output(zGetAppsByAppIdTasksByTaskIdEventsResponse) - -export const events = { - get: get22, -} - -export const post10 = oc +export const post11 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1060,20 +1079,19 @@ export const post10 = oc .input(z.object({ params: zPostAppsByAppIdTasksByTaskIdStopPath })) .output(zPostAppsByAppIdTasksByTaskIdStopResponse) -export const stop = { - post: post10, +export const stop2 = { + post: post11, } export const byTaskId = { - events, - stop, + stop: stop2, } export const tasks = { byTaskId, } -export const post11 = oc +export const post12 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1085,7 +1103,7 @@ export const post11 = oc .output(zPostAppsByAppIdVersionsByVersionIdRestoreResponse) export const restore = { - post: post11, + post: post12, } export const byVersionId = { @@ -1207,7 +1225,7 @@ export const advancedChat4 = { patch: patch9, } -export const post12 = oc +export const post13 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1219,7 +1237,7 @@ export const post12 = oc .output(zPostAppsByAppIdWebappAgentResetResponse) export const reset = { - post: post12, + post: post13, } export const get27 = oc @@ -1377,7 +1395,7 @@ export const workflow3 = { patch: patch14, } -export const post13 = oc +export const post14 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1389,7 +1407,7 @@ export const post13 = oc .output(zPostAppsByAppIdWebappResetResponse) export const reset2 = { - post: post13, + post: post14, } export const webapp = { @@ -1402,7 +1420,7 @@ export const webapp = { reset: reset2, } -export const post14 = oc +export const post15 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1416,14 +1434,14 @@ export const post14 = oc .output(zPostAppsByAppIdWorkflowRunResponse) export const run8 = { - post: post14, + post: post15, } export const workflow4 = { run: run8, } -export const post15 = oc +export const post16 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1435,7 +1453,7 @@ export const post15 = oc .output(zPostAppsByAppIdPublishResponse) export const publish = { - post: post15, + post: post16, } export const get32 = oc @@ -1518,7 +1536,7 @@ export const nodeTypes = { byNodeType, } -export const post16 = oc +export const post17 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1530,10 +1548,10 @@ export const post16 = oc .output(zPostOauthDeviceApproveResponse) export const approve = { - post: post16, + post: post17, } -export const post17 = oc +export const post18 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1545,10 +1563,10 @@ export const post17 = oc .output(zPostOauthDeviceCodeResponse) export const code = { - post: post17, + post: post18, } -export const post18 = oc +export const post19 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1560,7 +1578,7 @@ export const post18 = oc .output(zPostOauthDeviceDenyResponse) export const deny = { - post: post18, + post: post19, } export const get36 = oc @@ -1578,7 +1596,7 @@ export const lookup = { get: get36, } -export const post19 = oc +export const post20 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1590,7 +1608,7 @@ export const post19 = oc .output(zPostOauthDeviceTokenResponse) export const token = { - post: post19, + post: post20, } export const device = { @@ -1641,7 +1659,7 @@ export const permittedExternalApps = { byAppId: byAppId2, } -export const post20 = oc +export const post21 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1659,10 +1677,10 @@ export const post20 = oc .output(zPostWorkspacesByWorkspaceIdAppsAdvancedChatResponse) export const advancedChat5 = { - post: post20, + post: post21, } -export const post21 = oc +export const post22 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1674,14 +1692,14 @@ export const post21 = oc .output(zPostWorkspacesByWorkspaceIdAppsImportsByImportIdConfirmResponse) export const confirm = { - post: post21, + post: post22, } export const byImportId = { confirm, } -export const post22 = oc +export const post23 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1698,10 +1716,10 @@ export const post22 = oc .output(zPostWorkspacesByWorkspaceIdAppsImportsCheckResponse) export const check3 = { - post: post22, + post: post23, } -export const post23 = oc +export const post24 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1718,12 +1736,12 @@ export const post23 = oc .output(zPostWorkspacesByWorkspaceIdAppsImportsResponse) export const imports = { - post: post23, + post: post24, byImportId, check: check3, } -export const post24 = oc +export const post25 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -1741,7 +1759,7 @@ export const post24 = oc .output(zPostWorkspacesByWorkspaceIdAppsWorkflowResponse) export const workflow5 = { - post: post24, + post: post25, } export const apps2 = { @@ -1803,7 +1821,7 @@ export const cat = { * * Performs a side-effect-free comparison, equivalent to difyctl fs diff. POST is used because this is a structured query and an optional semantic summary can consume model quota. Automatic retries are not safe when include_semantic_summary=true because each retry can consume quota again. Requires an OAuth account bearer with WORKSPACE_READ. Authentication and workspace scope are checked before request validation, and Agent Knowledge Base membership is revalidated for every call. A hidden or missing Agent Knowledge Base or entry uses the same 404 response. These operations are read-only and do not emit mutation audit events. */ -export const post25 = oc +export const post26 = oc .route({ description: 'Performs a side-effect-free comparison, equivalent to difyctl fs diff. POST is used because this is a structured query and an optional semantic summary can consume model quota. Automatic retries are not safe when include_semantic_summary=true because each retry can consume quota again. Requires an OAuth account bearer with WORKSPACE_READ. Authentication and workspace scope are checked before request validation, and Agent Knowledge Base membership is revalidated for every call. A hidden or missing Agent Knowledge Base or entry uses the same 404 response. These operations are read-only and do not emit mutation audit events.', @@ -1823,7 +1841,7 @@ export const post25 = oc .output(zPostWorkspacesByWorkspaceIdKnowledgeFsKnowledgeSpacesByKnowledgeSpaceIdFsDiffResponse) export const diff = { - post: post25, + post: post26, } /** @@ -2060,7 +2078,7 @@ export const get47 = oc ) .output(zGetWorkspacesByWorkspaceIdMembersResponse) -export const post26 = oc +export const post27 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -2079,7 +2097,7 @@ export const post26 = oc export const members = { get: get47, - post: post26, + post: post27, byMemberId, } @@ -2103,7 +2121,7 @@ export const byCredentialId = { patch: patch16, } -export const post27 = oc +export const post28 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -2121,7 +2139,7 @@ export const post27 = oc .output(zPostWorkspacesByWorkspaceIdModelProvidersByProviderCredentialsResponse) export const credentials = { - post: post27, + post: post28, byCredentialId, } @@ -2149,7 +2167,7 @@ export const byCredentialId2 = { patch: patch17, } -export const post28 = oc +export const post29 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -2167,7 +2185,7 @@ export const post28 = oc .output(zPostWorkspacesByWorkspaceIdModelProvidersByProviderModelsCredentialsResponse) export const credentials2 = { - post: post28, + post: post29, byCredentialId: byCredentialId2, } @@ -2250,7 +2268,7 @@ export const byPluginId = { delete: delete5, } -export const post29 = oc +export const post30 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -2267,7 +2285,7 @@ export const post29 = oc .output(zPostWorkspacesByWorkspaceIdPluginsInstallResponse) export const install = { - post: post29, + post: post30, } export const get51 = oc @@ -2312,7 +2330,7 @@ export const byCredentialId3 = { patch: patch18, } -export const post30 = oc +export const post31 = oc .route({ inputStructure: 'detailed', method: 'POST', @@ -2330,7 +2348,7 @@ export const post30 = oc .output(zPostWorkspacesByWorkspaceIdToolProvidersByProviderCredentialsResponse) export const credentials3 = { - post: post30, + post: post31, byCredentialId: byCredentialId3, } diff --git a/packages/contracts/generated/api/openapi/types.gen.ts b/packages/contracts/generated/api/openapi/types.gen.ts index bbcf1743f39..849a517f96f 100644 --- a/packages/contracts/generated/api/openapi/types.gen.ts +++ b/packages/contracts/generated/api/openapi/types.gen.ts @@ -1093,6 +1093,7 @@ export type OpenApiErrorCode = | 'recipient_surface_mismatch' | 'request_entity_too_large' | 'run_not_found' + | 'run_not_stoppable' | 'secret_mask_not_secret' | 'secret_mask_unknown_id' | 'service_api_disabled' @@ -2619,6 +2620,34 @@ export type GetAppsByAppIdRunsByRunIdResponses = { export type GetAppsByAppIdRunsByRunIdResponse = GetAppsByAppIdRunsByRunIdResponses[keyof GetAppsByAppIdRunsByRunIdResponses] +export type GetAppsByAppIdRunsByRunIdEventsData = { + body?: never + path: { + app_id: string + run_id: string + } + query?: { + continue_on_pause?: boolean + include_state_snapshot?: boolean + } + url: '/apps/{app_id}/runs/{run_id}/events' +} + +export type GetAppsByAppIdRunsByRunIdEventsErrors = { + 422: ErrorBody + default: ErrorBody +} + +export type GetAppsByAppIdRunsByRunIdEventsError = + GetAppsByAppIdRunsByRunIdEventsErrors[keyof GetAppsByAppIdRunsByRunIdEventsErrors] + +export type GetAppsByAppIdRunsByRunIdEventsResponses = { + 200: EventStreamResponse +} + +export type GetAppsByAppIdRunsByRunIdEventsResponse = + GetAppsByAppIdRunsByRunIdEventsResponses[keyof GetAppsByAppIdRunsByRunIdEventsResponses] + export type GetAppsByAppIdRunsByRunIdNodesData = { body?: never path: { @@ -2643,6 +2672,30 @@ export type GetAppsByAppIdRunsByRunIdNodesResponses = { export type GetAppsByAppIdRunsByRunIdNodesResponse = GetAppsByAppIdRunsByRunIdNodesResponses[keyof GetAppsByAppIdRunsByRunIdNodesResponses] +export type PostAppsByAppIdRunsByRunIdStopData = { + body?: never + path: { + app_id: string + run_id: string + } + query?: never + url: '/apps/{app_id}/runs/{run_id}:stop' +} + +export type PostAppsByAppIdRunsByRunIdStopErrors = { + default: ErrorBody +} + +export type PostAppsByAppIdRunsByRunIdStopError = + PostAppsByAppIdRunsByRunIdStopErrors[keyof PostAppsByAppIdRunsByRunIdStopErrors] + +export type PostAppsByAppIdRunsByRunIdStopResponses = { + 200: TaskStopResponse +} + +export type PostAppsByAppIdRunsByRunIdStopResponse = + PostAppsByAppIdRunsByRunIdStopResponses[keyof PostAppsByAppIdRunsByRunIdStopResponses] + export type GetAppsByAppIdServiceApiData = { body?: never path: { @@ -2737,34 +2790,6 @@ export type PatchAppsByAppIdServiceApiAgentResponses = { export type PatchAppsByAppIdServiceApiAgentResponse = PatchAppsByAppIdServiceApiAgentResponses[keyof PatchAppsByAppIdServiceApiAgentResponses] -export type GetAppsByAppIdTasksByTaskIdEventsData = { - body?: never - path: { - app_id: string - task_id: string - } - query?: { - continue_on_pause?: boolean - include_state_snapshot?: boolean - } - url: '/apps/{app_id}/tasks/{task_id}/events' -} - -export type GetAppsByAppIdTasksByTaskIdEventsErrors = { - 422: ErrorBody - default: ErrorBody -} - -export type GetAppsByAppIdTasksByTaskIdEventsError = - GetAppsByAppIdTasksByTaskIdEventsErrors[keyof GetAppsByAppIdTasksByTaskIdEventsErrors] - -export type GetAppsByAppIdTasksByTaskIdEventsResponses = { - 200: EventStreamResponse -} - -export type GetAppsByAppIdTasksByTaskIdEventsResponse = - GetAppsByAppIdTasksByTaskIdEventsResponses[keyof GetAppsByAppIdTasksByTaskIdEventsResponses] - export type PostAppsByAppIdTasksByTaskIdStopData = { body?: never path: { diff --git a/packages/contracts/generated/api/openapi/zod.gen.ts b/packages/contracts/generated/api/openapi/zod.gen.ts index 8f152ebbea9..36b0c467ec1 100644 --- a/packages/contracts/generated/api/openapi/zod.gen.ts +++ b/packages/contracts/generated/api/openapi/zod.gen.ts @@ -1463,6 +1463,7 @@ export const zOpenApiErrorCode = z.enum([ 'recipient_surface_mismatch', 'request_entity_too_large', 'run_not_found', + 'run_not_stoppable', 'secret_mask_not_secret', 'secret_mask_unknown_id', 'service_api_disabled', @@ -1897,8 +1898,8 @@ export const zPermittedExternalAppsListQuery = z.object({ /** * TaskStopResponse * - * 200 body for POST /apps//tasks/:stop. The handler always returns - * {"result": "success"}, so `result` is required (no default) — the generated contract + * 200 body for POST /apps//runs/:stop and POST /apps//tasks/:stop. + * The handlers always return {"result": "success"}, so `result` is required (no default) — the generated contract * types it as a required `'success'` rather than an optional field. */ export const zTaskStopResponse = z.object({ @@ -2815,6 +2816,21 @@ export const zGetAppsByAppIdRunsByRunIdPath = z.object({ */ export const zGetAppsByAppIdRunsByRunIdResponse = zWorkflowRunDetailResponse +export const zGetAppsByAppIdRunsByRunIdEventsPath = z.object({ + app_id: z.string(), + run_id: z.string(), +}) + +export const zGetAppsByAppIdRunsByRunIdEventsQuery = z.object({ + continue_on_pause: z.boolean().optional().default(false), + include_state_snapshot: z.boolean().optional().default(false), +}) + +/** + * SSE event stream + */ +export const zGetAppsByAppIdRunsByRunIdEventsResponse = zEventStreamResponse + export const zGetAppsByAppIdRunsByRunIdNodesPath = z.object({ app_id: z.string(), run_id: z.string(), @@ -2825,6 +2841,16 @@ export const zGetAppsByAppIdRunsByRunIdNodesPath = z.object({ */ export const zGetAppsByAppIdRunsByRunIdNodesResponse = zWorkflowRunNodeExecutionListResponse +export const zPostAppsByAppIdRunsByRunIdStopPath = z.object({ + app_id: z.string(), + run_id: z.string(), +}) + +/** + * Run stopped + */ +export const zPostAppsByAppIdRunsByRunIdStopResponse = zTaskStopResponse + export const zGetAppsByAppIdServiceApiPath = z.object({ app_id: z.string(), }) @@ -2865,21 +2891,6 @@ export const zPatchAppsByAppIdServiceApiAgentPath = z.object({ */ export const zPatchAppsByAppIdServiceApiAgentResponse = zAgentServiceApi -export const zGetAppsByAppIdTasksByTaskIdEventsPath = z.object({ - app_id: z.string(), - task_id: z.string(), -}) - -export const zGetAppsByAppIdTasksByTaskIdEventsQuery = z.object({ - continue_on_pause: z.boolean().optional().default(false), - include_state_snapshot: z.boolean().optional().default(false), -}) - -/** - * SSE event stream - */ -export const zGetAppsByAppIdTasksByTaskIdEventsResponse = zEventStreamResponse - export const zPostAppsByAppIdTasksByTaskIdStopPath = z.object({ app_id: z.string(), task_id: z.string(), diff --git a/packages/contracts/non-json-openapi-responses.md b/packages/contracts/non-json-openapi-responses.md index 073394cb992..01b123a5700 100644 --- a/packages/contracts/non-json-openapi-responses.md +++ b/packages/contracts/non-json-openapi-responses.md @@ -37,7 +37,7 @@ This inventory records non-JSON runtime responses and their schema intent. Legac | console | GET | `/apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/events` | `text/event-stream` | `EventStreamResponse` | | console | GET | `/apps/{app_id}/workflows/published/runs/{run_id}/node-outputs/events` | `text/event-stream` | `EventStreamResponse` | | openapi | POST | `/apps/{app_id}/run` | `text/event-stream` | `EventStreamResponse` | -| openapi | GET | `/apps/{app_id}/tasks/{task_id}/events` | `text/event-stream` | `EventStreamResponse` | +| openapi | GET | `/apps/{app_id}/runs/{run_id}/events` | `text/event-stream` | `EventStreamResponse` | | service | GET | `/workflow/{task_id}/events` | `text/event-stream` | `EventStreamResponse` | | web | GET | `/workflow/{task_id}/events` | `text/event-stream` | `EventStreamResponse` | diff --git a/skills/difyctl/SKILL.md b/skills/difyctl/SKILL.md index 9689c1d901d..88b3de7cff2 100644 --- a/skills/difyctl/SKILL.md +++ b/skills/difyctl/SKILL.md @@ -9,7 +9,7 @@ difyctl talks to a Dify server. Business commands are not built in: the server p Always pass `--json`. Help and errors are text on a terminal and JSON in a pipe; `--json` (or `DIFY_OUTPUT=json`) forces JSON. Results and streams are always JSON. -Operations that remove or revoke have no confirmation prompt. Confirm with the human first. +Commands whose effect is destructive can't be undone and have no confirmation prompt. Check effect in help and confirm with the human first. Also confirm a server operation that shows no effect. ## Finding what the CLI can do @@ -20,7 +20,7 @@ Four views under `help`, covering local commands and server operations as one tr 1. `difyctl help --json`: the map. Start here. 2. `difyctl help --json`: everything under one namespace (`difyctl help get`). The bare namespace, `difyctl get --json`, prints the same. 3. `difyctl help --json`: search by plain words (`difyctl help upload file`), top 20 entries plus `total`. Widen the words when nothing fits. -4. `difyctl help --json`: one descriptor, dotted or spaced id. For an operation: `input` (JSON Schema), `bind`, `kind`, `examples` and `pins` (values difyctl fills in, currently `workspace_id`). ` --help` prints the same. +4. `difyctl help --json`: one descriptor, dotted or spaced id. For an operation: `input` (JSON Schema), `bind`, `kind`, `effect`, `examples` and `pins` (values difyctl fills in, currently `workspace_id`). ` --help` prints the same. `--all` adds operations the server marks internal. `--full` on the bare `help` prints every descriptor; rarely needed. diff --git a/skills/difyctl/references/operations.md b/skills/difyctl/references/operations.md index 340079f2d8e..00f42d457c3 100644 --- a/skills/difyctl/references/operations.md +++ b/skills/difyctl/references/operations.md @@ -16,6 +16,11 @@ Streaming ops fold to `{status, text, outputs?, total_tokens?, message_id?, conv After `submit.run.form`, run its `get.run.event` hint to follow the run; repeat if it suspends again. +To stop a run that is running or waiting on a form, use its `run_id` from the folded result. + +Workflow and advanced-chat runs stop with `difyctl stop run --app-id --run-id `. +Chat, agent-chat and completion replies stop with `difyctl stop task --app-id --task-id `, with `task_id` from the stream's events. + ## Errors and exit codes Stderr envelope: `{"error":{"code","message","hint"?,"details"?,"schema"?,"http_status"?,"server"?}}`. `code` is difyctl's (`server_error` for server rejections); the server's code is `error.server.code` (e.g. `dsl_invalid`). `details`: `[{loc,msg}]`.