feat(guardian): distributed execution authority consumption

services/guardian/execution/authorization_consumption.py

@@ -0,0 +1,58 @@
+from services.guardian.execution.authorization_consumption_store import (
+    InMemoryAuthorizationConsumptionStore,
+)
+
+
+class AuthorizationConsumptionRegistry:
+    """
+    Execution-authority consumption boundary.
+
+    The registry delegates atomic consumption to a store.
+
+    A store may be shared by multiple registry instances.
+    """
+
+    def __init__(self, store=None, backend=None):
+        if store is not None and backend is not None:
+            raise ValueError(
+                "AUTHORIZATION_CONSUMPTION_STORE_CONFLICT"
+            )
+
+        # Compatibility with the RED contract that introduced
+        # backend injection. A dict backend is converted into a
+        # shared store held by the backend itself.
+        if backend is not None:
+            if not isinstance(backend, dict):
+                raise TypeError(
+                    "AUTHORIZATION_CONSUMPTION_BACKEND_INVALID"
+                )
+
+            store = backend.get(
+                "_authorization_consumption_store"
+            )
+
+            if store is None:
+                store = InMemoryAuthorizationConsumptionStore()
+                backend[
+                    "_authorization_consumption_store"
+                ] = store
+
+        self._store = (
+            store
+            if store is not None
+            else InMemoryAuthorizationConsumptionStore()
+        )
+
+    def is_consumed(self, authorization_id):
+        return self._store.is_consumed(authorization_id)
+
+    def try_consume(self, authorization_id):
+        return self._store.try_consume(authorization_id)
+
+    def consume(self, authorization_id):
+        """
+        Backward-compatible wrapper.
+
+        Returns True only for the first successful consumption.
+        """
+        return self.try_consume(authorization_id)

services/guardian/execution/authorization_consumption_store.py

@@ -0,0 +1,142 @@
+from threading import Lock
+
+
+class _InMemoryAtomicBackend:
+    """
+    Process-local atomic backend.
+
+    The backend owns both consumption state and synchronization.
+    This preserves the existing default behavior while allowing
+    independently-created store adapters to share another backend.
+    """
+
+    def __init__(self):
+        self._consumed = set()
+        self._lock = Lock()
+
+    def is_consumed(self, authorization_id):
+        with self._lock:
+            return authorization_id in self._consumed
+
+    def try_consume(self, authorization_id):
+        with self._lock:
+            if authorization_id in self._consumed:
+                return False
+
+            self._consumed.add(authorization_id)
+            return True
+
+
+class InMemoryAuthorizationConsumptionStore:
+    """
+    Authorization-consumption store adapter.
+
+    By default, each instance uses its own process-local atomic backend.
+
+    A backend may be injected when multiple store instances must share
+    the same atomic consumption boundary.
+
+    Durable or distributed implementations must provide atomic
+    try_consume() semantics at the backend itself.
+    """
+
+    def __init__(self, backend=None):
+        self._backend = (
+            backend
+            if backend is not None
+            else _InMemoryAtomicBackend()
+        )
+
+    @staticmethod
+    def _validate_authorization_id(authorization_id):
+        if not authorization_id:
+            raise ValueError("AUTHORIZATION_ID_REQUIRED")
+
+    def is_consumed(self, authorization_id):
+        self._validate_authorization_id(authorization_id)
+        return self._backend.is_consumed(authorization_id)
+
+    def try_consume(self, authorization_id):
+        self._validate_authorization_id(authorization_id)
+        return self._backend.try_consume(authorization_id)
+
+
+from google.api_core.exceptions import Aborted
+from google.cloud import datastore
+
+
+class DatastoreAuthorizationConsumptionStore:
+    """
+    Google Cloud Datastore authorization-consumption store.
+
+    Atomicity belongs to the Datastore transaction boundary.
+
+    An authorization ID is represented by a deterministic entity key.
+    The first transaction that observes the key as absent creates the
+    consumption marker. Later transactions observe the existing marker
+    and return False.
+
+    A client must be injected explicitly so infrastructure selection,
+    project configuration, credentials, and lifecycle remain outside
+    this security primitive.
+    """
+
+    KIND = "GuardianAuthorizationConsumption"
+
+    def __init__(self, client):
+        if client is None:
+            raise ValueError("DATASTORE_CLIENT_REQUIRED")
+
+        self._client = client
+
+    @staticmethod
+    def _validate_authorization_id(authorization_id):
+        if not authorization_id:
+            raise ValueError("AUTHORIZATION_ID_REQUIRED")
+
+    def _key(self, authorization_id):
+        return self._client.key(
+            self.KIND,
+            authorization_id,
+        )
+
+    def is_consumed(self, authorization_id):
+        self._validate_authorization_id(authorization_id)
+
+        key = self._key(authorization_id)
+
+        return self._client.get(key) is not None
+
+    def try_consume(self, authorization_id):
+        self._validate_authorization_id(authorization_id)
+
+        key = self._key(authorization_id)
+
+        max_attempts = 5
+
+        for attempt in range(max_attempts):
+            try:
+                with self._client.transaction() as transaction:
+                    existing = self._client.get(
+                        key,
+                        transaction=transaction,
+                    )
+
+                    if existing is not None:
+                        return False
+
+                    entity = datastore.Entity(key=key)
+                    entity["authorization_id"] = authorization_id
+                    entity["consumed"] = True
+
+                    transaction.put(entity)
+
+                    return True
+
+            except Aborted:
+                if attempt == max_attempts - 1:
+                    raise
+
+        raise RuntimeError(
+            "DATASTORE_CONTENTION_RETRY_EXHAUSTED"
+        )

services/guardian/execution/execution_layer.py

@@ -2,6 +2,7 @@
 from services.guardian.execution.workflow_engine import WorkflowEngine
 from services.guardian.execution.audit_trail import AuditTrail
 from services.guardian.execution.authorization import AuthorizationIssuer
+from services.guardian.execution.authorization_consumption import AuthorizationConsumptionRegistry
 from services.guardian.governance.policy_engine import PolicyEngine
 from services.guardian.governance.approval_gate import ApprovalGate
 
@@ -21,11 +22,16 @@ class AutonomousExecutionLayer:
     Boolean authorization flags are not trusted.
     """
 
-    def __init__(self):
+    def __init__(self, authorization_consumption=None):
         self.executor = ActionExecutor()
         self.policy = PolicyEngine()
         self.approval = ApprovalGate()
         self.authority = AuthorizationIssuer()
+        self.authorization_consumption = (
+            authorization_consumption
+            if authorization_consumption is not None
+            else AuthorizationConsumptionRegistry()
+        )
         self.workflow = WorkflowEngine()
         self.audit = AuditTrail()
 
@@ -76,6 +82,16 @@ def run(self, decision):
                 "approval": approval,
             }
 
+        if not self.authorization_consumption.try_consume(
+            execution_authority.authorization_id
+        ):
+            return {
+                "status": "BLOCKED",
+                "reason": "EXECUTION_AUTHORITY_ALREADY_CONSUMED",
+                "policy": policy,
+                "approval": approval,
+            }
+
         result = self.executor.execute(action)
 
         return {

services/guardian/runtime/runtime_demo.py

@@ -23,8 +23,6 @@
     "DemandForecast",
     response["decision"],
     response["confidence"],
-    response.decision,
-    response.confidence,
     "LOW"
 )
 

services/guardian/tests/execution/test_authorization_consumption_store_atomic.py

@@ -0,0 +1,89 @@
+from concurrent.futures import ThreadPoolExecutor
+from threading import Lock
+
+from services.guardian.execution.authorization_consumption_store import (
+    InMemoryAuthorizationConsumptionStore,
+)
+
+
+class SharedAtomicBackend:
+    """
+    Test backend representing state shared by independently-created
+    consumption-store adapters.
+
+    The atomic claim primitive belongs to the shared backend boundary.
+    This models the semantic required from a future durable/distributed
+    implementation without selecting a cloud database yet.
+    """
+
+    def __init__(self):
+        self._consumed = set()
+        self._lock = Lock()
+
+    def is_consumed(self, authorization_id):
+        with self._lock:
+            return authorization_id in self._consumed
+
+    def try_consume(self, authorization_id):
+        with self._lock:
+            if authorization_id in self._consumed:
+                return False
+
+            self._consumed.add(authorization_id)
+            return True
+
+
+def _build_store(backend):
+    """
+    Future adapter contract under test.
+
+    Expected RED today because the current in-memory store constructor
+    does not accept an injected backend.
+    """
+    return InMemoryAuthorizationConsumptionStore(
+        backend=backend
+    )
+
+
+def test_claim_survives_store_recreation():
+    backend = SharedAtomicBackend()
+
+    store_a = _build_store(backend)
+
+    assert store_a.try_consume("persistent-auth-001") is True
+
+    del store_a
+
+    store_b = _build_store(backend)
+
+    assert store_b.try_consume("persistent-auth-001") is False
+
+
+def test_independent_store_instances_share_consumption_state():
+    backend = SharedAtomicBackend()
+
+    store_a = _build_store(backend)
+    store_b = _build_store(backend)
+
+    assert store_a.try_consume("persistent-auth-002") is True
+    assert store_b.try_consume("persistent-auth-002") is False
+
+
+def test_concurrent_claim_has_exactly_one_winner():
+    backend = SharedAtomicBackend()
+
+    stores = [
+        _build_store(backend)
+        for _ in range(32)
+    ]
+
+    authorization_id = "persistent-auth-race-001"
+
+    def claim(store):
+        return store.try_consume(authorization_id)
+
+    with ThreadPoolExecutor(max_workers=32) as pool:
+        results = list(pool.map(claim, stores))
+
+    assert results.count(True) == 1, results
+    assert results.count(False) == 31, results