diff --git a/packages/flow/docs/QUALIFICATION_PROJECT.md b/packages/flow/docs/QUALIFICATION_PROJECT.md index 91bd1d6cef..333676b4b5 100644 --- a/packages/flow/docs/QUALIFICATION_PROJECT.md +++ b/packages/flow/docs/QUALIFICATION_PROJECT.md @@ -144,6 +144,17 @@ openadapt-flow qualify certify bundle \ --json ``` +A bundle that embeds reusable subroutines (`subroutine//` +subflows, see `docs/design/SUBROUTINES.md`) also needs +`--subroutine-library subroutines.json`. Certification then admits every +embedded subroutine against that library and registers the workflow as a user +of those exact versions (`--subroutine-workflow-id`, default the workflow +name), writing the library back. Without the library, or when any use is +refused, certification fails with `subroutine_not_admitted`. A bundle with no +subroutines is certified exactly as before. `openadapt-flow qualify +--verify verify.yaml --certify` takes the same flag and refuses before running +any case when it is missing. + `qualify run` imports typed results from Desktop or another local, customer-controlled executor. Unsigned, stale, missing, symlinked, or hash-mismatched evidence is refused. Raw parameters, screenshots, credentials, diff --git a/packages/flow/docs/design/SUBROUTINES.md b/packages/flow/docs/design/SUBROUTINES.md new file mode 100644 index 0000000000..b46e7c5233 --- /dev/null +++ b/packages/flow/docs/design/SUBROUTINES.md @@ -0,0 +1,116 @@ +# Reusable subroutines + +Status: slices 1 and 2 (OpenAdapt#1892). Code: `openadapt_flow/subroutines.py`. + +Many workflows repeat the same few steps, such as searching for a record or +picking a date. A subroutine lifts such a sequence out of the workflows into +one typed unit. It is verified once and reused, and its boundaries are +checked every time it runs. + +## The unit + +A `Subroutine` has a signature and a body. + +- The signature names the typed inputs, the surface kind it runs on + (`web`, `windows`, ...), the entry checks and, optionally, the exit + checks. At least one entry check must read the live screen (text present, + or an anchor that resolves). A parameter comparison or an absence check + cannot show the run is on the right screen. +- The body is a `ProgramGraph` of action, branch and terminal states. + +Its identity is the SHA-256 of the canonical JSON of signature and body. +Any change makes a new subroutine. + +In this slice a body may only hold reversible, single-surface steps that read +nothing but their declared inputs. Writes (effects, API or submit bindings, +dialog answers, irreversible or unreviewed risk), reads, secrets, surface +switches, nested calls, loops, business decisions and a plaintext identity +band are refused. Every non-terminal state needs an outgoing edge, so a run +cannot leave the body without passing the exit checks. + +## Embedding and the run-time boundary + +`embed()` copies the subroutine into `Workflow.subflows` under +`subroutine//` and wraps it. An entry branch continues only +when every entry check holds and otherwise halts the run. With exit checks, +each success terminal becomes a branch that returns only when they hold. The +existing interpreter runs this unchanged, so no runtime code changes. + +## Library and re-qualification + +`SubroutineLibrary` stores subroutines by digest, keeps each name's versions +in order, and records which digest was qualified. A workflow is registered +as a user of the digests it embeds. Publishing a new version returns the +registered workflows still pinned to an older digest. Those workflows are +refused until they embed the new version and are registered again. + +## Admission + +`admit_workflow()` refuses a use, with a stable code, when: + +| Code | Meaning | +| --- | --- | +| `unknown_subroutine` | the library has no subroutine at that digest | +| `superseded` | a newer version of the name exists | +| `not_qualified` | that digest has no qualification record | +| `unregistered_use` | the workflow is not registered for that digest | +| `embedded_drift` | the embedded graph differs from what `embed()` produces | +| `surface_undeclared`, `surface_kind_mismatch`, `multi_surface` | the workflow does not run on the declared surface kind | +| `missing_input`, `input_type_mismatch`, `input_choices_mismatch`, `secret_input` | the workflow does not declare a matching input | +| `malformed_key` | a subflow key looks like a subroutine key but is not one | +| `library_missing` | certification was asked to admit a subroutine with no library (certification only) | + +Messages name subroutines, inputs and codes only. They never repeat screen +text or recorded values. + +## Certification + +`certify_project()` (so `flow qualify certify` and `flow qualify +--certify`) runs `certify_subroutine_uses()` on every workflow: + +- A workflow with no `subroutine/` subflow is certified exactly as before; + no library is read or written. +- A workflow that embeds one needs a library (`--subroutine-library + subroutines.json`). Without it, certification fails with + `subroutine_not_admitted` / `library_missing`, and the campaign form refuses + before running any case. +- With a library, certification first registers the workflow as the user of + the exact digests it embeds (replacing its older pins), then runs + `admit_workflow()`. Every refusal above still applies; each becomes a + `subroutine_not_admitted` qualification refusal whose + `details.subroutine_code` is the code from the table. The CLI writes the + library back atomically when registration changed it, and before the new + certificate is saved (`flow qualify certify` and the campaign alike). If + that write fails, the certificate is withdrawn, the bundle is saved + uncertified and the error is raised: no bundle is certified while the + library does not list it. + +Registration happens even when certification fails, so a later version of +the subroutine still lists the workflow for re-qualification. On a pass the +report gains nothing, so `current_certification_matches()` reproduces it +unchanged. Publishing a new version of a subroutine makes the next +certification of every registered user fail with `superseded` until it +embeds the new version. + +## Detection + +`mine_recurring_sequences()` proposes candidates from the programs alone. It +compares steps by shape: action kind, whether a parameter feeds the step, key +chord and accessibility role. It never compares labels, names, selectors, +URLs or application names. A candidate is a proposal. It is never admitted +or runnable. + +## Not in this slice + +- Re-checking admission at run time. A certification made before a newer + version was published stays valid for runs until the next certification; + the run gate does not yet consult the library. +- Binding the library digest into the certification record, and `flow + certify` (policy-only) does not run the gate. +- Renaming inputs at the call site. Today the caller must declare inputs + with the same names. +- Outputs (values a subroutine reads and hands back), writes, secrets and + multi-surface bodies. +- A sealed library. The library is a plain JSON file whose digests are + re-derived on load (an edited subroutine is refused), but the file as a + whole is not signed or sealed. diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index bd65924ff0..ebc334dd51 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -7086,11 +7086,28 @@ def _cmd_qualify(args: argparse.Namespace) -> int: if verb == "certify": policy = _qualification_policy(args) + library_path = getattr(args, "subroutine_library", None) + library = _load_subroutine_library(library_path) + library_before = library.to_json() if library is not None else "" report = certify_project( workflow, policy=policy, evidence_root=args.evidence_root, execution_profile=args.profile, + subroutine_library=library, + subroutine_workflow_id=getattr(args, "subroutine_workflow_id", None), + ) + from openadapt_flow.qualification import ( + persist_registration_before_certificate, + ) + + # The registration is written before the certificate is saved: a + # library write failure leaves the bundle uncertified, not certified + # with no dependent recorded for a later subroutine version. + persist_registration_before_certificate( + workflow, + args.bundle, + lambda: _save_subroutine_library(library, library_path, library_before), ) save_qualified_workflow(workflow, args.bundle) print(report.model_dump_json(indent=2) if args.json else report.render()) @@ -7099,6 +7116,43 @@ def _cmd_qualify(args: argparse.Namespace) -> int: raise SystemExit(f"unknown qualify command {verb!r}") +def _load_subroutine_library(path: Optional[str]) -> Any: + """Load ``--subroutine-library`` (None when not given); refuse a bad file.""" + + if not path: + return None + from openadapt_flow.subroutines import SubroutineError, load_library + + try: + return load_library(path) + except SubroutineError as exc: + raise SystemExit(f"subroutine library REFUSED: {exc}") from None + + +def _save_subroutine_library(library: Any, path: Optional[str], before: str) -> None: + """Write back the library when certification registered a workflow in it.""" + + if library is None or not path or library.to_json() == before: + return + from openadapt_flow.subroutines import save_library + + save_library(library, path) + + +def _add_subroutine_library_flags(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--subroutine-library", + default=None, + metavar="JSON", + help=( + "Subroutine library to admit embedded subroutines against. " + "Certifying registers this workflow in it, so a new version of a " + "subroutine it uses re-qualifies it. Required when the bundle " + "embeds a subroutine." + ), + ) + + def _cmd_qualify_proposal(args: argparse.Namespace) -> int: """Propose, accept, or locally admit pins mined from the compiled demo.""" @@ -8974,6 +9028,15 @@ def _cmd_qualify_campaign(args: argparse.Namespace) -> int: else Path("runs") / f"qualify-{bundle.resolve().name}-{time.strftime('%Y%m%d-%H%M%S')}" ) + library_path = getattr(args, "subroutine_library", None) + library = _load_subroutine_library(library_path) + library_state = {"before": library.to_json() if library is not None else ""} + + def persist_library() -> None: + _save_subroutine_library(library, library_path, library_state["before"]) + if library is not None: + library_state["before"] = library.to_json() + try: result = qualify( bundle, @@ -8984,13 +9047,17 @@ def _cmd_qualify_campaign(args: argparse.Namespace) -> int: certify=certify, run_params=run_params, campaign_id=args.campaign_id, + subroutine_library=library, + persist_subroutine_registration=persist_library, ) except QualifyConfigError as exc: print(str(exc), file=sys.stderr) return 2 except QualifyError as exc: + persist_library() print(f"qualify REFUSED: {exc}\nEvidence so far: {out}", file=sys.stderr) return 1 + persist_library() if not result.passed and result.partial: print( f"\nqualify PARTIAL: {len(result.rows)} trial(s) behaved as required in " @@ -10991,6 +11058,7 @@ def build_parser() -> argparse.ArgumentParser: default=None, help="Campaign id (default: verify.yaml or generated)", ) + _add_subroutine_library_flags(q) q.set_defaults(func=_cmd_qualify_campaign) q = qsub.add_parser("schema", help="Print the qualification-project JSON Schema") @@ -11451,6 +11519,13 @@ def build_parser() -> argparse.ArgumentParser: ) q.add_argument("--evidence-root", required=True) q.add_argument("--json", action="store_true") + _add_subroutine_library_flags(q) + q.add_argument( + "--subroutine-workflow-id", + default=None, + metavar="ID", + help="Id to register this workflow under (default: the workflow name)", + ) q.set_defaults(func=_cmd_qualify) p = sub.add_parser( diff --git a/packages/flow/openadapt_flow/qualification.py b/packages/flow/openadapt_flow/qualification.py index 9c660f640d..37b0002a4d 100644 --- a/packages/flow/openadapt_flow/qualification.py +++ b/packages/flow/openadapt_flow/qualification.py @@ -72,6 +72,7 @@ if TYPE_CHECKING: # pragma: no cover from openadapt_flow.ir import BusinessDecisionSpec, IdentityTemplate, Step, Workflow from openadapt_flow.policy import Policy + from openadapt_flow.subroutines import SubroutineLibrary QUALIFICATION_SCHEMA: Final[Literal["openadapt.qualification-project/v1"]] = ( @@ -1125,6 +1126,7 @@ class QualificationRefusalCode(str, Enum): JUDGMENT_CASE_EVIDENCE_UNVERIFIED = "judgment_case_evidence_unverified" BUSINESS_DECISION_DELIVERY_INVALID = "business_decision_delivery_invalid" POLICY_VIOLATION = "policy_violation" + SUBROUTINE_NOT_ADMITTED = "subroutine_not_admitted" class QualificationRefusal(BaseModel): @@ -5434,6 +5436,34 @@ def evaluate_qualification( ) +def subroutine_certification_refusals( + workflow: "Workflow", + library: Optional["SubroutineLibrary"], + *, + workflow_id: Optional[str] = None, +) -> list[QualificationRefusal]: + """Certification refusals for the subroutines ``workflow`` embeds. + + Runs :func:`openadapt_flow.subroutines.certify_subroutine_uses`, which + registers the workflow as a user of each embedded digest and then admits + every use. Empty for a workflow that embeds no subroutine. Messages name + subroutines, inputs and codes only. + """ + + from openadapt_flow.subroutines import certify_subroutine_uses + + admission = certify_subroutine_uses(workflow, library, workflow_id=workflow_id) + return [ + QualificationRefusal( + code=QualificationRefusalCode.SUBROUTINE_NOT_ADMITTED, + path=f"subflows.{refusal.subflow}", + message=refusal.message, + details={"subroutine_code": refusal.code}, + ) + for refusal in admission.refusals + ] + + def certify_project( workflow: "Workflow", *, @@ -5441,8 +5471,18 @@ def certify_project( evidence_root: Path | str, transition_predicate_vision: Any | None = None, execution_profile: Literal["standard", "regulated"] = "standard", + subroutine_library: Optional["SubroutineLibrary"] = None, + subroutine_workflow_id: Optional[str] = None, ) -> QualificationReport: - """Evaluate and persist the exact qualification decision in memory.""" + """Evaluate and persist the exact qualification decision in memory. + + A workflow that embeds subroutines (``subroutine//`` + subflows) is certified only when every use is admitted by + ``subroutine_library``; certifying also registers the workflow with the + library so a later version of any of those subroutines re-qualifies it. + Without a library such a workflow is refused. A workflow that embeds no + subroutine is evaluated exactly as before. + """ project = workflow.qualification report = evaluate_qualification( @@ -5451,6 +5491,16 @@ def certify_project( evidence_root=evidence_root, transition_predicate_vision=transition_predicate_vision, ) + subroutine_refusals = subroutine_certification_refusals( + workflow, subroutine_library, workflow_id=subroutine_workflow_id + ) + if subroutine_refusals: + report = report.model_copy( + update={ + "passed": False, + "refusals": [*report.refusals, *subroutine_refusals], + } + ) if project is not None: from openadapt_flow.policy import policy_contract_sha256 @@ -5693,6 +5743,31 @@ def current_certification_matches( ) +def persist_registration_before_certificate( + workflow: "Workflow", + bundle_dir: Path | str, + persist: Optional[Callable[[], None]], +) -> None: + """Write certification's subroutine registration before any certificate. + + ``certify_project`` registers the workflow with the subroutine library in + memory; ``persist`` writes that library. It must succeed before a new + certificate becomes durable, or a later version of a subroutine could not + list this workflow for re-qualification. When it raises, the certificate + is withdrawn in memory, ``bundle_dir`` is saved uncertified and the error + is re-raised, so no valid new certificate is left behind. + """ + + if persist is None: + return + try: + persist() + except BaseException: + _invalidate_certification(workflow) + save_qualified_workflow(workflow, bundle_dir) + raise + + def save_qualified_workflow( workflow: "Workflow", bundle_dir: Path | str, diff --git a/packages/flow/openadapt_flow/qualify/campaign.py b/packages/flow/openadapt_flow/qualify/campaign.py index e52dcaa7a2..0f8a591cf5 100644 --- a/packages/flow/openadapt_flow/qualify/campaign.py +++ b/packages/flow/openadapt_flow/qualify/campaign.py @@ -68,6 +68,7 @@ ) from openadapt_flow.qualify.faults import WebFaultDriver from openadapt_flow.qualify.keys import LocalTestKey, new_local_test_key +from openadapt_flow.subroutines import SubroutineLibrary, uses_subroutines from openadapt_flow.verification import VerificationTier #: (case id, expected outcome, oracle kind), in run order. @@ -1248,6 +1249,8 @@ def qualify( in_place: bool = False, write_campaign_files: bool = True, log: Callable[[str], None] = print, + subroutine_library: Optional[SubroutineLibrary] = None, + persist_subroutine_registration: Optional[Callable[[], None]] = None, ) -> QualifyResult: """Run the campaign; certify under ``certify`` (a policy name) if given. @@ -1260,6 +1263,12 @@ def qualify( ``evidence_root`` (default ``out``) is the root case evidence is relative to. ``write_campaign_files=False`` leaves out the campaign summary, timing, notes and ``QUALIFICATION.md`` (for packs that list every file). + ``subroutine_library`` is the library certification admits embedded + subroutines against (and registers this workflow with); a bundle that + embeds one is refused before any run when certifying without it. + ``persist_subroutine_registration`` writes that library; it runs right + after certification, before the certified bundle is saved, and when it + raises the certification is withdrawn and the bundle saved uncertified. """ from openadapt_flow.policy import load_policy @@ -1267,6 +1276,7 @@ def qualify( QualificationCaseKind, certify_project, evaluate_qualification, + persist_registration_before_certificate, record_case_results, save_qualified_workflow, ) @@ -1310,6 +1320,15 @@ def qualify( ) workflow = Workflow.load(bundle_src) + if ( + certify is not None + and subroutine_library is None + and uses_subroutines(workflow) + ): + raise QualifyConfigError( + "the bundle embeds subroutines; --certify needs --subroutine-library " + "to admit them" + ) save, notes = apply_verify_config(workflow, config) params = resolve_inputs(workflow, config, campaign_id=campaign_id) problems = check_ready(workflow, config, save, params) @@ -1651,7 +1670,15 @@ def authorize( ) if certify is not None: qreport = certify_project( - workflow, policy=load_policy(certify), evidence_root=root + workflow, + policy=load_policy(certify), + evidence_root=root, + subroutine_library=subroutine_library, + ) + # The registration is durable before any certified save below + # (business rules, the admission envelope, the final bundle). + persist_registration_before_certificate( + workflow, bundle_dir, persist_subroutine_registration ) else: qreport = evaluate_qualification(workflow, evidence_root=root) diff --git a/packages/flow/openadapt_flow/subroutines.py b/packages/flow/openadapt_flow/subroutines.py new file mode 100644 index 0000000000..1cd67049dc --- /dev/null +++ b/packages/flow/openadapt_flow/subroutines.py @@ -0,0 +1,1217 @@ +"""Reusable semantic subroutines: typed, content-addressed, admitted per use. + +Compiled programs repeat the same low-level steps across workflows (search +for a record, pick a date). A **subroutine** is one such sequence lifted out +of the workflows that share it, with a typed signature, so it is verified +once and reused. This module is the first slice of that idea. It makes the +unit and its admission real; it does not change the runtime. + +The unit + :class:`Subroutine` is a typed signature (:class:`SubroutineSignature`) + plus a body (:class:`~openadapt_flow.ir.ProgramGraph`). The signature + names its typed inputs, the surface kind it runs on, and the boundary + checks that must hold when it is entered and when it returns. Its identity + is a SHA-256 over the canonical JSON of signature and body + (:attr:`Subroutine.sha256`). Any change to a step, a check, an input or the + surface is a new subroutine with a new digest. + +What a body may hold (checked by :func:`body_problems`) + Only reversible, single-surface action steps, branches and terminals that + read nothing but declared inputs. Writes (effects, API bindings, submit + bindings, dialog responses, irreversible or unreviewed risk), reads, + secrets, surface switches, nested calls, loops, business decisions and a + plaintext identity band are refused. The boundary is narrow on purpose. + Widening it is later work, one refusal at a time. + +The library + :class:`SubroutineLibrary` stores subroutines by digest, keeps each name's + versions in order, and records which digest was qualified and under which + policy. Workflows that use a subroutine are registered with the library. + Publishing a new version of a name returns a + :class:`RequalificationPlan` that lists every registered workflow still + pinned to an older digest. Those workflows are refused until they are + re-embedded and re-qualified. + +Using one in a workflow + :func:`embed` copies a subroutine into ``Workflow.subflows`` under the + key ``subroutine//`` (:func:`subflow_key`) and wraps it: + + * an entry BRANCH that continues only when every entry check holds and + otherwise ends the whole run with a ``halt`` terminal; + * when exit checks are declared, each ``success`` terminal of the body + becomes a BRANCH that returns only when every exit check holds and + otherwise halts. + + The existing interpreter runs the wrapped graph unchanged, so the boundary + checks are enforced at run time by today's guard semantics (an unmet guard + takes the halt edge; nothing guesses). + +Admission + :func:`admit_workflow` finds every subroutine a workflow embeds (from the + subflow keys, never from application names) and admits each use only + when the digest is known to the library, is that name's current version, + is qualified, is registered for this workflow, the embedded graph is + byte-identical to what :func:`embed` produces, the workflow runs on the + declared surface kind, and the workflow supplies every required input + with the declared type. Any other case is a refusal with a stable code. + +Certification + :func:`certify_subroutine_uses` is the gate + :func:`openadapt_flow.qualification.certify_project` runs. It registers + the workflow as the user of the digests it embeds and then admits it, so + certifying is what puts a workflow on a subroutine's re-qualification + list. A workflow that embeds a subroutine is refused when no library is + supplied; one that embeds none is untouched. + +Detection + :func:`mine_recurring_sequences` proposes candidate subroutines from the + programs alone: it compares steps by shape (action kind, whether the step + is parameter-bound, key chord, accessibility role) and never by labels, + names, selectors, URLs or application names. A candidate is a proposal; + it is never admitted or runnable. + +Refusal messages name subroutines, inputs and codes only. They never echo +anchor text, recorded values or URL patterns. Deterministic and model-free; +no I/O beyond the JSON a caller asks for. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import tempfile +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Literal, Mapping, Optional + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from openadapt_flow.ir import ( + ActionKind, + ExecutionTargetKind, + ParamKind, + Predicate, + PredicateKind, + ProgramGraph, + State, + StateKind, + Step, + Transition, + Workflow, +) + +SUBROUTINE_SCHEMA: Literal["openadapt.subroutine/v1"] = "openadapt.subroutine/v1" +LIBRARY_SCHEMA: Literal["openadapt.subroutine-library/v1"] = ( + "openadapt.subroutine-library/v1" +) +CANDIDATE_SCHEMA: Literal["openadapt.subroutine-candidate/v1"] = ( + "openadapt.subroutine-candidate/v1" +) + +#: Prefix of the ``Workflow.subflows`` key an embedded subroutine lives under. +SUBFLOW_PREFIX = "subroutine/" +#: State ids the wrapper adds. A body may not use this prefix. +RESERVED_STATE_PREFIX = "__sr_" +_ENTRY = "__sr_entry__" +_ENTRY_REFUSED = "__sr_entry_refused__" +_RETURN = "__sr_return__" +_EXIT_REFUSED = "__sr_exit_refused__" + +_NAME_PATTERN = r"^[a-z][a-z0-9_]{0,63}$" +_INPUT_PATTERN = r"^[A-Za-z_][A-Za-z0-9_]{0,127}$" +_SHA256_PATTERN = r"^[0-9a-f]{64}$" + +#: Action kinds a subroutine body may perform in this slice. +_ALLOWED_ACTIONS = frozenset( + { + ActionKind.CLICK, + ActionKind.DOUBLE_CLICK, + ActionKind.RIGHT_CLICK, + ActionKind.DRAG, + ActionKind.TYPE, + ActionKind.SELECT_OPTION, + ActionKind.KEY, + ActionKind.HOTKEY, + ActionKind.WAIT, + ActionKind.SCROLL, + } +) +_ALLOWED_STATES = frozenset({StateKind.ACTION, StateKind.BRANCH, StateKind.TERMINAL}) + + +class SubroutineError(ValueError): + """A subroutine, library document or library call is not usable.""" + + +def _canonical(value: Any) -> bytes: + return json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + + +def _sha256(value: Any) -> str: + return hashlib.sha256(_canonical(value)).hexdigest() + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +# -- the unit ----------------------------------------------------------------- + + +class SubroutineInput(BaseModel): + """One typed input. Carries no example value, so no recorded data.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str = Field(pattern=_INPUT_PATTERN) + type: ParamKind = ParamKind.STRING + required: bool = True + choices: tuple[str, ...] = () + + @model_validator(mode="after") + def _choices_only_for_enum(self) -> "SubroutineInput": + if self.choices and self.type is not ParamKind.ENUM: + raise ValueError("only an enum input has choices") + if self.type is ParamKind.ENUM and not self.choices: + raise ValueError("an enum input needs its choices") + if len(set(self.choices)) != len(self.choices): + raise ValueError("enum choices must be unique") + return self + + +def _is_affirmative(predicate: Predicate) -> bool: + """True for a check that reads the live surface and can only pass on it. + + The same rule business decisions use: a TEXT_PRESENT with text or an + ANCHOR_RESOLVES with an anchor, alone or inside an AND. A parameter + comparison, an absence check, OR and NOT can hold without the expected + screen, so they do not count. + """ + + if predicate.kind is PredicateKind.TEXT_PRESENT: + return bool(predicate.text) + if predicate.kind is PredicateKind.ANCHOR_RESOLVES: + return predicate.anchor is not None + if predicate.kind is PredicateKind.AND: + return any(_is_affirmative(op) for op in predicate.operands) + return False + + +class SubroutineSignature(BaseModel): + """What a caller and the runtime can rely on at the subroutine boundary.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["openadapt.subroutine/v1"] = SUBROUTINE_SCHEMA + name: str = Field(pattern=_NAME_PATTERN) + intent: str = Field(min_length=1, max_length=240) + inputs: tuple[SubroutineInput, ...] = () + surface_kind: ExecutionTargetKind + entry_checks: tuple[Predicate, ...] = Field( + min_length=1, + description=( + "Checks on the live surface before the first step. All must hold, " + "and at least one must be affirmative (text present or an anchor " + "that resolves)." + ), + ) + exit_checks: tuple[Predicate, ...] = Field( + default=(), + description="Checks that must all hold before the subroutine returns.", + ) + + @model_validator(mode="after") + def _boundary(self) -> "SubroutineSignature": + names = [item.name for item in self.inputs] + if len(set(names)) != len(names): + raise ValueError("subroutine input names must be unique") + if not any(_is_affirmative(check) for check in self.entry_checks): + raise ValueError( + "a subroutine needs an affirmative entry check (text present or " + "an anchor that resolves); parameters, absence checks, OR and " + "NOT cannot show it is on the right screen" + ) + if self.exit_checks and not any( + _is_affirmative(check) for check in self.exit_checks + ): + raise ValueError("declared exit checks need one affirmative check") + return self + + def input(self, name: str) -> Optional[SubroutineInput]: + for item in self.inputs: + if item.name == name: + return item + return None + + +class Subroutine(BaseModel): + """A typed, content-addressed reusable unit: signature plus body.""" + + model_config = ConfigDict(extra="forbid") + + signature: SubroutineSignature + body: ProgramGraph + + @model_validator(mode="after") + def _well_formed(self) -> "Subroutine": + problems = body_problems(self) + if problems: + raise ValueError( + f"subroutine {self.signature.name!r} is not admissible: " + + "; ".join(problems) + ) + return self + + @property + def name(self) -> str: + return self.signature.name + + @property + def sha256(self) -> str: + """Content identity over signature and body (canonical JSON).""" + + return _sha256( + { + "signature": self.signature.model_dump(mode="json"), + "body": self.body.model_dump(mode="json"), + } + ) + + +# -- body checks ---------------------------------------------------------------- + + +def _param_refs(value: Any, problems: list[str], where: str) -> set[str]: + """Every parameter name a JSON value refers to, found generically. + + A key named ``param`` or ending in ``_param`` holds one name; a key named + ``params`` or ending in ``_params`` holds a list of names. Anything else + under such a key is refused, so a reference this walk does not + understand can never slip past the closure check. + """ + + found: set[str] = set() + if isinstance(value, dict): + for key, item in value.items(): + single = key == "param" or key.endswith("_param") + many = key == "params" or key.endswith("_params") + if (single or many) and item is not None: + if single and isinstance(item, str): + found.add(item) + elif ( + many + and isinstance(item, (list, tuple)) + and all(isinstance(name, str) for name in item) + ): + found.update(item) + else: + problems.append(f"{where}: unsupported parameter reference {key!r}") + continue + found |= _param_refs(item, problems, where) + elif isinstance(value, (list, tuple)): + for item in value: + found |= _param_refs(item, problems, where) + return found + + +def _step_problems(step: Step, where: str) -> list[str]: + problems: list[str] = [] + if step.action not in _ALLOWED_ACTIONS: + problems.append(f"{where}: action {step.action.value!r} is not allowed") + if step.effects: + problems.append(f"{where}: declares a system-of-record effect") + if step.api_binding is not None: + problems.append(f"{where}: carries an API binding") + if step.submit_binding is not None: + problems.append(f"{where}: carries a submit binding") + if step.dialog is not None: + problems.append(f"{where}: answers a dialog") + if step.risk != "reversible" or step.risk_review_required: + problems.append(f"{where}: is not a reviewed reversible step") + if step.secret: + problems.append(f"{where}: types a secret") + if step.read is not None or step.read_gate is not None: + problems.append(f"{where}: reads a record field") + if ( + step.surface is not None + or step.opens + or step.surface_origin is not None + or step.closes_surface + ): + problems.append(f"{where}: changes or names a surface") + return problems + + +#: Anchor fields that hold a plaintext identity band (screen text near the +#: target, or a record-row identity). A reusable unit may not carry either. +_IDENTITY_BAND_FIELDS = ("context_text", "structured_identity") + + +def _identity_band_paths(value: Any, path: str = "") -> list[str]: + """JSON paths of every non-empty identity band anywhere in ``value``. + + Walked generically, so an anchor inside a step, a step guard, a wait, + a transition guard, a boundary check or any nested predicate operand is + found the same way. Paths name fields only, never the band's text. + """ + + found: list[str] = [] + if isinstance(value, dict): + for key in sorted(value): + item = value[key] + here = f"{path}.{key}" if path else key + if key in _IDENTITY_BAND_FIELDS and item not in (None, ""): + found.append(here) + continue + found.extend(_identity_band_paths(item, here)) + elif isinstance(value, (list, tuple)): + for index, item in enumerate(value): + found.extend(_identity_band_paths(item, f"{path}[{index}]")) + return found + + +def body_problems(subroutine: Subroutine) -> list[str]: + """Why a subroutine cannot be admitted as a unit (empty when it can). + + Messages name state ids, step ids and input names only. + """ + + sig = subroutine.signature + body = subroutine.body + problems: list[str] = [] + states = body.states + if body.entry not in states: + problems.append("the body's entry state is not defined") + refs: set[str] = set() + for label, checks in (("entry", sig.entry_checks), ("exit", sig.exit_checks)): + for index, check in enumerate(checks): + dumped = check.model_dump(mode="json") + refs |= _param_refs(dumped, problems, "boundary check") + for band in _identity_band_paths(dumped): + problems.append( + f"{label} check {index}: keeps a plaintext identity band ({band})" + ) + has_success = False + for state_id, state in states.items(): + where = f"state {state_id!r}" + if state.id != state_id: + problems.append(f"{where}: key and id differ") + if state_id.startswith(RESERVED_STATE_PREFIX): + problems.append(f"{where}: uses the reserved prefix") + if state.kind not in _ALLOWED_STATES: + problems.append(f"{where}: kind {state.kind.value!r} is not allowed") + if state.kind is not StateKind.TERMINAL and not state.transitions: + # Falling off the graph returns to the caller without passing the + # exit checks, so every path must end at a terminal. + problems.append(f"{where}: has no outgoing transition") + for transition in state.transitions: + if transition.target not in states: + problems.append(f"{where}: transition to an undefined state") + if state.on_exception is not None and state.on_exception not in states: + problems.append(f"{where}: exception handler is not defined") + if state.kind is StateKind.ACTION: + if state.step is None: + problems.append(f"{where}: action state has no step") + else: + problems.extend(_step_problems(state.step, where)) + if state.kind is StateKind.TERMINAL and state.outcome in (None, "success"): + has_success = True + dumped_state = state.model_dump(mode="json") + refs |= _param_refs(dumped_state, problems, where) + for band in _identity_band_paths(dumped_state): + problems.append(f"{where}: keeps a plaintext identity band ({band})") + if not has_success: + problems.append("the body has no success terminal to return through") + declared = {item.name for item in sig.inputs} + undeclared = sorted(refs - declared) + if undeclared: + problems.append(f"the body reads undeclared inputs {undeclared}") + unused = sorted(declared - refs) + if unused: + problems.append(f"declared inputs are never read {unused}") + return problems + + +# -- embedding -------------------------------------------------------------------- + + +def subflow_key(subroutine: Subroutine) -> str: + """The ``Workflow.subflows`` key that pins a subroutine by digest.""" + + return f"{SUBFLOW_PREFIX}{subroutine.name}/{subroutine.sha256}" + + +def parse_subflow_key(key: str) -> Optional[tuple[str, str]]: + """``(name, sha256)`` for a subroutine subflow key, else None. + + A key that starts with the prefix but is malformed raises: a workflow + cannot hide a subroutine use behind a near-miss key. + """ + + if not key.startswith(SUBFLOW_PREFIX): + return None + parts = key[len(SUBFLOW_PREFIX) :].split("/") + if ( + len(parts) != 2 + or not re.fullmatch(_NAME_PATTERN, parts[0]) + or not re.fullmatch(_SHA256_PATTERN, parts[1]) + ): + raise SubroutineError("malformed subroutine subflow key") + return parts[0], parts[1] + + +def _all_of(checks: tuple[Predicate, ...], intent: str) -> Predicate: + if len(checks) == 1: + return checks[0].model_copy(deep=True) + return Predicate( + kind=PredicateKind.AND, + operands=[check.model_copy(deep=True) for check in checks], + intent=intent, + ) + + +def wrapped_body(subroutine: Subroutine) -> ProgramGraph: + """The body with its boundary checks as live guards around it.""" + + sig = subroutine.signature + states: dict[str, State] = {} + for state_id, state in subroutine.body.states.items(): + copy = state.model_copy(deep=True) + if ( + sig.exit_checks + and copy.kind is StateKind.TERMINAL + and copy.outcome in (None, "success") + ): + copy = State( + id=state_id, + kind=StateKind.BRANCH, + transitions=[ + Transition( + guard=_all_of(sig.exit_checks, f"{sig.name} exit checks"), + target=_RETURN, + label="exit checks hold", + ), + Transition(target=_EXIT_REFUSED, label="exit checks failed"), + ], + ) + states[state_id] = copy + states[_ENTRY] = State( + id=_ENTRY, + kind=StateKind.BRANCH, + transitions=[ + Transition( + guard=_all_of(sig.entry_checks, f"{sig.name} entry checks"), + target=subroutine.body.entry, + label="entry checks hold", + ), + Transition(target=_ENTRY_REFUSED, label="entry checks failed"), + ], + ) + states[_ENTRY_REFUSED] = State( + id=_ENTRY_REFUSED, + kind=StateKind.TERMINAL, + outcome="halt", + reason=f"subroutine {sig.name} entry checks did not hold", + ) + if sig.exit_checks: + states[_RETURN] = State(id=_RETURN, kind=StateKind.TERMINAL, outcome="success") + states[_EXIT_REFUSED] = State( + id=_EXIT_REFUSED, + kind=StateKind.TERMINAL, + outcome="halt", + reason=f"subroutine {sig.name} exit checks did not hold", + ) + return ProgramGraph(entry=_ENTRY, states=states) + + +def embed(workflow: Workflow, subroutine: Subroutine) -> tuple[Workflow, str]: + """Return a copy of ``workflow`` with the subroutine embedded, and its key. + + The caller adds the SUBFLOW_CALL (or loop body) that uses the key. An + existing subflow under the same key is replaced only by identical content. + """ + + key = subflow_key(subroutine) + graph = wrapped_body(subroutine) + existing = workflow.subflows.get(key) + if existing is not None and _sha256(existing.model_dump(mode="json")) != _sha256( + graph.model_dump(mode="json") + ): + raise SubroutineError( + f"workflow already embeds a different body under subroutine " + f"{subroutine.name!r}" + ) + copy = workflow.model_copy(deep=True) + copy.subflows = {**copy.subflows, key: graph} + return copy, key + + +# -- library ---------------------------------------------------------------------- + + +class SubroutineQualification(BaseModel): + """Evidence that one exact digest passed qualification.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + sha256: str = Field(pattern=_SHA256_PATTERN) + policy_name: str = Field(min_length=1, max_length=128) + evidence_sha256: str = Field( + pattern=_SHA256_PATTERN, + description="Digest of the retained qualification evidence.", + ) + qualified_at: str = Field(default_factory=_now) + + +class SubroutineUse(BaseModel): + """One workflow's pin on one subroutine digest.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + workflow_id: str = Field(min_length=1, max_length=256) + name: str = Field(pattern=_NAME_PATTERN) + sha256: str = Field(pattern=_SHA256_PATTERN) + + +class RequalificationPlan(BaseModel): + """What publishing a new version of a subroutine puts back in review.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + name: str + sha256: str + version: int + #: Registered uses pinned to an older digest of the same name. Each one is + #: refused until it is re-embedded and its workflow is re-qualified. + stale_uses: tuple[SubroutineUse, ...] = () + + @property + def workflows(self) -> tuple[str, ...]: + return tuple(sorted({use.workflow_id for use in self.stale_uses})) + + +class SubroutineLibrary: + """Content-addressed subroutines, their versions, qualifications and users. + + In memory, with :meth:`to_json` / :meth:`from_json`. Loading re-derives + every digest from content, so an edited document is refused rather than + trusted. + """ + + def __init__(self) -> None: + self._by_digest: dict[str, Subroutine] = {} + self._versions: dict[str, list[str]] = {} + self._qualified: dict[str, SubroutineQualification] = {} + self._uses: set[SubroutineUse] = set() + + # -- read -------------------------------------------------------------- + + def get(self, sha256: str) -> Optional[Subroutine]: + return self._by_digest.get(sha256) + + def versions(self, name: str) -> tuple[str, ...]: + return tuple(self._versions.get(name, ())) + + def current(self, name: str) -> Optional[Subroutine]: + versions = self._versions.get(name) + return self._by_digest[versions[-1]] if versions else None + + def qualification(self, sha256: str) -> Optional[SubroutineQualification]: + return self._qualified.get(sha256) + + def uses(self, name: Optional[str] = None) -> tuple[SubroutineUse, ...]: + selected = (u for u in self._uses if name is None or u.name == name) + return tuple(sorted(selected, key=lambda u: (u.name, u.workflow_id, u.sha256))) + + def dependents(self, name: str) -> tuple[str, ...]: + """Workflow ids registered as using any version of ``name``.""" + + return tuple(sorted({use.workflow_id for use in self.uses(name)})) + + # -- write ------------------------------------------------------------- + + def publish(self, subroutine: Subroutine) -> RequalificationPlan: + """Add a subroutine as its name's newest version (idempotent).""" + + digest = subroutine.sha256 + name = subroutine.name + versions = self._versions.setdefault(name, []) + if digest in self._by_digest: + if versions[-1] != digest: + raise SubroutineError( + f"subroutine {name!r} cannot go back to an older version; " + "publish a new change instead" + ) + else: + self._by_digest[digest] = Subroutine.model_validate( + subroutine.model_dump(mode="json") + ) + versions.append(digest) + stale = tuple(use for use in self.uses(name) if use.sha256 != digest) + return RequalificationPlan( + name=name, sha256=digest, version=len(versions), stale_uses=stale + ) + + def qualify( + self, sha256: str, *, policy_name: str, evidence_sha256: str + ) -> SubroutineQualification: + """Record that this exact digest passed qualification.""" + + if sha256 not in self._by_digest: + raise SubroutineError("cannot qualify a subroutine the library lacks") + record = SubroutineQualification( + sha256=sha256, policy_name=policy_name, evidence_sha256=evidence_sha256 + ) + self._qualified[sha256] = record + return record + + def register( + self, workflow_id: str, workflow: Workflow + ) -> tuple[SubroutineUse, ...]: + """Record every subroutine ``workflow`` embeds, replacing older pins. + + Registration is what lets a later version find this workflow. It does + not admit anything. + """ + + found = tuple( + SubroutineUse(workflow_id=workflow_id, name=name, sha256=digest) + for name, digest in embedded_uses(workflow) + ) + self._uses = {u for u in self._uses if u.workflow_id != workflow_id} + self._uses.update(found) + return found + + # -- persistence ------------------------------------------------------- + + def to_json(self) -> str: + payload = { + "schema": LIBRARY_SCHEMA, + "subroutines": { + digest: sub.model_dump(mode="json") + for digest, sub in sorted(self._by_digest.items()) + }, + "versions": {name: list(v) for name, v in sorted(self._versions.items())}, + "qualifications": [ + q.model_dump(mode="json") for _, q in sorted(self._qualified.items()) + ], + "uses": [u.model_dump(mode="json") for u in self.uses()], + } + return json.dumps(payload, indent=2, sort_keys=True) + + @classmethod + def from_json(cls, text: str) -> "SubroutineLibrary": + try: + raw = json.loads(text) + except json.JSONDecodeError as exc: + raise SubroutineError("subroutine library is not JSON") from exc + if not isinstance(raw, dict) or raw.get("schema") != LIBRARY_SCHEMA: + raise SubroutineError(f"not a subroutine library ({LIBRARY_SCHEMA})") + library = cls() + try: + for digest, data in raw.get("subroutines", {}).items(): + sub = Subroutine.model_validate(data) + if sub.sha256 != digest: + raise SubroutineError( + f"subroutine {sub.name!r} content does not match its digest" + ) + library._by_digest[digest] = sub + for name, versions in raw.get("versions", {}).items(): + if len(set(versions)) != len(versions): + raise SubroutineError(f"subroutine {name!r} repeats a version") + for digest in versions: + held = library._by_digest.get(digest) + if held is None or held.name != name: + raise SubroutineError( + f"subroutine {name!r} lists a version it does not hold" + ) + library._versions[name] = list(versions) + listed = {d for v in library._versions.values() for d in v} + if listed != set(library._by_digest): + raise SubroutineError("a stored subroutine is not in any version list") + for data in raw.get("qualifications", []): + record = SubroutineQualification.model_validate(data) + if record.sha256 not in library._by_digest: + raise SubroutineError("a qualification names an unknown digest") + library._qualified[record.sha256] = record + for data in raw.get("uses", []): + library._uses.add(SubroutineUse.model_validate(data)) + except SubroutineError: + raise + except ValueError as exc: + raise SubroutineError( + f"subroutine library is malformed ({type(exc).__name__})" + ) from None + return library + + +# -- admission -------------------------------------------------------------------- + +RefusalCode = Literal[ + "library_missing", + "malformed_key", + "unknown_subroutine", + "name_mismatch", + "superseded", + "not_qualified", + "unregistered_use", + "embedded_drift", + "surface_undeclared", + "multi_surface", + "surface_kind_mismatch", + "missing_input", + "input_type_mismatch", + "input_choices_mismatch", + "secret_input", +] + + +class SubroutineRefusal(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + code: RefusalCode + subflow: str + message: str + + +class SubroutineAdmission(BaseModel): + """Per-workflow verdict. ``admitted`` is True only with no refusals.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + workflow_id: str + uses: tuple[SubroutineUse, ...] = () + refusals: tuple[SubroutineRefusal, ...] = () + + @property + def admitted(self) -> bool: + return not self.refusals + + +def embedded_uses(workflow: Workflow) -> list[tuple[str, str]]: + """``(name, sha256)`` for every subroutine subflow in ``workflow``.""" + + found: list[tuple[str, str]] = [] + for key in sorted(workflow.subflows): + parsed = parse_subflow_key(key) + if parsed is not None: + found.append(parsed) + return found + + +def _caller_inputs_problems( + workflow: Workflow, sig: SubroutineSignature +) -> list[tuple[RefusalCode, str]]: + problems: list[tuple[RefusalCode, str]] = [] + secret = set(workflow.secret_params or []) + # Only parameters the caller declares count. A name the embedded body + # itself reads is not a value the caller supplies. + supplied = set(workflow.param_specs) | set(workflow.params) + for item in sig.inputs: + if item.name in secret: + problems.append( + ("secret_input", f"input {item.name!r} is a secret in the caller") + ) + continue + spec = workflow.param_specs.get(item.name) + if spec is None: + if item.name not in supplied: + if item.required: + problems.append( + ("missing_input", f"caller does not supply {item.name!r}") + ) + continue + if item.type is not ParamKind.STRING: + problems.append( + ( + "input_type_mismatch", + f"input {item.name!r} needs type {item.type.value!r}; " + "the caller declares no type", + ) + ) + continue + if spec.type is not item.type: + problems.append( + ( + "input_type_mismatch", + f"input {item.name!r} needs type {item.type.value!r}, " + f"caller has {spec.type.value!r}", + ) + ) + continue + if item.type is ParamKind.ENUM and not set(spec.choices) <= set(item.choices): + problems.append( + ( + "input_choices_mismatch", + f"caller allows choices for {item.name!r} that the " + "subroutine does not", + ) + ) + if item.required and not spec.required and spec.example is None: + problems.append( + ( + "missing_input", + f"input {item.name!r} is required but optional in the caller " + "with no default", + ) + ) + return problems + + +def admit_workflow( + workflow: Workflow, + library: SubroutineLibrary, + *, + workflow_id: Optional[str] = None, +) -> SubroutineAdmission: + """Admit every subroutine ``workflow`` embeds, or refuse with reasons. + + A workflow that embeds none is admitted with no uses: this gate adds + nothing to a workflow that does not use subroutines. + """ + + wid = workflow_id or workflow.name + refusals: list[SubroutineRefusal] = [] + uses: list[SubroutineUse] = [] + + def refuse(code: RefusalCode, key: str, message: str) -> None: + refusals.append(SubroutineRefusal(code=code, subflow=key, message=message)) + + registered = set(library.uses()) + for key in sorted(workflow.subflows): + try: + parsed = parse_subflow_key(key) + except SubroutineError: + refuse("malformed_key", key[:96], "malformed subroutine subflow key") + continue + if parsed is None: + continue + name, digest = parsed + use = SubroutineUse(workflow_id=wid, name=name, sha256=digest) + uses.append(use) + sub = library.get(digest) + if sub is None: + refuse("unknown_subroutine", key, f"library has no {name!r} at this digest") + continue + if sub.name != name: + refuse("name_mismatch", key, "digest belongs to a different subroutine") + continue + current = library.current(name) + if current is None or current.sha256 != digest: + refuse( + "superseded", + key, + f"{name!r} has a newer version; re-embed it and re-qualify " + "this workflow", + ) + if library.qualification(digest) is None: + refuse("not_qualified", key, f"this version of {name!r} is not qualified") + if use not in registered: + refuse( + "unregistered_use", + key, + "workflow is not registered as a user of this version, so a " + "change to it would not re-qualify this workflow", + ) + expected = wrapped_body(sub).model_dump(mode="json") + if _sha256(workflow.subflows[key].model_dump(mode="json")) != _sha256(expected): + refuse( + "embedded_drift", + key, + f"embedded {name!r} differs from the published version", + ) + sig = sub.signature + if workflow.surfaces: + refuse( + "multi_surface", + key, + "subroutines in multi-surface workflows are not supported yet", + ) + elif workflow.surface is None: + refuse("surface_undeclared", key, "workflow declares no surface kind") + elif workflow.surface != sig.surface_kind: + refuse( + "surface_kind_mismatch", + key, + f"{name!r} runs on {sig.surface_kind!r}, workflow on " + f"{workflow.surface!r}", + ) + for code, message in _caller_inputs_problems(workflow, sig): + refuse(code, key, message) + return SubroutineAdmission( + workflow_id=wid, uses=tuple(uses), refusals=tuple(refusals) + ) + + +def uses_subroutines(workflow: Workflow) -> bool: + """True when any subflow key claims to be a subroutine (even malformed).""" + + return any(key.startswith(SUBFLOW_PREFIX) for key in workflow.subflows) + + +def certify_subroutine_uses( + workflow: Workflow, + library: Optional[SubroutineLibrary], + *, + workflow_id: Optional[str] = None, +) -> SubroutineAdmission: + """The subroutine gate certification runs (``flow qualify``). + + Certification is where a workflow becomes a registered user of the exact + subroutine digests it embeds, so this registers the workflow with + ``library`` (replacing its older pins) and then runs :func:`admit_workflow`. + A later :meth:`SubroutineLibrary.publish` of any of those names therefore + lists this workflow for re-qualification, and every other refusal + (unknown, superseded, not qualified, edited body, surface, inputs) still + applies. + + Fail-closed: a workflow that embeds a subroutine is refused when no + library is supplied. A workflow that embeds none is admitted with no uses + and ``library`` is neither read nor changed. + """ + + wid = workflow_id or workflow.name + if not uses_subroutines(workflow): + return SubroutineAdmission(workflow_id=wid) + if library is None: + return SubroutineAdmission( + workflow_id=wid, + refusals=tuple( + SubroutineRefusal( + code="library_missing", + subflow=key[:96], + message=( + "workflow embeds a subroutine but no subroutine library " + "was supplied to check it against" + ), + ) + for key in sorted(workflow.subflows) + if key.startswith(SUBFLOW_PREFIX) + ), + ) + try: + library.register(wid, workflow) + except SubroutineError: + # A malformed key cannot be pinned; admission refuses it below. + pass + return admit_workflow(workflow, library, workflow_id=wid) + + +def load_library(path: Path | str) -> SubroutineLibrary: + """Read a library JSON file (digests are re-derived; edits are refused).""" + + try: + text = Path(path).read_text(encoding="utf-8") + except OSError as exc: + raise SubroutineError( + f"cannot read subroutine library ({type(exc).__name__})" + ) from None + return SubroutineLibrary.from_json(text) + + +def save_library(library: SubroutineLibrary, path: Path | str) -> None: + """Write a library JSON file atomically (temp file, then replace).""" + + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(prefix=".subroutines-", dir=target.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(library.to_json()) + handle.write("\n") + os.replace(tmp, target) + except BaseException: + if os.path.exists(tmp): + os.unlink(tmp) + raise + + +# -- detection ---------------------------------------------------------------------- + + +def step_shape(step: Step) -> str: + """An application-free token for comparing steps across workflows. + + Uses the action kind, whether a parameter feeds the step, the key chord + and the accessibility role. Never labels, names, selectors, text, URLs + or application names. + """ + + role = "" + if step.anchor is not None and step.anchor.structural is not None: + role = (step.anchor.structural.role or "").lower() + bound = "p" if (step.param or step.target_param) else "-" + chord = "+".join([*step.modifiers, step.key or ""]) if step.key else "" + return f"{step.action.value}|{bound}|{chord}|{role}" + + +class SubroutineOccurrence(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + workflow_id: str + step_ids: tuple[str, ...] + params: tuple[str, ...] = () + + +class SubroutineCandidate(BaseModel): + """A recurring step sequence. A proposal, never admitted or runnable.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["openadapt.subroutine-candidate/v1"] = CANDIDATE_SCHEMA + status: Literal["candidate"] = "candidate" + admitted: Literal[False] = False + runnable: Literal[False] = False + shape: tuple[str, ...] + occurrences: tuple[SubroutineOccurrence, ...] + + @property + def workflows(self) -> tuple[str, ...]: + return tuple(sorted({o.workflow_id for o in self.occurrences})) + + +def _straight_runs(workflow: Workflow) -> list[list[Step]]: + """Maximal straight-line action runs the program walks in order.""" + + if workflow.program is None: + return [list(workflow.steps)] if workflow.steps else [] + runs: list[list[Step]] = [] + for graph in (workflow.program, *workflow.subflows.values()): + incoming: dict[str, int] = defaultdict(int) + for state in graph.states.values(): + for transition in state.transitions: + incoming[transition.target] += 1 + incoming[graph.entry] += 1 + + def single_next(state: State) -> Optional[str]: + if len(state.transitions) == 1 and state.transitions[0].guard is None: + return state.transitions[0].target + return None + + continued: set[str] = set() + for state in graph.states.values(): + nxt = single_next(state) + if ( + state.kind is StateKind.ACTION + and nxt is not None + and incoming[nxt] == 1 + ): + continued.add(nxt) + for state_id, state in graph.states.items(): + if state.kind is not StateKind.ACTION or state_id in continued: + continue + run: list[Step] = [] + cursor: Optional[State] = state + seen: set[str] = set() + while ( + cursor is not None + and cursor.kind is StateKind.ACTION + and cursor.step is not None + and cursor.id not in seen + ): + seen.add(cursor.id) + run.append(cursor.step) + nxt = single_next(cursor) + if nxt is None or incoming[nxt] != 1: + break + cursor = graph.states.get(nxt) + if run: + runs.append(run) + return runs + + +def mine_recurring_sequences( + workflows: Mapping[str, Workflow], + *, + min_length: int = 3, + min_workflows: int = 2, + max_length: int = 32, +) -> list[SubroutineCandidate]: + """Propose step sequences that recur, by shape, across workflows. + + A sequence qualifies when at least ``min_workflows`` distinct workflows + contain it. A candidate that is part of a longer candidate found in the + same workflows is dropped. Output order is deterministic. + """ + + if min_length < 2 or min_workflows < 2: + raise SubroutineError("a subroutine needs 2+ steps seen in 2+ workflows") + hits: dict[tuple[str, ...], list[SubroutineOccurrence]] = defaultdict(list) + for wid in sorted(workflows): + for run in _straight_runs(workflows[wid]): + shapes = [step_shape(step) for step in run] + for length in range(min_length, min(max_length, len(run)) + 1): + for start in range(0, len(run) - length + 1): + window = run[start : start + length] + params = tuple( + name + for step in window + for name in (step.param, step.target_param) + if name + ) + hits[tuple(shapes[start : start + length])].append( + SubroutineOccurrence( + workflow_id=wid, + step_ids=tuple(step.id for step in window), + params=params, + ) + ) + found = { + shape: occ + for shape, occ in hits.items() + if len({o.workflow_id for o in occ}) >= min_workflows + } + + def inside(short: tuple[str, ...], long: tuple[str, ...]) -> bool: + n = len(short) + return any(long[i : i + n] == short for i in range(len(long) - n + 1)) + + kept: list[SubroutineCandidate] = [] + for shape in sorted(found, key=lambda s: (-len(s), s)): + wids = {o.workflow_id for o in found[shape]} + if any( + len(k.shape) > len(shape) + and inside(shape, k.shape) + and wids <= set(k.workflows) + for k in kept + ): + continue + kept.append(SubroutineCandidate(shape=shape, occurrences=tuple(found[shape]))) + return kept + + +def subroutine_from_steps( + signature: SubroutineSignature, steps: Iterable[Step] +) -> Subroutine: + """Build a straight-line subroutine from consecutive steps (deep copies).""" + + items = [step.model_copy(deep=True) for step in steps] + if not items: + raise SubroutineError("a subroutine needs at least one step") + ids = [step.id for step in items] + repeated = sorted({sid for sid in ids if ids.count(sid) > 1}) + if repeated: + # A repeated id would make a later step silently replace an earlier + # one, so the body would no longer be the steps it was built from. + raise SubroutineError(f"step ids must be unique; repeated {repeated}") + states: dict[str, State] = {} + end = "end" + for index, step in enumerate(items): + target = end if index + 1 == len(items) else items[index + 1].id + states[step.id] = State( + id=step.id, + kind=StateKind.ACTION, + step=step, + transitions=[Transition(target=target)], + ) + if end in states: + raise SubroutineError("a step id collides with the terminal id") + states[end] = State(id=end, kind=StateKind.TERMINAL, outcome="success") + return Subroutine( + signature=signature, body=ProgramGraph(entry=items[0].id, states=states) + ) diff --git a/packages/flow/tests/test_subroutine_certification.py b/packages/flow/tests/test_subroutine_certification.py new file mode 100644 index 0000000000..116d65ed29 --- /dev/null +++ b/packages/flow/tests/test_subroutine_certification.py @@ -0,0 +1,611 @@ +"""Certification wiring for reusable subroutines (OpenAdapt#1892, slice 2). + +``certify_project`` (and so ``flow qualify certify`` and ``flow qualify +--certify``) runs the subroutine gate: a workflow that embeds a subroutine is +certified only when every use is admitted by the supplied library, and +certifying registers the workflow so a new version of that subroutine puts +it back into re-qualification. A workflow that embeds no subroutine is +certified exactly as before. + +The real qualification fixture from ``test_qualification_project`` is reused, +so the passing case is a genuine signed campaign. No model calls. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openadapt_flow.ir import ParamSpec, Workflow +from openadapt_flow.policy import load_policy +from openadapt_flow.qualification import ( + QualificationRefusalCode, + certify_project, + current_certification_matches, +) +from openadapt_flow.subroutines import ( + SUBFLOW_PREFIX, + SubroutineError, + SubroutineLibrary, + certify_subroutine_uses, + embed, + load_library, + save_library, + uses_subroutines, +) +from openadapt_flow.verification import VerificationTier +from tests.test_qualification_project import ( + _configure, + _qualification_visual_fixture, + _record_passing_campaign, +) +from tests.test_qualification_project import ( + _workflow as _qualified_workflow, +) +from tests.test_subroutines import ( + EVIDENCE, + caller, + find_record, + key_step, + signature, + subroutine_from_steps, + type_step, +) + + +def codes(admission) -> set[str]: + return {refusal.code for refusal in admission.refusals} + + +# -- the gate --------------------------------------------------------------------- + + +def test_workflow_without_subroutines_never_touches_the_library(): + library = SubroutineLibrary() + before = library.to_json() + plain = Workflow(name="plain", surface="web") + assert not uses_subroutines(plain) + assert certify_subroutine_uses(plain, None).admitted + assert certify_subroutine_uses(plain, library).admitted + assert library.to_json() == before + + +def test_embedded_subroutine_without_a_library_is_refused(): + workflow = caller(find_record()) + admission = certify_subroutine_uses(workflow, None) + assert not admission.admitted + assert codes(admission) == {"library_missing"} + + +def test_certifying_registers_the_workflow_then_admits_it(): + sub = find_record() + workflow = caller(sub) + library = SubroutineLibrary() + library.publish(sub) + library.qualify(sub.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + assert library.dependents(sub.name) == () + + admission = certify_subroutine_uses(workflow, library) + + assert admission.admitted, admission.refusals + assert library.dependents(sub.name) == ("intake",) + # Run-time admission now finds the registration certification wrote. + from openadapt_flow.subroutines import admit_workflow + + assert admit_workflow(workflow, library).admitted + + +def test_certification_keeps_every_other_refusal(): + sub = find_record() + workflow = caller(sub) + library = SubroutineLibrary() + library.publish(sub) # published but never qualified + admission = certify_subroutine_uses(workflow, library) + assert codes(admission) == {"not_qualified"} + # Registration still happened, so a later version re-qualifies it. + assert library.dependents(sub.name) == ("intake",) + + unknown = certify_subroutine_uses(caller(find_record()), SubroutineLibrary()) + assert codes(unknown) == {"unknown_subroutine"} + + +def test_new_version_requalifies_a_certified_workflow(): + v1 = find_record() + workflow = caller(v1) + library = SubroutineLibrary() + library.publish(v1) + library.qualify(v1.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + assert certify_subroutine_uses(workflow, library, workflow_id="wf-1").admitted + + v2 = find_record( + steps=[ + key_step("focus", "G"), + type_step("enter", "query"), + key_step("go", "Enter"), + ] + ) + plan = library.publish(v2) + library.qualify(v2.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + + assert plan.workflows == ("wf-1",) + stale = certify_subroutine_uses(workflow, library, workflow_id="wf-1") + assert "superseded" in codes(stale) + # Re-embedding the new version and certifying again admits it. + assert certify_subroutine_uses(caller(v2), library, workflow_id="wf-1").admitted + + +def test_malformed_subroutine_key_is_refused_not_raised(): + workflow = caller(find_record()) + key = next(iter(workflow.subflows)) + workflow.subflows[f"{SUBFLOW_PREFIX}Bad/not-a-digest"] = workflow.subflows[key] + library = SubroutineLibrary() + admission = certify_subroutine_uses(workflow, library) + assert "malformed_key" in codes(admission) + + +def test_library_file_round_trip_and_tamper_refusal(tmp_path: Path): + sub = find_record() + library = SubroutineLibrary() + library.publish(sub) + library.qualify(sub.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + path = tmp_path / "lib" / "subroutines.json" + save_library(library, path) + assert load_library(path).to_json() == library.to_json() + assert [p.name for p in path.parent.iterdir()] == ["subroutines.json"] + + raw = json.loads(path.read_text()) + digest = next(iter(raw["subroutines"])) + raw["subroutines"][digest]["signature"]["intent"] = "edited" + path.write_text(json.dumps(raw)) + with pytest.raises(SubroutineError, match="does not match its digest"): + load_library(path) + with pytest.raises(SubroutineError, match="cannot read"): + load_library(tmp_path / "missing.json") + + +# -- certify_project on a real signed campaign ------------------------------------- + + +def _note_subroutine(): + return subroutine_from_steps( + signature( + name="enter_note", + intent="type the note into the focused field", + inputs=(signature().inputs[0].model_copy(update={"name": "note"}),), + surface_kind="citrix", + ), + [key_step("focus", "Tab"), type_step("enter", "note")], + ) + + +def _certifiable(tmp_path: Path, *, embed_subroutine: bool) -> tuple[Workflow, Path]: + workflow = _qualified_workflow() + workflow.surface = "citrix" + workflow.param_specs = {"note": ParamSpec(name="note", example="example")} + if embed_subroutine: + workflow, _ = embed(workflow, _note_subroutine()) + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "save.png").write_bytes(_qualification_visual_fixture()[1]) + workflow.save(bundle) + workflow = Workflow.load(bundle) + _configure(workflow, tier=VerificationTier.INDEPENDENT_SYSTEM) + evidence_root = tmp_path / "evidence" + _record_passing_campaign(workflow, evidence_root) + return workflow, evidence_root + + +def _subroutine_refusals(report) -> list: + return [ + r + for r in report.refusals + if r.code is QualificationRefusalCode.SUBROUTINE_NOT_ADMITTED + ] + + +def test_certify_project_without_subroutines_is_unchanged(tmp_path: Path): + workflow, evidence_root = _certifiable(tmp_path, embed_subroutine=False) + policy = load_policy("clinical-write") + report = certify_project(workflow, policy=policy, evidence_root=evidence_root) + assert report.passed, report.refusals + assert not _subroutine_refusals(report) + assert current_certification_matches(workflow, policy=policy) + + +def test_certify_project_refuses_subroutines_without_a_library(tmp_path: Path): + workflow, evidence_root = _certifiable(tmp_path, embed_subroutine=True) + policy = load_policy("clinical-write") + report = certify_project(workflow, policy=policy, evidence_root=evidence_root) + assert not report.passed + refusals = _subroutine_refusals(report) + assert [r.details["subroutine_code"] for r in refusals] == ["library_missing"] + # Everything else about the campaign passed; only the gate refused. + assert refusals == report.refusals + assert workflow.qualification is not None + certification = workflow.qualification.last_certification + assert certification is not None and certification.passed is False + assert certification.report_sha256 == report.report_sha256() + assert not current_certification_matches(workflow, policy=policy) + + +def test_certify_project_refuses_an_unqualified_subroutine(tmp_path: Path): + workflow, evidence_root = _certifiable(tmp_path, embed_subroutine=True) + library = SubroutineLibrary() + library.publish(_note_subroutine()) + report = certify_project( + workflow, + policy=load_policy("clinical-write"), + evidence_root=evidence_root, + subroutine_library=library, + ) + assert not report.passed + assert [r.details["subroutine_code"] for r in _subroutine_refusals(report)] == [ + "not_qualified" + ] + + +def test_certify_project_admits_and_registers_a_qualified_subroutine(tmp_path: Path): + workflow, evidence_root = _certifiable(tmp_path, embed_subroutine=True) + sub = _note_subroutine() + library = SubroutineLibrary() + library.publish(sub) + library.qualify(sub.sha256, policy_name="clinical-write", evidence_sha256=EVIDENCE) + policy = load_policy("clinical-write") + + report = certify_project( + workflow, + policy=policy, + evidence_root=evidence_root, + subroutine_library=library, + subroutine_workflow_id="qualified-write@1", + ) + + assert report.passed, report.refusals + assert library.dependents("enter_note") == ("qualified-write@1",) + assert current_certification_matches(workflow, policy=policy) + + # A new version of the subroutine lists this workflow and refuses it. + v2 = subroutine_from_steps( + sub.signature, + [ + key_step("focus", "Tab"), + key_step("clear", "Delete"), + type_step("enter", "note"), + ], + ) + plan = library.publish(v2) + assert plan.workflows == ("qualified-write@1",) + again = certify_project( + workflow, + policy=policy, + evidence_root=evidence_root, + subroutine_library=library, + subroutine_workflow_id="qualified-write@1", + ) + assert not again.passed + assert "superseded" in { + r.details["subroutine_code"] for r in _subroutine_refusals(again) + } + + +# -- the `flow qualify certify` command ------------------------------------------- + + +def _cli_bundle(tmp_path: Path) -> tuple[Path, Path]: + from openadapt_flow.qualification import save_qualified_workflow + + workflow, evidence_root = _certifiable(tmp_path, embed_subroutine=True) + save_qualified_workflow(workflow, tmp_path / "bundle") + return tmp_path / "bundle", evidence_root + + +def _certify_cli(bundle: Path, evidence_root: Path, *extra: str) -> int: + from openadapt_flow.__main__ import main + + return main( + [ + "qualify", + "certify", + str(bundle), + "--policy", + "clinical-write", + "--evidence-root", + str(evidence_root), + "--json", + *extra, + ] + ) + + +def test_cli_certify_refuses_an_embedded_subroutine_without_a_library( + tmp_path: Path, capsys +): + bundle, evidence_root = _cli_bundle(tmp_path) + assert _certify_cli(bundle, evidence_root) == 2 + out = json.loads(capsys.readouterr().out) + assert out["passed"] is False + assert {r["code"] for r in out["refusals"]} == {"subroutine_not_admitted"} + + +def test_cli_certify_admits_and_writes_the_registration_back(tmp_path: Path, capsys): + bundle, evidence_root = _cli_bundle(tmp_path) + sub = _note_subroutine() + library = SubroutineLibrary() + library.publish(sub) + library.qualify(sub.sha256, policy_name="clinical-write", evidence_sha256=EVIDENCE) + path = tmp_path / "subroutines.json" + save_library(library, path) + + code = _certify_cli( + bundle, + evidence_root, + "--subroutine-library", + str(path), + "--subroutine-workflow-id", + "qualified-write@1", + ) + + assert code == 0, capsys.readouterr() + assert json.loads(capsys.readouterr().out)["passed"] is True + assert load_library(path).dependents("enter_note") == ("qualified-write@1",) + + +def test_cli_certify_refuses_a_tampered_library_file(tmp_path: Path): + bundle, evidence_root = _cli_bundle(tmp_path) + path = tmp_path / "subroutines.json" + path.write_text('{"schema": "something-else"}') + with pytest.raises(SystemExit, match="subroutine library REFUSED"): + _certify_cli(bundle, evidence_root, "--subroutine-library", str(path)) + assert path.read_text() == '{"schema": "something-else"}' + + +# -- the `flow qualify --certify` campaign ---------------------------------------- + + +def test_campaign_certify_refuses_before_running_without_a_library(tmp_path: Path): + from openadapt_flow.qualify import QualifyConfigError, qualify + from tests.test_qualify_declarative import _config + from tests.test_qualify_declarative import _workflow as _campaign_workflow + + workflow = _campaign_workflow() + workflow.surface = "web" + workflow, _ = embed(workflow, find_record()) + bundle = tmp_path / "bundle" + workflow.save(bundle) + out = tmp_path / "out" + with pytest.raises(QualifyConfigError, match="--subroutine-library"): + qualify( + bundle, _config(), out=out, certify="clinical-write", log=lambda _l: None + ) + assert not out.exists() + + +# -- the registration is durable before the certificate --------------------------- +# +# A certificate saved before its registration is written leaves, on a library +# write failure, a certified bundle the library does not list: publishing a new +# version of the subroutine could not put it back into re-qualification. + + +@pytest.fixture +def read_only_dir(tmp_path: Path): + import os + import stat + + if hasattr(os, "geteuid") and os.geteuid() == 0: + pytest.skip("root ignores directory permissions") + directory = tmp_path / "library" + directory.mkdir() + yield directory + directory.chmod(stat.S_IRWXU) + + +def _qualified_library_file(directory: Path, *, read_only: bool = False) -> Path: + import stat + + sub = _note_subroutine() + library = SubroutineLibrary() + library.publish(sub) + library.qualify(sub.sha256, policy_name="clinical-write", evidence_sha256=EVIDENCE) + path = directory / "subroutines.json" + save_library(library, path) + if read_only: + directory.chmod(stat.S_IRUSR | stat.S_IXUSR) + return path + + +def test_cli_certify_leaves_no_certificate_when_the_registration_cannot_be_written( + tmp_path: Path, read_only_dir: Path +): + bundle, evidence_root = _cli_bundle(tmp_path) + path = _qualified_library_file(read_only_dir, read_only=True) + before = path.read_text() + + with pytest.raises(PermissionError): + _certify_cli( + bundle, + evidence_root, + "--subroutine-library", + str(path), + "--subroutine-workflow-id", + "qualified-write@1", + ) + + assert path.read_text() == before + assert load_library(path).dependents("enter_note") == () + saved = Workflow.load(bundle) + assert not current_certification_matches( + saved, policy=load_policy("clinical-write") + ) + assert saved.qualification is not None + assert saved.qualification.last_certification is None + + +def test_cli_certify_writes_the_registration_before_the_certificate( + tmp_path: Path, monkeypatch, capsys +): + import openadapt_flow.qualification as qualification_mod + import openadapt_flow.subroutines as subroutines_mod + + bundle, evidence_root = _cli_bundle(tmp_path) + path = _qualified_library_file(tmp_path) + events: list[str] = [] + real_save, real_library = ( + qualification_mod.save_qualified_workflow, + subroutines_mod.save_library, + ) + + def save_workflow(workflow, bundle_dir, **kw): + cert = workflow.qualification and workflow.qualification.last_certification + events.append("certificate" if cert is not None and cert.passed else "bundle") + return real_save(workflow, bundle_dir, **kw) + + def save_lib(library, target): + events.append("library") + return real_library(library, target) + + monkeypatch.setattr(qualification_mod, "save_qualified_workflow", save_workflow) + monkeypatch.setattr(subroutines_mod, "save_library", save_lib) + + code = _certify_cli( + bundle, + evidence_root, + "--subroutine-library", + str(path), + "--subroutine-workflow-id", + "qualified-write@1", + ) + + assert code == 0, capsys.readouterr() + assert events == ["library", "certificate"] + assert load_library(path).dependents("enter_note") == ("qualified-write@1",) + assert current_certification_matches( + Workflow.load(bundle), policy=load_policy("clinical-write") + ) + + +def _campaign_with_registration(monkeypatch, events: list[str], state: dict): + """The campaign's fakes, with certification that registers in the library. + + ``state["valid"]`` is whether the in-memory workflow holds a passing new + certificate; every bundle save records it. + """ + + from types import SimpleNamespace + + import openadapt_flow.qualification as qualification_mod + import openadapt_flow.subroutines as subroutines_mod + from tests.test_lookup_capabilities import _Campaign + + fake = _Campaign(monkeypatch) + real_save = qualification_mod.save_qualified_workflow + real_invalidate = qualification_mod._invalidate_certification + real_library = subroutines_mod.save_library + + def certify(workflow, *, subroutine_library=None, **_kw): + assert subroutine_library is not None + subroutine_library.register( + "campaign@1", embed(workflow, _note_subroutine())[0] + ) + state["valid"] = True + events.append("certify") + return SimpleNamespace( + passed=True, render=lambda: "PASS", model_dump=lambda **_k: {} + ) + + def save_workflow(workflow, bundle_dir, **kw): + events.append("certificate" if state.get("valid") else "bundle") + return real_save(workflow, bundle_dir, **kw) + + def invalidate(workflow): + state["valid"] = False + events.append("withdrawn") + return real_invalidate(workflow) + + def save_lib(library, target): + events.append("library") + return real_library(library, target) + + monkeypatch.setattr(qualification_mod, "certify_project", certify) + monkeypatch.setattr(qualification_mod, "save_qualified_workflow", save_workflow) + monkeypatch.setattr(qualification_mod, "_invalidate_certification", invalidate) + monkeypatch.setattr(subroutines_mod, "save_library", save_lib) + return fake + + +def _campaign_cli(monkeypatch, tmp_path: Path, fake, library: Path) -> int: + import openadapt_flow.qualify as qualify_pkg + from openadapt_flow.__main__ import main + from tests.test_lookup_capabilities import _config as _campaign_config + from tests.test_lookup_capabilities import _source_bundle + + monkeypatch.setattr( + qualify_pkg, "load_verify_config", lambda _p: _campaign_config() + ) + monkeypatch.setattr( + "openadapt_flow.qualify.campaign._default_hooks", + lambda _config, _hooks: fake.hooks(), + ) + return main( + [ + "qualify", + str(_source_bundle(tmp_path)), + "--verify", + str(tmp_path / "verify.yaml"), + "--certify", + "clinical-write", + "--out", + str(tmp_path / "out"), + "--subroutine-library", + str(library), + ] + ) + + +def test_campaign_certify_leaves_no_certificate_when_the_registration_cannot_be_written( + tmp_path: Path, read_only_dir: Path, monkeypatch +): + events: list[str] = [] + state: dict = {} + fake = _campaign_with_registration(monkeypatch, events, state) + path = _qualified_library_file(read_only_dir, read_only=True) + before = path.read_text() + + with pytest.raises(PermissionError): + _campaign_cli(monkeypatch, tmp_path, fake, path) + + after = events[events.index("certify") :] + assert after == ["certify", "library", "withdrawn", "bundle"] + assert path.read_text() == before + saved = Workflow.load(tmp_path / "out" / "bundle") + assert saved.qualification is not None + assert saved.qualification.last_certification is None + assert not current_certification_matches( + saved, policy=load_policy("clinical-write") + ) + assert not (tmp_path / "out" / "qualification" / "admission.local.json").exists() + + +def test_campaign_certify_writes_the_registration_before_the_certificate( + tmp_path: Path, monkeypatch +): + events: list[str] = [] + state: dict = {} + fake = _campaign_with_registration(monkeypatch, events, state) + path = _qualified_library_file(tmp_path) + + def no_envelope(**_kw): + raise ValueError("no admission envelope in this test") + + monkeypatch.setattr("openadapt_flow.qualify.result.local_admission", no_envelope) + + _campaign_cli(monkeypatch, tmp_path, fake, path) + + after = events[events.index("certify") :] + assert after[:2] == ["certify", "library"] + assert "certificate" in after + # Written once: the closing write-back finds nothing new. + assert events.count("library") == 1 + assert load_library(path).dependents("enter_note") == ("campaign@1",) diff --git a/packages/flow/tests/test_subroutines.py b/packages/flow/tests/test_subroutines.py new file mode 100644 index 0000000000..8aa4a08947 --- /dev/null +++ b/packages/flow/tests/test_subroutines.py @@ -0,0 +1,783 @@ +"""Reusable semantic subroutines (OpenAdapt#1892, first slice). + +A subroutine is a typed, content-addressed unit that workflows embed by +digest. These tests pin its identity, the narrow body it may hold, the +library's version and re-qualification bookkeeping, the per-use admission +refusals, the live boundary checks the runtime enforces through the existing +interpreter, and application-free detection of recurring sequences. + +Backend and vision are faked (reused from test_replayer). No model calls. +""" + +from __future__ import annotations + +import json +from functools import partial + +import pytest +from pydantic import ValidationError + +from openadapt_flow.bundle_validation import validate_workflow +from openadapt_flow.ir import ( + ActionKind, + Anchor, + ParamKind, + ParamSpec, + Predicate, + PredicateKind, + ProgramGraph, + State, + StateKind, + Step, + StructuralLocator, + SurfaceSpec, + Transition, + Workflow, +) +from openadapt_flow.runtime.effects import Effect, EffectKind +from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.subroutines import ( + SUBFLOW_PREFIX, + Subroutine, + SubroutineError, + SubroutineInput, + SubroutineLibrary, + SubroutineSignature, + admit_workflow, + body_problems, + embed, + mine_recurring_sequences, + parse_subflow_key, + step_shape, + subflow_key, + subroutine_from_steps, + wrapped_body, +) +from tests.test_replayer import FakeBackend, FakeVision, Match, make_png + +EVIDENCE = "e" * 64 + +# -- builders ----------------------------------------------------------------- + + +def text(value: str) -> Predicate: + return Predicate(kind=PredicateKind.TEXT_PRESENT, text=value) + + +def key_step(step_id: str, key: str) -> Step: + return Step(id=step_id, intent=f"press {key}", action=ActionKind.KEY, key=key) + + +def type_step(step_id: str, param: str) -> Step: + return Step(id=step_id, intent=f"type {param}", action=ActionKind.TYPE, param=param) + + +def anchored_click(step_id: str, label: str, role: str = "button") -> Step: + return Step( + id=step_id, + intent=f"click {label}", + action=ActionKind.CLICK, + anchor=Anchor( + template="templates/btn.png", + region=(100, 100, 50, 20), + click_point=(110, 105), + ocr_text=label, + structural=StructuralLocator(role=role, name=label), + ), + ) + + +def signature(**overrides) -> SubroutineSignature: + fields = dict( + name="find_record", + intent="search for a record and open it", + inputs=(SubroutineInput(name="query"),), + surface_kind="web", + entry_checks=(text("Find record"),), + ) + fields.update(overrides) + return SubroutineSignature(**fields) + + +def find_record(**overrides) -> Subroutine: + steps = overrides.pop( + "steps", + [key_step("focus", "F"), type_step("enter", "query"), key_step("go", "Enter")], + ) + return subroutine_from_steps(signature(**overrides), steps) + + +def caller(sub: Subroutine, *, name: str = "intake", **fields) -> Workflow: + base = Workflow( + name=name, + surface=fields.pop("surface", "web"), + param_specs=fields.pop( + "param_specs", {"query": ParamSpec(name="query", example="x")} + ), + **fields, + ) + embedded, key = embed(base, sub) + embedded.program = ProgramGraph( + entry="call", + states={ + "call": State( + id="call", + kind=StateKind.SUBFLOW_CALL, + subflow=key, + transitions=[Transition(target="done")], + ), + "done": State(id="done", kind=StateKind.TERMINAL, outcome="success"), + }, + ) + return embedded + + +def ready_library(sub: Subroutine, *workflows: Workflow) -> SubroutineLibrary: + library = SubroutineLibrary() + library.publish(sub) + library.qualify(sub.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + for workflow in workflows: + library.register(workflow.name, workflow) + return library + + +def codes(admission) -> set[str]: + return {refusal.code for refusal in admission.refusals} + + +# -- the unit ------------------------------------------------------------------- + + +def test_entry_check_must_read_the_live_screen(): + for weak in ( + Predicate(kind=PredicateKind.PARAM_EQUALS, param="query", value="x"), + Predicate(kind=PredicateKind.TEXT_ABSENT, text="Error"), + Predicate(kind=PredicateKind.NOT, operands=[text("Error")]), + Predicate(kind=PredicateKind.OR, operands=[text("A"), text("B")]), + ): + with pytest.raises(ValidationError, match="affirmative entry check"): + signature(entry_checks=(weak,)) + with pytest.raises(ValidationError): + signature(entry_checks=()) + # An AND that includes one live check is accepted. + signature( + entry_checks=( + Predicate( + kind=PredicateKind.AND, + operands=[ + text("Find record"), + Predicate(kind=PredicateKind.TEXT_ABSENT, text="Error"), + ], + ), + ) + ) + + +def test_digest_is_content_identity(): + a, b = find_record(), find_record() + assert a.sha256 == b.sha256 and len(a.sha256) == 64 + changed = [ + find_record(entry_checks=(text("Search"),)), + find_record(surface_kind="windows"), + find_record(inputs=(SubroutineInput(name="query", type=ParamKind.DATE),)), + find_record( + steps=[ + key_step("focus", "G"), + type_step("enter", "query"), + key_step("go", "Enter"), + ] + ), + find_record(exit_checks=(text("Record"),)), + ] + digests = {a.sha256, *(sub.sha256 for sub in changed)} + assert len(digests) == 1 + len(changed) + + +@pytest.mark.parametrize( + ("step", "fragment"), + [ + ( + Step( + id="w", + intent="save", + action=ActionKind.KEY, + key="s", + effects=[ + Effect( + kind=EffectKind.RECORD_WRITTEN, entity="row", match={"id": "1"} + ) + ], + ), + "system-of-record effect", + ), + ( + Step( + id="w", + intent="save", + action=ActionKind.KEY, + key="s", + risk="irreversible", + ), + "reviewed reversible", + ), + ( + Step( + id="w", + intent="save", + action=ActionKind.KEY, + key="s", + risk_review_required=True, + ), + "reviewed reversible", + ), + ( + Step( + id="w", intent="pw", action=ActionKind.TYPE, param="query", secret=True + ), + "secret", + ), + ( + Step(id="w", intent="go", action=ActionKind.KEY, key="x", surface="main"), + "surface", + ), + ], +) +def test_body_refuses_writes_reads_secrets_and_surface_changes(step, fragment): + steps = [type_step("enter", "query"), step] + with pytest.raises(ValidationError, match=fragment): + find_record(steps=steps) + + +def test_body_refuses_a_plaintext_identity_band(): + click = anchored_click("pick", "Open") + click.anchor.context_text = "row text" + with pytest.raises(ValidationError, match="plaintext identity band"): + find_record(steps=[type_step("enter", "query"), click]) + + +def _banded_anchor(field: str) -> Anchor: + anchor = anchored_click("x", "Open").anchor + assert anchor is not None + setattr(anchor, field, "Jane Roe 1970-01-01") + return anchor + + +def _resolves(anchor: Anchor) -> Predicate: + return Predicate(kind=PredicateKind.ANCHOR_RESOLVES, anchor=anchor) + + +@pytest.mark.parametrize("field", ["context_text", "structured_identity"]) +@pytest.mark.parametrize( + "where", + [ + "entry_check", + "entry_check_nested", + "exit_check", + "exit_check_nested_not", + "step_guard", + "step_wait_until", + "transition_guard", + ], +) +def test_identity_band_is_refused_in_every_predicate_anchor(where, field): + """Codex review of #1987: anchors inside predicates were not checked.""" + + from openadapt_flow.ir import Guard + + banded = _resolves(_banded_anchor(field)) + steps = [ + key_step("focus", "F"), + type_step("enter", "query"), + key_step("go", "Enter"), + ] + if where == "entry_check": + build = partial(find_record, entry_checks=(text("Find record"), banded)) + elif where == "entry_check_nested": + nested = Predicate( + kind=PredicateKind.AND, + operands=[ + text("Find record"), + Predicate(kind=PredicateKind.NOT, operands=[banded]), + ], + ) + build = partial(find_record, entry_checks=(nested,)) + elif where == "exit_check": + build = partial(find_record, exit_checks=(banded,)) + elif where == "exit_check_nested_not": + build = partial( + find_record, + exit_checks=( + text("Record"), + Predicate( + kind=PredicateKind.OR, + operands=[Predicate(kind=PredicateKind.NOT, operands=[banded])], + ), + ), + ) + elif where == "step_guard": + steps[2].guard = Guard(predicate=banded) + build = partial(find_record, steps=steps) + elif where == "step_wait_until": + steps[2].wait_until = banded + build = partial(find_record, steps=steps) + else: + body = ProgramGraph( + entry="enter", + states={ + "enter": State( + id="enter", + kind=StateKind.ACTION, + step=type_step("enter", "query"), + transitions=[Transition(target="b")], + ), + "b": State( + id="b", + kind=StateKind.BRANCH, + transitions=[ + Transition(guard=banded, target="end"), + Transition(target="end"), + ], + ), + "end": State(id="end", kind=StateKind.TERMINAL, outcome="success"), + }, + ) + build = partial(Subroutine, signature=signature(), body=body) + with pytest.raises(ValidationError, match="plaintext identity band") as exc: + build() + # The refusal names the field, never the band's text. + assert field in str(exc.value) + assert "Jane Roe" not in str(exc.value) + + +def test_duplicate_step_ids_are_refused_not_dropped(): + """Codex review of #1987: a repeated id silently replaced an earlier step.""" + + with pytest.raises(SubroutineError, match="step ids must be unique"): + find_record( + steps=[ + type_step("enter", "query"), + key_step("duplicate", "F"), + key_step("duplicate", "Enter"), + ] + ) + # Unique ids keep every action, in order. + sub = find_record( + steps=[type_step("enter", "query"), key_step("a", "F"), key_step("b", "Enter")] + ) + order, cursor = [], sub.body.entry + while sub.body.states[cursor].kind is StateKind.ACTION: + state = sub.body.states[cursor] + order.append((state.step.action, state.step.key or state.step.param)) + cursor = state.transitions[0].target + assert order == [ + (ActionKind.TYPE, "query"), + (ActionKind.KEY, "F"), + (ActionKind.KEY, "Enter"), + ] + + +def test_body_reads_exactly_its_declared_inputs(): + with pytest.raises(ValidationError, match="undeclared inputs \\['other'\\]"): + find_record(steps=[type_step("enter", "query"), type_step("more", "other")]) + with pytest.raises(ValidationError, match="never read \\['query'\\]"): + find_record(steps=[key_step("focus", "F")]) + # A parameter read only by a boundary check counts as read too. + with pytest.raises(ValidationError, match="undeclared inputs \\['mode'\\]"): + find_record( + steps=[type_step("enter", "query"), key_step("go", "Enter")], + entry_checks=( + Predicate( + kind=PredicateKind.AND, + operands=[ + text("Find record"), + Predicate( + kind=PredicateKind.PARAM_EQUALS, param="mode", value="a" + ), + ], + ), + ), + ) + + +def test_body_refuses_nesting_reserved_ids_and_dangling_edges(): + sig = signature() + nested = ProgramGraph( + entry="t", + states={ + "t": State( + id="t", + kind=StateKind.ACTION, + step=type_step("t", "query"), + transitions=[Transition(target="c")], + ), + "c": State( + id="c", + kind=StateKind.SUBFLOW_CALL, + subflow="other", + transitions=[Transition(target="end")], + ), + "end": State(id="end", kind=StateKind.TERMINAL, outcome="success"), + }, + ) + with pytest.raises(ValidationError, match="'subflow_call' is not allowed"): + Subroutine(signature=sig, body=nested) + reserved = ProgramGraph( + entry="__sr_x", + states={ + "__sr_x": State( + id="__sr_x", + kind=StateKind.ACTION, + step=type_step("t", "query"), + transitions=[Transition(target="missing")], + ), + }, + ) + fall_off = ProgramGraph( + entry="t", + states={ + "t": State(id="t", kind=StateKind.ACTION, step=type_step("t", "query")), + "end": State(id="end", kind=StateKind.TERMINAL, outcome="success"), + }, + ) + with pytest.raises(ValidationError, match="no outgoing transition"): + Subroutine(signature=sig, body=fall_off) + with pytest.raises(ValidationError) as exc: + Subroutine(signature=sig, body=reserved) + message = str(exc.value) + assert "reserved prefix" in message + assert "undefined state" in message + assert "no success terminal" in message + + +# -- embedding and the live boundary checks --------------------------------------- + + +@pytest.fixture() +def bundle(tmp_path): + bundle_dir = tmp_path / "bundle" + (bundle_dir / "templates").mkdir(parents=True) + (bundle_dir / "templates" / "btn.png").write_bytes(make_png((50, 20))) + return bundle_dir + + +def test_embedded_subroutine_is_a_valid_program(): + sub = find_record(exit_checks=(text("Record"),)) + workflow = caller(sub) + key = subflow_key(sub) + assert key == f"{SUBFLOW_PREFIX}find_record/{sub.sha256}" + assert parse_subflow_key(key) == ("find_record", sub.sha256) + report = validate_workflow(workflow) + structure = [i for i in report.issues if i.category == "structure"] + assert structure == [] + # Re-embedding the same content is idempotent; different content refuses. + again, _ = embed(workflow, sub) + assert again.subflows[key] == workflow.subflows[key] + tampered = workflow.model_copy(deep=True) + tampered.subflows[key].states["focus"].step.key = "Z" + with pytest.raises(SubroutineError, match="different body"): + embed(tampered, sub) + + +def test_malformed_subroutine_key_is_refused(): + with pytest.raises(SubroutineError): + parse_subflow_key(f"{SUBFLOW_PREFIX}find_record/short") + assert parse_subflow_key("plain") is None + + +def _run(workflow, bundle, run_dir, screen_text): + vision = FakeVision() + vision.text_results = {t: Match((10, 10), (0, 0, 5, 5)) for t in screen_text} + backend = FakeBackend() + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + workflow, params={"query": "abc"}, bundle_dir=bundle, run_dir=run_dir + ) + return report, backend + + +def test_runtime_runs_body_only_when_entry_check_holds(bundle, tmp_path): + workflow = caller(find_record()) + report, backend = _run(workflow, bundle, tmp_path / "ok", {"Find record"}) + assert report.success is True + assert [a[0] for a in backend.actions] == ["press", "type", "press"] + + report, backend = _run(workflow, bundle, tmp_path / "wrong", set()) + assert report.success is False + assert report.terminal_outcome == "halt" + assert backend.actions == [] + + +def test_runtime_halts_when_exit_check_fails(bundle, tmp_path): + workflow = caller(find_record(exit_checks=(text("Record"),))) + report, backend = _run(workflow, bundle, tmp_path / "ok", {"Find record", "Record"}) + assert report.success is True + report, backend = _run(workflow, bundle, tmp_path / "bad", {"Find record"}) + assert report.success is False + assert report.terminal_outcome == "halt" + # The body ran; returning without the exit check is what halted. + assert len(backend.actions) == 3 + + +# -- library -------------------------------------------------------------------- + + +def test_library_versions_requalification_and_no_going_back(): + v1 = find_record() + a, b = caller(v1, name="intake"), caller(v1, name="referral") + other = caller(find_record(name="pick_date"), name="schedule") + library = ready_library(v1, a, b) + library.publish(find_record(name="pick_date")) + library.register("schedule", other) + assert library.dependents("find_record") == ("intake", "referral") + + again = library.publish(v1) + assert again.version == 1 and again.stale_uses == () + + v2 = find_record(entry_checks=(text("Search"),)) + plan = library.publish(v2) + assert plan.version == 2 + assert plan.workflows == ("intake", "referral") + assert {u.sha256 for u in plan.stale_uses} == {v1.sha256} + assert library.current("find_record").sha256 == v2.sha256 + assert library.versions("find_record") == (v1.sha256, v2.sha256) + + with pytest.raises(SubroutineError, match="older version"): + library.publish(v1) + with pytest.raises(SubroutineError, match="lacks"): + library.qualify("0" * 64, policy_name="standard", evidence_sha256=EVIDENCE) + + +def test_library_round_trips_and_refuses_edits(): + sub = find_record() + workflow = caller(sub) + library = ready_library(sub, workflow) + text_doc = library.to_json() + loaded = SubroutineLibrary.from_json(text_doc) + assert loaded.to_json() == text_doc + assert admit_workflow(workflow, loaded).admitted + + edited = json.loads(text_doc) + body = edited["subroutines"][sub.sha256]["body"] + body["states"]["focus"]["step"]["key"] = "Z" + with pytest.raises(SubroutineError, match="does not match its digest"): + SubroutineLibrary.from_json(json.dumps(edited)) + + orphan = json.loads(text_doc) + orphan["versions"] = {} + with pytest.raises(SubroutineError, match="not in any version list"): + SubroutineLibrary.from_json(json.dumps(orphan)) + + stray = json.loads(text_doc) + stray["qualifications"][0]["sha256"] = "1" * 64 + with pytest.raises(SubroutineError, match="unknown digest"): + SubroutineLibrary.from_json(json.dumps(stray)) + + with pytest.raises(SubroutineError, match="not a subroutine library"): + SubroutineLibrary.from_json(json.dumps({"schema": "other"})) + + +# -- admission ---------------------------------------------------------------------- + + +def test_admits_a_registered_qualified_current_use(): + sub = find_record() + workflow = caller(sub) + admission = admit_workflow(workflow, ready_library(sub, workflow)) + assert admission.admitted, admission.refusals + assert [u.sha256 for u in admission.uses] == [sub.sha256] + + +def test_workflow_without_subroutines_is_unaffected(): + plain = Workflow(name="plain", steps=[key_step("a", "A")]) + admission = admit_workflow(plain, SubroutineLibrary()) + assert admission.admitted and admission.uses == () + + +def test_refuses_unknown_unqualified_and_unregistered(): + sub = find_record() + workflow = caller(sub) + assert codes(admit_workflow(workflow, SubroutineLibrary())) == { + "unknown_subroutine" + } + + library = SubroutineLibrary() + library.publish(sub) + assert codes(admit_workflow(workflow, library)) == { + "not_qualified", + "unregistered_use", + } + library.qualify(sub.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + assert codes(admit_workflow(workflow, library)) == {"unregistered_use"} + library.register(workflow.name, workflow) + assert admit_workflow(workflow, library).admitted + + +def test_changing_a_subroutine_requalifies_every_user(): + v1 = find_record() + a, b = caller(v1, name="intake"), caller(v1, name="referral") + library = ready_library(v1, a, b) + assert admit_workflow(a, library).admitted and admit_workflow(b, library).admitted + + v2 = find_record(exit_checks=(text("Record"),)) + plan = library.publish(v2) + library.qualify(v2.sha256, policy_name="standard", evidence_sha256=EVIDENCE) + assert plan.workflows == ("intake", "referral") + for workflow in (a, b): + assert codes(admit_workflow(workflow, library)) == {"superseded"} + + # Re-embedding v2 alone is not enough: the workflow must be registered + # again (re-qualified) for the new version. + a2 = caller(v2, name="intake") + assert codes(admit_workflow(a2, library)) == {"unregistered_use"} + library.register("intake", a2) + assert admit_workflow(a2, library).admitted + assert codes(admit_workflow(b, library)) == {"superseded"} + + +def test_refuses_an_edited_embedded_body(): + sub = find_record() + workflow = caller(sub) + library = ready_library(sub, workflow) + key = subflow_key(sub) + edited = workflow.model_copy(deep=True) + edited.subflows[key].states["go"].step.key = "Tab" + assert codes(admit_workflow(edited, library)) == {"embedded_drift"} + skipped = workflow.model_copy(deep=True) + skipped.subflows[key].entry = "focus" # bypass the entry check + assert codes(admit_workflow(skipped, library)) == {"embedded_drift"} + + +def test_refuses_surface_mismatches(): + sub = find_record() + for surface, expected in ( + ("windows", "surface_kind_mismatch"), + (None, "surface_undeclared"), + ): + workflow = caller(sub, surface=surface) + library = ready_library(sub, workflow) + assert codes(admit_workflow(workflow, library)) == {expected} + multi = caller(sub, surfaces=[SurfaceSpec(id="main", kind="web")]) + assert codes(admit_workflow(multi, ready_library(sub, multi))) == {"multi_surface"} + + +def test_refuses_inputs_the_caller_cannot_supply(): + dated = find_record(inputs=(SubroutineInput(name="query", type=ParamKind.DATE),)) + + def verdict(sub, **fields): + workflow = caller(sub, **fields) + return codes(admit_workflow(workflow, ready_library(sub, workflow))) + + assert verdict(find_record(), param_specs={}) == {"missing_input"} + assert verdict(dated) == {"input_type_mismatch"} + assert verdict(dated, param_specs={}, params={"query": "2026-01-01"}) == { + "input_type_mismatch" + } + assert ( + verdict( + dated, + param_specs={ + "query": ParamSpec(name="query", type=ParamKind.DATE, example="d") + }, + ) + == set() + ) + assert verdict( + find_record(), + param_specs={"query": ParamSpec(name="query", required=False)}, + ) == {"missing_input"} + assert verdict(find_record(), secret_params=["query"]) == {"secret_input"} + + enum_sub = find_record( + inputs=(SubroutineInput(name="query", type=ParamKind.ENUM, choices=("a", "b")),) + ) + wide = { + "query": ParamSpec( + name="query", type=ParamKind.ENUM, choices=["a", "c"], example="a" + ) + } + narrow = { + "query": ParamSpec( + name="query", type=ParamKind.ENUM, choices=["a"], example="a" + ) + } + assert verdict(enum_sub, param_specs=wide) == {"input_choices_mismatch"} + assert verdict(enum_sub, param_specs=narrow) == set() + + +def test_refusal_messages_do_not_echo_screen_text(): + sub = find_record( + steps=[type_step("enter", "query"), anchored_click("pick", "Row Label Marker")], + entry_checks=(text("Sample Clinic"),), + ) + workflow = caller(sub, surface="windows") + edited = workflow.model_copy(deep=True) + edited.subflows[subflow_key(sub)].states["pick"].step.key = "x" + admission = admit_workflow(edited, SubroutineLibrary()) + library = ready_library(sub) + admission2 = admit_workflow(edited, library) + dumped = admission.model_dump_json() + admission2.model_dump_json() + assert "Marker" not in dumped and "Sample" not in dumped + + +# -- detection ------------------------------------------------------------------------ + + +def _linear(name, *steps): + return Workflow(name=name, surface="web", steps=list(steps)) + + +def test_mines_a_sequence_shared_across_workflows_by_shape_only(): + # Same shape, different labels: detection never compares labels. + a = _linear( + "a", + key_step("a0", "Escape"), + anchored_click("a1", "Search patients", role="searchbox"), + type_step("a2", "name"), + key_step("a3", "Enter"), + anchored_click("a4", "Done", role="button"), + ) + b = _linear( + "b", + anchored_click("b1", "Find a person", role="searchbox"), + type_step("b2", "who"), + key_step("b3", "Enter"), + key_step("b4", "Tab"), + ) + candidates = mine_recurring_sequences({"a": a, "b": b}) + assert len(candidates) == 1 + (candidate,) = candidates + assert candidate.admitted is False and candidate.runnable is False + assert candidate.workflows == ("a", "b") + assert len(candidate.shape) == 3 + assert {o.step_ids for o in candidate.occurrences} == { + ("a1", "a2", "a3"), + ("b1", "b2", "b3"), + } + assert {o.params for o in candidate.occurrences} == {("name",), ("who",)} + blob = json.dumps(candidate.model_dump(mode="json")) + assert "Search" not in blob and "Find" not in blob + + +def test_repeats_inside_one_workflow_are_not_enough(): + steps = [ + type_step(f"s{i}", "q") if i % 2 else key_step(f"s{i}", "Enter") + for i in range(8) + ] + assert mine_recurring_sequences({"only": _linear("only", *steps)}) == [] + with pytest.raises(SubroutineError): + mine_recurring_sequences({}, min_workflows=1) + + +def test_mines_straight_runs_inside_program_graphs(): + sub = find_record() + a, b = caller(sub, name="a"), caller(sub, name="b") + candidates = mine_recurring_sequences({"a": a, "b": b}) + assert candidates and candidates[0].workflows == ("a", "b") + assert [step_shape(s.step) for s in sub.body.states.values() if s.step] == list( + candidates[0].shape + ) + + +def test_wrapped_body_and_problems_are_deterministic(): + sub = find_record(exit_checks=(text("Record"),)) + assert wrapped_body(sub) == wrapped_body(sub) + assert body_problems(sub) == []