feat: add cluster execution owner contract

.github/workflows/ci.yml

@@ -0,0 +1,21 @@
+name: CI
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.8', '3.11']
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: python -m pip install --upgrade pip
+      - run: python -m pip install -r requirements.txt
+      - run: python -m unittest discover -s tests -v

README.md

@@ -28,3 +28,21 @@ controller.add_handler(r'/test/(.*)', Handler)
 
 controller.start()
 ```
+
+## Execute-only ownership
+
+Cluster-local resource owners can request execution without PlatformGateway
+usage recording. The shared handler recognizes only this contract:
+
+```http
+X-Ace-Execution-Owner: x402
+```
+
+`BaseHandler.record()` skips its `/record` call for that owner. Missing, empty,
+or unknown owner values keep the normal recording path. Workers must still avoid
+task persistence and callbacks before publishing asynchronous operations through
+an execute-only owner.
+
+This is a reserved cluster-internal header. Every public ingress or gateway must
+remove client-supplied `X-Ace-Execution-Owner` before forwarding to a worker.
+Only direct cluster-local resource-owner calls may carry it.

acedatacloud_scaffold/__init__.py

@@ -1,4 +1,5 @@
 from .controllers import BaseController
+from .execution import EXECUTION_OWNER_HEADER, X402_EXECUTION_OWNER, get_execution_owner, is_execute_only
 from .handlers import BaseHandler
 from .handlers.callback import CallbackHandler
 from .handlers.health import HealthHandler
@@ -9,6 +10,10 @@
     'BaseController',
     'BaseHandler',
     'CallbackHandler',
+    'EXECUTION_OWNER_HEADER',
     'HealthHandler',
     'HybridHandler',
+    'X402_EXECUTION_OWNER',
+    'get_execution_owner',
+    'is_execute_only',
 ]

acedatacloud_scaffold/execution.py

@@ -0,0 +1,19 @@
+EXECUTION_OWNER_HEADER = 'X-Ace-Execution-Owner'
+X402_EXECUTION_OWNER = 'x402'
+
+
+def get_execution_owner(headers):
+    if headers is None:
+        return None
+    value = next(
+        (value for key, value in headers.items() if str(key).lower() == EXECUTION_OWNER_HEADER.lower()),
+        None,
+    )
+    if not isinstance(value, str):
+        return None
+    owner = value.strip().lower()
+    return owner or None
+
+
+def is_execute_only(headers):
+    return get_execution_owner(headers) == X402_EXECUTION_OWNER

acedatacloud_scaffold/handlers/base.py

@@ -1,5 +1,6 @@
 from loguru import logger
 from acedatacloud_scaffold.exceptions import APIException
+from acedatacloud_scaffold.execution import X402_EXECUTION_OWNER, get_execution_owner
 import json
 from acedatacloud_scaffold.settings import \
     ERROR_STATUS_API_ERROR, ERROR_CODE_API_ERROR, RECORD_SERVER_URL
@@ -93,6 +94,12 @@ def get_record_response(self):
 
     @logger.catch
     def record(self, extra_data={}):
+        execution_owner = get_execution_owner(self.request.headers)
+        if execution_owner == X402_EXECUTION_OWNER:
+            self.logger.debug('x402 execution owner: skip platform usage record')
+            return
+        if execution_owner is not None:
+            self.logger.warning(f'unknown execution owner {execution_owner}: keep platform usage record')
         record_server_url = RECORD_SERVER_URL
         self.logger.debug(f'record url {record_server_url}')
         data = {

tests/test_execution_owner.py

@@ -0,0 +1,69 @@
+from types import SimpleNamespace
+from unittest import TestCase
+from unittest.mock import Mock, patch
+
+from acedatacloud_scaffold.execution import get_execution_owner, is_execute_only
+from acedatacloud_scaffold.handlers.base import BaseHandler
+from tornado.httputil import HTTPHeaders
+
+
+class ExecutionOwnerTest(TestCase):
+    def test_x402_owner_is_case_insensitive(self):
+        headers = {'x-ace-execution-owner': ' X402 '}
+
+        self.assertEqual(get_execution_owner(headers), 'x402')
+        self.assertTrue(is_execute_only(headers))
+
+    def test_tornado_headers_preserve_owner_contract(self):
+        headers = HTTPHeaders({'x-ace-execution-owner': 'X402'})
+
+        self.assertEqual(get_execution_owner(headers), 'x402')
+        self.assertTrue(is_execute_only(headers))
+
+    def test_unknown_or_missing_owner_is_not_execute_only(self):
+        self.assertFalse(is_execute_only({}))
+        self.assertFalse(is_execute_only({'X-Ace-Execution-Owner': 'router'}))
+        self.assertFalse(is_execute_only({'X-Ace-Execution-Owner': ''}))
+
+    @patch('acedatacloud_scaffold.handlers.base.requests.post')
+    def test_x402_owner_skips_platform_record(self, post):
+        handler = self.make_handler({'X-Ace-Execution-Owner': 'x402'})
+
+        BaseHandler.record(handler)
+
+        post.assert_not_called()
+
+    @patch('acedatacloud_scaffold.handlers.base.requests.post')
+    def test_normal_and_unknown_owners_keep_platform_record(self, post):
+        post.return_value = SimpleNamespace(status_code=200)
+        normal_handler = self.make_handler({})
+        unknown_handler = self.make_handler({'X-Ace-Execution-Owner': 'unknown'})
+
+        BaseHandler.record(normal_handler)
+        BaseHandler.record(unknown_handler)
+
+        self.assertEqual(post.call_count, 2)
+        normal_handler.logger.warning.assert_not_called()
+        unknown_handler.logger.warning.assert_called_once_with(
+            'unknown execution owner unknown: keep platform usage record'
+        )
+
+    @staticmethod
+    def make_handler(headers):
+        handler = SimpleNamespace(
+            request=SimpleNamespace(headers=headers),
+            logger=Mock(),
+            trace_id='trace',
+            payflow=None,
+            get_record_trace_id=Mock(return_value='trace'),
+            get_record_application_id=Mock(return_value=None),
+            get_record_api_id=Mock(return_value=None),
+            get_record_task_id=Mock(return_value='task'),
+            get_record_user_id=Mock(return_value=None),
+            get_record_credential_id=Mock(return_value=None),
+            get_record_authorization_id=Mock(return_value=None),
+            get_record_actor_user_id=Mock(return_value=None),
+            get_record_request=Mock(return_value={}),
+            get_record_response=Mock(return_value={'status': 200}),
+        )
+        return handler