From 5aa336b529854a8e024c12e685cc2b7feaabb8c4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 16:55:31 +0800 Subject: [PATCH] feat(p5): add bounded GCS artifact ports Co-Authored-By: Codex --- README.md | 8 + src/alpaca_platform/__init__.py | 8 + src/alpaca_platform/gcs_p5_artifacts.py | 226 ++++++++++++++++++ .../p5_default_parked_scheduler.py | 1 + src/alpaca_platform/shadow_receipt_store.py | 18 +- tests/test_gcs_p5_artifacts.py | 159 ++++++++++++ tests/test_p5_default_parked_scheduler.py | 27 +++ 7 files changed, 443 insertions(+), 4 deletions(-) create mode 100644 src/alpaca_platform/gcs_p5_artifacts.py create mode 100644 tests/test_gcs_p5_artifacts.py diff --git a/README.md b/README.md index 66e3c56..5fb10e2 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,14 @@ digest 当授权，也没有 reset breaker 的接口。 policy gate、QSL 风控内核每周期产生的 envelope、可信身份与审计/告警。完成这些外部步骤前，不能 添加定时调度或把此接口接到任何 paper/live 路径。 +`GcsP5ShadowArtifactReader` 与 `GcsCreateOnlyShadowReceiptStore` 现提供了对应的**受限 +GCS 传输适配器**，但仍未部署或自动实例化。调用方必须显式注入已经绑定 workload +identity 的 bucket client；适配器不会读取环境变量、服务帐号或 bucket 名称。reader 只读取 +精确的 `p5-inputs/.json`，不会列出或猜测“最新”工件；store 只对精确的 +`p5-receipts/.json` 执行带 `if_generation_match=0` 的 create-only 写入，并在 +冲突时重新读取对账。它不删除、不覆盖，也不产生定时任务、网络配置、券商访问或 P4/P6 +权限。存储不可用会闭合为 `PARKED/receipt_store_unavailable`。 + ```bash python -m alpaca_platform.shadow_ledger --input cycle.json --output receipt.json ``` diff --git a/src/alpaca_platform/__init__.py b/src/alpaca_platform/__init__.py index 30524f7..5609f5f 100644 --- a/src/alpaca_platform/__init__.py +++ b/src/alpaca_platform/__init__.py @@ -1,5 +1,10 @@ """Bounded P4/P5 execution gateway primitives.""" +from .gcs_p5_artifacts import ( + GcsCreateOnlyShadowReceiptStore, + GcsP5ArtifactError, + GcsP5ShadowArtifactReader, +) from .p5_default_parked_scheduler import ( P5_DEFAULT_PARKED_SCHEDULER_STATUS_SCHEMA, P5_DEFAULT_PARKED_SCHEDULER_SUMMARY_SCHEMA, @@ -69,6 +74,9 @@ "POLICY_GATE_RECEIPT_SCHEMA", "SCHEDULER_RESULT_SCHEMA", "CreateOnlyShadowReceiptStore", + "GcsCreateOnlyShadowReceiptStore", + "GcsP5ArtifactError", + "GcsP5ShadowArtifactReader", "InMemoryRestrictedP5ShadowArtifactReader", "InMemoryShadowReceiptStore", "P5DefaultParkedSchedulerError", diff --git a/src/alpaca_platform/gcs_p5_artifacts.py b/src/alpaca_platform/gcs_p5_artifacts.py new file mode 100644 index 0000000..fdba96e --- /dev/null +++ b/src/alpaca_platform/gcs_p5_artifacts.py @@ -0,0 +1,226 @@ +"""Narrow Google Cloud Storage adapters for P5 shadow artifacts. + +The P5 controller intentionally owns all candidate, policy, receipt, and risk +validation. These adapters only move one already-bounded JSON snapshot and +one already-validated admission through GCS. They never discover a latest +cycle, list a bucket, accept a URL, read credentials from configuration, or +touch a broker. + +Production code injects a ``google.cloud.storage.Bucket`` bound to a workload +identity. Keeping the bucket client injected makes the data boundary explicit +and keeps this module testable without a cloud SDK or network access. +""" + +from __future__ import annotations + +import copy +import json +import re +from collections.abc import Mapping +from typing import Any, Protocol + +from .p5_default_parked_scheduler import P5ShadowArtifactSnapshot, RestrictedP5ArtifactReadError +from .shadow_receipt_store import ShadowReceiptStoreError, validate_shadow_receipt_admission + +_CYCLE_ID = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$") +_PREFIX = re.compile(r"^[a-z0-9][a-z0-9._/-]*[a-z0-9]$") +_REQUEST_FIELDS = { + "cycle_id", + "forward_observation", + "policy_gate_receipt", + "risk_control", + "deployment_bundle_sha256", + "risk_gate_decision", + "prior_receipt", +} + + +class GcsP5ArtifactError(ShadowReceiptStoreError): + """A sanitized GCS adapter error safe for the P5 fail-closed boundary.""" + + +class _Blob(Protocol): + def download_as_bytes(self) -> bytes: + """Return the complete immutable JSON object.""" + + def upload_from_string(self, data: str, **kwargs: Any) -> None: + """Create one JSON object, honoring the supplied generation precondition.""" + + +class _Bucket(Protocol): + def blob(self, blob_name: str) -> _Blob: + """Return a handle for exactly one caller-derived object name.""" + + +def _cycle_id(value: Any) -> str: + if not isinstance(value, str) or not _CYCLE_ID.fullmatch(value): + raise GcsP5ArtifactError("P5 cycle_id must be a lowercase immutable identity") + return value + + +def _prefix(value: Any, *, label: str) -> str: + if not isinstance(value, str): + raise GcsP5ArtifactError(f"{label} must be a safe relative object prefix") + normalized = value.strip().strip("/") + if ( + not normalized + or "//" in normalized + or ".." in normalized + or any(part == "." for part in normalized.split("/")) + or not _PREFIX.fullmatch(normalized) + ): + raise GcsP5ArtifactError(f"{label} must be a safe relative object prefix") + return normalized + + +def _object_name(prefix: str, cycle_id: str) -> str: + return f"{prefix}/{cycle_id}.json" + + +def _is_status(error: BaseException, expected: int) -> bool: + values = [getattr(error, name, None) for name in ("code", "status_code")] + response = getattr(error, "response", None) + if response is not None: + values.extend(getattr(response, name, None) for name in ("code", "status_code")) + for value in values: + if callable(value): + try: + value = value() + except TypeError: + continue + if value == expected: + return True + return False + + +def _json_object(value: Any, *, label: str) -> dict[str, Any]: + if not isinstance(value, (bytes, bytearray)): + raise GcsP5ArtifactError(f"{label} must be UTF-8 JSON bytes") + if len(value) > 1_000_000: + raise GcsP5ArtifactError(f"{label} exceeds the bounded P5 artifact size") + try: + decoded = json.loads(bytes(value).decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise GcsP5ArtifactError(f"{label} is not valid JSON") from exc + if not isinstance(decoded, dict): + raise GcsP5ArtifactError(f"{label} must be a JSON object") + return decoded + + +def _snapshot_from_request(value: Mapping[str, Any], *, expected_cycle_id: str) -> P5ShadowArtifactSnapshot: + missing = sorted(_REQUEST_FIELDS - set(value)) + unknown = sorted(set(value) - _REQUEST_FIELDS) + if missing or unknown: + raise GcsP5ArtifactError("P5 artifact request has an invalid field set") + if _cycle_id(value["cycle_id"]) != expected_cycle_id: + raise GcsP5ArtifactError("P5 artifact request cycle_id does not match the requested cycle") + # Copy before handing the values to the controller so an SDK/fake caller + # cannot mutate the received snapshot after this adapter returns it. + try: + copied = copy.deepcopy(dict(value)) + except Exception as exc: # pragma: no cover - JSON input is normally copyable + raise GcsP5ArtifactError("P5 artifact request cannot be isolated") from exc + return P5ShadowArtifactSnapshot( + cycle_id=expected_cycle_id, + forward_observation=copied["forward_observation"], + policy_gate_receipt=copied["policy_gate_receipt"], + risk_control=copied["risk_control"], + deployment_bundle_sha256=copied["deployment_bundle_sha256"], + risk_gate_decision=copied["risk_gate_decision"], + prior_receipt=copied["prior_receipt"], + ) + + +class GcsP5ShadowArtifactReader: + """Read exactly one immutable P5 input request by cycle id. + + A missing request is a normal ``None`` result. Any other storage or shape + issue is deliberately collapsed to :class:`RestrictedP5ArtifactReadError`; + the default-PARKED controller turns it into a sanitized status without + publishing provider paths or error details. + """ + + def __init__(self, bucket: _Bucket, *, request_prefix: str = "p5-inputs") -> None: + if not callable(getattr(bucket, "blob", None)): + raise GcsP5ArtifactError("P5 artifact bucket must provide blob()") + self._bucket = bucket + self._request_prefix = _prefix(request_prefix, label="P5 request_prefix") + + def read_snapshot(self, *, cycle_id: str) -> P5ShadowArtifactSnapshot | None: + normalized_cycle_id = _cycle_id(cycle_id) + try: + payload = self._bucket.blob( + _object_name(self._request_prefix, normalized_cycle_id) + ).download_as_bytes() + except Exception as exc: # noqa: BLE001 - provider exceptions must fail closed. + if _is_status(exc, 404): + return None + raise RestrictedP5ArtifactReadError("P5 artifact snapshot is unavailable") from None + try: + return _snapshot_from_request( + _json_object(payload, label="P5 artifact request"), + expected_cycle_id=normalized_cycle_id, + ) + except GcsP5ArtifactError as exc: + raise RestrictedP5ArtifactReadError("P5 artifact snapshot is invalid") from exc + + +class GcsCreateOnlyShadowReceiptStore: + """Persist validated P5 admissions with GCS generation-match creation. + + The runtime identity needs only object read and object create on this + prefix. It never lists, overwrites, or deletes objects. A precondition + collision returns ``False`` so the P5 admission layer can re-read and + reconcile the immutable stored admission. + """ + + def __init__(self, bucket: _Bucket, *, receipt_prefix: str = "p5-receipts") -> None: + if not callable(getattr(bucket, "blob", None)): + raise GcsP5ArtifactError("P5 receipt bucket must provide blob()") + self._bucket = bucket + self._receipt_prefix = _prefix(receipt_prefix, label="P5 receipt_prefix") + + def read(self, cycle_id: str) -> dict[str, Any] | None: + normalized_cycle_id = _cycle_id(cycle_id) + try: + payload = self._bucket.blob( + _object_name(self._receipt_prefix, normalized_cycle_id) + ).download_as_bytes() + except Exception as exc: # noqa: BLE001 - provider exceptions must fail closed. + if _is_status(exc, 404): + return None + raise GcsP5ArtifactError("P5 receipt object is unavailable") from None + try: + admission = validate_shadow_receipt_admission( + _json_object(payload, label="P5 receipt admission") + ) + except GcsP5ArtifactError: + raise + except ShadowReceiptStoreError as exc: + raise GcsP5ArtifactError("P5 receipt object is invalid") from exc + if admission["cycle_id"] != normalized_cycle_id: + raise GcsP5ArtifactError("P5 receipt object cycle_id does not match the requested cycle") + return admission + + def create_if_absent(self, admission: Mapping[str, Any]) -> bool: + normalized = validate_shadow_receipt_admission(admission) + payload = json.dumps( + normalized, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ) + try: + self._bucket.blob( + _object_name(self._receipt_prefix, normalized["cycle_id"]) + ).upload_from_string( + payload, + content_type="application/json", + if_generation_match=0, + ) + except Exception as exc: # noqa: BLE001 - provider exceptions must fail closed. + if _is_status(exc, 412): + return False + raise GcsP5ArtifactError("P5 receipt object could not be created") from None + return True diff --git a/src/alpaca_platform/p5_default_parked_scheduler.py b/src/alpaca_platform/p5_default_parked_scheduler.py index e0934bf..9069a6f 100644 --- a/src/alpaca_platform/p5_default_parked_scheduler.py +++ b/src/alpaca_platform/p5_default_parked_scheduler.py @@ -71,6 +71,7 @@ "risk_gate_decision_invalid", "risk_gate_decision_mismatch", "risk_gate_decision_prohibited", + "receipt_store_unavailable", "receipt_conflict", } diff --git a/src/alpaca_platform/shadow_receipt_store.py b/src/alpaca_platform/shadow_receipt_store.py index 3d20470..9a37304 100644 --- a/src/alpaca_platform/shadow_receipt_store.py +++ b/src/alpaca_platform/shadow_receipt_store.py @@ -64,6 +64,7 @@ "risk_gate_decision_invalid", "risk_gate_decision_mismatch", "risk_gate_decision_prohibited", + "receipt_store_unavailable", "receipt_conflict", } @@ -374,10 +375,19 @@ def persist_shadow_cycle_outcome( admission_sha256=None, ) cycle_id = admission["cycle_id"] - existing = store.read(cycle_id) - created = existing is None and store.create_if_absent(admission) - - stored = store.read(cycle_id) + try: + existing = store.read(cycle_id) + created = existing is None and store.create_if_absent(admission) + stored = store.read(cycle_id) + except ShadowReceiptStoreError: + return _persistence_result( + cycle_id=cycle_id, + computed_at=admission["computed_at"], + status="PARKED", + reason_code="receipt_store_unavailable", + shadow_receipt_sha256=None, + admission_sha256=None, + ) if stored is None: _fail("create-only shadow receipt store did not retain or expose the cycle admission") stored_admission = validate_shadow_receipt_admission(stored) diff --git a/tests/test_gcs_p5_artifacts.py b/tests/test_gcs_p5_artifacts.py new file mode 100644 index 0000000..d02ec5e --- /dev/null +++ b/tests/test_gcs_p5_artifacts.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from alpaca_platform import gcs_p5_artifacts +from alpaca_platform.gcs_p5_artifacts import ( + GcsCreateOnlyShadowReceiptStore, + GcsP5ArtifactError, + GcsP5ShadowArtifactReader, +) +from alpaca_platform.p5_default_parked_scheduler import RestrictedP5ArtifactReadError +from alpaca_platform.shadow_receipt_store import ShadowReceiptStoreError + + +class FakeGcsError(Exception): + def __init__(self, status_code: int) -> None: + super().__init__(f"provider status {status_code}") + self.status_code = status_code + + +class FakeBlob: + def __init__(self, bucket: FakeBucket, name: str) -> None: + self._bucket = bucket + self._name = name + + def download_as_bytes(self) -> bytes: + self._bucket.downloaded.append(self._name) + if self._bucket.read_error is not None: + raise self._bucket.read_error + try: + return self._bucket.objects[self._name] + except KeyError as exc: + raise FakeGcsError(404) from exc + + def upload_from_string(self, data: str, **kwargs: Any) -> None: + self._bucket.uploaded.append((self._name, data, kwargs)) + if self._bucket.write_error is not None: + raise self._bucket.write_error + if kwargs.get("if_generation_match") == 0 and self._name in self._bucket.objects: + raise FakeGcsError(412) + self._bucket.objects[self._name] = data.encode("utf-8") + + +class FakeBucket: + def __init__(self) -> None: + self.objects: dict[str, bytes] = {} + self.downloaded: list[str] = [] + self.uploaded: list[tuple[str, str, dict[str, Any]]] = [] + self.read_error: Exception | None = None + self.write_error: Exception | None = None + + def blob(self, blob_name: str) -> FakeBlob: + return FakeBlob(self, blob_name) + + +def snapshot_payload(cycle_id: str) -> dict[str, object]: + return { + "cycle_id": cycle_id, + "forward_observation": {"opaque": "forward"}, + "policy_gate_receipt": {"opaque": "policy"}, + "risk_control": {"opaque": "risk"}, + "deployment_bundle_sha256": "a" * 64, + "risk_gate_decision": {"opaque": "decision"}, + "prior_receipt": None, + } + + +def test_reader_reads_one_exact_cycle_without_listing_or_discovery(): + bucket = FakeBucket() + cycle_id = "tqqq_core_only_p2_v5_shadow_20260820" + payload = snapshot_payload(cycle_id) + bucket.objects[f"p5-inputs/{cycle_id}.json"] = json.dumps(payload).encode("utf-8") + + snapshot = GcsP5ShadowArtifactReader(bucket).read_snapshot(cycle_id=cycle_id) + + assert snapshot is not None + assert snapshot.cycle_id == cycle_id + assert snapshot.forward_observation == {"opaque": "forward"} + assert bucket.downloaded == [f"p5-inputs/{cycle_id}.json"] + + +def test_reader_missing_or_malformed_objects_fail_closed(): + bucket = FakeBucket() + cycle_id = "tqqq_core_only_p2_v5_shadow_20260820" + reader = GcsP5ShadowArtifactReader(bucket) + + assert reader.read_snapshot(cycle_id=cycle_id) is None + + bucket.objects[f"p5-inputs/{cycle_id}.json"] = b"[]" + with pytest.raises(RestrictedP5ArtifactReadError, match="snapshot is invalid"): + reader.read_snapshot(cycle_id=cycle_id) + + bucket.read_error = FakeGcsError(503) + with pytest.raises(RestrictedP5ArtifactReadError, match="snapshot is unavailable"): + reader.read_snapshot(cycle_id=cycle_id) + + +def test_reader_rejects_unsafe_identifiers_and_prefixes(): + with pytest.raises(GcsP5ArtifactError, match="cycle_id"): + GcsP5ShadowArtifactReader(FakeBucket()).read_snapshot(cycle_id="../latest") + with pytest.raises(GcsP5ArtifactError, match="request_prefix"): + GcsP5ShadowArtifactReader(FakeBucket(), request_prefix="p5-inputs/../other") + with pytest.raises(GcsP5ArtifactError, match="request_prefix"): + GcsP5ShadowArtifactReader(FakeBucket(), request_prefix="p5-inputs/./other") + + +def test_receipt_store_is_create_only_and_uses_a_generation_precondition(monkeypatch): + bucket = FakeBucket() + cycle_id = "tqqq_core_only_p2_v5_shadow_20260820" + admission = {"cycle_id": cycle_id, "opaque": "already validated upstream"} + monkeypatch.setattr( + gcs_p5_artifacts, + "validate_shadow_receipt_admission", + lambda value: dict(value), + ) + store = GcsCreateOnlyShadowReceiptStore(bucket) + + assert store.read(cycle_id) is None + assert store.create_if_absent(admission) is True + assert store.create_if_absent(admission) is False + assert store.read(cycle_id) == admission + assert [item[0] for item in bucket.uploaded] == [ + f"p5-receipts/{cycle_id}.json", + f"p5-receipts/{cycle_id}.json", + ] + assert bucket.uploaded[0][2] == { + "content_type": "application/json", + "if_generation_match": 0, + } + assert json.loads(bucket.objects[f"p5-receipts/{cycle_id}.json"]) == admission + + +def test_receipt_store_returns_bounded_errors_for_unavailable_or_invalid_objects(monkeypatch): + bucket = FakeBucket() + cycle_id = "tqqq_core_only_p2_v5_shadow_20260820" + admission = {"cycle_id": cycle_id} + store = GcsCreateOnlyShadowReceiptStore(bucket) + monkeypatch.setattr( + gcs_p5_artifacts, + "validate_shadow_receipt_admission", + lambda value: dict(value), + ) + + bucket.write_error = FakeGcsError(503) + with pytest.raises(ShadowReceiptStoreError, match="could not be created"): + store.create_if_absent(admission) + + bucket.write_error = None + bucket.objects[f"p5-receipts/{cycle_id}.json"] = b"{}" + monkeypatch.setattr( + gcs_p5_artifacts, + "validate_shadow_receipt_admission", + lambda _value: (_ for _ in ()).throw(ShadowReceiptStoreError("invalid admission")), + ) + with pytest.raises(GcsP5ArtifactError, match="receipt object is invalid"): + store.read(cycle_id) diff --git a/tests/test_p5_default_parked_scheduler.py b/tests/test_p5_default_parked_scheduler.py index b5bcf62..11c959e 100644 --- a/tests/test_p5_default_parked_scheduler.py +++ b/tests/test_p5_default_parked_scheduler.py @@ -21,6 +21,7 @@ validate_p5_default_parked_scheduler_status, validate_p5_default_parked_scheduler_summary, ) +from alpaca_platform.shadow_receipt_store import ShadowReceiptStoreError def sha(character: str) -> str: @@ -192,6 +193,14 @@ def create_if_absent(self, admission): # type: ignore[no-untyped-def] return super().create_if_absent(admission) +class UnavailableStore: + def read(self, cycle_id: str): # type: ignore[no-untyped-def] + raise ShadowReceiptStoreError("storage unavailable") + + def create_if_absent(self, admission): # type: ignore[no-untyped-def] + raise ShadowReceiptStoreError("storage unavailable") + + def test_default_without_reader_is_parked_and_never_touches_store(): store = SpyStore() @@ -302,6 +311,24 @@ def test_complete_snapshot_records_once_then_reconciles_with_sanitized_dedup_sum assert forbidden not in rendered +def test_receipt_store_unavailability_parks_a_ready_snapshot_without_raising(): + reader = InMemoryRestrictedP5ShadowArtifactReader() + snapshot = ready_snapshot() + reader.put_snapshot(snapshot) + + outcome = run_p5_default_parked_shadow_cycle( + cycle_id=snapshot.cycle_id, + computed_at="2026-08-20T20:00:00Z", + artifact_reader=reader, + receipt_store=UnavailableStore(), + ) + + assert outcome.status["status"] == "PARKED" + assert outcome.status["reason_code"] == "receipt_store_unavailable" + assert outcome.status["shadow_receipt_sha256"] is None + assert outcome.status["admission_sha256"] is None + + def test_unavailable_reader_is_parked_without_exposing_adapter_error_or_using_store(): class UnavailableReader: def read_snapshot(self, *, cycle_id: str) -> P5ShadowArtifactSnapshot | None: