feat: history/queue align with ComfyUI (outputs_to_execute, extra_data, GET /queue);

Change-Id: I1f9890aa249a2aaff0fdbc10bcda3e130b70a4e3
Co-developed-by: Cursor <noreply@cursor.com>
Made-with: Cursor

Author: yinsu.zs

src/code/agent/routes/gateway_routes.py

@@ -47,7 +47,7 @@ def __init__(self):
         self.prompt_handler = PromptHandler(task_manager)
         self.serverless_handler = ServerlessHandler()
         self.history_handler = HistoryHandler()
-        self.interrupt_handler = InterruptHandler(task_manager)
+        self.interrupt_handler = InterruptHandler()
         self.userdata_handler = UserdataHandler()
         self.ws_handler = WsHandler()
         self.serverless_ws_handler = ServerlessWsHandler(constants.GPU_FUNCTION_URL)

src/code/agent/services/fc_openapi/__init__.py

@@ -1,3 +1,3 @@
-from .fc_stop_client import stop_async_task
+from .fc_client import stop_async_task
 
 __all__ = ["stop_async_task"]

src/code/agent/services/fc_openapi/fc_client.py

@@ -0,0 +1,102 @@
+"""
+FC OpenAPI 客户端
+用于在 CPU 侧调用函数计算 OpenAPI，操作 GPU 上的异步任务。
+使用实例 RAM 角色凭证：优先从请求头（FC 注入）获取，否则从环境变量获取。
+"""
+from flask import request
+
+from alibabacloud_tea_openapi.client import Client as OpenApiClient
+from alibabacloud_tea_openapi import models as open_api_models
+from alibabacloud_tea_util import models as util_models
+
+import constants
+from utils.logger import log
+
+
+def _get_credentials():
+    """
+    获取阿里云凭证（与 ServerlessApiService.get_credentials 一致）。
+    优先从请求头获取（FC 实例 RAM 角色注入），否则从环境变量获取。
+    Returns:
+        tuple: (access_key_id, access_key_secret, security_token)
+    """
+    ak = ""
+    sk = ""
+    sts = ""
+    try:
+        ak = request.headers.get(constants.HEADER_KEY_ACCESS_KEY_ID, "")
+        sk = request.headers.get(constants.HEADER_KEY_ACCESS_KEY_SECRET, "")
+        sts = request.headers.get(constants.HEADER_KEY_SECURITY_TOKEN, "")
+    except Exception as e:
+        log("WARNING", f"get credentials from header failed: {e}")
+    if not ak or not sk:
+        ak = getattr(constants, "ALIBABA_CLOUD_ACCESS_KEY_ID", "") or ""
+        sk = getattr(constants, "ALIBABA_CLOUD_ACCESS_KEY_SECRET", "") or ""
+        sts = getattr(constants, "ALIBABA_CLOUD_SECURITY_TOKEN", "") or ""
+    return ak, sk, sts
+
+
+def _create_client(ak: str, sk: str, sts: str, endpoint: str) -> OpenApiClient:
+    """
+    构造 FC OpenApiClient，供各 API 调用方复用，避免重复构建 Config。
+    """
+    config = open_api_models.Config(
+        access_key_id=ak,
+        access_key_secret=sk,
+        security_token=sts or None,
+    )
+    config.endpoint = endpoint
+    return OpenApiClient(config)
+
+
+def stop_async_task(task_id: str) -> bool:
+    """
+    调用 FC OpenAPI StopAsyncTask 停止指定异步任务（带签名，使用实例 RAM 角色凭证）。
+
+    Args:
+        task_id: 异步任务 ID（与我们的 task_id/prompt_id 一致）。
+
+    Returns:
+        bool: 请求成功且为 2xx 返回 True，否则 False。
+    """
+    account_id = getattr(constants, "FC_ACCOUNT_ID", "") or ""
+    region = getattr(constants, "FC_REGION", "cn-hangzhou") or "cn-hangzhou"
+    cpu_function_name = getattr(constants, "FC_FUNCTION_NAME", "") or ""
+
+    if not account_id or not cpu_function_name:
+        return False
+
+    ak, sk, sts = _get_credentials()
+    if not ak or not sk:
+        log("WARNING", "StopAsyncTask: no credentials (header or env)")
+        return False
+
+    gpu_function_name = cpu_function_name.replace("-gw", "")
+    endpoint = f"{account_id}.{region}.fc.aliyuncs.com"
+
+    try:
+        client = _create_client(ak, sk, sts, endpoint)
+
+        params = open_api_models.Params(
+            action="StopAsyncTask",
+            version="2023-03-30",
+            protocol="HTTPS",
+            method="PUT",
+            auth_type="AK",
+            style="FC",
+            pathname=f"/2023-03-30/functions/{gpu_function_name}/async-tasks/{task_id}/stop",
+            req_body_type="json",
+            body_type="json",
+        )
+        req = open_api_models.OpenApiRequest(query={"qualifier": "LATEST"})
+        runtime = util_models.RuntimeOptions(read_timeout=5000, connect_timeout=5000)
+
+        resp = client.call_api(params, req, runtime)
+        status_code = resp.get("statusCode") or 0
+        if status_code >= 200 and status_code < 300:
+            return True
+        log("WARNING", f"StopAsyncTask failed: task_id={task_id}, status={status_code}, body={resp.get('body', '')[:200]}")
+        return False
+    except Exception as e:
+        log("WARNING", f"StopAsyncTask request error: task_id={task_id}, error={e}")
+        return False

src/code/agent/services/gateway/handlers/interrupt_handler.py

@@ -1,31 +1,21 @@
 """
 Interrupt Handler
-处理 POST /api/interrupt 请求，与 ComfyUI 行为对齐
+处理 POST /api/interrupt 请求
 """
-from flask import Response
-
-from utils.logger import log
+from flask import jsonify
 
 
 class InterruptHandler:
-    """处理 POST /api/interrupt：中断当前用户正在执行的任务，并通知 GPU/后端真正停止"""
-
-    def __init__(self, task_manager):
-        self.task_manager = task_manager
+    """处理 POST /api/interrupt"""
 
     def handle_post(self):
         """
-        处理 POST /api/interrupt 请求
-        1) 先对当前用户每个 RUNNING task_id 调用 FC StopAsyncTask，在平台侧停止 GPU 任务
-        2) 再取消当前用户所有 RUNNING 任务（停轮询、清理本地状态）
-        返回 200（与 ComfyUI POST /interrupt 对齐）
+        POST /api/interrupt 暂不支持：FC StopAsyncTask 会带来非预期体验，
+        禁止客户端中止正在运行的任务，返回 403。
         """
-        # 函数计算 StopAsyncTask 存在 bug，等 FC 修复后再上线
-        # from services.fc_openapi.fc_stop_client import stop_async_task
-        # running_ids = self.task_manager.get_current_user_running_task_ids()
-        # for task_id in running_ids:
-        #     stop_async_task(task_id)
-        count = self.task_manager.interrupt_current_user()
-        if count > 0:
-            log("INFO", f"Interrupted {count} running task(s) for current user")
-        return Response(status=200)
+        return jsonify({
+            "error": {
+                "type": "not_supported",
+                "message": "Interrupting a running task is not supported.",
+            }
+        }), 403

src/code/agent/services/gateway/handlers/queue_handler.py

@@ -99,7 +99,7 @@ def handle_post_request(self):
         if "clear" in request_data and request_data["clear"]:
             # 清空队列：先对当前用户 PENDING 任务并行调 StopAsyncTask（数量上限 + 总超时），再清本地 PENDING（与 ComfyUI 只清 pending 对齐，不停 RUNNING）
             log("INFO", f"Clearing task queue")
-            from services.fc_openapi.fc_stop_client import stop_async_task
+            from services.fc_openapi.fc_client import stop_async_task
 
             pending_ids = self.task_manager.get_current_user_pending_task_ids()
             ids_to_stop = pending_ids[:MAX_PENDING_STOP_ON_CLEAR]

src/code/agent/test/unit/services/fc_openapi/fc_client_test.py

@@ -0,0 +1,73 @@
+"""
+FC OpenAPI 客户端单测
+"""
+import pytest
+from unittest.mock import patch, MagicMock
+
+import constants
+from services.fc_openapi.fc_client import stop_async_task
+
+
+class TestStopAsyncTaskConfig:
+    """配置缺失时不应发请求"""
+
+    def test_returns_false_when_fc_account_id_empty(self):
+        with patch.object(constants, "FC_ACCOUNT_ID", ""), \
+             patch.object(constants, "FC_FUNCTION_NAME", "my-func-gw"), \
+             patch("services.fc_openapi.fc_client.OpenApiClient") as mock_client:
+            result = stop_async_task("task-123")
+            assert result is False
+            mock_client.assert_not_called()
+
+    def test_returns_false_when_fc_function_name_empty(self):
+        with patch.object(constants, "FC_ACCOUNT_ID", "123456"), \
+             patch.object(constants, "FC_FUNCTION_NAME", ""), \
+             patch("services.fc_openapi.fc_client.OpenApiClient") as mock_client:
+            result = stop_async_task("task-123")
+            assert result is False
+            mock_client.assert_not_called()
+
+    def test_returns_false_when_no_credentials(self):
+        with patch.object(constants, "FC_ACCOUNT_ID", "123456"), \
+             patch.object(constants, "FC_FUNCTION_NAME", "my-func-gw"), \
+             patch("services.fc_openapi.fc_client._get_credentials", return_value=("", "", "")), \
+             patch("services.fc_openapi.fc_client.OpenApiClient") as mock_client:
+            result = stop_async_task("task-123")
+            assert result is False
+            mock_client.assert_not_called()
+
+
+class TestStopAsyncTaskRequest:
+    """配置与凭证齐全时调用 OpenAPI 且 path/query 正确"""
+
+    @pytest.fixture(autouse=True)
+    def set_config(self):
+        with patch.object(constants, "FC_ACCOUNT_ID", "1338904783509062"), \
+             patch.object(constants, "FC_REGION", "cn-hangzhou"), \
+             patch.object(constants, "FC_FUNCTION_NAME", "art-speed-test-mpbx-gw-prod"), \
+             patch("services.fc_openapi.fc_client._get_credentials", return_value=("ak", "sk", "sts")):
+            yield
+
+    def test_put_url_and_qualifier(self, set_config):
+        mock_client_instance = MagicMock()
+        mock_client_instance.call_api.return_value = {"statusCode": 200}
+        with patch("services.fc_openapi.fc_client.OpenApiClient", return_value=mock_client_instance):
+            result = stop_async_task("task-123")
+
+        assert result is True
+        mock_client_instance.call_api.assert_called_once()
+        call_args = mock_client_instance.call_api.call_args
+        params = call_args[0][0]
+        req = call_args[0][1]
+        assert "2023-03-30/functions/art-speed-test-mpbx-prod/async-tasks/task-123/stop" in (params.pathname or "")
+        assert getattr(req, "query", {}).get("qualifier") == "LATEST"
+
+    def test_gpu_function_name_strips_gw(self, set_config):
+        mock_client_instance = MagicMock()
+        mock_client_instance.call_api.return_value = {"statusCode": 200}
+        with patch("services.fc_openapi.fc_client.OpenApiClient", return_value=mock_client_instance):
+            stop_async_task("tid")
+        params = mock_client_instance.call_api.call_args[0][0]
+        pathname = params.pathname or ""
+        assert "art-speed-test-mpbx-prod" in pathname
+        assert "art-speed-test-mpbx-gw-prod" not in pathname or "/functions/art-speed-test-mpbx-prod/" in pathname

src/code/agent/test/unit/services/gateway/interrupt_handler_test.py

@@ -1,36 +1,37 @@
 """
 Interrupt Handler 单元测试
-测试 POST /api/interrupt 与 ComfyUI 对齐
+POST /api/interrupt 暂不支持，返回 403
 """
 import pytest
-from unittest.mock import Mock, patch
+from unittest.mock import Mock
+from flask import Flask
 
 from services.gateway.handlers.interrupt_handler import InterruptHandler
 
 
 @pytest.fixture
-def mock_task_manager():
-    return Mock()
+def app():
+    return Flask(__name__)
 
 
 @pytest.fixture
-def interrupt_handler(mock_task_manager):
-    return InterruptHandler(mock_task_manager)
-
-
-def test_handle_post_calls_interrupt_current_user_and_returns_200(interrupt_handler, mock_task_manager):
-    """POST /api/interrupt 调用 interrupt_current_user 并返回 200"""
-    mock_task_manager.get_current_user_running_task_ids.return_value = []
-    mock_task_manager.interrupt_current_user.return_value = 1
-    with patch("services.fc_openapi.fc_stop_client.stop_async_task"):
-        response = interrupt_handler.handle_post()
-    assert response.status_code == 200
-    mock_task_manager.interrupt_current_user.assert_called_once()
-
-
-def test_handle_post_calls_interrupt_current_user_when_running_exists(interrupt_handler, mock_task_manager):
-    """POST /api/interrupt 调用 interrupt_current_user 并返回 200（StopAsyncTask 临时注释期间仅本地中断）"""
-    mock_task_manager.interrupt_current_user.return_value = 2
-    response = interrupt_handler.handle_post()
-    assert response.status_code == 200
-    mock_task_manager.interrupt_current_user.assert_called_once()
+def interrupt_handler():
+    return InterruptHandler()
+
+
+def test_handle_post_returns_403(interrupt_handler, app):
+    """POST /api/interrupt 返回 403，不允许客户端中止正在运行的任务"""
+    with app.test_request_context("/api/interrupt", method="POST"):
+        response, status_code = interrupt_handler.handle_post()
+    assert status_code == 403
+    data = response.get_json()
+    assert data["error"]["type"] == "not_supported"
+
+
+def test_handle_post_returns_error_message(interrupt_handler, app):
+    """POST /api/interrupt 错误体包含可读的 message"""
+    with app.test_request_context("/api/interrupt", method="POST"):
+        response, status_code = interrupt_handler.handle_post()
+    data = response.get_json()
+    assert "message" in data["error"]
+    assert len(data["error"]["message"]) > 0

src/code/agent/test/unit/services/gateway/queue_handler_test.py

@@ -33,7 +33,7 @@ def test_handle_post_clear_calls_stop_async_task_per_pending_then_clear_queue(qu
     """POST /api/queue clear 先对每个 PENDING 调 stop_async_task，再 clear_queue"""
     mock_task_manager.get_current_user_pending_task_ids.return_value = ["tid1", "tid2"]
     mock_task_manager.clear_queue.return_value = 2
-    with patch("services.fc_openapi.fc_stop_client.stop_async_task") as mock_stop:
+    with patch("services.fc_openapi.fc_client.stop_async_task") as mock_stop:
         with app.test_request_context("/api/queue", method="POST", json={"clear": True}):
             response = queue_handler.handle_post_request()
     assert response.status_code == 200
@@ -47,7 +47,7 @@ def test_handle_post_clear_with_no_pending_still_clears_queue(queue_handler, moc
     """POST /api/queue clear 若无 PENDING 仍会调用 clear_queue"""
     mock_task_manager.get_current_user_pending_task_ids.return_value = []
     mock_task_manager.clear_queue.return_value = 0
-    with patch("services.fc_openapi.fc_stop_client.stop_async_task") as mock_stop:
+    with patch("services.fc_openapi.fc_client.stop_async_task") as mock_stop:
         with app.test_request_context("/api/queue", method="POST", json={"clear": True}):
             response = queue_handler.handle_post_request()
     assert response.status_code == 200
@@ -61,7 +61,7 @@ def test_handle_post_clear_caps_stop_async_task_calls(queue_handler, mock_task_m
     pending_ids = [f"tid_{i}" for i in range(over_limit)]
     mock_task_manager.get_current_user_pending_task_ids.return_value = pending_ids
     mock_task_manager.clear_queue.return_value = over_limit
-    with patch("services.fc_openapi.fc_stop_client.stop_async_task") as mock_stop:
+    with patch("services.fc_openapi.fc_client.stop_async_task") as mock_stop:
         with app.test_request_context("/api/queue", method="POST", json={"clear": True}):
             response = queue_handler.handle_post_request()
     assert response.status_code == 200
@@ -77,7 +77,7 @@ def test_handle_post_clear_logs_warning_when_stop_partially_fails(queue_handler,
     def stop_side_effect(task_id):
         return task_id == "tid1"
 
-    with patch("services.fc_openapi.fc_stop_client.stop_async_task", side_effect=stop_side_effect):
+    with patch("services.fc_openapi.fc_client.stop_async_task", side_effect=stop_side_effect):
         with patch("services.gateway.handlers.queue_handler.log") as mock_log:
             with app.test_request_context("/api/queue", method="POST", json={"clear": True}):
                 response = queue_handler.handle_post_request()
@@ -93,7 +93,7 @@ def test_handle_post_clear_no_warning_when_all_stop_succeed(queue_handler, mock_
     mock_task_manager.get_current_user_pending_task_ids.return_value = ["tid1", "tid2"]
     mock_task_manager.clear_queue.return_value = 2
 
-    with patch("services.fc_openapi.fc_stop_client.stop_async_task", return_value=True):
+    with patch("services.fc_openapi.fc_client.stop_async_task", return_value=True):
         with patch("services.gateway.handlers.queue_handler.log") as mock_log:
             with app.test_request_context("/api/queue", method="POST", json={"clear": True}):
                 response = queue_handler.handle_post_request()