feat(api): let agents edit, test and publish workflow apps over openapi

DSL export returns a draft_hash covering the graph, features, environment
and conversation variables; an import that sends an older hash fails
instead of wiping newer edits. A blank secret imported over an app keeps
the stored value. The incompatible-node error names the nodes.

New openapi ops: draft test runs, run history with node steps, publish,
version list and restore, and environment variables (secrets masked on
read). Publish shares one service method with the console.

Author: GareArc

api/controllers/console/app/workflow.py

@@ -87,7 +87,6 @@
 from graphon.variables import SecretVariable, SegmentType, VariableBase
 from graphon.variables.exc import VariableError
 from libs import helper
-from libs.datetime_utils import naive_utc_now
 from libs.helper import TimestampField, dump_response, to_timestamp, uuid_value
 from libs.login import login_required
 from models import Account, App
@@ -101,10 +100,7 @@
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_ref_service import WorkflowRefService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
-from services.workflow_variable_reference_validator import (
-    format_variable_reference_errors,
-    validate_variable_references,
-)
+from services.workflow_variable_reference_validator import advisory_variable_reference_warning
 
 logger = logging.getLogger(__name__)
 
@@ -1274,21 +1270,6 @@ def post(self, payload: DraftWorkflowNodeRunPayload, current_user: Account, app_
         ).model_dump(mode="json")
 
 
-def _advisory_variable_reference_warning(graph_text: str | None) -> str | None:
-    """Return a non-blocking publish warning. A checker failure must not fail publish."""
-    if not graph_text:
-        return None
-    try:
-        graph = json.loads(graph_text)
-        if not isinstance(graph, dict):
-            return None
-        issues = validate_variable_references(graph)
-        return format_variable_reference_errors(issues) if issues else None
-    except Exception:
-        logger.warning("Skipped advisory variable reference check", exc_info=True)
-        return None
-
-
 @console_ns.route("/apps/<uuid:app_id>/workflows/publish")
 class PublishedWorkflowApi(Resource):
     @console_ns.doc("get_published_workflow")
@@ -1337,25 +1318,18 @@ def post(self, args: PublishWorkflowPayload, current_user: Account, app_model: A
 
         workflow_service = WorkflowService()
         with sessionmaker(db.engine).begin() as session:
-            workflow = workflow_service.publish_workflow(
+            workflow = workflow_service.publish_app_workflow(
                 session=session,
                 app_model=app_model,
                 account=current_user,
                 marked_name=args.marked_name or "",
                 marked_comment=args.marked_comment or "",
             )
 
-            # Update app_model within the same session to ensure atomicity
-            app_model_in_session = session.get(App, app_model.id)
-            if app_model_in_session:
-                app_model_in_session.workflow_id = workflow.id
-                app_model_in_session.updated_by = current_user.id
-                app_model_in_session.updated_at = naive_utc_now()
-
             workflow_created_at = TimestampField().format(workflow.created_at)
             graph_text = workflow.graph
 
-        warning = _advisory_variable_reference_warning(graph_text)
+        warning = advisory_variable_reference_warning(graph_text)
         payload: dict[str, object] = {
             "result": "success",
             "created_at": workflow_created_at,

api/controllers/openapi/__init__.py

@@ -61,6 +61,9 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     DeviceMutateResponse,
     DevicePollRequest,
     DeviceTokenResponse,
+    DraftWorkflowRunPayload,
+    EnvVariableListResponse,
+    EnvVariableSetPayload,
     FileUploadPayload,
     FormSubmitResponse,
     HealthResponse,
@@ -77,13 +80,20 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     OpenApiFormSubmitPayload,
     PermittedExternalAppsListQuery,
     PermittedExternalAppsListResponse,
+    PublishPayload,
+    PublishResponse,
+    RestoreResponse,
     RevokeResponse,
+    RunListQuery,
+    RunListResponse,
     ServerVersionResponse,
     SessionListQuery,
     SessionListResponse,
     SessionRow,
     TaskStopResponse,
     UsageInfo,
+    VersionListQuery,
+    VersionListResponse,
     WorkflowRunData,
     WorkflowRunPayload,
     WorkspaceDetailResponse,
@@ -93,6 +103,7 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     WorkspaceSummaryResponse,
 )
 from fields.file_fields import FileResponse
+from fields.workflow_run_fields import WorkflowRunDetailResponse, WorkflowRunNodeExecutionListResponse
 from services.entities.dsl_entities import CheckDependenciesResult, Import
 
 register_schema_models(
@@ -108,13 +119,18 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     DevicePollRequest,
     DeviceLookupQuery,
     DeviceMutateRequest,
+    DraftWorkflowRunPayload,
     FileUploadPayload,
     MemberInvitePayload,
     MemberListQuery,
     MemberRoleUpdatePayload,
     OpenApiFormSubmitPayload,
+    EnvVariableSetPayload,
     PermittedExternalAppsListQuery,
+    PublishPayload,
+    RunListQuery,
     SessionListQuery,
+    VersionListQuery,
     WorkflowRunPayload,
     WorkspaceListQuery,
 )
@@ -160,6 +176,13 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     FileResponse,
     ServerVersionResponse,
     HealthResponse,
+    RunListResponse,
+    PublishResponse,
+    VersionListResponse,
+    RestoreResponse,
+    EnvVariableListResponse,
+    WorkflowRunDetailResponse,
+    WorkflowRunNodeExecutionListResponse,
 )
 # Standalone definition for contract codegen; ErrorBody.code stays an open
 # string on the wire so old clients keep parsing future codes.
@@ -169,6 +192,7 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     account,
     app_dsl,
     app_run,
+    app_workflow,
     apps,
     apps_permitted_external,
     files,
@@ -186,6 +210,7 @@ def finish_document(self, document: dict[str, object]) -> dict[str, object]:
     "account",
     "app_dsl",
     "app_run",
+    "app_workflow",
     "apps",
     "apps_permitted_external",
     "files",

api/controllers/openapi/_errors.py

@@ -68,6 +68,12 @@ class OpenApiErrorCode(StrEnum):
     RECIPIENT_SURFACE_MISMATCH = "recipient_surface_mismatch"
     CATALOG_STALE = "catalog_stale"
     TRIGGER_WORKFLOW_SERVICE_MODE_UNAVAILABLE = "trigger_workflow_service_mode_unavailable"
+    RUN_NOT_FOUND = "run_not_found"
+    VERSION_NOT_FOUND = "version_not_found"
+    VERSION_NOT_RESTORABLE = "version_not_restorable"
+    DRAFT_NOT_FOUND = "draft_not_found"
+    SECRET_MASK_UNKNOWN_ID = "secret_mask_unknown_id"
+    SECRET_MASK_NOT_SECRET = "secret_mask_not_secret"
 
 
 class ErrorDetail(BaseModel):
@@ -270,3 +276,39 @@ class InvalidFilePart(OpenApiError):  # noqa: N818
     code = 422
     error_code = OpenApiErrorCode.INVALID_PARAM
     description = "A request part could not be used"
+
+
+class RunNotFound(OpenApiError):  # noqa: N818
+    code = 404
+    error_code = OpenApiErrorCode.RUN_NOT_FOUND
+    description = "No run with this id exists for this app."
+
+
+class VersionNotFound(OpenApiError):  # noqa: N818
+    code = 404
+    error_code = OpenApiErrorCode.VERSION_NOT_FOUND
+    description = "No version with this id exists for this app."
+
+
+class VersionNotRestorable(OpenApiError):  # noqa: N818
+    code = 400
+    error_code = OpenApiErrorCode.VERSION_NOT_RESTORABLE
+    description = "Only a published version can be restored."
+
+
+class DraftNotFound(OpenApiError):  # noqa: N818
+    code = 404
+    error_code = OpenApiErrorCode.DRAFT_NOT_FOUND
+    description = "This app has no draft."
+
+
+class SecretMaskUnknownId(OpenApiError):  # noqa: N818
+    code = 400
+    error_code = OpenApiErrorCode.SECRET_MASK_UNKNOWN_ID
+    description = "The value is the masked secret placeholder, but env_id is not an existing secret."
+
+
+class SecretMaskNotSecret(OpenApiError):  # noqa: N818
+    code = 400
+    error_code = OpenApiErrorCode.SECRET_MASK_NOT_SECRET
+    description = "The value is the masked secret placeholder, which only a secret value_type can keep."

api/controllers/openapi/_models.py

@@ -12,7 +12,9 @@
 from controllers.common.human_input import HumanInputFormSubmitPayload
 from controllers.openapi._upload import UploadPart, UploadParts
 from enums import DeploymentEdition
-from libs.helper import EmailStr, UUIDStr, UUIDStrOrEmpty, uuid_value
+from fields.workflow_run_fields import WorkflowRunPaginationResponse
+from graphon.variables import SegmentType
+from libs.helper import EmailStr, UUIDStr, UUIDStrOrEmpty, to_timestamp, uuid_value
 from models.model import AppMode
 from services.app_dsl_service import Import
 
@@ -343,21 +345,23 @@ class RunPayloadBase(BaseModel):
     """What every run takes; each mode's payload adds its own fields and forbids the rest."""
 
     inputs: dict[str, Any] = Field(
+        default_factory=dict,
         description=(
             "Variables declared by the app. The exact shape is per app: read `input_schema` from "
             "describe.console_app. A file variable takes a Dify file mapping (remote url or upload id) here, "
-            "or a local file in `files`, not both."
-        )
+            "or a local path in `files`, not both."
+        ),
     )
     files: UploadParts | None = Field(
         default=None,
         description=(
-            "Local files keyed by the app's file variable name; the server uploads each one and sets "
-            "`inputs[<name>]`. Send a list (part name `files[<name>][]`) for a file-list variable"
+            "Local file paths keyed by the app's file variable name; each file is uploaded and becomes "
+            "that variable's value. Give a list of paths for a file-list variable"
         ),
     )
     attachments: list[UploadPart] | None = Field(
-        default=None, description="Local files attached to the run itself (the app's `sys.files`), not to a variable"
+        default=None,
+        description="Local file paths attached to the run itself (the app's `sys.files`), not to a variable",
     )
     workspace_id: UUIDStrOrEmpty | None = Field(default=None, description="Workspace that owns the app")
 
@@ -366,6 +370,10 @@ class WorkflowRunPayload(RunPayloadBase, _WorkflowVersionFields):
     model_config = ConfigDict(extra="forbid")
 
 
+class DraftWorkflowRunPayload(RunPayloadBase):
+    model_config = ConfigDict(extra="forbid")
+
+
 class ChatRunPayload(RunPayloadBase, _ConversationFields):
     model_config = ConfigDict(extra="forbid")
 
@@ -498,6 +506,11 @@ class AppDslImportPayload(BaseModel):
     icon: str | None = Field(None)
     icon_background: str | None = Field(None)
     app_id: str | None = Field(None, description="Existing app ID to overwrite (workflow/advanced-chat apps only)")
+    draft_hash: str | None = Field(
+        None,
+        description="draft_hash from the export or restore this import is based on. The import fails if the "
+        "draft's graph, features, environment variables or conversation variables changed since. Requires app_id",
+    )
 
     @model_validator(mode="after")
     def _validate_source_by_mode(self) -> AppDslImportPayload:
@@ -521,6 +534,11 @@ class AppDslExportResponse(BaseModel):
     """Export DSL response."""
 
     data: str = Field(..., description="DSL YAML string")
+    draft_hash: str | None = Field(
+        None,
+        description="Hash of the draft's graph, features, environment variables and conversation variables; "
+        "pass it to the import to refuse overwriting newer edits",
+    )
 
 
 class AppDslImportResponse(Import, Hinted):
@@ -540,7 +558,7 @@ class OpenApiFormSubmitPayload(HumanInputFormSubmitPayload):
 
     files: UploadParts | None = Field(
         default=None,
-        description="Local files keyed by the form's file input name, same convention as the run ops' `files`",
+        description="Local file paths keyed by the form's file input name, same convention as the run ops' `files`",
     )
 
 
@@ -550,3 +568,99 @@ class HumanInputFormDefinitionResponse(BaseModel):
     resolved_default_values: dict[str, str]
     user_actions: list[dict[str, Any]] = Field(default_factory=list)
     expiration_time: int | None = None
+
+
+class RunListQuery(BaseModel):
+    last_id: UUIDStr | None = Field(None, description="Cursor: id of the last run on the previous page")
+    limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
+    status: Literal["running", "succeeded", "failed", "stopped", "partial-succeeded"] | None = None
+    triggered_from: Literal["debugging", "app-run"] | None = Field(
+        None, description="debugging: draft test runs; app-run: real use. Omitted: debugging, as in the console"
+    )
+
+
+class RunListResponse(WorkflowRunPaginationResponse, Hinted):
+    """Cursor page of runs; `hints` carries the next page."""
+
+
+class PublishPayload(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    marked_name: str = Field("", max_length=20, description="Version name")
+    marked_comment: str = Field("", max_length=100, description="Version note")
+
+
+class PublishResponse(BaseModel):
+    version_id: str
+    created_at: int
+    warning: str | None = Field(None, description="Variable references that may read a skipped branch")
+
+
+class VersionListQuery(PageQuery):
+    named_only: bool = Field(False, description="Only versions that have a name")
+
+
+class VersionRow(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    marked_name: str
+    marked_comment: str
+    created_by: str | None = None
+    created_at: int
+
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _timestamp(cls, value: datetime | int) -> int | None:
+        return to_timestamp(value)
+
+
+class VersionListResponse(Hinted):
+    """Page of published versions, newest first; there is no total, `hints` carries the next page."""
+
+    page: int
+    limit: int
+    has_more: bool
+    data: list[VersionRow]
+
+
+class EnvVariableRow(BaseModel):
+    id: str = Field(description="What set and delete take to address this variable")
+    name: str = Field(description="What nodes use to refer to this variable")
+    description: str = ""
+    value_type: str
+    value: Any = Field(description="The value; a secret with a value is masked, an empty one reads as empty")
+
+
+class EnvVariableListResponse(BaseModel):
+    data: list[EnvVariableRow]
+
+
+class EnvVariableValueType(StrEnum):
+    """Value types the draft environment-variable ``set`` op accepts.
+
+    A curated subset of ``SegmentType``: what the console's environment-variable editor
+    allows (``ENVIRONMENT_VARIABLE_SUPPORTED_TYPES`` in controllers/console/app/workflow.py).
+    Members reference ``SegmentType.*.value`` so the subset relationship is type-checked.
+    """
+
+    STRING = SegmentType.STRING.value
+    NUMBER = SegmentType.NUMBER.value
+    SECRET = SegmentType.SECRET.value
+
+
+class EnvVariableSetPayload(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    name: str = Field(description="Variable name")
+    value_type: EnvVariableValueType = Field(description="string, number or secret")
+    value: Any = Field(description="The value; sending the masked value of an existing secret keeps the stored one")
+    description: str = Field("", description="What the variable is for")
+
+
+class RestoreResponse(BaseModel):
+    result: Literal["success"]
+    draft_hash: str = Field(
+        description="Hash of the restored draft's graph, features, environment variables and conversation "
+        "variables; pass it to a DSL import as draft_hash"
+    )