feat: carry managed Execute authority through BYOC

openadapt_flow/connector/client.py

@@ -28,6 +28,8 @@
 
 from openadapt_flow.hosted import HostedError
 
+MANAGED_DELIVERY_AUTHORITY_CAPABILITY = "managed_delivery_authority_v1"
+
 
 class ConnectorClientError(HostedError):
     """An outbound control-plane call failed."""
@@ -88,7 +90,12 @@ def poll(self, wait_s: int) -> Optional[dict[str, Any]]:
         """Long-poll for the next leased job. Returns the poll envelope
         ``{"job": {...}}`` or None on a 204 (no work in the wait window)."""
         resp = self._client.post(
-            "/api/connector/poll", json={"wait": wait_s}, headers=self._bearer()
+            "/api/connector/poll",
+            json={
+                "wait": wait_s,
+                "capabilities": [MANAGED_DELIVERY_AUTHORITY_CAPABILITY],
+            },
+            headers=self._bearer(),
         )
         if resp.status_code == 204:
             return None

openadapt_flow/connector/executor.py

@@ -41,13 +41,21 @@
 import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Callable, Optional
+from typing import Any, Callable, Mapping, Optional
+from urllib.parse import urlsplit
 
 from openadapt_flow.connector.config import ConnectorSettings
 from openadapt_flow.connector.protocol import ByocGovernanceError, ByocJob
 from openadapt_flow.connector.storage import CustomerStorage, extract_bundle_archive
 from openadapt_flow.failure_signals import automation_failure_signal
 from openadapt_flow.ir import RunReport
+from openadapt_flow.runner.dispatch_envelope import (
+    write_managed_dispatch_envelope_value,
+)
+from openadapt_flow.runtime.durable.authority import (
+    REMOTE_AUTHORITY_TOKEN_ENV,
+    REMOTE_AUTHORITY_URL_ENV,
+)
 
 #: A run-gate refusal (fail-closed admission denied) exits 2 before the replay
 #: creates report.json.
@@ -64,7 +72,9 @@ class RunOutcome:
 
 #: A runner maps argv -> RunOutcome. The default shells the governed ``run`` CLI
 #: in a child process; tests inject a fake to avoid launching a GUI.
-Runner = Callable[[list[str], Path], RunOutcome]
+Runner = Callable[[list[str], Path, Mapping[str, str]], RunOutcome]
+
+_MANAGED_DELIVERY_AUTHORITY_PATH = "/api/internal/managed-delivery-permit"
 
 
 @dataclass
@@ -104,6 +114,7 @@ def build_run_argv(
     bundle_dir: Path,
     run_dir: Path,
     params_file: Optional[Path],
+    managed_dispatch_file: Optional[Path] = None,
 ) -> list[str]:
     """The exact governed CLI invocation for a verified BYOC dispatch.
 
@@ -132,14 +143,23 @@ def build_run_argv(
         argv += ["--params-file", str(params_file)]
     if job.target_url:
         argv += ["--url", job.target_url]
+    if managed_dispatch_file is not None:
+        argv += ["--managed-dispatch-file", str(managed_dispatch_file)]
     if settings.allow_unencrypted:
         # Local escape hatch, mirroring the governed run CLI; default OFF.
         argv.append("--allow-unencrypted")
     return argv
 
 
-def _subprocess_runner(argv: list[str], run_dir: Path) -> RunOutcome:
-    proc = subprocess.run(argv, capture_output=True, text=True)  # nosec - fixed argv
+def _subprocess_runner(
+    argv: list[str], run_dir: Path, child_env: Mapping[str, str]
+) -> RunOutcome:
+    proc = subprocess.run(  # nosec - fixed argv and exact process-local env
+        argv,
+        capture_output=True,
+        text=True,
+        env=dict(child_env),
+    )
     report_path = run_dir / "report.json"
     report: dict[str, Any] = {}
     if report_path.is_file():
@@ -267,6 +287,44 @@ def _write_policy_audit(job: ByocJob, run_dir: Path) -> None:
         pass
 
 
+def _assert_managed_authority_is_pinned(
+    job: ByocJob, settings: ConnectorSettings
+) -> None:
+    """Keep the run capability on the enrolled control-plane origin."""
+
+    if job.managed_dispatch is None:
+        return
+    try:
+        configured = urlsplit(settings.control_plane_url)
+        delivered = urlsplit(job.managed_delivery_authority_url or "")
+        configured_port = configured.port or (
+            443 if configured.scheme == "https" else None
+        )
+        delivered_port = delivered.port or (
+            443 if delivered.scheme == "https" else None
+        )
+    except ValueError as exc:
+        raise ByocGovernanceError(
+            "byoc managed delivery authority URL is invalid"
+        ) from exc
+    if (
+        configured.scheme != "https"
+        or not configured.hostname
+        or configured.username is not None
+        or configured.password is not None
+        or bool(configured.query)
+        or bool(configured.fragment)
+        or delivered.scheme != "https"
+        or delivered.hostname != configured.hostname
+        or delivered_port != configured_port
+        or delivered.path != _MANAGED_DELIVERY_AUTHORITY_PATH
+    ):
+        raise ByocGovernanceError(
+            "byoc managed delivery authority is not pinned to the enrolled "
+            "control plane (fail closed)"
+        )
+
+
 def execute_job(
     job: ByocJob,
     settings: ConnectorSettings,
@@ -284,6 +342,7 @@ def execute_job(
     # 1. Governance gates (fail closed BEFORE any bundle is fetched or run).
     try:
         job.ensure_governed(require_run_token=require_run_token)
+        _assert_managed_authority_is_pinned(job, settings)
     except ByocGovernanceError as exc:
         return ExecutionResult("failed", {}, None, job.report_ref(), str(exc))
     if not _grounding_env_available(job):
@@ -332,10 +391,32 @@ def execute_job(
 
         try:
             params_file = _write_params_file(job.params, run_dir)
+            managed_dispatch_file = None
+            child_env = os.environ.copy()
+            # Never let a long-running Connector inherit stale authority from
+            # its service environment. Add the exact run capability only for
+            # the child that also receives the matching dispatch envelope.
+            child_env.pop(REMOTE_AUTHORITY_URL_ENV, None)
+            child_env.pop(REMOTE_AUTHORITY_TOKEN_ENV, None)
+            if job.managed_dispatch is not None:
+                managed_dispatch_file = write_managed_dispatch_envelope_value(
+                    run_dir / "managed-dispatch.json", job.managed_dispatch
+                )
+                child_env[REMOTE_AUTHORITY_URL_ENV] = str(
+                    job.managed_delivery_authority_url
+                )
+                child_env[REMOTE_AUTHORITY_TOKEN_ENV] = str(job.run_token)
 
             # 3. The governed, fail-closed child invocation.
-            argv = build_run_argv(job, settings, Path(bundle_dir), run_dir, params_file)
-            outcome = runner(argv, run_dir)
+            argv = build_run_argv(
+                job,
+                settings,
+                Path(bundle_dir),
+                run_dir,
+                params_file,
+                managed_dispatch_file,
+            )
+            outcome = runner(argv, run_dir, child_env)
         except Exception as exc:
             return ExecutionResult(
                 "failed",

openadapt_flow/connector/protocol.py

@@ -21,10 +21,12 @@
 from __future__ import annotations
 
 from typing import Any, Optional
+from urllib.parse import urlsplit
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 
 from openadapt_flow.ir import ExecutionTargetKind
+from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope
 
 
 class ByocJobParseError(ValueError):
@@ -100,9 +102,15 @@ class ByocJob(BaseModel):
     bundle_download_url: Optional[str] = None
 
     # --- Governed callback binding (fail-closed) -------------------------------
-    #: Run-scoped HMAC capability presented as ``x-run-token`` on the PHI-free
-    #: callback. Not a secret value — proves this run, forbids forging another.
+    #: Run-scoped bearer capability presented as ``x-run-token`` on the PHI-free
+    #: callback. It proves this run and must remain private.
     run_token: Optional[str] = None
+    #: Exact Cloud-issued, one-run authority. It stays PHI-free and is written
+    #: to a private local file before the governed child starts.
+    managed_dispatch: Optional[ManagedDispatchEnvelope] = None
+    #: HTTPS endpoint that issues monotonic per-action delivery permits for the
+    #: exact managed dispatch. The run-scoped ``run_token`` authenticates it.
+    managed_delivery_authority_url: Optional[str] = None
     bundle_version_id: Optional[str] = None
     runtime_validation_id: Optional[str] = None
     #: SHA-256 of the exact approved sanitized derivative ZIP bytes staged at
@@ -162,6 +170,46 @@ def ensure_governed(self, *, require_run_token: bool = True) -> None:
                 "byoc dispatch is missing a run-scoped callback token; refusing "
                 "to run a job whose outcome we could not report (fail closed)"
             )
+        if (self.managed_dispatch is None) != (
+            self.managed_delivery_authority_url is None
+        ):
+            raise ByocGovernanceError(
+                "byoc managed dispatch and delivery authority must be supplied "
+                "together (fail closed)"
+            )
+        if self.managed_dispatch is not None:
+            try:
+                authorization = self.managed_dispatch.exact_authorization()
+            except ValueError as exc:
+                raise ByocGovernanceError(
+                    "byoc managed dispatch has an invalid authorization binding"
+                ) from exc
+            if self.managed_dispatch.run_id != self.run_id:
+                raise ByocGovernanceError(
+                    "byoc managed dispatch belongs to a different run (fail closed)"
+                )
+            if not authorization.approval_source.startswith("hosted:"):
+                raise ByocGovernanceError(
+                    "byoc managed dispatch lacks hosted authorization provenance"
+                )
+            try:
+                parsed = urlsplit(self.managed_delivery_authority_url or "")
+            except ValueError as exc:
+                raise ByocGovernanceError(
+                    "byoc managed delivery authority URL is invalid"
+                ) from exc
+            if (
+                parsed.scheme != "https"
+                or not parsed.hostname
+                or parsed.username is not None
+                or parsed.password is not None
+                or bool(parsed.query)
+                or bool(parsed.fragment)
+            ):
+                raise ByocGovernanceError(
+                    "byoc managed delivery authority must be a credential-free "
+                    "HTTPS endpoint without query or fragment"
+                )
         if not self.bundle_version_id or not self.runtime_validation_id:
             raise ByocGovernanceError(
                 "byoc dispatch is missing its immutable bundle-version or "

openadapt_flow/runner/dispatch_envelope.py

@@ -93,18 +93,18 @@ def exact_authorization(self) -> GovernedRunAuthorization:
         return self.authorization
 
 
-def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path:
-    """Write one already verified dispatch without following a path."""
-
-    run_id = verified.payload.run_id
-    authorization = verified.payload.authorization
-    envelope = ManagedDispatchEnvelope(
-        run_id=run_id,
-        bundle_content_digest=authorization.bundle_content_digest,
-        runtime_inputs_digest=authorization.runtime_inputs_digest,
-        authorization=authorization,
-        dispatch_binding_sha256=verified.payload.dispatch_binding_sha256,
-    )
+def write_managed_dispatch_envelope_value(
+    path: Path, envelope: ManagedDispatchEnvelope
+) -> Path:
+    """Write one strictly parsed dispatch envelope without following a path.
+
+    This is the shared process-boundary writer for both push runners and the
+    customer-controlled outbound-pull Connector.  The caller must already hold
+    the strict :class:`ManagedDispatchEnvelope` model; this function rechecks
+    its internal authorization binding before any bytes reach disk.
+    """
+
+    envelope.exact_authorization()
     path = Path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
     flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
@@ -136,6 +136,21 @@ def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") ->
     return path
 
 
+def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path:
+    """Write one already verified runner dispatch without following a path."""
+
+    run_id = verified.payload.run_id
+    authorization = verified.payload.authorization
+    envelope = ManagedDispatchEnvelope(
+        run_id=run_id,
+        bundle_content_digest=authorization.bundle_content_digest,
+        runtime_inputs_digest=authorization.runtime_inputs_digest,
+        authorization=authorization,
+        dispatch_binding_sha256=verified.payload.dispatch_binding_sha256,
+    )
+    return write_managed_dispatch_envelope_value(path, envelope)
+
+
 def _read_managed_dispatch_envelope(path: Path) -> ManagedDispatchEnvelope:
     """Read only a regular, private file owned by this effective user."""
 

openadapt_flow/runtime/durable/authority.py

@@ -31,7 +31,7 @@
 from typing import Any, Iterator, Literal, Optional
 from urllib.error import HTTPError, URLError
 from urllib.parse import urlparse
-from urllib.request import Request, urlopen
+from urllib.request import HTTPRedirectHandler, Request, build_opener
 
 from pydantic import BaseModel, ConfigDict
 
@@ -70,6 +70,27 @@
 }
 
 
+class _RefuseRemoteAuthorityRedirects(HTTPRedirectHandler):
+    """Keep the runner credential on the configured authority origin."""
+
+    def redirect_request(
+        self,
+        req: Request,
+        fp: Any,
+        code: int,
+        msg: str,
+        headers: Any,
+        newurl: str,
+    ) -> None:
+        raise HTTPError(
+            req.full_url,
+            code,
+            "remote delivery authority redirects are refused",
+            headers,
+            fp,
+        )
+
+
 def _is_windows() -> bool:
     return os.name == "nt"
 
@@ -1645,7 +1666,8 @@ def _require_remote_delivery_permit(
                 response_bytes = self._remote_transport(url, headers, body)
             else:
                 request = Request(url, data=body, headers=headers, method="POST")
-                with urlopen(request, timeout=10) as response:  # nosec B310 - HTTPS above
+                opener = build_opener(_RefuseRemoteAuthorityRedirects())
+                with opener.open(request, timeout=10) as response:  # nosec B310 - HTTPS above
                     if not 200 <= response.status < 300:
                         raise DurableAuthorityBusy("remote delivery authority refused")
                     response_bytes = response.read(