Reusable subroutines: typed unit, fail-closed admission, certification wiring (part of #1892)

packages/flow/docs/QUALIFICATION_PROJECT.md

@@ -144,6 +144,17 @@ openadapt-flow qualify certify bundle \
   --json
 ```
 
+A bundle that embeds reusable subroutines (`subroutine/<name>/<sha256>`
+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 <bundle>
+--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,

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/<name>/<sha256>` 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 <bundle>
+--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.

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(

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,15 +5436,53 @@ 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",
     *,
     policy: "Policy",
     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/<name>/<sha256>``
+    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,