feat: enforce signed qualification admission before actuation

openadapt_flow/__main__.py

@@ -56,6 +56,7 @@
 
 import argparse
 import json
+import os
 import sys
 from contextlib import contextmanager
 from pathlib import Path
@@ -1566,6 +1567,12 @@ def _cmd_run(args: argparse.Namespace) -> int:
         resolve_execution_profile,
     )
     from openadapt_flow.ir import Workflow
+    from openadapt_flow.qualification_admission import (
+        QualificationAdmissionError,
+        expected_from_payload,
+        load_qualification_signer_trust,
+        verify_qualification_admission,
+    )
     from openadapt_flow.run_gate import (
         build_qualification_case_authorization,
         build_runtime_authorization,
@@ -1706,6 +1713,29 @@ def _cmd_run(args: argparse.Namespace) -> int:
                 "run REFUSED: managed dispatch binding is invalid. Nothing was executed."
             )
             return 2
+        if authorization.qualification_admission is None:
+            print(
+                "run REFUSED: managed production dispatch has no signed "
+                "qualification admission. Nothing was executed."
+            )
+            return 2
+        try:
+            signer_trust = load_qualification_signer_trust(
+                os.environ.get("OPENADAPT_QUALIFICATION_SIGNERS_JSON", "")
+            )
+            verify_qualification_admission(
+                authorization.qualification_admission,
+                trusted_signers=signer_trust,
+                expected=expected_from_payload(
+                    authorization.qualification_admission.payload
+                ),
+            )
+        except QualificationAdmissionError:
+            print(
+                "run REFUSED: qualification admission is not signed by an "
+                "active trusted authority. Nothing was executed."
+            )
+            return 2
         local_authorization = build_runtime_authorization(
             workflow,
             report,

openadapt_flow/connector/__init__.py

@@ -19,7 +19,11 @@
 ``connector enroll`` is retained for mock/development control planes.
 """
 
-from openadapt_flow.connector.client import ConnectorClient, ConnectorClientError
+from openadapt_flow.connector.client import (
+    MANAGED_QUALIFICATION_AUTHORITY_CAPABILITY,
+    ConnectorClient,
+    ConnectorClientError,
+)
 from openadapt_flow.connector.config import (
     ConnectorConfigError,
     ConnectorSettings,
@@ -65,6 +69,7 @@
     "GroundingModel",
     "InMemoryCustomerStorage",
     "LocalCustomerStorage",
+    "MANAGED_QUALIFICATION_AUTHORITY_CAPABILITY",
     "build_run_argv",
     "build_storage",
     "connector_config_path",

openadapt_flow/connector/client.py

@@ -28,7 +28,7 @@
 
 from openadapt_flow.hosted import HostedError
 
-MANAGED_DELIVERY_AUTHORITY_CAPABILITY = "managed_delivery_authority_v1"
+MANAGED_QUALIFICATION_AUTHORITY_CAPABILITY = "managed_qualification_authority_v2"
 
 
 class ConnectorClientError(HostedError):
@@ -93,7 +93,7 @@ def poll(self, wait_s: int) -> Optional[dict[str, Any]]:
             "/api/connector/poll",
             json={
                 "wait": wait_s,
-                "capabilities": [MANAGED_DELIVERY_AUTHORITY_CAPABILITY],
+                "capabilities": [MANAGED_QUALIFICATION_AUTHORITY_CAPABILITY],
             },
             headers=self._bearer(),
         )

openadapt_flow/execution_profiles.py

@@ -167,6 +167,26 @@ def execution_profile_contract(
     return _CONTRACTS[resolve_execution_profile(value)]
 
 
+def requires_signed_qualification_admission(
+    value: ExecutionProfile | str | None,
+    *,
+    will_actuate: bool,
+) -> bool:
+    """Return whether this invocation must carry production qualification.
+
+    Recording, compilation, linting, qualification, and report-only simulation
+    do not cross an actuation boundary. They stay available without production
+    authority. Demo is the one named execution profile that can actuate without
+    a signed qualification admission. Every real Standard or Regulated action
+    requires that admission, independent of whether Cloud or the customer owns
+    the runner.
+    """
+
+    if not will_actuate or value is None:
+        return False
+    return execution_profile_contract(value).production
+
+
 def required_effect_tier(
     workflow: Workflow,
     profile: ExecutionProfile | str,