flow: run an admitted API step under a governed run and Standard/Regulated

CAPABILITIES.md

@@ -183,6 +183,7 @@ improvising or retrying when something is off.
 | Fall back from the API step to page structure when the API is unavailable, before anything is sent | In review ([#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269)) | Synthetic | [#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269) |
 | Approve an API step's exact request under per-item approval: one digest binds the request bytes, the record and the record check, and a changed request is not sent | In review ([#1355](https://github.com/OpenAdaptAI/OpenAdapt/pull/1355)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#under-an-approval-mode) |
 | Run an API step in a durable run: checkpoint the span after the save is sent, never resend or redo a save that may have landed, and reconcile it by reading the record back | In review ([#1358](https://github.com/OpenAdaptAI/OpenAdapt/pull/1358)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#in-a-durable-run) |
+| Run an admitted API step under a governed run authorization and the Standard or Regulated profile: the authorization names the API step by digest, the person who admits it names the right-record parameter, and the outcome reads `VERIFIED` | In review ([#1361](https://github.com/OpenAdaptAI/OpenAdapt/pull/1361)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#under-governance) |
 | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) |
 | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) |
 | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) |

llms-full.txt

@@ -539,6 +539,7 @@ improvising or retrying when something is off.
 | Fall back from the API step to page structure when the API is unavailable, before anything is sent | In review ([#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269)) | Synthetic | [#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269) |
 | Approve an API step's exact request under per-item approval: one digest binds the request bytes, the record and the record check, and a changed request is not sent | In review ([#1355](https://github.com/OpenAdaptAI/OpenAdapt/pull/1355)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#under-an-approval-mode) |
 | Run an API step in a durable run: checkpoint the span after the save is sent, never resend or redo a save that may have landed, and reconcile it by reading the record back | In review ([#1358](https://github.com/OpenAdaptAI/OpenAdapt/pull/1358)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#in-a-durable-run) |
+| Run an admitted API step under a governed run authorization and the Standard or Regulated profile: the authorization names the API step by digest, the person who admits it names the right-record parameter, and the outcome reads `VERIFIED` | In review ([#1361](https://github.com/OpenAdaptAI/OpenAdapt/pull/1361)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#under-governance) |
 | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) |
 | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) |
 | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) |

packages/flow/docs/API_TIER.md

@@ -17,8 +17,8 @@ with the same record check. The harness and the earlier spike are in
 openadapt-flow record --backend web --url https://app.example/ --capture-network --out rec/
 openadapt-flow compile rec/ --out bundle/ --name referral
 openadapt-flow api show bundle/
-openadapt-flow api admit bundle/        # a person, at a terminal
-openadapt-flow run bundle/ --approval-mode not_gated ...
+openadapt-flow api admit bundle/ --identity subject_name=lname   # a person, at a terminal
+openadapt-flow run bundle/ ...
 ```
 
 1. **Record with `--capture-network`.** Flow keeps the page's requests next
@@ -31,7 +31,9 @@ openadapt-flow run bundle/ --approval-mode not_gated ...
    `api admit` asks a person to type `admit`. Until then, the API step never
    runs. An agent, model, or service name can't admit it, and the admission
    is bound to the exact proposal. If the binding changes, it needs a new
-   admission.
+   admission. `--identity KEY=PARAM` names the parameter that identifies the
+   record (for example `subject_name=lname` or `record_id=mrn`); Standard
+   and Regulated runs need one (see [Under governance](#under-governance)).
 4. **Run.** An admitted API step runs in place of the GUI steps it covers.
    The run report marks those steps `actuation: "api_folded"`, and the save
    step is marked `actuation: "api"`. When the API step doesn't run, the
@@ -165,14 +167,58 @@ collateral-loss guard) can't be proven that way after the fact, so the pause
 stays for a person to check the record. The same limit applies to a GUI
 save.
 
+## Under governance
+
+Every `openadapt-flow run` carries a run authorization, and Standard and
+Regulated runs add the run gate's profile checks. An API step runs there
+under these rules:
+
+- **The authorization names it.** When the run gate admits the run, the
+  authorization lists each admitted API step (`api_steps`) with the digest
+  of the binding the person admitted, the folded-checks digest and the
+  record check's contract hashes. The runtime runs an API step only when
+  all three still match. A step the authorization doesn't list, such as one
+  admitted after the run was authorized, runs on the GUI. An API step that
+  no person admitted is GUI-only for the run gate too, so a proposal alone
+  never blocks a run.
+- **Right record.** A Standard or Regulated save needs an identity contract
+  on every path it can run. For an API step it's the parameter named at
+  admission (`api admit --identity`): the request must send that
+  parameter's run value as-is in a form field, and the record check must
+  select the record by the same parameter (in `match`, or in `new_records`
+  for a create). The runtime checks both sides before the request is sent,
+  and the outcome check checks them again. A form field's name is the
+  page's, not the record's (`form_lname` versus `lname`), so the field and
+  the record check are tied by the parameter, not by name. An API step
+  admitted without one is refused by the run gate's identity check, which
+  names the step and the `--identity` flag.
+- **Outcome.** The covered steps are skips, and the outcome check accepts
+  them only when the API step's own save in the same report ran on the API
+  path and succeeded, and every step it covers was folded. The save is then
+  checked like any other API action: delivered, confirmed by the record
+  check at the profile's evidence tier, with exact identity evidence. A run
+  like that reads `VERIFIED`.
+
+Not yet covered, so the GUI steps run with that reason:
+
+- **Qualification cases.** Qualification needs representative and fault
+  cases on the API path for an admitted API step, and a case run doesn't
+  run API steps yet. So a bundle with an admitted API step can't complete
+  qualification today, and a run under a production qualification admission
+  uses the GUI steps.
+- **Managed (Cloud) dispatch.** Cloud's run authorization doesn't list API
+  steps yet.
+
 ## When it doesn't run
 
 The API step runs only in a plain linear run. It doesn't run, and the GUI
 steps run instead, when:
 
 - an approval mode is on and the API actuator can't prepare the request
   before sending it (a deployment's own actuator without `prepare`);
-- the run is governed by qualification;
+- the run's authorization didn't admit that API step, or it is a
+  qualification case, a managed (Cloud) dispatch, or a run under a
+  production qualification admission (see [Under governance](#under-governance));
 - the step isn't admitted, or the program's steps, on-screen checks or the
   save's record check changed since it was proposed or admitted;
 - the save checks the right record on screen and the API step names no

packages/flow/openadapt_flow/cli_api.py

@@ -61,6 +61,18 @@ def register_api_parser(sub: argparse._SubParsersAction) -> None:
     admit.add_argument("bundle", help="Bundle directory")
     admit.add_argument("--step", default=None, help="Admit only this save step")
     admit.add_argument("--by", default=None, help="Your name (default: the login name)")
+    admit.add_argument(
+        "--identity",
+        action="append",
+        default=[],
+        metavar="KEY=PARAM",
+        help=(
+            "The parameter that identifies the record, for example "
+            "subject_name=lname or record_id=mrn (repeatable). The request must "
+            "send it and the record check must select by it. Standard and "
+            "Regulated runs need one."
+        ),
+    )
     admit.set_defaults(func=cmd_api_admit)
 
 
@@ -169,23 +181,32 @@ def cmd_api_admit(
     if not found:
         print("api admit: no proposed API step to admit", file=sys.stderr)
         return 2
-    for step in found:
-        print(describe_api_step(step))
-        print()
+    identity: dict[str, str] = {}
+    for item in args.identity or []:
+        key, sep, param = item.partition("=")
+        if not sep or not key.strip() or not param.strip():
+            print(f"api admit: --identity {item!r} is not KEY=PARAM", file=sys.stderr)
+            return 2
+        identity[key.strip()] = param.strip()
+    # Build the admission first, so the person reads exactly what is admitted.
+    try:
+        updated, steps = admit_api_step(
+            workflow, admitted_by=args.by, step_id=args.step, identity=identity
+        )
+    except (ApiProposalError, ValueError) as exc:
+        print(f"api admit: {exc}", file=sys.stderr)
+        return 2
+    for step in updated.steps:
+        if step.id in steps:
+            print(describe_api_step(step))
+            print()
     answer = ask(
         f"Type '{ADMIT_WORD}' to let these requests save in place of the GUI steps "
         "(the record check still confirms every save): "
     )
     if answer.strip().lower() != ADMIT_WORD:
         print("Not admitted. Nothing changed.")
         return 1
-    try:
-        updated, steps = admit_api_step(
-            workflow, admitted_by=args.by, step_id=args.step
-        )
-    except ApiProposalError as exc:
-        print(f"api admit: {exc}", file=sys.stderr)
-        return 2
     updated.save(bundle)
     print(f"Admitted the API step on {', '.join(steps)}.")
     return 0

packages/flow/openadapt_flow/compiler/api_proposal.py

@@ -55,6 +55,7 @@
     ActionKind,
     ApiAdmission,
     ApiBinding,
+    ApiIdentityBinding,
     Step,
     WebFormField,
     WebFormPrecheck,
@@ -434,6 +435,12 @@ def describe_api_step(step: Step) -> str:
     lines.append(
         f"  record check after: {len(b.effects)} effect(s) copied from the save step"
     )
+    for identity in b.identity:
+        lines.append(
+            f"  right record ({identity.key}): the request sends {{{identity.param}}} "
+            f"in {', '.join(p.removeprefix('/body/') for p in identity.request_pointers)} "
+            f"and the record check selects {identity.effect_field} = {{{identity.param}}}"
+        )
     if spec.screen_only_checks:
         lines.append(f"  screen-only checks, not repeated: {spec.screen_only_checks}")
     if spec.other_writes_seen:
@@ -446,18 +453,70 @@ def describe_api_step(step: Step) -> str:
     return "\n".join(lines)
 
 
+def web_form_identity(
+    binding: ApiBinding, effects: list[Any], key: str, param: str
+) -> ApiIdentityBinding:
+    """The right-record contract for one parameter of a ``web_form`` save.
+
+    The person admitting the API step names the parameter that identifies
+    the record (``key=param``). The request must send that parameter's run
+    value as-is in at least one form field, and the record check must select
+    the record by the same parameter. Raises :class:`ApiProposalError` when
+    either side is missing.
+    """
+
+    spec = binding.web_form
+    assert spec is not None
+    pointers = [
+        f"/body/{field.name}"
+        for field in spec.fields
+        if field.source == "param" and field.param == param
+    ]
+    if not pointers:
+        raise ApiProposalError(
+            f"the request sends no form field with the run value {{{param}}}, so it "
+            f"can't identify the record by {param}"
+        )
+    selectors = sorted(
+        {
+            name
+            for effect in effects
+            for row in [effect.match, *effect.new_records]
+            for name, expr in row.items()
+            if getattr(expr, "param", None) == param
+        }
+    )
+    if not selectors:
+        raise ApiProposalError(
+            f"the record check selects no field by {{{param}}}, so it can't confirm "
+            f"the record the request names by {param}"
+        )
+    return ApiIdentityBinding(
+        key=key,  # type: ignore[arg-type]
+        param=param,
+        effect_field=selectors[0],
+        request_pointers=pointers[:8],
+    )
+
+
 def admit_api_step(
     workflow: Workflow,
     *,
     admitted_by: Optional[str] = None,
     step_id: Optional[str] = None,
     now: Optional[datetime] = None,
+    identity: Optional[dict[str, str]] = None,
 ) -> tuple[Workflow, list[str]]:
     """Record a person's admission of the proposed API steps.
 
     Copies the save step's current record check into the binding first (so a
     check added after compile is the one admitted), then binds the admission
     to the exact binding. The caller saves the workflow and asks the person.
+
+    ``identity`` maps a right-record signal (``subject_name``, ``record_id``,
+    ...) to the parameter that carries it (:func:`web_form_identity`).
+    Standard and Regulated runs need one: their run gate refuses a save
+    whose API step names no right-record contract.
     """
 
     from openadapt_flow.runtime.effects.effect import is_agent_identity
@@ -485,8 +544,16 @@ def admit_api_step(
                     f"{step.id} has no record check; an API step must be confirmed by one "
                     "(add it, for example with record --verify or check mark, then admit)"
                 )
+            identities = [
+                web_form_identity(b, list(step.effects), key, param)
+                for key, param in sorted((identity or {}).items())
+            ]
             clean = b.model_copy(
-                update={"effects": list(step.effects), "admission": None}
+                update={
+                    "effects": list(step.effects),
+                    "admission": None,
+                    "identity": identities or list(b.identity),
+                }
             )
             b = clean.model_copy(
                 update={

packages/flow/openadapt_flow/execution_profiles.py

@@ -406,6 +406,11 @@ def _api_identity_evidence_is_exact(
     ):
         return False
 
+    from openadapt_flow.runtime.actuators.web_form import (
+        effect_selector_rows,
+        web_form_pointer_template,
+    )
+
     for identity in binding.identity:
         if identity.param not in text_params or text_params[identity.param] == "":
             return False
@@ -414,11 +419,18 @@ def _api_identity_evidence_is_exact(
             tuple(field.split(".")) == effect_path
             and expression.param == identity.param
             for effect in effects
-            for field, expression in effect.match.items()
+            for row in effect_selector_rows(effect)
+            for field, expression in row.items()
         ):
             return False
         token = "{" + identity.param + "}"
         for pointer in identity.request_pointers:
+            if binding.kind == "web_form":
+                # The form field sends the parameter the record check
+                # selects on (see web_form_pointer_template).
+                if web_form_pointer_template(binding, pointer) != token:
+                    return False
+                continue
             pointer_parts = [
                 item.replace("~1", "/").replace("~0", "~")
                 for item in pointer.lstrip("/").split("/")
@@ -1812,6 +1824,37 @@ def _reported_parameter_predicate_value(
             return None if value is None else not value
         return None
 
+    # GUI steps an admitted API step ran in place of (docs/API_TIER.md): each
+    # is a skip only when the API step's own save in this report ran on the
+    # API path and every step it covers was folded. The save itself is then
+    # classified like any other API action below.
+    api_folded_ids: set[str] = set()
+    if workflow.program is None:
+        results_by_id = {result.step_id: result for result, _step in paired_results}
+        for step in workflow.steps:
+            binding = step.api_binding
+            if (
+                binding is None
+                or binding.kind != "web_form"
+                or binding.web_form is None
+                or not binding.admitted()
+            ):
+                continue
+            save_result = results_by_id.get(step.id)
+            covered = list(binding.web_form.replaces_steps[:-1])
+            if (
+                save_result is not None
+                and save_result.ok
+                and save_result.actuation == "api"
+                and binding.web_form.replaces_steps[-1] == step.id
+                and all(
+                    results_by_id.get(step_id) is not None
+                    and results_by_id[step_id].actuation == "api_folded"
+                    for step_id in covered
+                )
+            ):
+                api_folded_ids.update(covered)
+
     for result, step in paired_results:
         if result.skipped or result.exception_handled:
             if (
@@ -1840,7 +1883,15 @@ def _reported_parameter_predicate_value(
                     workflow.program is not None
                     and result.actuation == "human_attended_skip"
                 )
-                if not (declared_guard_skip or authenticated_attended_skip):
+                folded_into_api_step = (
+                    result.actuation == "api_folded"
+                    and result.step_id in api_folded_ids
+                )
+                if not (
+                    declared_guard_skip
+                    or authenticated_attended_skip
+                    or folded_into_api_step
+                ):
                     return ExecutionOutcome.COMPLETED_UNVERIFIED
             continue
         if not result.ok: