From ac85fb41abf05fe1e180756ab6fc980dd513e517 Mon Sep 17 00:00:00 2001 From: GareArc Date: Tue, 13 Oct 2026 00:47:51 -0700 Subject: [PATCH] refactor(api): publish over openapi through the console's own steps and advisory warning --- api/controllers/console/app/workflow.py | 22 +----- api/controllers/openapi/app_workflow.py | 14 +++- api/services/workflow/app_publish.py | 53 ------------- .../workflow_variable_reference_validator.py | 19 +++++ .../controllers/console/app/test_workflow.py | 7 +- .../controllers/openapi/test_app_workflow.py | 76 +++++++++++++++++++ 6 files changed, 112 insertions(+), 79 deletions(-) delete mode 100644 api/services/workflow/app_publish.py diff --git a/api/controllers/console/app/workflow.py b/api/controllers/console/app/workflow.py index 255cc684dd0..77ca8d0fe3b 100644 --- a/api/controllers/console/app/workflow.py +++ b/api/controllers/console/app/workflow.py @@ -101,10 +101,7 @@ from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, 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__) @@ -1281,21 +1278,6 @@ class DraftWorkflowNodeRunApi(Resource): ).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//workflows/publish") class PublishedWorkflowApi(Resource): @console_ns.doc("get_published_workflow") @@ -1362,7 +1344,7 @@ class PublishedWorkflowApi(Resource): 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, diff --git a/api/controllers/openapi/app_workflow.py b/api/controllers/openapi/app_workflow.py index e3018dd41aa..2a6ebe0ef48 100644 --- a/api/controllers/openapi/app_workflow.py +++ b/api/controllers/openapi/app_workflow.py @@ -63,16 +63,17 @@ from fields.workflow_run_fields import ( ) from graphon.variables import SecretVariable, VariableBase from graphon.variables.exc import VariableError +from libs.datetime_utils import naive_utc_now from libs.helper import to_timestamp -from models import AppMode +from models import App, AppMode from models.workflow import Workflow, WorkflowDataError, WorkflowRun from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError -from services.workflow.app_publish import publish_app_workflow, variable_reference_warning from services.workflow.dsl_import import stored_secret_ids from services.workflow.graph_check import CONTAINER_NODE_TYPES, GRAPH_MODES from services.workflow.graph_diff import draft_token from services.workflow_run_service import WorkflowRunListArgs from services.workflow_service import WorkflowService +from services.workflow_variable_reference_validator import advisory_variable_reference_warning _RUN_READ_GUARDS: Final = account_app_guards( RBACPermission.APP_CREATE_AND_MANAGEMENT, scope=Scope.APPS_READ, roles=None, modes=GRAPH_MODES @@ -200,7 +201,7 @@ class AppPublishApi(Resource): ) def post(self, ctx: Context, app_id: str, *, body: PublishPayload): try: - workflow = publish_app_workflow( + workflow = WorkflowService().publish_workflow( session=ctx.session, app_model=ctx.app, account=ctx.account, @@ -209,10 +210,15 @@ class AppPublishApi(Resource): ) except ValueError as exc: raise BadRequest(str(exc)) from exc + app_in_session = ctx.session.get(App, ctx.app.id) + if app_in_session: + app_in_session.workflow_id = workflow.id + app_in_session.updated_by = ctx.account.id + app_in_session.updated_at = naive_utc_now() return PublishResponse( version_id=workflow.id, created_at=to_timestamp(workflow.created_at), - warning=variable_reference_warning(workflow.graph), + warning=advisory_variable_reference_warning(workflow.graph), ) diff --git a/api/services/workflow/app_publish.py b/api/services/workflow/app_publish.py deleted file mode 100644 index bd924d091a2..00000000000 --- a/api/services/workflow/app_publish.py +++ /dev/null @@ -1,53 +0,0 @@ -"""Publishing an app's draft workflow for openapi, as the console publish route does.""" - -from __future__ import annotations - -import json -import logging - -from sqlalchemy.orm import Session - -from libs.datetime_utils import naive_utc_now -from models import Account, App -from models.workflow import Workflow -from services.workflow_service import WorkflowService -from services.workflow_variable_reference_validator import ( - format_variable_reference_errors, - validate_variable_references, -) - -logger = logging.getLogger(__name__) - - -def publish_app_workflow( - *, session: Session, app_model: App, account: Account, marked_name: str, marked_comment: str -) -> Workflow: - """Publish the draft as a new version and make the app use it.""" - workflow = WorkflowService().publish_workflow( - session=session, - app_model=app_model, - account=account, - marked_name=marked_name, - marked_comment=marked_comment, - ) - app_in_session = session.get(App, app_model.id) - if app_in_session: - app_in_session.workflow_id = workflow.id - app_in_session.updated_by = account.id - app_in_session.updated_at = naive_utc_now() - return workflow - - -def variable_reference_warning(graph_text: str | None) -> str | None: - """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 diff --git a/api/services/workflow_variable_reference_validator.py b/api/services/workflow_variable_reference_validator.py index dd1409e4257..258bcb8aeb9 100644 --- a/api/services/workflow_variable_reference_validator.py +++ b/api/services/workflow_variable_reference_validator.py @@ -1,5 +1,7 @@ from __future__ import annotations +import json +import logging import re from collections import defaultdict, deque from collections.abc import Callable, Iterator, Mapping, Sequence @@ -29,6 +31,8 @@ _TEMPLATE_REFERENCE_PATTERN = re.compile( ) _MAX_REPORTED_ISSUES = 10 +logger = logging.getLogger(__name__) + def _ids(items: object, key: str) -> frozenset[str]: if not isinstance(items, list): @@ -206,6 +210,21 @@ def format_variable_reference_errors(issues: Sequence[VariableReferenceIssue]) - ) +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 + + def _reachable_from(starts: Sequence[str], successors: Mapping[str, list[str]]) -> set[str]: visited: set[str] = set() queue: deque[str] = deque(starts) diff --git a/api/tests/unit_tests/controllers/console/app/test_workflow.py b/api/tests/unit_tests/controllers/console/app/test_workflow.py index a74e12c3f3a..446f09b9492 100644 --- a/api/tests/unit_tests/controllers/console/app/test_workflow.py +++ b/api/tests/unit_tests/controllers/console/app/test_workflow.py @@ -23,6 +23,7 @@ from graphon.variables.variables import RAGPipelineVariable from models.account import Account from models.model import App, AppMode from models.workflow import Workflow, WorkflowType +from services import workflow_variable_reference_validator as variable_reference_validator from services.app.access import AppAccessFilter from tests.unit_tests.config_override import apply_config_overrides @@ -167,10 +168,12 @@ def test_publish_workflow_returns_success( graph={"clean": "{}", "empty": None, "non-object": "[]", "invalid-json": "{"}.get(advisory, json.dumps(graph)), ) if advisory == "checker-error": - monkeypatch.setattr(workflow_module, "validate_variable_references", Mock(side_effect=RuntimeError("checker"))) + monkeypatch.setattr( + variable_reference_validator, "validate_variable_references", Mock(side_effect=RuntimeError("checker")) + ) elif advisory == "formatter-error": monkeypatch.setattr( - workflow_module, "format_variable_reference_errors", Mock(side_effect=RuntimeError("format")) + variable_reference_validator, "format_variable_reference_errors", Mock(side_effect=RuntimeError("format")) ) session = Mock() session.get.return_value = app_model diff --git a/api/tests/unit_tests/controllers/openapi/test_app_workflow.py b/api/tests/unit_tests/controllers/openapi/test_app_workflow.py index 4a0eef6a093..cdee88e6215 100644 --- a/api/tests/unit_tests/controllers/openapi/test_app_workflow.py +++ b/api/tests/unit_tests/controllers/openapi/test_app_workflow.py @@ -1,4 +1,6 @@ +import json from collections.abc import Callable +from datetime import datetime from types import SimpleNamespace from typing import Protocol, cast from unittest.mock import Mock @@ -6,6 +8,7 @@ from unittest.mock import Mock import pytest from flask import Flask from pydantic import ValidationError +from werkzeug.exceptions import BadRequest from controllers.openapi import app_workflow from controllers.openapi._errors import ( @@ -21,6 +24,8 @@ from controllers.openapi._models import ( AdvancedChatNodeRunPayload, EnvVariableSetPayload, NodeRunPayload, + PublishPayload, + PublishResponse, RestoreResponse, RunListQuery, VersionListQuery, @@ -28,6 +33,7 @@ from controllers.openapi._models import ( from controllers.openapi._upload import file_fields from controllers.openapi.app_workflow import ( AppEnvItemApi, + AppPublishApi, AppRunNodeListApi, AppVersionListApi, AppVersionRestoreApi, @@ -293,3 +299,73 @@ def test_run_draft_node_ends_the_read_transaction_before_uploading( names = [call[0] for call in calls.mock_calls] assert names == (["end", "merge"] if ends_transaction else ["merge"]) + + +def _branch_skipping_graph() -> str: + return json.dumps( + { + "nodes": [ + {"id": "start", "data": {"type": "start"}}, + {"id": "branch", "data": {"type": "if-else"}}, + {"id": "producer", "data": {"type": "code", "title": "Producer"}}, + {"id": "consumer", "data": {"type": "answer", "title": "Consumer", "answer": "{{#producer.text#}}"}}, + ], + "edges": [ + {"source": "start", "target": "branch"}, + {"source": "branch", "target": "producer", "sourceHandle": "true"}, + {"source": "branch", "target": "consumer", "sourceHandle": "false"}, + ], + } + ) + + +@pytest.mark.parametrize("graph", ["{}", "{", None], ids=["clean", "invalid-json", "empty"]) +def test_publish_points_the_app_at_the_new_version( + app: Flask, monkeypatch: pytest.MonkeyPatch, graph: str | None +) -> None: + service = _fake_workflow_service(monkeypatch) + service.publish_workflow.return_value = SimpleNamespace( + id="version-1", created_at=datetime(2026, 10, 1, 12, 0, 0), graph=graph + ) + ctx = _app_context() + stored_app = SimpleNamespace(workflow_id=None, updated_by=None, updated_at=None) + cast(Mock, ctx.session).get.return_value = stored_app + + api = AppPublishApi() + with app.test_request_context("/openapi/v1/apps/app-1:publish", method="POST"): + response = cast(_EndpointView, api.post).__handler__(api, ctx, "app-1", body=PublishPayload(marked_name="v2")) + + assert service.publish_workflow.call_args.kwargs["marked_name"] == "v2" + assert (stored_app.workflow_id, stored_app.updated_by) == ("version-1", "account-1") + assert stored_app.updated_at is not None + assert cast(PublishResponse, response).version_id == "version-1" + assert cast(PublishResponse, response).warning is None + + +def test_publish_warns_about_references_to_a_skipped_branch(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: + service = _fake_workflow_service(monkeypatch) + service.publish_workflow.return_value = SimpleNamespace( + id="version-1", created_at=datetime(2026, 10, 1, 12, 0, 0), graph=_branch_skipping_graph() + ) + + api = AppPublishApi() + with app.test_request_context("/openapi/v1/apps/app-1:publish", method="POST"): + response = cast(_EndpointView, api.post).__handler__(api, _app_context(), "app-1", body=PublishPayload()) + + warning = cast(PublishResponse, response).warning + assert warning is not None + assert "Consumer" in warning + assert "Producer" in warning + + +def test_publish_refusal_is_a_bad_request(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: + service = _fake_workflow_service(monkeypatch) + service.publish_workflow.side_effect = ValueError("Node l (llm): Missing provider or model configuration") + ctx = _app_context() + + api = AppPublishApi() + with app.test_request_context("/openapi/v1/apps/app-1:publish", method="POST"): + with pytest.raises(BadRequest, match="Missing provider"): + cast(_EndpointView, api.post).__handler__(api, ctx, "app-1", body=PublishPayload()) + + cast(Mock, ctx.session).get.assert_not_called()