abrichr · Nov 5, 2026 · Nov 6, 2026 · Nov 6, 2026 · Nov 6, 2026 · Nov 6, 2026
@@ -158,7 +158,15 @@ def describe(self) -> dict[str, str]:
     def refresh(self, entry: Any, run_id: str) -> None:
         return None
 
-    def execute(self, entry: Any, run_id: str, inputs: dict[str, str], raw: Any) -> Any:
+    def execute(
+        self,
+        entry: Any,
+        run_id: str,
+        inputs: dict[str, str],
+        raw: Any,
+        *,
+        admitted_bundle_sha256: str | None = None,
+    ) -> Any:
         from openadapt_flow.agent.service import EngineResult
 
         sandbox = {"engine": "simulated", "case": "worklist"}

@@ -79,7 +79,9 @@ def __init__(self, tmp: Path, reason: str = "unexpected_screen") -> None:
         self.tmp = tmp
         self.reason = reason
 
-    def execute(self, entry, run_id, inputs, raw) -> EngineResult:
+    def execute(
+        self, entry, run_id, inputs, raw, *, admitted_bundle_sha256=None
+    ) -> EngineResult:
         run_dir = self.tmp / "runs" / run_id
         run_dir.mkdir(parents=True, exist_ok=True)
         return EngineResult(

@@ -75,7 +75,9 @@ class ReplayEngine:
     def __init__(self, tmp: Path, desk: RepairDesk, bundle: Path, held: HeldPage) -> None:
         self.tmp, self.desk, self.bundle, self.held = tmp, desk, bundle, held
 
-    def execute(self, entry, run_id, inputs, raw) -> EngineResult:
+    def execute(
+        self, entry, run_id, inputs, raw, *, admitted_bundle_sha256=None
+    ) -> EngineResult:
         bundle = self.desk.active_bundle(entry.name, self.bundle)
         run_dir = self.tmp / "runs" / run_id
         app = fx.DialogApp()

@@ -2865,6 +2865,16 @@ def _cmd_replay(args: argparse.Namespace) -> int:
     refuse_unadmitted_draft(bundle)
     run_dir = Path(args.run_dir) if args.run_dir else _default_run_dir()
     workflow = Workflow.load(bundle)
+    # Replay reloads the bundle: it must still be the intake-admitted bytes,
+    # and the authorization `run` built must name them (#1890).
+    refused_admission = _admitted_bundle_refusal(
+        args,
+        bundle,
+        workflow,
+        getattr(args, "_governed_run_authorization", None),
+    )
+    if refused_admission is not None:
+        return refused_admission
     # `undo run` only: the person-approved undo steps of this same bundle
     # (openadapt_flow.cli_undo), run through the normal replay path.
     workflow = getattr(args, "_undo_workflow", None) or workflow
@@ -3974,6 +3984,51 @@ def _json(path: str) -> Any:
     return 0 if receipt["outcome"] == "verified" else 1
 
 
+def _admitted_bundle_refusal(
+    args: argparse.Namespace,
+    bundle: Path,
+    workflow: Any = None,
+    authorization: Any = None,
+) -> Optional[int]:
+    """Refuse unless the bundle is the exact one an intake grant admitted.
+
+    ``--admitted-bundle-sha256`` carries the intake grant's digest (#1890)
+    from the agent runner into this process. The digest is recomputed from
+    the files and, when given, from the in-memory workflow and compared with
+    the run authorization's bundle digest, so an edit after the intake gate
+    can neither be admitted nor authorized. Returns an exit code to stop
+    with, or None to continue.
+    """
+
+    admitted = getattr(args, "admitted_bundle_sha256", None)
+    if not admitted:
+        return None
+    from openadapt_flow.bundle_validation import (
+        compute_content_digest,
+        compute_file_hashes,
+    )
+    from openadapt_flow.intake.param_gate import bundle_sha256
+
+    try:
+        matches = bundle_sha256(bundle) == admitted
+        if matches and workflow is not None:
+            matches = (
+                compute_content_digest(workflow, compute_file_hashes(workflow, bundle))
+                == admitted
+            )
+    except Exception:  # noqa: BLE001 - unreadable bundle is not the admitted one
+        matches = False
+    if matches and authorization is not None:
+        matches = getattr(authorization, "bundle_content_digest", None) == admitted
+    if matches:
+        return None
+    print(
+        "run REFUSED: the workflow bundle is not the bundle the intake grant "
+        "admitted. Nothing was executed."
+    )
+    return 2
+
+
 def _cmd_run(args: argparse.Namespace) -> int:
     """Execute a bundle under a named deployment profile -- FAIL-CLOSED.
 
@@ -4016,18 +4071,28 @@ def _cmd_run(args: argparse.Namespace) -> int:
     from openadapt_flow.skills import refuse_unqualified_skill
 
     refuse_unqualified_skill(bundle)
+    from openadapt_flow.admitted_composition import is_process_contract_artifact
+    from openadapt_flow.variants import is_variant_set_artifact
+
+    if getattr(args, "admitted_bundle_sha256", None) and (
+        is_composition_artifact(bundle)
+        or is_process_contract_artifact(bundle)
+        or is_variant_set_artifact(bundle)
+    ):
+        # An intake grant admits one workflow bundle, never a parent artifact.
+        print(
+            "run REFUSED: the workflow bundle is not the bundle the intake "
+            "grant admitted. Nothing was executed."
+        )
+        return 2
     if is_composition_artifact(bundle):
         from openadapt_flow.cli_compose import cmd_run_composition
 
         return cmd_run_composition(args, run_child=_cmd_run)
-    from openadapt_flow.admitted_composition import is_process_contract_artifact
-
     if is_process_contract_artifact(bundle):
         from openadapt_flow.cli_process import cmd_run_process
 
         return cmd_run_process(args, run_child=_cmd_run)
-    from openadapt_flow.variants import is_variant_set_artifact
-
     if is_variant_set_artifact(bundle):
         from openadapt_flow.cli_variants import cmd_run_variants
 
@@ -4045,6 +4110,9 @@ def _cmd_run(args: argparse.Namespace) -> int:
             f"{_SAFE_BUNDLE_LOAD_GUIDANCE} Nothing was executed."
         )
         return 2
+    refused_admission = _admitted_bundle_refusal(args, bundle, workflow)
+    if refused_admission is not None:
+        return refused_admission
 
     qualification_case = getattr(args, "_qualification_case_execution", None)
     gate_params = (
@@ -4359,6 +4427,17 @@ def _cmd_run(args: argparse.Namespace) -> int:
         args._delivery_authority_kind = "customer_local"
         args._remote_delivery_run_id = None
 
+    # The authorization this run hands to replay must name the admitted
+    # digest, and the bundle must still be those bytes (#1890).
+    refused_admission = _admitted_bundle_refusal(
+        args,
+        bundle,
+        workflow,
+        getattr(args, "_governed_run_authorization", None),
+    )
+    if refused_admission is not None:
+        return refused_admission
+
     refused = _bind_approval_mode(args)
     if refused is not None:
         return refused
@@ -10054,6 +10133,9 @@ def build_parser() -> argparse.ArgumentParser:
         metavar="VERSION",
         help="Refuse unless the bundle's compiler version equals this",
     )
+    # Set by the agent runner for an intake-granted run (#1890): the exact
+    # bundle digest the grant admitted. Not an operator knob.
+    p.add_argument("--admitted-bundle-sha256", default=None, help=argparse.SUPPRESS)
     p.add_argument("--managed-dispatch-file", default=None, help=argparse.SUPPRESS)
     p.add_argument(
         "--qualification-trust",

@@ -241,6 +241,11 @@ def __init__(
         self._entries_by_id = {
             entry.legacy_id: entry for entry in self.catalog.values() if entry.legacy_id
         }
+        #: Operator-registered, single-use intake grants per workflow. Never
+        #: accepted from tool arguments (#1890).
+        from openadapt_flow.intake.param_gate import IntakeGrantRegistry
+
+        self.intake_grants = IntakeGrantRegistry()
 
     def add_catalog_entry(self, entry: CatalogEntry) -> None:
         """Add a workflow while the server runs (a program taught over MCP).
@@ -266,6 +271,7 @@ def service(self) -> RunService:
                 allow_run=self.allow_run,
                 require_all_inputs=not self.allow_recorded_defaults,
                 repair_desk=self.repair_desk,
+                intake_grants=self.intake_grants,
             )
         return self._service
 
@@ -979,6 +985,54 @@ def _legacy_from_record(run_id: str, record: dict) -> dict:
         payload.pop("workflow_id", None)
         return payload
 
+    def register_intake_provenance(
+        self,
+        workflow: str,
+        path: Path | str,
+        *,
+        document_sha256: Optional[str] = None,
+    ) -> Any:
+        """Grant one run of an operator bundle against one intake document.
+
+        ``path`` is the ``*.provenance.json`` written by ``openadapt-flow
+        intake``. The grant is bound to the provenance's source document(s)
+        and to the digest of the bundle that will run, expires, and is used
+        up by the first run that passes. Tool callers can't supply or replace
+        it (#1890). Workflows taught over MCP have no bundle the gate can
+        read, so registering for one is refused rather than silently
+        unchecked.
+        """
+
+        from openadapt_flow.intake.param_gate import IntakeGrantError
+
+        entry = self.catalog.get(workflow)
+        if entry is None:
+            entry = next(
+                (e for e in self.catalog.values() if e.legacy_id == workflow), None
+            )
+        if entry is not None:
+            if entry.info is None:
+                raise BridgeError(
+                    "intake provenance can't be checked for a workflow taught "
+                    "over MCP; save it as an operator bundle first"
+                )
+            info = entry.info
+            names = [entry.name, entry.legacy_id or ""]
+        elif workflow in self.workflows:
+            info = self.workflows[workflow]
+            names = [workflow]
+        else:
+            raise BridgeError("unknown workflow for intake provenance")
+        bundle_dir = Path(info.bundle_dir)
+        if self.repair_desk is not None and entry is not None:
+            bundle_dir = self.repair_desk.active_bundle(entry.name, bundle_dir)
+        try:
+            return self.intake_grants.register(
+                names, path, bundle_dir, document_sha256=document_sha256
+            )
+        except IntakeGrantError as exc:
+            raise BridgeError(str(exc)) from exc
+
     def _run(self, workflow_id: str, arguments: dict) -> dict:
         info = self._require_workflow(workflow_id)
         if not self.allow_run:
@@ -1007,11 +1061,22 @@ def _run(self, workflow_id: str, arguments: dict) -> dict:
                 "arguments do not match the declared workflow parameter schema"
             )
         params = dict(arguments)
+        names = [workflow_id] + [
+            name for name, e in self.catalog.items() if e.legacy_id == workflow_id
+        ]
+        intake = self.intake_grants.gate(
+            names, info.bundle_dir, params, param_types=info.param_types
+        )
+        if not intake.ok:
+            raise BridgeError(intake.reason)
+        # Run with the proven intake values, not the caller's spelling of them.
+        params = intake.values
         outcome = self.runner.run(
             workflow=workflow_id,
             bundle_dir=info.bundle_dir,
             params=params,
             url_override=url_override,
+            admitted_bundle_sha256=intake.bundle_sha256,
         )
         attention = (
             attention_for(self.attended, outcome.run_dir)

@@ -117,10 +117,14 @@ def _param_types(workflow: Any) -> dict[str, dict[str, Any]]:
             getattr(spec, "type", None), "value", getattr(spec, "type", None)
         )
         choices = [str(choice) for choice in (getattr(spec, "choices", None) or [])]
-        types[str(name)] = {
+        entry: dict[str, Any] = {
             "type": str(kind) if kind else "string",
             "choices": choices,
         }
+        intake_field = getattr(spec, "intake_field", None)
+        if intake_field:
+            entry["intake_field"] = str(intake_field)
+        types[str(name)] = entry
     return types