diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 4376ed1a6..41c2670cb 100644 --- a/CAPABILITIES.md +++ b/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) | diff --git a/llms-full.txt b/llms-full.txt index 9e97777a5..4c747535c 100644 --- a/llms-full.txt +++ b/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) | diff --git a/packages/flow/docs/API_TIER.md b/packages/flow/docs/API_TIER.md index c245cdc4c..8753108fc 100644 --- a/packages/flow/docs/API_TIER.md +++ b/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,6 +167,48 @@ 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 @@ -172,7 +216,9 @@ 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 diff --git a/packages/flow/openadapt_flow/cli_api.py b/packages/flow/openadapt_flow/cli_api.py index c15022a43..b62a06825 100644 --- a/packages/flow/openadapt_flow/cli_api.py +++ b/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,9 +181,25 @@ 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): " @@ -179,13 +207,6 @@ def cmd_api_admit( 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 diff --git a/packages/flow/openadapt_flow/compiler/api_proposal.py b/packages/flow/openadapt_flow/compiler/api_proposal.py index 8be653a45..16363a245 100644 --- a/packages/flow/openadapt_flow/compiler/api_proposal.py +++ b/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={ diff --git a/packages/flow/openadapt_flow/execution_profiles.py b/packages/flow/openadapt_flow/execution_profiles.py index 75a303f0a..c4c7ef511 100644 --- a/packages/flow/openadapt_flow/execution_profiles.py +++ b/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: diff --git a/packages/flow/openadapt_flow/policy.py b/packages/flow/openadapt_flow/policy.py index 454b469ae..e158e076e 100644 --- a/packages/flow/openadapt_flow/policy.py +++ b/packages/flow/openadapt_flow/policy.py @@ -191,12 +191,17 @@ def executable_actuation_paths(step: Step) -> tuple[EffectPath, ...]: policy, and reporting. An absent API binding means GUI-only. The default API binding remains dual-path because a proven pre-delivery ``UNAVAILABLE`` result falls through to GUI. ``on_unavailable='halt'`` removes that GUI - fallback and makes the binding API-only. + fallback and makes the binding API-only. A ``web_form`` binding that no + person admitted is GUI-only: the runtime never runs it. """ binding = step.api_binding if binding is None: return ("gui",) + if binding.kind == "web_form" and not binding.admitted(): + # A compiled API step that no person admitted never runs: the + # replayer runs the recorded GUI steps (docs/API_TIER.md). + return ("gui",) if binding.on_unavailable == "halt": return ("api",) return ("gui", "api") diff --git a/packages/flow/openadapt_flow/run_gate.py b/packages/flow/openadapt_flow/run_gate.py index 7323b1cf8..2c1f6bb40 100644 --- a/packages/flow/openadapt_flow/run_gate.py +++ b/packages/flow/openadapt_flow/run_gate.py @@ -86,6 +86,7 @@ from openadapt_flow.runtime.authorization import ( GovernedRunAuthorization, UnverifiedWriteApproval, + admitted_api_steps, interstitial_declarations_digest, runtime_inputs_digest, ) @@ -680,11 +681,24 @@ def _gate_identity( True, f"{total}/{total} entity-sensitive/consequential action(s) identity-armed", ) + api_steps = [ + s.id + for s in unarmed + if s.api_binding is not None + and s.api_binding.kind == "web_form" + and not s.api_binding.identity + ] + hint = ( + f"; the API step on {', '.join(api_steps)} names no right-record " + "parameter (admit it with openadapt-flow api admit --identity KEY=PARAM)" + if api_steps + else "" + ) return _result( GATE_IDENTITY, False, f"{len(unarmed)}/{total} entity-sensitive/consequential action(s) are " - "UNARMED -- would act with no verified target identity", + f"UNARMED -- would act with no verified target identity{hint}", [s.id for s in unarmed], ) @@ -1112,6 +1126,7 @@ def build_runtime_authorization( required_identity_step_ids=tuple(report.required_identity_step_ids), unverified_write_approvals=tuple(approvals), approval_source=approval_source, + api_steps=admitted_api_steps(workflow), ) diff --git a/packages/flow/openadapt_flow/runtime/actuators/web_form.py b/packages/flow/openadapt_flow/runtime/actuators/web_form.py index e356b8f90..3ab2b94d8 100644 --- a/packages/flow/openadapt_flow/runtime/actuators/web_form.py +++ b/packages/flow/openadapt_flow/runtime/actuators/web_form.py @@ -186,6 +186,36 @@ def folded_checks_digest(steps: Any) -> str: return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() +def web_form_pointer_template(binding: Any, pointer: str) -> Optional[str]: + """The template a ``web_form`` save sends at ``/body/``. + + ``"{param}"`` when that form field sends a run parameter as-is, else None. + A form field's name is the page's, not the record's (``form_lname`` + versus ``lname``), so an identity binding ties the field to the record + check by the parameter both use, not by name. + """ + + spec = getattr(binding, "web_form", None) + if spec is None: + return None + root, _, name = pointer.lstrip("/").partition("/") + if root != "body": + return None + for field in spec.fields: + if field.name == name and field.source == "param" and field.param: + return "{" + field.param + "}" + return None + + +def effect_selector_rows(effect: Any) -> list[Any]: + """The selector rows of a record check: ``match``, then ``new_records``. + + A create's record check names the new record in ``new_records``. + """ + + return [effect.match, *getattr(effect, "new_records", [])] + + # -- transport ------------------------------------------------------------------- diff --git a/packages/flow/openadapt_flow/runtime/authorization.py b/packages/flow/openadapt_flow/runtime/authorization.py index 83f1d2936..5748c7b77 100644 --- a/packages/flow/openadapt_flow/runtime/authorization.py +++ b/packages/flow/openadapt_flow/runtime/authorization.py @@ -15,7 +15,7 @@ import threading from datetime import datetime, timezone from pathlib import Path -from typing import Any, Literal, Mapping, TypeAlias, Union, cast +from typing import Any, Literal, Mapping, Optional, TypeAlias, Union, cast from uuid import uuid4 from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator @@ -339,6 +339,55 @@ class UnverifiedWriteApproval(BaseModel): effect_contract_hashes: tuple[str, ...] = Field(min_length=1) +class AdmittedApiStep(BaseModel): + """One API step (``web_form`` binding) admitted into a governed run. + + Binds the exact binding a person admitted (``admission_digest``), the + on-screen checks of the GUI steps it stands for (``folded_checks_digest``) + and its record check (``effect_contract_hashes``). The runtime runs an API + step under a governed authorization only when all three still match. + """ + + model_config = ConfigDict(frozen=True) + + step_id: str + admission_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + folded_checks_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + effect_contract_hashes: tuple[str, ...] = Field(min_length=1) + + +def admitted_api_steps(workflow: Any) -> tuple[AdmittedApiStep, ...]: + """The admitted ``web_form`` API steps of ``workflow``, as run bindings. + + Only a binding a person admitted, whose record check is still the save + step's own, is listed. + """ + + rows: list[AdmittedApiStep] = [] + 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() + or not binding.effects + or list(binding.effects) != list(step.effects) + ): + continue + rows.append( + AdmittedApiStep( + step_id=step.id, + admission_digest=binding.admission_digest(), + folded_checks_digest=binding.web_form.folded_checks_digest, + effect_contract_hashes=tuple( + effect.contract_hash() for effect in binding.effects + ), + ) + ) + return tuple(rows) + + class GovernedRunAuthorization(BaseModel): """Ephemeral capability carrying admission decisions into replay. @@ -459,6 +508,12 @@ class GovernedRunAuthorization(BaseModel): #: the runtime then gates like ``each_item``. It is a runner-local field: #: it is not part of the Cloud dispatch binding digest. approval_mode: Literal["each_item", "problems_only", "not_gated"] | None = None + #: API steps admitted into this run (docs/API_TIER.md). An API step not + #: listed here, or whose binding, folded checks or record check changed, + #: runs on the GUI. Cloud's dispatch never carries any yet, so a managed + #: run stays on the GUI. Left out when empty, so other authorizations + #: (and the dispatch binding digest) serialize exactly as before. + api_steps: tuple[AdmittedApiStep, ...] = Field(default_factory=tuple) @model_serializer(mode="wrap") def _serialize_compatible(self, handler: Any) -> dict[str, Any]: @@ -467,8 +522,39 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: data: dict[str, Any] = handler(self) if self.approval_mode is None: data.pop("approval_mode", None) + if not self.api_steps: + data.pop("api_steps", None) return data + def admitted_api_step(self, step: Any, workflow: Any) -> Optional[str]: + """Why ``step``'s API step can't run under this authorization, or None.""" + + from openadapt_flow.runtime.actuators.web_form import folded_checks_digest + + binding = step.api_binding + row = next((item for item in self.api_steps if item.step_id == step.id), None) + if row is None or binding is None or binding.web_form is None: + return "this run's authorization admitted no API step for this save" + ids = [s.id for s in workflow.steps] + try: + start = ids.index(binding.web_form.replaces_steps[0]) + end = ids.index(step.id) + except ValueError: + return "the program's steps no longer match the admitted API step" + if ( + row.admission_digest != binding.admission_digest() + or row.folded_checks_digest != binding.web_form.folded_checks_digest + or row.folded_checks_digest + != folded_checks_digest(workflow.steps[start:end]) + or list(row.effect_contract_hashes) + != [effect.contract_hash() for effect in step.effects] + ): + return ( + "the API step, its folded checks or its record check differ from " + "the ones this run's authorization admitted" + ) + return None + @model_validator(mode="after") def _qualification_binding_is_complete(self) -> "GovernedRunAuthorization": if (self.qualification_admission is None) != ( diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index c3f59a856..5a3baebd0 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -2444,8 +2444,6 @@ def _web_form_span_at( "approval mode needs the exact request before the save, and " "this API actuator can't prepare it, so the GUI steps run" ) - elif self.governed_authorization is not None: - reason = "a governed run uses its qualified GUI path" elif not binding.admitted(): reason = "the API step is not admitted (openadapt-flow api admit)" elif save.identity_armed and not binding.identity: @@ -2453,7 +2451,8 @@ def _web_form_span_at( # must check it too, so it needs a right-record contract. reason = ( "the save checks the right record on screen, and the API " - "step names no right-record contract" + "step names no right-record contract (openadapt-flow api " + "admit --identity)" ) elif list(binding.effects) != list(save.effects): reason = ( @@ -2471,6 +2470,10 @@ def _web_form_span_at( "step was proposed, and the API step would skip it " "(openadapt-flow api propose)" ) + elif self.governed_authorization is not None and ( + governed_reason := self._governed_web_form_refusal(workflow, save) + ): + reason = governed_reason elif self._actuator_for(binding) is None: reason = "there is no live browser session for the API step" else: @@ -2478,6 +2481,33 @@ def _web_form_span_at( return end_index, save, reason return None + def _governed_web_form_refusal( + self, workflow: Workflow, save: Step + ) -> Optional[str]: + """Why a governed run can't use this API step, or None when it can. + + The run's authorization must have admitted this exact API step. A + qualification case runs the path it names, and a managed (Cloud) + dispatch or a production qualification admission doesn't bind API + steps yet, so those run the GUI steps. + """ + + authorization = self.governed_authorization + assert authorization is not None + if authorization.qualification_case_id is not None: + return ( + "a qualification case runs the GUI path it names; qualifying " + "an API step's path is not supported yet" + ) + if self.managed_dispatch_binding is not None: + return "a managed dispatch doesn't bind API steps yet" + if authorization.production_qualification_admission_id is not None: + return ( + "a production qualification admission doesn't cover API steps " + "yet, so the qualified GUI path runs" + ) + return authorization.admitted_api_step(save, workflow) + def _resume_inside_web_form_span( self, workflow: Workflow, step_index: int, prior: list[StepResult] ) -> Optional[StepResult]: @@ -8470,6 +8500,12 @@ def _api_request_pointer_value(binding: ApiBinding, pointer: str) -> Any: if pointer.startswith("/url/"): return binding.url_template + if binding.kind == "web_form": + from openadapt_flow.runtime.actuators.web_form import ( + web_form_pointer_template, + ) + + return web_form_pointer_template(binding, pointer) root_name, *tokens = pointer.lstrip("/").split("/") if root_name == "body": value: Any = binding.body_template @@ -8518,6 +8554,14 @@ def _api_identity_refusal( self.governed_authorization is not None and self.governed_authorization.requires_verified_identity(step.id) ) + if step.api_binding is not None and step.api_binding.kind == "web_form": + # A form save whose GUI step checks the right record, or whose + # admission named a right-record parameter, is checked the same + # way in every run, governed or not (its durable checkpoint + # needs the evidence too). + authorization_requires = authorization_requires or bool( + step.identity_armed or step.api_binding.identity + ) project = workflow.qualification policy = project.identity_policies.get(step.id) if project is not None else None if not authorization_requires and policy is None: @@ -8560,9 +8604,17 @@ def _api_identity_refusal( ] exact_token = "{" + identity.param + "}" effect_target_path = tuple(identity.effect_field.split(".")) + # A form field's name is the page's, not the record's (form_lname + # versus lname): for a form save the field is bound to the record + # check by sending the same parameter the check selects on. + names_must_match = binding.kind != "web_form" pointers_are_bound = all( template is not None - and self._api_request_pointer_field_path(pointer) == effect_target_path + and ( + not names_must_match + or self._api_request_pointer_field_path(pointer) + == effect_target_path + ) and ( isinstance(template, str) and self._url_path_targets_param(template, identity.param) @@ -8579,8 +8631,14 @@ def _api_identity_refusal( f"({step.intent}): identity parameter {identity.param!r} is " "not bound at every declared target-bearing request pointer" ) + from openadapt_flow.runtime.actuators.web_form import ( + effect_selector_rows, + ) + effect_exprs = [ - effect.match.get(identity.effect_field) for effect in effects + row.get(identity.effect_field) + for effect in effects + for row in effect_selector_rows(effect) ] if not any( expr is not None and expr.param == identity.param diff --git a/packages/flow/tests/test_api_step_durable.py b/packages/flow/tests/test_api_step_durable.py index 0cdaf4fe6..82b80900d 100644 --- a/packages/flow/tests/test_api_step_durable.py +++ b/packages/flow/tests/test_api_step_durable.py @@ -20,7 +20,7 @@ import requests from openadapt_flow.compiler.api_proposal import admit_api_step -from openadapt_flow.ir import Workflow +from openadapt_flow.ir import WebFormField, Workflow from openadapt_flow.mockmed.fault_server import serve as fault_serve from openadapt_flow.runtime.actuators import ( ActuationStatus, @@ -295,3 +295,53 @@ def test_a_gui_fallback_that_halted_at_the_save_resumes_on_the_gui(fx): assert second.calls == 0 assert ("press", "Enter") in [(a[0], a[1]) for a in fx.backend.actions] assert [r["note"] for r in fx.records()] == ["gui"] + + +def test_an_identity_checked_save_with_a_right_record_contract_runs_durably(tmp_path): + # The referral's case: the save checks the right record on screen and the + # API step was admitted with --identity. The API step checks the request + # and the record check select the same record, so the durable checkpoint + # has its identity evidence and the run never fails after the save. + from openadapt_flow.runtime.effects import Effect, EffectKind, ValueExpr + + wf = _span_workflow(admitted=False) + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={**TARGET, "note": ValueExpr(param="note")}, + expected_count=1, + timeout_s=2.0, + ) + save = wf.steps[-1] + spec = save.api_binding.web_form.model_copy( + update={ + "fields": [ + *save.api_binding.web_form.fields, + WebFormField(name="form_note", source="param", param="note"), + ] + } + ) + binding = save.api_binding.model_copy( + update={"effects": [effect], "web_form": spec} + ) + save = save.model_copy( + update={"identity_armed": True, "effects": [effect], "api_binding": binding} + ) + wf = wf.model_copy( + update={"steps": [*wf.steps[:-1], save], "params": {**wf.params, "note": "api"}} + ) + wf, _ = admit_api_step( + wf, admitted_by="reviewer", identity={"subject_name": "note"} + ) + fx = Fixture(tmp_path, wf) + try: + actuator = ScriptedActuator(fx.url, ActuationStatus.ACTUATED) + report = fx.run(actuator) + assert report.success is True, [r.error for r in report.results] + assert actuator.calls == 1 + save_result = report.results[-1] + assert save_result.actuation == "api" + assert save_result.identity is not None + assert save_result.identity.status == "verified" + assert [c.actuation for c in fx.store().checkpoints()] == ["api_folded", "api"] + finally: + fx.stop() diff --git a/packages/flow/tests/test_api_step_governed.py b/packages/flow/tests/test_api_step_governed.py new file mode 100644 index 000000000..d26917e7b --- /dev/null +++ b/packages/flow/tests/test_api_step_governed.py @@ -0,0 +1,323 @@ +"""An API step under the default production profile (docs/API_TIER.md). + +The default profile is Regulated, durable, governed by a run authorization, +and approval-gated per item. The API step runs there only when the run's +authorization admitted that exact API step (its binding, its folded-checks +digest and its record check); a person approves the exact request; the save +is checked with the same record check as the GUI save. + +The record system is a scripted verifier at INDEPENDENT_SYSTEM; the values +are made up. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import pytest + +from openadapt_flow.approval_mode.gate import ApprovalGate +from openadapt_flow.approval_mode.item_approval import ItemApprovalStore +from openadapt_flow.approval_mode.model import ItemDecisionKind +from openadapt_flow.compiler.api_proposal import admit_api_step +from openadapt_flow.execution_profiles import ExecutionProfile +from openadapt_flow.ir import ( + ActionKind, + ApiBinding, + Step, + WebFormField, + WebFormSpec, + Workflow, +) +from openadapt_flow.run_gate import build_runtime_authorization +from openadapt_flow.runtime.actuators.web_form import ( + WebFormActuator, + folded_checks_digest, +) +from openadapt_flow.runtime.effects import ValueExpr +from tests.test_api_step_approval import APPROVE, Person +from tests.test_approval_mode_replayer import RecordSystem, _create_workflow +from tests.test_execution_profiles import ( + _KEY, + Replayer, + _gate, + _ReadyVision, + _sealed, +) +from tests.test_replayer import FakeBackend + +PARAMS = {"lname": "Marsh"} +FORM = """
+ + +
""" + + +class Transport: + base_url = "http://app" + + def __init__(self) -> None: + self.posts: list[tuple[str, str]] = [] + + def get(self, path: str) -> Any: + return _Resp(200, FORM) + + def post_form(self, path: str, body: str, *, referer: str) -> Any: + self.posts.append((path, body)) + return _Resp(200, "saved") + + +class _Resp: + def __init__(self, status: int, text: str) -> None: + self.status, self.text, self.headers = status, text, {} + + +def _workflow(*, admitted: bool = True, identity: bool = True) -> Workflow: + """Type the last name, then submit; the API step stands for both.""" + + base = _create_workflow("api-referral") + submit = base.steps[0] + # The record check finds the new record by the last name this run typed. + effect = submit.effects[0].model_copy( + update={"new_records": [{"lname": ValueExpr(param="lname")}]} + ) + submit = submit.model_copy(update={"effects": [effect]}) + typing = Step( + id="type_lname", + intent="type the last name", + action=ActionKind.TYPE, + text="{lname}", + ) + spec = WebFormSpec( + form_path="/new.php", + fields=[ + WebFormField(name="csrf", source="session"), + WebFormField(name="lname", source="param", param="lname"), + ], + controls=["csrf", "lname"], + replaces_steps=[typing.id, submit.id], + folded_checks_digest=folded_checks_digest([typing]), + ) + submit = submit.model_copy( + update={ + "api_binding": ApiBinding( + kind="web_form", + url_template="/save.php", + effects=list(submit.effects), + web_form=spec, + ) + } + ) + wf = Workflow( + name=base.name, + steps=[typing, submit], + params={"lname": "Marsh"}, + ) + if admitted: + wf, _ = admit_api_step( + wf, + admitted_by="reviewer", + identity={"subject_name": "lname"} if identity else None, + ) + return wf + + +class Item: + def __init__(self, tmp_path, *, workflow: Optional[Workflow] = None) -> None: + self.tmp_path = tmp_path + self.workflow, self.bundle = _sealed( + tmp_path, workflow or _workflow(), encrypted=True + ) + self.verifier = RecordSystem() + self.gate_result = _gate( + self.workflow, + self.bundle, + ExecutionProfile.REGULATED, + verifier=self.verifier, + durable=True, + ) + assert self.gate_result.passed, self.gate_result.render() + self.authorization = build_runtime_authorization( + self.workflow, self.gate_result, params=PARAMS + ) + self.run_dir = tmp_path / "run" + self.person = Person([APPROVE]) + self.store = ItemApprovalStore(self.run_dir) + self.person.store = self.store + self.transport = Transport() + self.backend = FakeBackend() + + def run(self, **overrides: Any): + gate = ApprovalGate( + "each_item", + approval_store=self.store, + approval_wait_s=30.0, + approval_poll_s=1.0, + sleep=self.person.sleep, + monotonic=self.person.monotonic, + ) + kwargs: dict[str, Any] = dict( + vision=_ReadyVision(), + effect_verifier=self.verifier, + governed_authorization=self.authorization, + durable=True, + checkpoint_key=_KEY, + require_settled=True, + approval_gate=gate, + web_form_actuator=WebFormActuator(self.transport), + ) + kwargs.update(overrides) + return Replayer(self.backend, **kwargs).run( + self.workflow, + params=PARAMS, + bundle_dir=self.bundle, + run_dir=self.run_dir, + ) + + +def test_the_default_profile_runs_the_admitted_api_step(tmp_path): + item = Item(tmp_path) + report = item.run() + + assert report.success is True, ( + [ + (r.step_id, r.ok, r.actuation, r.api_fallback, r.error, r.effect_results) + for r in report.results + ], + report.execution_outcome, + report.transaction_outcome, + ) + assert item.backend.actions == [] + assert [path for path, _ in item.transport.posts] == ["/save.php"] + assert [r.actuation for r in report.results] == ["api_folded", "api"] + assert report.transaction_outcome == "VERIFIED" + assert report.approval.decision is ItemDecisionKind.APPROVED + (request,) = item.store.requests() + assert request.form_state_kind == "api_request" + + +def test_an_api_step_admitted_without_a_right_record_parameter_is_refused(tmp_path): + from openadapt_flow.execution_profiles import ExecutionProfile as P + + workflow, bundle = _sealed(tmp_path, _workflow(identity=False), encrypted=True) + gate = _gate(workflow, bundle, P.REGULATED, verifier=RecordSystem(), durable=True) + assert not gate.passed + assert "api admit --identity" in gate.render() + + +def test_an_unadmitted_api_step_leaves_the_run_on_the_gui(tmp_path): + item = Item(tmp_path, workflow=_workflow(admitted=False)) + assert item.authorization.api_steps == () + report = item.run() + + assert item.transport.posts == [] + assert "not admitted" in report.results[0].api_fallback + assert report.results[0].actuation != "api_folded" + + +def test_an_api_step_the_authorization_did_not_admit_runs_on_the_gui(tmp_path): + item = Item(tmp_path) + item.authorization = item.authorization.model_copy(update={"api_steps": ()}) + report = item.run() + + assert item.transport.posts == [] + assert "admitted no API step" in report.results[0].api_fallback + + +@pytest.mark.parametrize( + "update, reason", + [ + ({"qualification_case_id": "case-1"}, "qualification case"), + ( + {"production_qualification_admission_id": "admission-1"}, + "production qualification admission", + ), + ], +) +def test_profiles_that_dont_bind_api_steps_yet_say_so(tmp_path, update, reason): + item = Item(tmp_path) + replayer = Replayer( + FakeBackend(), + vision=_ReadyVision(), + governed_authorization=item.authorization.model_copy(update=update), + web_form_actuator=WebFormActuator(item.transport), + ) + span = replayer._web_form_span_at(item.workflow, 0, durable=True) + assert span is not None and reason in span[2] + + +def test_a_managed_dispatch_doesnt_bind_api_steps_yet(tmp_path): + item = Item(tmp_path) + replayer = Replayer( + FakeBackend(), + vision=_ReadyVision(), + governed_authorization=item.authorization, + web_form_actuator=WebFormActuator(item.transport), + ) + replayer.managed_dispatch_binding = object() + span = replayer._web_form_span_at(item.workflow, 0, durable=True) + assert span is not None and "managed dispatch" in span[2] + + +def test_the_authorization_lists_api_steps_only_when_there_are_some(tmp_path): + item = Item(tmp_path) + (row,) = item.authorization.api_steps + assert row.step_id == "submit" + assert "api_steps" in item.authorization.model_dump(mode="json") + bare = item.authorization.model_copy(update={"api_steps": ()}) + assert "api_steps" not in bare.model_dump(mode="json") + + +def _admit_cli(tmp_path, *identity: str, answer: str = "admit"): + import argparse + + from openadapt_flow.cli_api import cmd_api_admit + + bundle = tmp_path / "cli-bundle" + _workflow(admitted=False).save(bundle) + args = argparse.Namespace( + bundle=str(bundle), step=None, by="reviewer", identity=list(identity) + ) + code = cmd_api_admit(args, ask=lambda _prompt: answer, interactive=lambda: True) + return code, Workflow.load(bundle) + + +def test_api_admit_names_the_right_record_parameter(tmp_path, capsys): + code, workflow = _admit_cli(tmp_path, "subject_name=lname") + assert code == 0 + binding = workflow.steps[-1].api_binding + assert binding.admitted() + (identity,) = binding.identity + assert (identity.key, identity.param, identity.effect_field) == ( + "subject_name", + "lname", + "lname", + ) + assert identity.request_pointers == ["/body/lname"] + assert "right record (subject_name)" in capsys.readouterr().out + + +@pytest.mark.parametrize( + "identity, message", + [ + ("subject_name=mrn", "sends no form field"), + ("subject_name", "is not KEY=PARAM"), + ("not_a_signal=lname", "api admit:"), + ], +) +def test_api_admit_refuses_a_parameter_that_cant_identify_the_record( + tmp_path, capsys, identity, message +): + code, workflow = _admit_cli(tmp_path, identity) + assert code == 2 + assert message in capsys.readouterr().err + assert not workflow.steps[-1].api_binding.admitted() + + +def test_the_record_check_must_select_by_the_identity_parameter(): + from openadapt_flow.compiler.api_proposal import ApiProposalError, web_form_identity + + save = _workflow(admitted=False).steps[-1] + unrelated = save.effects[0].model_copy(update={"new_records": []}) + with pytest.raises(ApiProposalError, match="selects no field"): + web_form_identity(save.api_binding, [unrelated], "subject_name", "lname")