Feat: published agent version control (#13410)

### What problem does this PR solve?

Feat: published agent version control

### Type of change


- [x] New Feature (non-breaking change which adds functionality)

api/apps/canvas_app.py

@@ -75,6 +75,7 @@ async def rm():
 @login_required
 async def save():
     req = await get_request_json()
+    req['release'] = bool(req.get("release", ""))
     try:
         req["dsl"] = CanvasReplicaService.normalize_dsl(req["dsl"])
     except ValueError as e:

api/apps/sdk/session.py

@@ -83,7 +83,9 @@ async def create(tenant_id, chat_id):
 @manager.route("/agents/<agent_id>/sessions", methods=["POST"])  # noqa: F821
 @token_required
 async def create_agent_session(tenant_id, agent_id):
-    user_id = request.args.get("user_id", tenant_id)
+    req = await get_request_json()
+    user_id = req.get("user_id") or request.args.get("user_id", tenant_id)
+    release_mode = req.get("release", request.args.get("release", False))
     e, cvs = UserCanvasService.get_by_id(agent_id)
     if not e:
         return get_error_data_result("Agent not found.")
@@ -92,6 +94,8 @@ async def create_agent_session(tenant_id, agent_id):
     if not isinstance(cvs.dsl, str):
         cvs.dsl = json.dumps(cvs.dsl, ensure_ascii=False)
 
+    if release_mode and not bool(cvs.release):
+        raise PermissionError("No available published version")
     session_id = get_uuid()
     canvas = Canvas(cvs.dsl, tenant_id, agent_id, canvas_id=cvs.id)
     canvas.reset()
@@ -986,15 +990,35 @@ async def agent_bot_completions(agent_id):
         return get_error_data_result(message='Authentication error: API key is invalid!"')
 
     if req.get("stream", True):
-        resp = Response(agent_completion(objs[0].tenant_id, agent_id, **req), mimetype="text/event-stream")
+        async def stream():
+            try:
+                async for answer in agent_completion(objs[0].tenant_id, agent_id, **req):
+                    yield answer
+            except Exception as e:
+                logging.exception(e)
+                error_result = get_error_data_result(message=str(e) or "Unknown error")
+                yield "data:" + json.dumps(
+                    {
+                        "event": "message",
+                        "data": {"content": f"Error {error_result['code']}: {error_result['message']}\n\n"},
+                        **error_result,
+                    },
+                    ensure_ascii=False,
+                ) + "\n\n"
+
+        resp = Response(stream(), mimetype="text/event-stream")
         resp.headers.add_header("Cache-control", "no-cache")
         resp.headers.add_header("Connection", "keep-alive")
         resp.headers.add_header("X-Accel-Buffering", "no")
         resp.headers.add_header("Content-Type", "text/event-stream; charset=utf-8")
         return resp
 
-    async for answer in agent_completion(objs[0].tenant_id, agent_id, **req):
-        return get_result(data=answer)
+    try:
+        async for answer in agent_completion(objs[0].tenant_id, agent_id, **req):
+            return get_result(data=answer)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message=str(e) or "Unknown error")
 
     return None
 

api/db/db_models.py

@@ -1032,6 +1032,7 @@ class UserCanvas(DataBaseModel):
     title = CharField(max_length=255, null=True, help_text="Canvas title")
 
     permission = CharField(max_length=16, null=False, help_text="me|team", default="me", index=True)
+    release = BooleanField(null=False, help_text="is released", default=False, index=True)
     description = TextField(null=True, help_text="Canvas description")
     canvas_type = CharField(max_length=32, null=True, help_text="Canvas type", index=True)
     canvas_category = CharField(max_length=32, null=False, default="agent_canvas", help_text="Canvas category: agent_canvas|dataflow_canvas", index=True)
@@ -1407,6 +1408,7 @@ def migrate_db():
     alter_db_add_column(migrator, "task", "task_type", CharField(max_length=32, null=False, default=""))
     alter_db_add_column(migrator, "task", "priority", IntegerField(default=0))
     alter_db_add_column(migrator, "user_canvas", "permission", CharField(max_length=16, null=False, help_text="me|team", default="me", index=True))
+    alter_db_add_column(migrator, "user_canvas", "release", BooleanField(null=False, help_text="is released", default=False, index=True))
     alter_db_add_column(migrator, "llm", "is_tools", BooleanField(null=False, help_text="support tools", default=False))
     alter_db_add_column(migrator, "mcp_server", "variables", JSONField(null=True, help_text="MCP Server variables", default=dict))
     alter_db_rename_column(migrator, "task", "process_duation", "process_duration")

api/db/services/canvas_service.py

@@ -195,10 +195,12 @@ async def completion(tenant_id, agent_id, session_id=None, **kwargs):
     inputs = kwargs.get("inputs", {})
     user_id = kwargs.get("user_id", "")
     custom_header = kwargs.get("custom_header", "")
+    release_mode = str(kwargs.get("release", "")).strip().lower()
 
     if session_id:
         e, conv = API4ConversationService.get_by_id(session_id)
-        assert e, "Session not found!"
+        if not e:
+            raise LookupError("Session not found!")
         if not conv.message:
             conv.message = []
         if not isinstance(conv.dsl, str):
@@ -206,10 +208,15 @@ async def completion(tenant_id, agent_id, session_id=None, **kwargs):
         canvas = Canvas(conv.dsl, tenant_id, agent_id, canvas_id=agent_id, custom_header=custom_header)
     else:
         e, cvs = UserCanvasService.get_by_id(agent_id)
-        assert e, "Agent not found."
-        assert cvs.user_id == tenant_id, "You do not own the agent."
+        if not e:
+            raise LookupError("Agent not found.")
+        if cvs.user_id != tenant_id:
+            raise PermissionError("You do not own the agent.")
+        if release_mode == "true" and not bool(cvs.release):
+            raise PermissionError("No available published version")
         if not isinstance(cvs.dsl, str):
             cvs.dsl = json.dumps(cvs.dsl, ensure_ascii=False)
+
         session_id=get_uuid()
         canvas = Canvas(cvs.dsl, tenant_id, agent_id, canvas_id=cvs.id, custom_header=custom_header)
         canvas.reset()