feat: select the strongest qualified effect verifier

docs/EFFECT_KIT.md

@@ -26,9 +26,12 @@ from the reference apps.
    at-most-once counts, idempotency keys, and `{param: ...}` references that
    bind to the run's governed parameters. Contracts are substrate-neutral.
 2. **The deployment declares WHERE truth lives** — the `effects:` section of
-   `deployment.yaml` wires exactly one `EffectVerifier` (REST / GraphQL /
-   FHIR / SQL / file / email / document / document-hash, or a registered
-   plugin adapter) plus its secret-isolated auth.
+   `deployment.yaml` wires one `EffectVerifier` (REST / GraphQL / FHIR / SQL /
+   file / email / document / document-hash, or a registered plugin adapter)
+   plus its secret-isolated auth. When more than one reviewed read boundary is
+available, `candidates:` selects the strongest evidence tier for each resolved
+effect before input. It does not downgrade after input: an unavailable selected proof
+   halts or enters reconciliation.
 3. **The runtime refuses to guess.** Every verdict is CONFIRMED / REFUTED /
    INDETERMINATE; both non-confirmed verdicts HALT. A step that declares
    effects with no verifier configured HALTs. An escalated failure emits a
@@ -123,6 +126,39 @@ Any kind additionally accepts the evidence-minimization fields
 `evidence_redact_fields` / `evidence_keep_fields` (see "Evidence minimization"
 below).
 
+### Candidate selection when there is no database connection
+
+A database connection is not required. Configure the strongest qualified
+read boundary that the workflow has: REST/FHIR/GraphQL, read-only SQL, a file
+or report export, a separately authenticated read-only session through a
+plugin, or a persisted-state re-acquisition. Do not configure a same-surface
+screen read-back as proof of a consequential write.
+
+For more than one reviewed boundary, use `effects.candidates` instead of
+`effects.kind`. Each candidate has the normal `EffectsConfig` fields. Flow
+constructs every candidate before actuation, then selects the lowest numeric
+`VerificationTier` for each resolved effect; declaration order resolves a tie.
+This makes the choice deterministic and reviewable. A missing secret, an
+invalid config, or an invalid plugin tier refuses the run before input. The
+on-screen candidate is tier 3 only for that exact effect when its read-back
+reopens persisted state through a different path. It is tier 4 for a
+same-surface read-back. After the action, Flow does not fall back to a weaker
+candidate if the selected verifier is unavailable. It records the unavailable
+proof and halts or creates the normal reconciliation task.
+
+```yaml
+effects:
+  candidates:
+    - kind: document             # independent export arrival (tier 1)
+      root: /secure/exports
+      file_pattern: "confirmation-*.json"
+      document_format: json
+    - kind: onscreen             # lower-tier persisted-state read-back
+```
+
+The single `kind:` form remains the recommended configuration when one
+qualified verifier exists and remains fully compatible with prior deployments.
+
 The `sql` kind refuses to construct unless `sql_query` passes the read-only
 statement filter (single statement, `SELECT`/`WITH` leading keyword, no
 comments, no mutating/DDL/control keywords or known side-effecting functions,

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 "