From e752011b469fa23e56596f0ab456f8d414d1e8c0 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 13:17:11 -0500 Subject: [PATCH 1/4] Add typed reusable subroutines with fail-closed admission (part of #1892) A subroutine is a typed signature (inputs, surface kind, entry and exit checks) plus a ProgramGraph body, identified by the SHA-256 of both. Workflows embed it under subroutine//, wrapped so the existing interpreter halts when an entry or exit check does not hold. The library keeps versions, qualifications and registered users. Publishing a new version lists every workflow pinned to an older one, and admission refuses those uses until they are re-embedded and registered again. Admission also refuses unknown, unqualified or unregistered digests, edited embedded bodies, surface and input mismatches. Detection proposes recurring sequences by step shape only, never by labels or application names. --- packages/flow/docs/design/SUBROUTINES.md | 81 ++ packages/flow/openadapt_flow/subroutines.py | 1084 +++++++++++++++++++ packages/flow/tests/test_subroutines.py | 661 +++++++++++ 3 files changed, 1826 insertions(+) create mode 100644 packages/flow/docs/design/SUBROUTINES.md create mode 100644 packages/flow/openadapt_flow/subroutines.py create mode 100644 packages/flow/tests/test_subroutines.py diff --git a/packages/flow/docs/design/SUBROUTINES.md b/packages/flow/docs/design/SUBROUTINES.md new file mode 100644 index 0000000000..af5269a1d9 --- /dev/null +++ b/packages/flow/docs/design/SUBROUTINES.md @@ -0,0 +1,81 @@ +# Reusable subroutines + +Status: first slice (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 | + +Messages name subroutines, inputs and codes only. They never repeat screen +text or recorded values. + +## 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 + +- Wiring admission and registration into `flow qualify` and certification. +- 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, on-disk library. The library is in memory with JSON + import and export. diff --git a/packages/flow/openadapt_flow/subroutines.py b/packages/flow/openadapt_flow/subroutines.py new file mode 100644 index 0000000000..eded4bb1ab --- /dev/null +++ b/packages/flow/openadapt_flow/subroutines.py @@ -0,0 +1,1084 @@ +"""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. + +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 re +from collections import defaultdict +from datetime import datetime, timezone +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") + for anchor in (step.anchor, step.drag_end_anchor): + if anchor is not None and ( + anchor.context_text is not None or anchor.structured_identity is not None + ): + problems.append(f"{where}: keeps a plaintext identity band") + return problems + + +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 check in (*sig.entry_checks, *sig.exit_checks): + refs |= _param_refs(check.model_dump(mode="json"), problems, "boundary check") + 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 + refs |= _param_refs(state.model_dump(mode="json"), problems, where) + 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[ + "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) + ) + + +# -- 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") + 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_subroutines.py b/packages/flow/tests/test_subroutines.py new file mode 100644 index 0000000000..3f051617e9 --- /dev/null +++ b/packages/flow/tests/test_subroutines.py @@ -0,0 +1,661 @@ +"""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 + +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 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) == [] From 12608874eebc0c854e0afe551dbf119ef591df05 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 20:36:46 -0500 Subject: [PATCH 2/4] Subroutines: refuse identity bands in every predicate anchor and repeated step ids (part of #1892) Addresses the two P2 findings in the Codex review of #1987: - body_problems now walks every state and boundary check generically for a non-empty Anchor.context_text / structured_identity (step guards, waits, transition guards, entry/exit checks and nested operands). Refusals name the field path only, never the band text. - subroutine_from_steps rejects repeated step ids instead of letting a later step silently replace an earlier one. 15 new regression tests; all fail on e752011b4. --- packages/flow/openadapt_flow/subroutines.py | 54 +++++++-- packages/flow/tests/test_subroutines.py | 122 ++++++++++++++++++++ 2 files changed, 168 insertions(+), 8 deletions(-) diff --git a/packages/flow/openadapt_flow/subroutines.py b/packages/flow/openadapt_flow/subroutines.py index eded4bb1ab..c84b4e34b4 100644 --- a/packages/flow/openadapt_flow/subroutines.py +++ b/packages/flow/openadapt_flow/subroutines.py @@ -333,14 +333,37 @@ def _step_problems(step: Step, where: str) -> list[str]: or step.closes_surface ): problems.append(f"{where}: changes or names a surface") - for anchor in (step.anchor, step.drag_end_anchor): - if anchor is not None and ( - anchor.context_text is not None or anchor.structured_identity is not None - ): - problems.append(f"{where}: keeps a plaintext identity band") 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). @@ -354,8 +377,14 @@ def body_problems(subroutine: Subroutine) -> list[str]: if body.entry not in states: problems.append("the body's entry state is not defined") refs: set[str] = set() - for check in (*sig.entry_checks, *sig.exit_checks): - refs |= _param_refs(check.model_dump(mode="json"), problems, "boundary check") + 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}" @@ -381,7 +410,10 @@ def body_problems(subroutine: Subroutine) -> list[str]: problems.extend(_step_problems(state.step, where)) if state.kind is StateKind.TERMINAL and state.outcome in (None, "success"): has_success = True - refs |= _param_refs(state.model_dump(mode="json"), problems, where) + 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} @@ -1066,6 +1098,12 @@ def subroutine_from_steps( 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): diff --git a/packages/flow/tests/test_subroutines.py b/packages/flow/tests/test_subroutines.py index 3f051617e9..8aa4a08947 100644 --- a/packages/flow/tests/test_subroutines.py +++ b/packages/flow/tests/test_subroutines.py @@ -12,6 +12,7 @@ from __future__ import annotations import json +from functools import partial import pytest from pydantic import ValidationError @@ -254,6 +255,127 @@ def test_body_refuses_a_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")]) From 0e2bf83e5f3fd0e761674faf89cbc1ca66b8ef28 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 20:48:27 -0500 Subject: [PATCH 3/4] Certification runs the subroutine gate: register, then admit (part of #1892) certify_project (flow qualify certify, flow qualify --certify) now runs certify_subroutine_uses on every workflow: - no subroutine/ subflows: unchanged, library neither read nor written; - embedded subroutine without a library: refused (subroutine_not_admitted / library_missing); the campaign form refuses before running any case; - with a library: register the workflow as the user of the exact digests it embeds, then admit_workflow; every refusal maps to subroutine_not_admitted with details.subroutine_code. CLI: --subroutine-library JSON on qualify certify and the campaign, plus --subroutine-workflow-id on certify; the library is written back atomically when registration changed it. A passing report gains nothing, so current_certification_matches reproduces it unchanged. 15 tests in test_subroutine_certification.py, including a real signed campaign; the 7 wiring tests fail on the previous head. --- packages/flow/docs/QUALIFICATION_PROJECT.md | 11 + packages/flow/docs/design/SUBROUTINES.md | 39 +- packages/flow/openadapt_flow/__main__.py | 57 +++ packages/flow/openadapt_flow/qualification.py | 52 ++- .../flow/openadapt_flow/qualify/campaign.py | 19 +- packages/flow/openadapt_flow/subroutines.py | 95 +++++ .../tests/test_subroutine_certification.py | 382 ++++++++++++++++++ 7 files changed, 649 insertions(+), 6 deletions(-) create mode 100644 packages/flow/tests/test_subroutine_certification.py 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 index af5269a1d9..d0d8bf2f4c 100644 --- a/packages/flow/docs/design/SUBROUTINES.md +++ b/packages/flow/docs/design/SUBROUTINES.md @@ -1,6 +1,6 @@ # Reusable subroutines -Status: first slice (OpenAdapt#1892). Code: `openadapt_flow/subroutines.py`. +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 @@ -58,10 +58,36 @@ refused until they embed the new version and are registered again. | `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. + +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 @@ -72,10 +98,15 @@ or runnable. ## Not in this slice -- Wiring admission and registration into `flow qualify` and certification. +- 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, on-disk library. The library is in memory with JSON - import and export. +- 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..a83836d785 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -7086,19 +7086,62 @@ 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), ) save_qualified_workflow(workflow, args.bundle) + _save_subroutine_library(library, library_path, library_before) print(report.model_dump_json(indent=2) if args.json else report.render()) return 0 if report.passed else 2 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 +9017,9 @@ 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_before = library.to_json() if library is not None else "" try: result = qualify( bundle, @@ -8984,13 +9030,16 @@ def _cmd_qualify_campaign(args: argparse.Namespace) -> int: certify=certify, run_params=run_params, campaign_id=args.campaign_id, + subroutine_library=library, ) except QualifyConfigError as exc: print(str(exc), file=sys.stderr) return 2 except QualifyError as exc: + _save_subroutine_library(library, library_path, library_before) print(f"qualify REFUSED: {exc}\nEvidence so far: {out}", file=sys.stderr) return 1 + _save_subroutine_library(library, library_path, library_before) if not result.passed and result.partial: print( f"\nqualify PARTIAL: {len(result.rows)} trial(s) behaved as required in " @@ -10991,6 +11040,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 +11501,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..b8febec708 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 diff --git a/packages/flow/openadapt_flow/qualify/campaign.py b/packages/flow/openadapt_flow/qualify/campaign.py index e52dcaa7a2..73111c4f05 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,7 @@ def qualify( in_place: bool = False, write_campaign_files: bool = True, log: Callable[[str], None] = print, + subroutine_library: Optional[SubroutineLibrary] = None, ) -> QualifyResult: """Run the campaign; certify under ``certify`` (a policy name) if given. @@ -1260,6 +1262,9 @@ 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. """ from openadapt_flow.policy import load_policy @@ -1310,6 +1315,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 +1665,10 @@ 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, ) else: qreport = evaluate_qualification(workflow, evidence_root=root) diff --git a/packages/flow/openadapt_flow/subroutines.py b/packages/flow/openadapt_flow/subroutines.py index c84b4e34b4..1cd67049dc 100644 --- a/packages/flow/openadapt_flow/subroutines.py +++ b/packages/flow/openadapt_flow/subroutines.py @@ -55,6 +55,14 @@ 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 @@ -71,9 +79,12 @@ 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 @@ -742,6 +753,7 @@ def from_json(cls, text: str) -> "SubroutineLibrary": # -- admission -------------------------------------------------------------------- RefusalCode = Literal[ + "library_missing", "malformed_key", "unknown_subroutine", "name_mismatch", @@ -936,6 +948,89 @@ def refuse(code: RefusalCode, key: str, message: str) -> None: ) +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 ---------------------------------------------------------------------- diff --git a/packages/flow/tests/test_subroutine_certification.py b/packages/flow/tests/test_subroutine_certification.py new file mode 100644 index 0000000000..87fd70e814 --- /dev/null +++ b/packages/flow/tests/test_subroutine_certification.py @@ -0,0 +1,382 @@ +"""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() From a92e6f42f7d80253cedf3315d8118bae3c6abb3a Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 22:00:39 -0500 Subject: [PATCH 4/4] Certification writes the subroutine registration before the certificate (part of #1892) flow qualify certify saved the certified bundle, then wrote the library; the campaign (flow qualify --certify) saved it inside qualify() before the CLI wrote the library. With a read-only library directory the CLI raised PermissionError but left a bundle that passed current_certification_matches while the library listed no dependent, so a new subroutine version could not put it back into re-qualification. persist_registration_before_certificate() runs the library write right after certify_project and before any certified save. On failure it withdraws the certification, saves the bundle uncertified and re-raises. Both CLI paths use it; the campaign takes it as qualify(persist_subroutine_registration=...) and writes the library once. 4 regression tests (fail on the previous head): read-only library directory for certify and for the campaign leaves no valid certificate and an unchanged library; on success the library is written before the certificate. --- packages/flow/docs/design/SUBROUTINES.md | 6 +- packages/flow/openadapt_flow/__main__.py | 26 +- packages/flow/openadapt_flow/qualification.py | 25 ++ .../flow/openadapt_flow/qualify/campaign.py | 10 + .../tests/test_subroutine_certification.py | 229 ++++++++++++++++++ 5 files changed, 291 insertions(+), 5 deletions(-) diff --git a/packages/flow/docs/design/SUBROUTINES.md b/packages/flow/docs/design/SUBROUTINES.md index d0d8bf2f4c..b46e7c5233 100644 --- a/packages/flow/docs/design/SUBROUTINES.md +++ b/packages/flow/docs/design/SUBROUTINES.md @@ -79,7 +79,11 @@ text or recorded values. `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. + 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 diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index a83836d785..ebc334dd51 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -7097,8 +7097,19 @@ def _cmd_qualify(args: argparse.Namespace) -> int: 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) - _save_subroutine_library(library, library_path, library_before) print(report.model_dump_json(indent=2) if args.json else report.render()) return 0 if report.passed else 2 @@ -9019,7 +9030,13 @@ def _cmd_qualify_campaign(args: argparse.Namespace) -> int: ) library_path = getattr(args, "subroutine_library", None) library = _load_subroutine_library(library_path) - library_before = library.to_json() if library is not None else "" + 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, @@ -9031,15 +9048,16 @@ def _cmd_qualify_campaign(args: argparse.Namespace) -> int: 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: - _save_subroutine_library(library, library_path, library_before) + persist_library() print(f"qualify REFUSED: {exc}\nEvidence so far: {out}", file=sys.stderr) return 1 - _save_subroutine_library(library, library_path, library_before) + persist_library() if not result.passed and result.partial: print( f"\nqualify PARTIAL: {len(result.rows)} trial(s) behaved as required in " diff --git a/packages/flow/openadapt_flow/qualification.py b/packages/flow/openadapt_flow/qualification.py index b8febec708..37b0002a4d 100644 --- a/packages/flow/openadapt_flow/qualification.py +++ b/packages/flow/openadapt_flow/qualification.py @@ -5743,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 73111c4f05..0f8a591cf5 100644 --- a/packages/flow/openadapt_flow/qualify/campaign.py +++ b/packages/flow/openadapt_flow/qualify/campaign.py @@ -1250,6 +1250,7 @@ def qualify( 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. @@ -1265,6 +1266,9 @@ def qualify( ``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 @@ -1272,6 +1276,7 @@ def qualify( QualificationCaseKind, certify_project, evaluate_qualification, + persist_registration_before_certificate, record_case_results, save_qualified_workflow, ) @@ -1670,6 +1675,11 @@ def authorize( 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) if certify is not None and qreport.passed and config.business_rules: diff --git a/packages/flow/tests/test_subroutine_certification.py b/packages/flow/tests/test_subroutine_certification.py index 87fd70e814..116d65ed29 100644 --- a/packages/flow/tests/test_subroutine_certification.py +++ b/packages/flow/tests/test_subroutine_certification.py @@ -380,3 +380,232 @@ def test_campaign_certify_refuses_before_running_without_a_library(tmp_path: Pat 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",)