fix: add durable atomic execution claims (#340)

* feat: define runtime selectable allowlist contract

Co-Authored-By: Codex <noreply@openai.com>

* ci: pin QPK sync action dependencies

Co-Authored-By: Codex <noreply@openai.com>

* fix: add atomic durable execution claims

Co-Authored-By: Codex <noreply@openai.com>

* test: prove durable claims fail closed and single winner

Co-Authored-By: Codex <noreply@openai.com>

* fix: implement atomic create for cloud object stores

Co-Authored-By: Codex <noreply@openai.com>

* test: keep risk snapshot fixture unexpired in CI

Co-Authored-By: Codex <noreply@openai.com>

---------

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

.github/workflows/open-downstream-qpk-pin-prs.yml

@@ -27,9 +27,9 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 30
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
 
-      - uses: actions/setup-python@v6
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.1.0
         with:
           python-version: "3.11"
 

.github/workflows/update-qpk-pin.yml

@@ -26,9 +26,9 @@ jobs:
     timeout-minutes: 15
     if: github.event_name == 'push' && github.ref == 'refs/heads/main'
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
 
-      - uses: actions/setup-python@v6
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.1.0
         with:
           python-version: "3.11"
 
@@ -79,7 +79,7 @@ jobs:
         id: create_pin_pr
         if: steps.update.outputs.changed == 'true' && steps.verify.outcome == 'success'
         continue-on-error: true
-        uses: peter-evans/create-pull-request@v7
+        uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7.0.8
         with:
           token: ${{ secrets.QSL_REPO_SYNC_TOKEN || github.token }}
           commit-message: "chore: advance staged QPK pin"

docs/qpk_repo_sync_auth.zh-CN.md

@@ -2,6 +2,12 @@
 
 `open-downstream-qpk-pin-prs` workflow 需要 `QSL_REPO_SYNC_TOKEN` 才能跨仓 push 分支并开 PR。
 
+## 当前发布链说明
+
+QPK 发布采用 release-set 两阶段流程：先生成候选 QPK 版本并验证依赖闭包，再由下游仓库分别通过 CI 后更新正式 release-set。旧的 `auto/qpk-pin-sync-*` 和孤立的 `auto/qpk-pin-update` 分支不再作为发布输入；如果发现没有对应开放 PR 的孤立自动分支，应删除后再重新触发发布流程。
+
+跨仓同步 token 只用于创建下游 PR，不用于部署、运行时交易或修改任何 broker 权限。优先使用仅限目标仓库的 fine-grained token；不要把个人长期 token 写入仓库文件、workflow 日志或 issue。
+
 ## 为什么用仓库级 Secret（不是 Org Secret）
 
 `QuantPlatformKit` 是 **public** 仓库。在 GitHub Free org 上，org-level secret 对 public 仓的注入不可靠（workflow 里会静默变成空字符串）。**仓库级 secret** 是唯一已验证稳定的方案。

docs/runtime_selectable_allowlist_v1.zh-CN.md

@@ -0,0 +1,24 @@
+# Runtime-selectable allowlist v1
+
+`runtime-selectable allowlist` 描述某个平台在当前部署中可以选择的策略 profile；它不等于策略生命周期状态，也不授予 live 权限。
+
+## 边界
+
+- catalog 负责描述策略生命周期和证据状态；
+- allowlist 负责描述 runtime 可选择的 profile；
+- authority policy、Risk Gate 和 broker permission 负责决定是否可以执行；
+- 缺少 allowlist、证据或 authority 时，profile 必须保持 `PARKED`/`live_candidate`，不能自动执行。
+
+```json
+{
+  "schema": "qsl.runtime_selectable_allowlist.v1",
+  "platform": "example",
+  "domain": "us_equity",
+  "profiles": ["example_profile"],
+  "source_digest": "<sha256>",
+  "generated_at": "<utc>",
+  "permission_effect": "none"
+}
+```
+
+迁移期间，旧 `get_runtime_enabled_profiles()` 只能作为一次性读取适配器；新代码不得继续写入旧 status 字段。迁移完成后，各 broker/runtime 应只消费本 allowlist，并独立校验 authority、Risk Gate、回滚引用和 broker 权限。

src/quant_platform_kit/cloud/aws_provider.py

@@ -115,6 +115,24 @@ def write_text(self, uri: str, data: str, content_type: str = "text/plain") -> s
         self.client.put_object(Bucket=bucket, Key=key, Body=data.encode("utf-8"), ContentType=content_type)
         return uri
 
+    def create_text(self, uri: str, data: str, content_type: str = "text/plain") -> bool:
+        """Create an object only if it does not already exist."""
+        bucket, key = self._parse_uri(uri)
+        try:
+            self.client.put_object(
+                Bucket=bucket,
+                Key=key,
+                Body=data.encode("utf-8"),
+                ContentType=content_type,
+                IfNoneMatch="*",
+            )
+            return True
+        except self.client.exceptions.ClientError as exc:
+            code = exc.response.get("Error", {}).get("Code")
+            if code in {"PreconditionFailed", "412", "ConditionalRequestConflict"}:
+                return False
+            raise
+
     def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str:
         bucket, key = self._parse_uri(uri)
         self.client.put_object(Bucket=bucket, Key=key, Body=data, ContentType=content_type)

src/quant_platform_kit/cloud/azure_provider.py

@@ -161,6 +161,23 @@ def write_text(self, uri: str, data: str, content_type: str = "text/plain") -> s
         )
         return uri
 
+    def create_text(self, uri: str, data: str, content_type: str = "text/plain") -> bool:
+        """Create a blob only if it does not already exist."""
+        _, container, blob = self._parse_uri(uri)
+        try:
+            self._get_blob_client(container, blob).upload_blob(
+                data.encode("utf-8"),
+                overwrite=False,
+                content_settings={"content_type": content_type},
+            )
+            return True
+        except Exception as exc:
+            from azure.core.exceptions import ResourceExistsError
+
+            if isinstance(exc, ResourceExistsError):
+                return False
+            raise
+
     def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str:
         account, container, blob = self._parse_uri(uri)
         self._get_blob_client(container, blob).upload_blob(

src/quant_platform_kit/cloud/gcp_provider.py

@@ -128,6 +128,17 @@ def write_text(self, uri: str, data: str, content_type: str = "text/plain") -> s
         self.client.bucket(bucket).blob(blob).upload_from_string(data, content_type=content_type)
         return uri
 
+    def create_text(self, uri: str, data: str, content_type: str = "text/plain") -> bool:
+        from google.api_core.exceptions import Conflict, PreconditionFailed
+        bucket, blob = self._parse_uri(uri)
+        try:
+            self.client.bucket(bucket).blob(blob).upload_from_string(
+                data, content_type=content_type, if_generation_match=0
+            )
+            return True
+        except (Conflict, PreconditionFailed):
+            return False
+
     def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str:
         bucket, blob = self._parse_uri(uri)
         self.client.bucket(bucket).blob(blob).upload_from_string(data, content_type=content_type)

src/quant_platform_kit/cloud/local_provider.py

@@ -108,6 +108,19 @@ def write_text(self, uri: str, data: str, content_type: str = "text/plain") -> s
         path.write_text(data)
         return uri
 
+    def create_text(self, uri: str, data: str, content_type: str = "text/plain") -> bool:
+        path = self._to_local_path(uri)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        except FileExistsError:
+            return False
+        with os.fdopen(fd, "w") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        return True
+
     def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str:
         path = self._to_local_path(uri)
         path.parent.mkdir(parents=True, exist_ok=True)

src/quant_platform_kit/cloud/ports.py

@@ -83,6 +83,10 @@ def write_text(self, uri: str, data: str, content_type: str = "text/plain") -> s
         """写入文本对象，返回 URI。"""
         ...
 
+    def create_text(self, uri: str, data: str, content_type: str = "text/plain") -> bool:
+        """仅创建文本对象；对象已存在时返回 False。"""
+        ...
+
     def write_bytes(self, uri: str, data: bytes, content_type: str = "application/octet-stream") -> str:
         """写入二进制对象，返回 URI。"""
         ...

src/quant_platform_kit/common/execution_state.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 import tempfile
 from collections.abc import Callable, Mapping
@@ -161,6 +162,36 @@ def record_marker(
             path.parent.mkdir(parents=True, exist_ok=True)
             path.write_text(encoded, encoding="utf-8")
 
+    def claim_marker(self, marker_key: str, *, metadata: Mapping[str, Any] | None = None) -> bool:
+        """Atomically reserve an execution identity before broker submission."""
+        if not str(marker_key or "").strip():
+            raise ValueError("execution claim requires a marker key")
+        payload = json.dumps({
+            "schema_version": "execution_claim.v1",
+            "marker_key": str(marker_key),
+            "claimed_at": datetime.now(timezone.utc).isoformat(),
+            "state": "claimed",
+            "metadata": dict(metadata or {}),
+        }, ensure_ascii=False, indent=2, sort_keys=True)
+        if self.cloud_prefix_uri:
+            create = getattr(self._object_store(), "create_text", None)
+            if not callable(create):
+                raise RuntimeError("cloud execution store lacks atomic create-only support")
+            return bool(create(self._cloud_uri(marker_key), payload, "application/json"))
+        if self.local_dir:
+            path = self._local_path(marker_key)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            try:
+                fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            except FileExistsError:
+                return False
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                handle.write(payload)
+                handle.flush()
+                os.fsync(handle.fileno())
+            return True
+        raise RuntimeError("execution state store has no durable claim backend")
+
     def has_prior_execution_report(
         self,
         *,