From f26e32b914fe6958d1348893d37873f33bfea359 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 12:40:49 -0400 Subject: [PATCH 1/4] WIP: record-selecting clicks follow the run's value (Step.target_param) Adds Step.target_param (dropped from serialization when None, so existing bundle digests and contract hashes are unchanged) and runtime.target_param: a click that selects a record looks for the run's value instead of the demonstrated label, drops the demonstrated crop, and halts before acting when the value is empty or the identity evidence does not embed the parameter. Loop induction (search-process-return, row-varying bodies) is not written yet. Untested. Co-Authored-By: Claude Opus 5.5 --- packages/flow/openadapt_flow/ir.py | 33 ++++ .../flow/openadapt_flow/runtime/replayer.py | 19 ++ .../openadapt_flow/runtime/target_param.py | 167 ++++++++++++++++++ 3 files changed, 219 insertions(+) create mode 100644 packages/flow/openadapt_flow/runtime/target_param.py diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index aee7c6ba8..7cad0b997 100644 --- a/packages/flow/openadapt_flow/ir.py +++ b/packages/flow/openadapt_flow/ir.py @@ -1636,6 +1636,19 @@ class Step(BaseModel): "never replayed directly." ), ) + target_param: Optional[str] = Field( + default=None, + description=( + "CLICK/DOUBLE_CLICK only: the run parameter whose value names the" + " record this click selects (a search result, a worklist row)." + " When the run's value differs from the demonstrated one, the" + " runtime looks for the run's value instead of the demonstrated" + " label and skips the demonstrated crop; the identity check then" + " verifies the row in parameter mode. A step without identity" + " evidence, or with an empty value, halts before acting. None" + " (the default) clicks the demonstrated target, as before." + ), + ) key: Optional[str] = None # for KEY/HOTKEY, e.g. "Enter" or "s" modifiers: list[str] = Field( default_factory=list, @@ -1801,6 +1814,26 @@ def _validate_surface_contract(self) -> "Step": ) return self + @model_serializer(mode="wrap") + def _serialize_compatible(self, handler: Any) -> dict[str, Any]: + """Keep steps without a selected-record parameter byte-identical.""" + data: dict[str, Any] = handler(self) + if self.target_param is None: + data.pop("target_param", None) + return data + + @model_validator(mode="after") + def _validate_target_param(self) -> "Step": + if self.target_param is None: + return self + if self.action not in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK): + raise ValueError("target_param is valid only on CLICK/DOUBLE_CLICK") + if self.anchor is None: + raise ValueError("a target_param click needs an anchor") + if not self.target_param.strip(): + raise ValueError("target_param must name a parameter") + return self + @model_validator(mode="after") def _validate_rich_action_contract(self) -> "Step": if self.action is not ActionKind.DRAG and self.drag_end_anchor is not None: diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 49681fc5d..461f060f3 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -7283,6 +7283,21 @@ def _run_step_timed( if missed is not None and missed[0] == step.id: self._scroll_missed_target = None target_missed = missed[1] + if step.target_param is not None: + # A record-selecting click acts on the run's record: look for the + # run's value; the identity check verifies the row (param mode). + from openadapt_flow.runtime.target_param import bind_target_param + + step, refusal = bind_target_param(step, params, workflow) + if refusal is not None: + return StepResult( + step_id=step.id, + intent=step.intent, + ok=False, + error=refusal, + safety_halt=True, + failure_category="safety_halt", + ) router = self._surface_router if router is None: if step.action is ActionKind.SWITCH_SURFACE: @@ -17784,6 +17799,10 @@ def _asset_bytes( workflow: Optional[Workflow] = None, ) -> Optional[bytes]: """Return the verified run snapshot for governed asset consumers.""" + if not rel: + # No crop: a record-selecting click rebound to another record + # (runtime.target_param) carries none. + return None if self.governed_authorization is not None: data = self._governed_asset_snapshot.get(rel) expected = self._governed_asset_hashes.get(rel) diff --git a/packages/flow/openadapt_flow/runtime/target_param.py b/packages/flow/openadapt_flow/runtime/target_param.py new file mode 100644 index 000000000..0981e8005 --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/target_param.py @@ -0,0 +1,167 @@ +"""Point a record-selecting click at the run's record (``Step.target_param``). + +A demonstrated click on a search result or a worklist row targets the row the +person clicked. When a run (or a loop iteration) works on a different record, +the click must land on that record's row instead. ``Step.target_param`` names +the run parameter whose value identifies the row. + +:func:`bind_target_param` returns the step the runtime acts on: + +- the run's value equals the demonstrated one: the step is unchanged; +- otherwise the anchor looks for the run's value: its OCR label becomes the + value, the demonstrated crop is dropped (it shows the demonstrated row), and + a structural locator is kept only when its accessible name was exactly the + demonstrated value (a CSS selector or automation id points at the old row, + so it is dropped). + +It refuses (the step halts before any input) when: + +- the run has no value, or an empty one, for the parameter; +- the anchor carries no identity evidence, or its identity evidence does not + embed the parameter: the identity check could then not tell the run's + record from the demonstrated one; +- the identity band embeds another parameter the run gives no value for. + +The identity check that runs before every click then verifies the live row in +parameter mode: the run's value must be in the row, and the rest of the row +must match the recorded row with every embedded parameter substituted. No +model calls. +""" + +from __future__ import annotations + +import re +from typing import Mapping, Optional + +from openadapt_flow.ir import Anchor, Step, Workflow + +#: Template path used for a rebound anchor: no crop, so the template rungs are +#: skipped (the recorded crop shows the demonstrated row). +NO_TEMPLATE = "" + + +def _text(value: object) -> str: + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + return str(value) + + +def _squash(text: str) -> str: + return "".join(text.casefold().split()) + + +def demo_value(workflow: Workflow, name: str) -> str: + """The demonstrated value of ``name`` (empty when there is none).""" + + if name in workflow.params: + return _text(workflow.params[name]) + spec = workflow.param_specs.get(name) + if spec is not None and spec.example is not None: + return _text(spec.example) + return "" + + +def embedded_identity_params(anchor: Anchor, examples: Mapping[str, str]) -> set[str]: + """Parameters whose demonstrated value the anchor's identity band embeds.""" + + from openadapt_flow.runtime import identity as _id + + names: set[str] = set() + tmpl = anchor.identity_template + if tmpl is not None: + names.update(tmpl.param_token_indices) + names.update(tmpl.structured_params) + names.update(getattr(tmpl, "context_params", None) or []) + for band in (anchor.context_text, anchor.structured_identity): + if band: + names.update( + _id.embedded_params(band, {k: v for k, v in examples.items() if v}) + ) + return names + + +def has_identity_evidence(anchor: Anchor) -> bool: + tmpl = anchor.identity_template + return bool( + anchor.context_text + or anchor.structured_identity + or (tmpl is not None and (tmpl.tokens or tmpl.structured)) + ) + + +def bind_target_param( + step: Step, + params: Mapping[str, object], + workflow: Workflow, +) -> tuple[Step, Optional[str]]: + """Return ``(step to act on, refusal)``. A refusal means: halt, no input.""" + + name = step.target_param + if name is None: + return step, None + anchor = step.anchor + if anchor is None: # the IR validator already refuses this + return step, f"Step '{step.id}' selects a record but has no anchor" + value = _text(params.get(name)).strip() + if not value: + return step, ( + f"Step '{step.id}' ({step.intent}) selects the record named by " + f"parameter '{name}', but this run gives no value for it; " + "refusing to click" + ) + if not has_identity_evidence(anchor): + return step, ( + f"Step '{step.id}' ({step.intent}) selects a record by parameter " + f"'{name}' but carries no identity evidence to verify the row; " + "refusing to click" + ) + examples = { + key: demo_value(workflow, key) + for key in set(workflow.params) | set(workflow.param_specs) + } + embedded = embedded_identity_params(anchor, examples) + if name not in embedded: + return step, ( + f"Step '{step.id}' ({step.intent}) selects a record by parameter " + f"'{name}', but its identity evidence does not contain that " + "parameter, so the row could not be verified; refusing to click" + ) + missing = sorted( + other + for other in embedded + if not _text(params.get(other)).strip() and not examples.get(other) + ) + if missing: + return step, ( + f"Step '{step.id}' ({step.intent}) verifies the row by " + f"{', '.join(repr(m) for m in missing)}, which this run does not " + "supply; refusing to click" + ) + demo = examples.get(name, "") + if demo and _squash(value) == _squash(demo): + return step, None + + ocr_text = value + structural = None + if anchor.structural is not None and anchor.structural.name and demo: + if _squash(anchor.structural.name) == _squash(demo): + structural = anchor.structural.model_copy( + update={"name": value, "selector": None, "automation_id": None} + ) + if anchor.ocr_text and demo and _squash(anchor.ocr_text) != _squash(demo): + # Keep the label's own words when the demonstrated value is a whole + # word run inside it ("Open Alice" -> "Open Bob"); otherwise look for + # the value alone. + pattern = re.compile(re.escape(demo), re.IGNORECASE) + if len(pattern.findall(anchor.ocr_text)) == 1: + ocr_text = pattern.sub(lambda _m: value, anchor.ocr_text) + bound_anchor = anchor.model_copy( + update={ + "template": NO_TEMPLATE, + "ocr_text": ocr_text, + "structural": structural, + } + ) + return step.model_copy(update={"anchor": bound_anchor}), None From bff0a447fd01738692a23fe87c4dcb7e7feb140c Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 13:47:15 -0400 Subject: [PATCH 2/4] flow: test record-selecting clicks; require a verified row, never heal from it A target_param click now halts on an unreadable or abstained identity even when the step is reversible, skips the hover shortcut, and never heals its anchor from another record's row. Adds loop_shapes helpers (not wired yet). Co-Authored-By: Claude Opus 5.5 --- .../openadapt_flow/compiler/loop_shapes.py | 436 ++++++++++++++++++ .../flow/openadapt_flow/runtime/replayer.py | 12 +- packages/flow/tests/test_target_param.py | 252 ++++++++++ 3 files changed, 698 insertions(+), 2 deletions(-) create mode 100644 packages/flow/openadapt_flow/compiler/loop_shapes.py create mode 100644 packages/flow/tests/test_target_param.py diff --git a/packages/flow/openadapt_flow/compiler/loop_shapes.py b/packages/flow/openadapt_flow/compiler/loop_shapes.py new file mode 100644 index 000000000..e8d089f65 --- /dev/null +++ b/packages/flow/openadapt_flow/compiler/loop_shapes.py @@ -0,0 +1,436 @@ +"""Loops that are not back-to-back repeats, and their per-item checks. + +:mod:`openadapt_flow.compiler.induction` collapses a body that repeats step +for step. Real worklists rarely look like that: + +- **search, open, act, return.** Each row is found, opened, worked and left + ("Back to list"). The last row usually ends without the return, so the + repeats are not identical. +- **row-varying bodies.** A field is filled for some rows and left empty for + others; a step appears only for rows in some state. + +This module holds the deterministic, model-free pieces induction uses for +those shapes: + +- :func:`segment` finds the iterations that start at a position: chunks that + begin with the same step and share a core, the last of which may stop + before the trailing return; +- :func:`merge` lines up the iterations' steps into body positions (present + in every iteration, or only in some); +- :func:`band_identity` decides whether a record-selecting click can verify + each row's record, and which extra row columns that needs; +- :func:`bind_effect_columns` rewrites a write's record check so each row is + checked against its own values, not the demonstrated row's. + +Nothing here emits a program or decides certification; induction does. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from difflib import SequenceMatcher +from typing import Any, Iterator, Optional, Sequence + +from openadapt_flow.ir import Anchor + +#: A full iteration may carry this many steps outside the shared core before +#: the chunk is taken for unrelated work (scaled up for long bodies). +MAX_EXTRA_STEPS = 3 + + +@dataclass +class Position: + """One body position: its signature and where each iteration has it.""" + + sig: Any + at: dict[int, int] = field(default_factory=dict) # iteration -> index + + +def merge(seqs: Sequence[Sequence[Any]]) -> list[Position]: + """Line up several signature sequences into ordered body positions. + + The first sequence seeds the positions; each later one is folded in with + a ``difflib`` alignment. A replaced run keeps both sides as separate + positions (each present only where it was seen). + """ + + if not seqs: + return [] + positions = [Position(sig=s, at={0: j}) for j, s in enumerate(seqs[0])] + for t in range(1, len(seqs)): + b = list(seqs[t]) + sm = SequenceMatcher(a=[p.sig for p in positions], b=b, autojunk=False) + merged: list[Position] = [] + for tag, i1, i2, j1, j2 in sm.get_opcodes(): + if tag == "equal": + for k in range(i2 - i1): + positions[i1 + k].at[t] = j1 + k + merged.append(positions[i1 + k]) + continue + if tag in ("delete", "replace"): + merged.extend(positions[i1:i2]) + if tag in ("insert", "replace"): + merged.extend(Position(sig=b[j], at={t: j}) for j in range(j1, j2)) + positions = merged + return positions + + +@dataclass +class Segmentation: + """Iterations found at one start position of one trace.""" + + #: ``(start, end)`` of each iteration, end exclusive. The last one may + #: stop before the trailing return. + chunks: list[tuple[int, int]] + #: Positions of the merged body (over the full iterations only). + positions: list[Position] + #: Indices into ``positions`` of the core shared by every full iteration. + core: list[int] + #: Indices into ``positions`` the last iteration did not reach (a suffix + #: of the core): the return to the list. + tail: list[int] + + @property + def end(self) -> int: + return self.chunks[-1][1] + + +def _match_last( + sigs: Sequence[Any], start: int, positions: list[Position], core: set[int] +) -> tuple[int, int]: + """Greedy match of the last iteration against the body. + + Returns ``(end, reached)``: the trace index after the last matched step + and the body index matching stopped at. Skipping an optional position is + allowed; skipping a core position is not (that step then ends the + iteration). + """ + + ptr = 0 + j = start + while j < len(sigs) and ptr < len(positions): + hit = None + for q in range(ptr, len(positions)): + if positions[q].sig == sigs[j]: + hit = q + break + if q in core: + break + if hit is None: + break + ptr = hit + 1 + j += 1 + return j, ptr + + +def segment(sigs: Sequence[Any], start: int) -> Iterator[Segmentation]: + """Candidate iterations starting at ``start``, most iterations first. + + Every occurrence of ``sigs[start]`` after ``start`` may begin a new + iteration. A candidate needs two or more iterations, a core of at least + two steps shared by every full iteration, few extra steps per iteration, + and a last iteration that reaches the whole core except, possibly, a + trailing suffix. Callers check the steps themselves (that the core varies + by row, that the tail only navigates). + """ + + s0 = sigs[start] + cands = [p for p in range(start, len(sigs)) if sigs[p] == s0] + for k in range(len(cands), 1, -1): + full = [(cands[m], cands[m + 1]) for m in range(k - 1)] + positions = merge([list(sigs[a:b]) for a, b in full]) + n_full = len(full) + core = [q for q, pos in enumerate(positions) if len(pos.at) == n_full] + if len(core) < 2: + continue + allowed = max(MAX_EXTRA_STEPS, len(core) // 2) + if any((b - a) - len(core) > allowed for a, b in full): + continue + last_start = cands[k - 1] + end, reached = _match_last(sigs, last_start, positions, set(core)) + tail = [q for q in core if q >= reached] + # The last iteration must reach every core position before the tail. + if any(q not in tail and q >= reached for q in core): + continue + if tail and tail != core[len(core) - len(tail) :]: + continue + if len(tail) >= len(core) - 1: + continue + yield Segmentation( + chunks=full + [(last_start, end)], + positions=positions, + core=core, + tail=tail, + ) + + +# --------------------------------------------------------------------------- +# Identity of the row a record-selecting click opens +# --------------------------------------------------------------------------- + + +@dataclass +class BandIdentity: + """Whether a record-selecting click can verify each row's record.""" + + ok: bool + reason: str = "" + #: Extra row columns the identity band needs: name -> per-iteration + #: value (None where the band was not readable for that iteration). + identity_columns: dict[str, dict[int, Optional[str]]] = field( + default_factory=dict + ) + #: Extra columns' demonstrated values (from the template iteration). + identity_examples: dict[str, str] = field(default_factory=dict) + #: Hashed bands: token indices to index under each column name. + token_indices: dict[str, list[int]] = field(default_factory=dict) + basis: str = "" + + +def _squash(text: str) -> str: + return "".join(str(text).casefold().split()) + + +def _band(anchor: Anchor) -> Optional[str]: + return anchor.structured_identity or anchor.context_text + + +def band_identity( + *, + column: str, + template_iter: int, + anchors: dict[int, Anchor], + row_values: dict[int, dict[str, str]], + same_trace: dict[int, int], + column_prefix: str, +) -> BandIdentity: + """Can the click at this body position verify the record of every row? + + Args: + column: The row column whose value selects the record. + template_iter: The iteration the body's step is taken from. + anchors: The click's anchor in each iteration that has it. + row_values: Each iteration's row column values. + same_trace: Iteration -> trace index (hashed bands compare only + within one trace, which shares one salt). + column_prefix: Prefix for extra identity column names. + + A readable band must embed the column's value; any other text that + changes from row to row becomes an identity column whose value each row + must supply. A hashed band must contain the value's tokens, carry no + hashed structured identity (that cannot follow a new row), and change + from row to row only where the row's columns explain it. + """ + + from openadapt_flow.runtime import identity as _id + + anchor = anchors[template_iter] + value = row_values[template_iter].get(column, "") + if not value: + return BandIdentity(ok=False, reason="the demonstrated row has no value") + band = _band(anchor) + if band: + if column not in _id.embedded_params(band, {column: value}): + return BandIdentity( + ok=False, + reason=( + "the row's identity text does not contain the value that " + "selects it" + ), + ) + return _readable_band_columns( + band=band, + template_iter=template_iter, + anchors=anchors, + row_values=row_values, + column_prefix=column_prefix, + ) + tmpl = anchor.identity_template + if tmpl is None or not tmpl.tokens: + return BandIdentity(ok=False, reason="the click carries no identity evidence") + if tmpl.structured and column not in tmpl.structured_params: + return BandIdentity( + ok=False, + reason=( + "its hashed structured identity names the demonstrated record " + "and cannot follow another row; induce from the recordings" + ), + ) + indices = _hashed_indices(tmpl, value) + if indices is None: + return BandIdentity( + ok=False, + reason="the row's hashed identity does not contain the selecting value", + ) + explained = { + tmpl.tokens[j].r + for j in indices + } + base = {t.r for j, t in enumerate(tmpl.tokens) if j not in indices} + for it, other in anchors.items(): + if it == template_iter or same_trace.get(it) != same_trace[template_iter]: + continue + otmpl = other.identity_template + if otmpl is None or otmpl.salt != tmpl.salt: + continue + other_value = row_values[it].get(column, "") + other_idx = _hashed_indices(otmpl, other_value) or [] + rest = {t.r for j, t in enumerate(otmpl.tokens) if j not in other_idx} + if rest != base: + return BandIdentity( + ok=False, + reason=( + "its hashed identity changes from row to row beyond the " + "selecting value (for example a date of birth), and hashed " + "text cannot name it; induce from the recordings" + ), + ) + del explained + return BandIdentity( + ok=True, token_indices={column: indices}, basis="hashed identity band" + ) + + +def _hashed_indices(tmpl: Any, value: str) -> Optional[list[int]]: + from openadapt_flow.runtime import identity as _id + from openadapt_flow.runtime import identity_template as _it + + salt = _it._salt_bytes(tmpl.salt) + out: list[int] = [] + for tok in _id.tokenize(value): + key = _it._hash(salt, _id.squash(tok)) + hits = [j for j, t in enumerate(tmpl.tokens) if t.r == key and j not in out] + if not hits: + return None + out.append(hits[0]) + return sorted(out) or None + + +def _readable_band_columns( + *, + band: str, + template_iter: int, + anchors: dict[int, Anchor], + row_values: dict[int, dict[str, str]], + column_prefix: str, +) -> BandIdentity: + """Name each run of band text that changes by row and no column explains.""" + + def tokens(text: str, values: dict[str, str]) -> list[str]: + # Replace each column value by a placeholder so explained text aligns. + out = text + for name, val in sorted(values.items(), key=lambda kv: -len(kv[1] or "")): + if val and len(_squash(val)) >= 2: + out = _replace_ci(out, val, f"\x00{name}\x00") + return out.split() + + base = tokens(band, row_values[template_iter]) + varying: set[int] = set() + aligned: dict[int, list[tuple[int, int, int, int]]] = {} + for it, other in anchors.items(): + if it == template_iter: + continue + oband = _band(other) + if not oband: + continue + other_toks = tokens(oband, row_values.get(it, {})) + ops = SequenceMatcher(a=base, b=other_toks, autojunk=False).get_opcodes() + aligned[it] = [(i1, i2, j1, j2) for tag, i1, i2, j1, j2 in ops if tag != "equal"] + for tag, i1, i2, _j1, _j2 in ops: + if tag == "insert": + return BandIdentity( + ok=False, + reason="another row's identity text has words the demonstrated row lacks", + ) + if tag != "equal": + varying.update(range(i1, i2)) + if any(base[j].startswith("\x00") for j in varying): + return BandIdentity( + ok=False, reason="a row column's value does not line up across rows" + ) + runs: list[list[int]] = [] + for j in sorted(varying): + if runs and runs[-1][-1] == j - 1: + runs[-1].append(j) + else: + runs.append([j]) + result = BandIdentity(ok=True, basis="readable identity band") + for n, run in enumerate(runs, start=1): + name = f"{column_prefix}{n}" + lo, hi = run[0], run[-1] + 1 + result.identity_examples[name] = " ".join(base[lo:hi]) + per_iter: dict[int, Optional[str]] = {template_iter: " ".join(base[lo:hi])} + for it, other in anchors.items(): + if it == template_iter: + continue + oband = _band(other) + if not oband: + per_iter[it] = None + continue + other_toks = tokens(oband, row_values.get(it, {})) + got: Optional[list[str]] = None + for i1, i2, j1, j2 in aligned.get(it, []): + if i1 <= lo and hi <= i2 and (i1, i2) == (lo, hi): + got = other_toks[j1:j2] + if got is None: + # This row's text matched the template here. + got = base[lo:hi] + per_iter[it] = " ".join(got) + result.identity_columns[name] = per_iter + return result + + +def _replace_ci(text: str, old: str, new: str) -> str: + import re + + return re.sub(re.escape(old), lambda _m: new, text, flags=re.IGNORECASE) + + +# --------------------------------------------------------------------------- +# Per-row record checks +# --------------------------------------------------------------------------- + + +def bind_effect_columns(effects: list[Any], template_values: dict[str, str]) -> tuple[list[Any], set[str]]: + """Rebind a write's record check to the row's columns. + + A literal equal to the template row's value of a column becomes a + reference to that column, so each iteration checks the record it wrote. + Returns the rebound effects and the columns they now reference + (including columns they referenced already). + """ + + from openadapt_flow.runtime.effects.effect import ValueExpr + + by_value: dict[str, list[str]] = {} + for name, val in template_values.items(): + if val: + by_value.setdefault(val, []).append(name) + used: set[str] = set() + + def rebind(expr: Any) -> Any: + if expr is None: + return None + if expr.param is not None: + used.add(expr.param) + return expr + names = by_value.get(expr.literal or "") + if names is not None and len(names) == 1: + used.add(names[0]) + return ValueExpr(param=names[0]) + return expr + + out = [] + for eff in effects: + updates: dict[str, Any] = { + "match": {k: rebind(v) for k, v in eff.match.items()}, + "new_records": [ + {k: rebind(v) for k, v in rec.items()} for rec in eff.new_records + ], + } + if eff.value is not None: + updates["value"] = rebind(eff.value) + if eff.idempotency_key is not None: + updates["idempotency_key"] = rebind(eff.idempotency_key) + out.append(eff.model_copy(update=updates, deep=True)) + return out, used diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 461f060f3..4c5be41ad 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -8663,6 +8663,10 @@ def _run_step_on_active_surface( and matched_region is not None and resolution.rung not in ("template", "structural") and step.anchor is not None + # A click rebound to the run's record (runtime.target_param) + # found another record's row; that row must never become the + # step's recorded anchor. + and not (step.target_param is not None and not step.anchor.template) ): # Heal from the PRE-action frame: that is the frame the # anchor was resolved against (the action may have navigated @@ -10228,9 +10232,11 @@ def _identity_gate_error( if qualification is not None else None ) + # A click that selects a record by a run value (Step.target_param) + # is judged by its identity band, never by hover evidence alone. hover_check = ( self._hover_check(step, resolution, frame_png, workflow, bundle_dir) - if qualified_policy is None + if qualified_policy is None and step.target_param is None else None ) if hover_check is None and ( @@ -10376,7 +10382,7 @@ def _identity_gate_error( + " — refusing to act; run aborted" ) elif check.status in ("unreadable", "abstain") and ( - step.risk == "irreversible" or governed + step.risk == "irreversible" or governed or step.target_param is not None ): reason = ( "rests on a glyph-confusable identifier OCR may have " @@ -10389,6 +10395,8 @@ def _identity_gate_error( "requires verified identity under the governed run policy" if governed else "is irreversible" + if step.risk == "irreversible" + else "selects a record by a run value" ) error = ( f"Step '{step.id}' ({step.intent}) {scope} and its target " diff --git a/packages/flow/tests/test_target_param.py b/packages/flow/tests/test_target_param.py new file mode 100644 index 000000000..2fc284510 --- /dev/null +++ b/packages/flow/tests/test_target_param.py @@ -0,0 +1,252 @@ +"""A record-selecting click follows the run's record (``Step.target_param``). + +The click on a search result or worklist row must land on the row that shows +the run's value, and the identity check must still verify that row before +any input. Anything that would let the click act on an unverified row halts. +""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.ir import ( + ActionKind, + Anchor, + ParamSpec, + StructuralLocator, + Step, + Workflow, +) +from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.runtime.target_param import NO_TEMPLATE, bind_target_param +from openadapt_flow.runtime.identity_template import build_identity_template +from tests.test_replayer import ( + FakeBackend, + Match, + OcrLine, + bundle, + resolving_vision, + run_dir, +) + +__all__ = ["bundle", "run_dir"] # pytest fixtures re-exported + + +def _row_click( + *, + context: str | None = "Open chart for Phil (active)", + ocr_text: str = "Phil", + structural: StructuralLocator | None = None, + target_param: str | None = "patient", +) -> Step: + return Step( + id="s_open", + intent="open the patient's chart", + action=ActionKind.CLICK, + anchor=Anchor( + template="templates/btn.png", + region=(100, 100, 50, 20), + click_point=(110, 105), + ocr_text=ocr_text, + context_text=context, + structural=structural, + ), + target_param=target_param, + ) + + +def _wf(step: Step, **params: str) -> Workflow: + return Workflow(name="open", params=params or {"patient": "Phil"}, steps=[step]) + + +# -------------------------------------------------------------------------- +# IR: additive and validated +# -------------------------------------------------------------------------- + + +def test_unset_target_param_is_not_serialized(): + step = _row_click(target_param=None) + assert "target_param" not in step.model_dump(mode="json") + assert _row_click().model_dump(mode="json")["target_param"] == "patient" + + +def test_target_param_round_trips(): + step = _row_click() + assert Step.model_validate(step.model_dump(mode="json")) == step + + +def test_target_param_only_on_clicks(): + with pytest.raises(ValueError, match="CLICK"): + Step( + id="t", + intent="type", + action=ActionKind.TYPE, + text="x", + target_param="patient", + ) + + +# -------------------------------------------------------------------------- +# Binding rules +# -------------------------------------------------------------------------- + + +def test_demo_value_keeps_the_step_unchanged(): + step = _row_click() + bound, refusal = bind_target_param(step, {"patient": "Phil"}, _wf(step)) + assert refusal is None + assert bound is step + + +def test_other_value_looks_for_the_value_without_the_demo_crop(): + step = _row_click( + structural=StructuralLocator(role="link", name="Phil", selector="#row-17") + ) + bound, refusal = bind_target_param(step, {"patient": "Susan"}, _wf(step)) + assert refusal is None + assert bound.anchor.template == NO_TEMPLATE + assert bound.anchor.ocr_text == "Susan" + assert bound.anchor.structural.name == "Susan" + assert bound.anchor.structural.selector is None # pointed at the old row + # Identity evidence is untouched: the gate substitutes the run's value. + assert bound.anchor.context_text == step.anchor.context_text + assert step.anchor.ocr_text == "Phil" # the bundle's step is not mutated + + +def test_label_around_the_value_keeps_its_words(): + step = _row_click(ocr_text="Open Phil") + bound, _ = bind_target_param(step, {"patient": "Susan"}, _wf(step)) + assert bound.anchor.ocr_text == "Open Susan" + + +def test_structural_name_that_is_not_the_value_is_dropped(): + step = _row_click(structural=StructuralLocator(role="link", name="Belford, Phil")) + bound, _ = bind_target_param(step, {"patient": "Susan"}, _wf(step)) + assert bound.anchor.structural is None + + +@pytest.mark.parametrize("value", [None, "", " "]) +def test_missing_value_refuses(value): + step = _row_click() + params = {} if value is None else {"patient": value} + _, refusal = bind_target_param(step, params, _wf(step)) + assert refusal is not None and "no value" in refusal + + +def test_no_identity_evidence_refuses(): + step = _row_click(context=None) + _, refusal = bind_target_param(step, {"patient": "Susan"}, _wf(step)) + assert refusal is not None and "no identity evidence" in refusal + + +def test_identity_band_without_the_parameter_refuses(): + step = _row_click(context="Open chart for Alice (active)") + _, refusal = bind_target_param(step, {"patient": "Susan"}, _wf(step)) + assert refusal is not None and "does not contain" in refusal + + +def test_unsupplied_embedded_identity_parameter_refuses(): + """The hashed band also embeds a date of birth the run must name.""" + step = _row_click(context=None) + step.anchor.identity_template = build_identity_template( + "Open chart for Phil 1972-02-09", + param_examples={"patient": "Phil", "dob": "1972-02-09"}, + salt_hex="00" * 16, + ) + workflow = Workflow( + name="open", + params={"patient": "Phil"}, + param_specs={"dob": ParamSpec(name="dob", required=True)}, + steps=[step], + ) + _, ok = bind_target_param( + step, {"patient": "Susan", "dob": "1980-01-01"}, workflow + ) + assert ok is None + _, refusal = bind_target_param(step, {"patient": "Susan"}, workflow) + assert refusal is not None and "'dob'" in refusal + + +# -------------------------------------------------------------------------- +# Replay through the interpreter +# -------------------------------------------------------------------------- + + +def _ocr_vision(row_text: str): + vision = resolving_vision() + vision.template_results = [] # the demo crop must not be consulted + vision.text_results = { + "Susan": Match(point=(120, 140), region=(100, 132, 60, 16), confidence=0.95) + } + vision.ocr_lines = [OcrLine(row_text, region=(170, 132, 240, 16))] + return vision + + +def test_replay_clicks_the_run_s_row_and_verifies_it(bundle, run_dir): + step = _row_click() + vision = _ocr_vision("Open chart for Susan (active)") + backend = FakeBackend() + report = Replayer(backend, vision=vision).run( + _wf(step), params={"patient": "Susan"}, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True, report.results[0].error + assert vision.template_calls == [] + assert "Susan" in vision.text_calls + assert ("click", 120, 140, False) in backend.actions + identity = report.results[0].identity + assert identity.status == "verified" + assert identity.mode == "param" + assert identity.param == "patient" + + +def test_replay_halts_on_a_row_naming_another_record(bundle, run_dir): + step = _row_click() + vision = _ocr_vision("Open chart for Susan (archived) Getting, Robert") + backend = FakeBackend() + report = Replayer(backend, vision=vision).run( + _wf(step), params={"patient": "Susan"}, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False, report.results[0].identity + assert backend.actions == [] + assert report.results[0].identity.status == "mismatch" + + +def test_replay_refusal_sends_no_input(bundle, run_dir): + step = _row_click(context="Open chart for Alice (active)") + vision = _ocr_vision("Open chart for Susan (active)") + backend = FakeBackend() + report = Replayer(backend, vision=vision).run( + _wf(step), params={"patient": "Susan"}, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert backend.actions == [] + result = report.results[0] + assert result.safety_halt is True + assert "does not contain" in (result.error or "") + + +def test_replay_unreadable_row_halts_a_reversible_selection(bundle, run_dir): + """An ordinary reversible click proceeds on an unreadable band; a click + that selects a record by a run value must verify the row first.""" + step = _row_click() + vision = _ocr_vision("Open chart for Susan (active)") + vision.ocr_lines = [] + backend = FakeBackend() + report = Replayer(backend, vision=vision).run( + _wf(step), params={"patient": "Susan"}, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert backend.actions == [] + result = report.results[0] + assert result.identity.status == "unreadable" + assert "selects a record by a run value" in (result.error or "") + + +def test_healing_never_adopts_another_record_s_row(bundle, run_dir): + step = _row_click() + vision = _ocr_vision("Open chart for Susan (active)") + report = Replayer(FakeBackend(), vision=vision).run( + _wf(step), params={"patient": "Susan"}, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert report.results[0].heal is None From c1d5df37621832326482d1ac65fd444d1fb57cc3 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 14:19:02 -0400 Subject: [PATCH 3/4] flow: induce search-open-act-return and row-varying loops, checked per row Loop induction now finds worklists whose rows are not back-to-back repeats: search a record, open it, work it, return to the list (the last row may skip the return, which then runs only when the list is not on screen), and rows that differ (a field some rows lack, a write some rows get behind a screen check a person accepts). A click whose label holds a typed row value selects that row's record (Step.target_param); identity text that changes by row becomes a required identity column. A body that writes is emitted only when every write has a record check bound to the row's values and either an identity-checked selection precedes it or the check counts a new record. Every loop carries a LoopCertificate. The sealed-leaf integrity check compares the program's own step for a rebound click. Recordings compile with identity bands readable and induction hashes them once the row columns are known. Tests run held-out rows through the interpreter against an in-memory app and check its records. Co-Authored-By: Claude Opus 5.5 --- CAPABILITIES.md | 1 + .../flow/openadapt_flow/compiler/compile.py | 8 +- .../flow/openadapt_flow/compiler/induction.py | 838 +++++++++++++++--- .../openadapt_flow/compiler/loop_shapes.py | 60 +- .../flow/openadapt_flow/runtime/replayer.py | 17 +- packages/flow/tests/test_loop_shapes.py | 682 ++++++++++++++ packages/flow/tests/test_target_param.py | 8 +- 7 files changed, 1469 insertions(+), 145 deletions(-) create mode 100644 packages/flow/tests/test_loop_shapes.py diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 874ac4595..becc526c3 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -164,6 +164,7 @@ improvising or retrying when something is off. | Learn one program with branches from two or more recordings, or say which recordings are missing | Released | Synthetic | `openadapt-flow induce`, `disambiguate`; [CAPABILITIES.md (flow)](packages/flow/docs/CAPABILITIES.md) | | Learn a guarded branch from several halts fixed the same way | Released | Synthetic | `openadapt-flow teach-branch`, `learning/branch_teach.py` | | Run a program over a worklist | Released | Real app | `openadapt-flow for-each` | +| Learn a worklist loop from recordings that search a record, open it, work it and go back to the list, even when the last row skips the return or rows differ (a field some rows lack, a write some rows get behind a screen check a person accepts). Each pass selects the row's own record by its values, verifies it, and checks that row's write | On main ([#1435](https://github.com/OpenAdaptAI/OpenAdapt/pull/1435)) | Synthetic (held-out rows run through the interpreter against an in-memory app) | `openadapt-flow induce`; `Step.target_param` | | Typed business decisions inside a run | Released | Synthetic | `business_decision` states, `openadapt-flow business-decisions` | | Watch many real runs in the background and mine candidate branches (shadow mode) | On main ([#1212](https://github.com/OpenAdaptAI/OpenAdapt/pull/1212)) | Synthetic | [SHADOW_MINING.md](packages/capture/docs/SHADOW_MINING.md), [LEARN_FROM_MANY_RUNS.md](packages/flow/docs/LEARN_FROM_MANY_RUNS.md) | | Fill a workflow's inputs from a referral PDF, fax image or email | Released | Synthetic | `openadapt-flow intake`, [INTAKE.md](packages/flow/docs/INTAKE.md) | diff --git a/packages/flow/openadapt_flow/compiler/compile.py b/packages/flow/openadapt_flow/compiler/compile.py index 0433bf997..f4fdf3221 100644 --- a/packages/flow/openadapt_flow/compiler/compile.py +++ b/packages/flow/openadapt_flow/compiler/compile.py @@ -1923,6 +1923,7 @@ def _compile_recording( pixel_verify_enabled: bool = False, bind_effect_params: bool = False, drop_redundant_keys: bool = False, + phi_free_identity: bool = True, ) -> Workflow: """Compile a recording directory into a workflow bundle. @@ -3471,8 +3472,11 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: # (openadapt_flow.runtime.identity_template). Runs LAST so param-hygiene # lint and optional model annotation still see the plaintext. Backward # compatible: bundles compiled before this carry the plaintext fields and - # replay unchanged. - _phi_free_identity(workflow) + # replay unchanged. Multi-trace induction compiles with this off and + # hashes the bands itself once the loop's row columns are known + # (compiler.induction.phi_free_program_identity). + if phi_free_identity: + _phi_free_identity(workflow) # PHI governance manifest (audit REM-1): classify the bundle so an operator # inventory and the pre-commit/CI guard can act on it. diff --git a/packages/flow/openadapt_flow/compiler/induction.py b/packages/flow/openadapt_flow/compiler/induction.py index 08c33ca17..778236843 100644 --- a/packages/flow/openadapt_flow/compiler/induction.py +++ b/packages/flow/openadapt_flow/compiler/induction.py @@ -23,17 +23,20 @@ :class:`~openadapt_flow.ir.Relation`, its body the repeated subflow (Rousillon/Helena "for every row ..."). - **LIMITS (loop inference is a PROTOTYPE -- keep the claim honest).** The only - loop shape detected is a *consecutive* repeated sub-sequence - (:func:`_reduce_trace`). It does NOT recognize a **search -> process -> - return** loop (navigate away and back each iteration), **pagination** (a - "next page" step between bodies), or a **per-row conditional body** (the body - differs row to row). When the repeated body contains a CONSEQUENTIAL - (irreversible / effect-bearing) action, whether those repeats are a - data-driven worklist or a fixed unrolled sequence cannot be certified from - trace *shape* alone -- inducing the wrong loop would repeat an irreversible - action -- so induction emits an :class:`Uncertainty` (``ambiguous_loop``) - and REFUSES rather than emit a possibly-wrong loop. + Besides back-to-back repeats, :mod:`.loop_shapes` finds **search -> open + -> act -> return** worklists (the last row may skip the return, which then + runs only when the list is not on screen) and bodies that **vary by row** + (a typed field some rows lack runs only when the row has a value; a + reversible step some rows have runs when its target shows; a write some + rows get needs a screen check a person accepts). A click whose label holds + a typed row value selects that row's record (``Step.target_param``) and is + identity-checked per row. A body that WRITES is emitted only when every + write has a record check bound to the row's values and either an + identity-checked selection precedes it or the check counts a new record; + otherwise induction emits an ``ambiguous_loop`` :class:`Uncertainty` and + REFUSES. Every emitted loop carries a :class:`LoopCertificate`. + Not recognized yet: **pagination** (a "next page" step between rows) and a + list nested inside each row. * **Branches.** A step present/divergent in some traces but not others UNDER A DETECTABLE CONDITION is a ``branch`` state with guarded transitions -- the guard is *proposed and flagged for confirmation* (Skill-DisCo: divergences @@ -208,7 +211,8 @@ class Uncertainty(BaseModel): description=( "'ambiguous_branch' | 'ambiguous_loop' | 'unconfirmed_branch' | " "'unconfirmed_optional' | 'underdetermined_order' | " - "'ambiguous_selection' | 'alignment_failure'" + "'ambiguous_selection' | 'alignment_failure' | " + "'unconfirmed_loop_branch'" ) ) location: str = Field(description="Column / position the ambiguity is at") @@ -337,6 +341,30 @@ class NarratedGuard(BaseModel): detail: str = "" +class LoopCertificate(BaseModel): + """Why an induced loop may run once per row: its bound and its checks. + + Written for every emitted loop. ``bound`` is the termination argument; + ``selections`` names each click that selects a row's record and what its + identity check compares; ``writes`` names each write and the row values + its record check is bound to. + """ + + location: str + relation: str + bound: str + max_iterations: int + row_columns: list[str] = Field(default_factory=list) + identity_columns: list[str] = Field( + default_factory=list, + description="Row columns the identity check needs beyond the typed ones.", + ) + selections: list[str] = Field(default_factory=list) + writes: list[str] = Field(default_factory=list) + conditional_steps: list[str] = Field(default_factory=list) + returns_to_list: bool = False + + class InductionResult(BaseModel): """The induced program (or an honest refusal) plus the full audit trail.""" @@ -381,6 +409,10 @@ class InductionResult(BaseModel): default_factory=list, description="One verdict per underdetermined point probing looked at.", ) + loop_certificates: list[LoopCertificate] = Field( + default_factory=list, + description="Bound and per-row checks of every emitted loop.", + ) @property def underdetermined(self) -> bool: @@ -496,7 +528,48 @@ def _as_workflow(trace: TraceInput) -> Workflow: recording = Path(trace) with tempfile.TemporaryDirectory(prefix="induce-boot-") as tmp: - return compile_recording(recording, Path(tmp) / "bundle", name=recording.name) + # Identity bands stay readable until induction has named every row + # column; induce_program hashes them before returning. + return compile_recording( + recording, + Path(tmp) / "bundle", + name=recording.name, + phi_free_identity=False, + ) + + +def phi_free_program_identity(workflow: Workflow) -> None: + """Replace readable identity bands in an induced program with templates. + + The same PHI-at-rest rule ``compile`` applies (audit REM-2), run after + induction so each template indexes the loop's row columns: a record + selected per row is then verified against that row's values. + """ + from openadapt_flow.runtime.identity_template import ( + build_identity_template, + new_salt_hex, + ) + from openadapt_flow.traversal import iter_workflow_steps + + salt = new_salt_hex() + examples = {k: str(v) for k, v in workflow.params.items()} + for step in iter_workflow_steps(workflow): + anchor = step.anchor + if anchor is None or anchor.identity_template is not None: + continue + if not (anchor.context_text or anchor.structured_identity): + continue + template = build_identity_template( + anchor.context_text, + structured_identity=anchor.structured_identity, + param_examples=examples, + salt_hex=salt, + ) + if template is None: + continue + anchor.identity_template = template + anchor.context_text = None + anchor.structured_identity = None _LEADING_VERBS = frozenset( @@ -536,6 +609,17 @@ def _field_key(step: Step) -> str: if struct is not None and struct.role: return f"select:{struct.role}" return _norm_intent(step) + if step.action in (ActionKind.TYPE, ActionKind.SELECT_OPTION): + if step.field_label: + return step.field_label + if step.text and step.text in step.intent and len(step.text.strip()) >= 2: + # A compiled TYPE step's intent quotes the typed value ("type + # 'Phil'"); the value is what varies, not the field. + return _norm_intent( + step.model_copy( + update={"intent": step.intent.replace(step.text, "")} + ) + ) if step.anchor is not None and step.anchor.ocr_text: return step.anchor.ocr_text return _norm_intent(step) @@ -576,12 +660,52 @@ def _value(step: Step) -> Optional[str]: # =========================================================================== +#: How many steps back a typed value can name the row a click selects. +_SELECTION_LOOKBACK = 6 + + +def _trace_sigs(steps: list[Step]) -> list[tuple[str, str]]: + """Alignment signatures for one trace. + + A click whose label contains a value typed shortly before it (a search, + then a click on the result) selects the record that value names. Its + label differs row to row, so its signature is the value-free + ``select:typed`` instead of the label. + """ + sigs: list[tuple[str, str]] = [] + for i, step in enumerate(steps): + sig = _sig(step) + if step.action in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK) and not sig[ + 1 + ].startswith("select:"): + label = _squash_text(_value(step) or "") + for prev in steps[max(0, i - _SELECTION_LOOKBACK) : i]: + typed = ( + _squash_text(prev.text or "") + if prev.action + in ( + ActionKind.TYPE, + ActionKind.SELECT_OPTION, + ) + else "" + ) + if len(typed) >= 2 and typed in label: + sig = (step.action.value, "select:typed") + break + sigs.append(sig) + return sigs + + +def _squash_text(text: str) -> str: + return "".join(text.casefold().split()) + + class _SingleTok: kind = "single" - def __init__(self, step: Step): + def __init__(self, step: Step, sig: Optional[tuple] = None): self.step = step - self.align_sig: tuple = _sig(step) + self.align_sig: tuple = sig if sig is not None else _sig(step) @property def value(self) -> Optional[str]: @@ -591,11 +715,49 @@ def value(self) -> Optional[str]: class _LoopTok: kind = "loop" - def __init__(self, body_steps: list[Step], iterations: list[list[Step]]): + def __init__( + self, + body_steps: list[Step], + iterations: list[list[Step]], + *, + iteration_sigs: Optional[list[list[tuple]]] = None, + core_sigs: Optional[tuple] = None, + short_last: bool = False, + ): self.body_steps = body_steps self.iterations = iterations # per-iteration list of steps - body_sig = tuple(_sig(s) for s in body_steps) - self.align_sig = ("__loop__", body_sig) + self.iteration_sigs = iteration_sigs or [ + [_sig(s) for s in it] for it in iterations + ] + del core_sigs # kept for callers; alignment uses the semantic core + self.align_sig = ("__loop__", self._semantic_core()) + #: The last iteration stopped before the body's trailing return. + self.short_last = short_last + + def _semantic_core(self) -> tuple: + """The steps every row has that carry its meaning. + + Typed values, record selections and writes present in every + iteration, in order. Navigation and row-varying steps are left out, + so two traces of the same worklist align even when one ended on the + list and the other on a record, or when their rows differed. + """ + from openadapt_flow.compiler.loop_shapes import merge + + n_it = len(self.iterations) + core = [] + for pos in merge(self.iteration_sigs): + if len(pos.at) < n_it: + continue + it0 = next(iter(pos.at)) + step = self.iterations[it0][pos.at[it0]] + if ( + step.action in (ActionKind.TYPE, ActionKind.SELECT_OPTION) + or pos.sig[1].startswith("select:") + or _step_consequential(step) + ): + core.append(pos.sig) + return tuple(core) or tuple(self.iteration_sigs[0]) @property def count(self) -> int: @@ -615,18 +777,12 @@ def _reduce_trace(steps: list[Step]) -> list: everything else stays a ``_SingleTok``. Smallest repeating block wins (the tightest loop body). Model-free and purely structural. - LIMITS (loop detection is a PROTOTYPE). This recognizes ONLY a body that - repeats BACK-TO-BACK in one trace. It does NOT recognize: - * search -> process -> return loops (each iteration navigates away and - back, so the body is not literally consecutive); - * pagination (a "next page" step separates the per-page bodies); - * per-row CONDITIONAL bodies (the body differs from row to row). - A repeat is also, on shape alone, indistinguishable from a fixed unrolled - sequence of distinct steps. The caller (:func:`_emit_loop`) therefore - REFUSES (emits an ``ambiguous_loop`` Uncertainty) rather than emit a loop - when the repeated body is CONSEQUENTIAL, where guessing wrong would repeat - an irreversible action.""" - sigs = [_sig(s) for s in steps] + When no back-to-back repeat starts at a step, :func:`_loose_loop` looks + for rows that start with the same step and share a core (search -> open + -> act -> return, rows that differ). Pagination is not recognized. The + caller (:func:`_emit_loop`) refuses a body that writes unless each write + is checked per row.""" + sigs = _trace_sigs(steps) n = len(steps) toks: list = [] i = 0 @@ -641,20 +797,97 @@ def _reduce_trace(steps: list[Step]) -> list: if reps >= 2: found = (length, reps) break + loose = _loose_loop(steps, sigs, i) + if found is not None and loose is not None: + # A back-to-back repeat can be the full rows of a worklist whose + # last row stopped before the return; keep the longer reading. + if sum(len(it) for it in loose.iterations) > found[0] * found[1]: + found = None if found is not None: length, reps = found body_steps = steps[i : i + length] iterations = [ steps[i + r * length : i + r * length + length] for r in range(reps) ] - toks.append(_LoopTok(body_steps, iterations)) + toks.append( + _LoopTok( + body_steps, + iterations, + iteration_sigs=[ + sigs[i + r * length : i + r * length + length] + for r in range(reps) + ], + ) + ) i += length * reps - else: - toks.append(_SingleTok(steps[i])) - i += 1 + continue + if loose is not None: + toks.append(loose) + i = sum(len(it) for it in loose.iterations) + i + continue + toks.append(_SingleTok(steps[i], sigs[i])) + i += 1 return toks +def _navigation_only(step: Step) -> bool: + """A step that only moves around: no typed value, no write.""" + return step.action not in ( + ActionKind.TYPE, + ActionKind.SELECT_OPTION, + ActionKind.READ, + ) and not _step_consequential(step) + + +def _loose_loop(steps: list[Step], sigs: list, start: int) -> Optional["_LoopTok"]: + """Iterations that are not back-to-back repeats (see :mod:`.loop_shapes`). + + Accepted only when the shared core carries a value that changes from row + to row (that is what makes the repeats a worklist) and the trailing + steps the last row skipped only navigate (the return to the list). + """ + from openadapt_flow.compiler.loop_shapes import merge, segment + + for seg in segment(sigs, start): + chunks = [steps[a:b] for a, b in seg.chunks] + chunk_sigs = [list(sigs[a:b]) for a, b in seg.chunks] + full = chunks[:-1] + varies = False + # Values compare across every iteration, the short last one included. + for pos in merge(chunk_sigs): + if len(pos.at) < 2: + continue + vals = {_value(chunks[it][j]) for it, j in pos.at.items()} + it0 = next(iter(pos.at)) + step0 = chunks[it0][pos.at[it0]] + if len(vals) > 1 and step0.action in ( + ActionKind.TYPE, + ActionKind.SELECT_OPTION, + ActionKind.CLICK, + ActionKind.DOUBLE_CLICK, + ): + varies = True + break + if not varies: + continue + # Only trailing navigation is a return; anything the last row + # skipped before it is a row-varying step (decided at emission). + nav_tail = 0 + for q in reversed(seg.tail): + if not _navigation_only(full[0][seg.positions[q].at[0]]): + break + nav_tail += 1 + core_sigs = tuple(seg.positions[q].sig for q in seg.core) + return _LoopTok( + chunks[0], + chunks, + iteration_sigs=chunk_sigs, + core_sigs=core_sigs, + short_last=bool(nav_tail), + ) + return None + + # =========================================================================== # Multiple-trace alignment (incremental LCS merge of reduced token sequences) # =========================================================================== @@ -855,6 +1088,9 @@ def _instruction_for( return f"Record how '{name}' is identified each run." if kind == "alignment_failure": return "Record traces of the same task." + if kind == "unconfirmed_loop_branch": + name = field or "the write" + return f"Record rows with and without {name} under a visible condition." return "Record a demonstration that resolves this uncertainty." @@ -1041,6 +1277,7 @@ def induce_program( screen_texts: Optional[Callable[[int, str], Optional[Sequence[str]]]] = None, interpreter: Optional[Any] = None, accepted_narration: Optional[dict[str, str]] = None, + accepted_loop_guards: Optional[dict[str, str]] = None, ) -> InductionResult: """Induce a parameterized :class:`ProgramGraph` from multiple demonstrations of the same task -- the RFC §3 induction loop (step [4] + [5]). @@ -1078,6 +1315,10 @@ def induce_program( narrated guards a person accepted. The guard id pins the exact guard the report showed; a different guard at that location is not accepted. A guard a recording contradicts stays refused. + accepted_loop_guards: ``{":p@": person}`` + -- a write a loop body makes for some rows only, run when that + screen text shows. The report proposes the text; a person + accepts it. Without acceptance such a loop is refused. Returns: An :class:`InductionResult` with the induced program (or a refusal) and @@ -1120,6 +1361,8 @@ def induce_program( seen_div_groups: set[int] = set() optional_nodes: dict[int, _Node] = {} divergences: dict[int, tuple[list[_Column], int]] = {} + # Loop identity columns no demonstration shows (each run's rows name them). + identity_only: set[str] = set() for idx, col in enumerate(columns): present = sorted(col.tokens) @@ -1140,7 +1383,18 @@ def induce_program( # --- loop ---------------------------------------------------------- if col.kind == "loop": node, dec = _emit_loop( - idx, col, present, n, taken, param_specs, data_sources, result, propose + idx, + col, + present, + n, + taken, + param_specs, + data_sources, + result, + propose, + screen_texts=screen_texts, + accepted_loop_guards=accepted_loop_guards, + identity_only=identity_only, ) decisions.append(dec) if node is not None: @@ -1205,149 +1459,525 @@ def induce_program( params={ key: runtime_param_text(spec.example) if spec.example is not None else "" for key, spec in param_specs.items() + if key not in identity_only }, data_sources=data_sources, guard_sources=guard_sources, ) + if any(not isinstance(t, Workflow) for t in traces): + phi_free_program_identity(workflow) + program = workflow.program result.program = program result.workflow = workflow return result def _emit_loop( - idx, col, present, n, taken, param_specs, data_sources, result, propose=None + idx, + col, + present, + n, + taken, + param_specs, + data_sources, + result, + propose=None, + *, + screen_texts=None, + accepted_loop_guards=None, + identity_only=None, ): + """Emit a worklist LOOP for an aligned loop column, or refuse. + + The iterations of every trace are lined up into body positions + (:func:`.loop_shapes.merge`). Then: + + * a typed value that changes by row is a row column; + * a click whose label contains a row column's value selects that row's + record (``Step.target_param``); its identity band must embed the value, + and other band text that changes by row becomes an identity column; + * steps the last row skipped at the end of the body (the return to the + list) move to the front, behind a branch that runs them only when the + body's first control is not on screen; + * a step present for some rows only is guarded: a typed value on its + column being non-empty, a reversible step on its own target, a write + only on screen text a person accepted; + * a write in the body needs, per row, a record check bound to the row's + columns, and either an identity-checked selection of the row before it + or a record check that counts a new record. + + Anything else is an ``ambiguous_loop`` (or ``ambiguous_selection`` / + ``unconfirmed_loop_branch``) uncertainty and no loop is emitted. + """ + from openadapt_flow.compiler.loop_shapes import ( + band_identity, + bind_effect_columns, + merge, + ) + loop_toks = [col.tokens[t] for t in present] - body_steps = loop_toks[0].body_steps - field = _field_key(body_steps[0]) if body_steps else f"row_{idx}" counts = [tok.count for tok in loop_toks] + location = f"loop_{idx}" + + # (trace, iteration, steps, sigs, stopped-before-return) + iters: list[tuple[int, int, list[Step], list, bool]] = [] + for t, tok in zip(present, loop_toks): + for m, (steps_m, sigs_m) in enumerate(zip(tok.iterations, tok.iteration_sigs)): + iters.append((t, m, steps_m, sigs_m, tok.short_last and m == tok.count - 1)) + positions = merge([x[3] for x in iters]) + n_iter = len(iters) + short = {k for k, x in enumerate(iters) if x[4]} + + def step_at(q: int, k: int) -> Step: + return iters[k][2][positions[q].at[k]] + + # The return to the list: trailing steps only the short last rows + # skipped, and only steps that navigate. A write is never moved. + tail: list[int] = [] + for q in range(len(positions) - 1, -1, -1): + absent = set(range(n_iter)) - set(positions[q].at) + k0 = next(iter(positions[q].at)) + if absent and absent <= short and _navigation_only(step_at(q, k0)): + tail.insert(0, q) + else: + break + body_qs = [q for q in range(len(positions)) if q not in tail] + optional = [q for q in body_qs if len(positions[q].at) < n_iter] + + rep_i = max(range(len(loop_toks)), key=lambda i: loop_toks[i].count) + rep_t = present[rep_i] + cands = [k for k, x in enumerate(iters) if x[0] == rep_t] + tmpl_k = max( + cands, + key=lambda k: (sum(1 for p in positions if k in p.at), k not in short, -k), + ) + + def src_k(q: int) -> int: + at = positions[q].at + return tmpl_k if tmpl_k in at else min(at) + + def rep_step(q: int) -> Step: + return step_at(q, src_k(q)) - # LOOP HONESTY (RFC §3 [5]; see module LIMITS). Trace SHAPE alone cannot - # certify that a consecutive repeat is a data-driven worklist rather than a - # fixed unrolled sequence -- and it recognizes ONLY consecutive repeats (not - # search->process->return, pagination, or per-row conditional bodies). When - # the repeated body contains a CONSEQUENTIAL action, guessing wrong repeats - # an irreversible / effect-bearing write, so REFUSE (emit an Uncertainty) - # rather than emit a possibly-wrong loop. - if any(_step_consequential(s) for s in body_steps): - location = f"loop_{idx}" + first_q = body_qs[0] if body_qs else 0 + field = _field_key(rep_step(first_q)) if positions else f"row_{idx}" + + def refuse(kind: str, detail: str, *, record_field: str = field): question = _confirmation_question( location, AmbiguityKind.OPTIONAL_DIALOG, prompt=( - f"A consecutive repeat of {counts} step(s) over '{field}' was " - "observed, and its body performs a CONSEQUENTIAL (irreversible " - "or effect-bearing) action. Is this a data-driven loop over a " - "worklist, or a fixed sequence of distinct steps? (Trace shape " - "cannot distinguish them, and a wrong loop repeats the " - "irreversible action.)" + f"A repeated body over '{record_field}' was observed (counts " + f"{counts}), but it cannot run once per row safely: {detail}" ), - evidence=f"repeated consequential body '{field}' x{counts} at column {idx}", + evidence=f"repeated body '{record_field}' x{counts} at column {idx}", consequential=True, ) _add_uncertainty( result, - kind="ambiguous_loop", + kind=kind, location=location, - detail=( - f"repeated body '{field}' (counts {counts}) contains a " - "consequential action; a worklist loop vs a fixed unrolled " - "sequence is not distinguishable from trace shape -- " - "refusing to guess a loop over an irreversible step." - ), + detail=detail, consequential=True, question=question, - field=field, + field=record_field, propose=propose, ) - dec = ColumnDecision( + return None, ColumnDecision( index=idx, kind="ambiguous_loop", align_sig=str(col.align_sig), - field=field, + field=record_field, present_in=present, counts=counts, - note=f"underdetermined consequential loop over '{field}' (counts {counts})", + note=f"underdetermined loop over '{record_field}' (counts {counts}): {detail}", + ) + + # --- row columns: typed values that change by row -------------------- + columns: dict[int, str] = {} + row_vals: dict[int, dict[str, str]] = {k: {} for k in range(n_iter)} + loop_taken = set(taken) + for q in body_qs: + st = rep_step(q) + if st.action not in (ActionKind.TYPE, ActionKind.SELECT_OPTION) or st.secret: + continue + vals = {k: (_value(step_at(q, k)) or "") for k in positions[q].at} + if len({vals.get(k, "") for k in range(n_iter)}) <= 1: + continue + name = _param_name_for(_field_key(st), loop_taken, f"row_{idx}_{q}") + loop_taken.add(name) + columns[q] = name + for k in range(n_iter): + row_vals[k][name] = vals.get(k, "") + + # --- record-selecting clicks ----------------------------------------- + targets: dict[int, str] = {} + identity_cols: dict[str, dict[int, Optional[str]]] = {} + identity_examples: dict[str, Optional[str]] = {} + token_indices: dict[int, dict[str, list[int]]] = {} + selection_notes: list[str] = [] + for q in body_qs: + st = rep_step(q) + if st.action not in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK): + continue + vals = {k: _squash_text(_value(step_at(q, k)) or "") for k in positions[q].at} + if len(set(vals.values())) <= 1: + continue + hit = next( + ( + name + for name in columns.values() + if all( + row_vals[k][name] and _squash_text(row_vals[k][name]) in vals[k] + for k in positions[q].at + ) + ), + None, ) - return None, dec + click_field = _field_key(st) + if hit is None: + return refuse( + "ambiguous_selection", + f"the clicked row '{click_field}' changes by row but no typed " + "row value names it, so the click cannot be pointed at each " + "row's record", + record_field=click_field, + ) + anchors = { + k: step_at(q, k).anchor + for k in positions[q].at + if step_at(q, k).anchor is not None + } + bi = band_identity( + column=hit, + template_iter=src_k(q), + anchors=anchors, + row_values=row_vals, + same_trace={k: iters[k][0] for k in range(n_iter)}, + column_prefix=f"{hit}_identity_", + ) + if not bi.ok: + return refuse( + "ambiguous_selection", + f"the click on '{click_field}' selects each row's record by " + f"'{hit}', but {bi.reason}", + record_field=click_field, + ) + for name, per_iter in bi.identity_columns.items(): + identity_cols[name] = per_iter + identity_examples[name] = bi.identity_examples.get(name) + loop_taken.add(name) + for k in range(n_iter): + value = per_iter.get(k) + if value is not None: + row_vals[k][name] = value + if bi.token_indices: + token_indices[q] = bi.token_indices + targets[q] = hit + extra = ( + f", also by {', '.join(bi.identity_columns)}" if bi.identity_columns else "" + ) + selection_notes.append(f"{click_field}: by '{hit}'{extra} ({bi.basis})") + + # --- writes: a per-row record check and a per-row identity ----------- + effects_out: dict[int, list] = {} + write_notes: list[str] = [] + all_cols = set(row_vals[tmpl_k]) | set(identity_cols) | set(columns.values()) + for q in body_qs: + st = rep_step(q) + if not _step_consequential(st): + continue + name = _field_key(st) + if not st.effects: + return refuse( + "ambiguous_loop", + f"the body writes at '{name}' with no record check, so a " + "row's write could not be confirmed per row", + ) + rebound, used = bind_effect_columns(list(st.effects), row_vals[src_k(q)]) + used &= all_cols + if not used: + return refuse( + "ambiguous_loop", + f"the record check of '{name}' names no row value, so it " + "would check the demonstrated row on every row", + ) + selected = any(tq < q and tq not in optional for tq in targets) + counts_new = any( + getattr(e.kind, "value", e.kind) == "exact_new_set" + or ( + getattr(e.kind, "value", e.kind) == "record_written" + and (e.count_new_only or e.idempotency_key is not None) + ) + for e in rebound + ) + if not (selected or counts_new): + return refuse( + "ambiguous_loop", + f"nothing before '{name}' selects the row's record with an " + "identity check, and its record check does not count a new " + "record, so a write could land on the wrong record", + ) + effects_out[q] = rebound + basis = "identity-checked selection" if selected else "new-record check" + write_notes.append( + f"{name}: record check on {', '.join(sorted(used))} ({basis})" + ) + + # --- steps present for some rows only ---------------------------------- + guards: dict[int, Guard] = {} + conditional_notes: list[str] = [] + for q in optional: + st = rep_step(q) + name = _field_key(st) + if q in columns: + guards[q] = Guard( + predicate=Predicate( + kind=PredicateKind.NOT, + operands=[ + Predicate( + kind=PredicateKind.PARAM_EQUALS, + param=columns[q], + value="", + ) + ], + intent=f"the row has a value for '{columns[q]}'", + ), + on_unmet="skip", + ) + conditional_notes.append(f"{name}: when the row has '{columns[q]}'") + continue + if q in targets: + return refuse( + "ambiguous_loop", + f"the record-selecting click '{name}' happens for some rows only", + ) + if not _step_consequential(st): + guards[q] = optional_guard(st) + conditional_notes.append(f"{name}: when its target is on screen") + continue + text = _loop_branch_text(q, positions, iters, screen_texts) + loc = f"{location}:p{q}" + if text is not None and f"{loc}@{text}" in (accepted_loop_guards or {}): + guards[q] = Guard( + predicate=Predicate( + kind=PredicateKind.TEXT_PRESENT, + text=text, + intent=f"'{text}' is on screen (accepted by " + f"{accepted_loop_guards[f'{loc}@{text}']})", + ), + on_unmet="skip", + ) + conditional_notes.append(f"{name}: when '{text}' is on screen (accepted)") + continue + detail = f"the write '{name}' happens for some rows only" + ( + f"; the screen showed '{text}' exactly on those rows. Accept " + f"the guard as '{loc}@{text}' to run it only then" + if text is not None + else "; no screen text tells those rows apart" + ) + if text is not None: + result.proposed.append( + Proposal( + target=loc, + kind="guard", + content=f"run '{name}' only when '{text}' is on screen", + source="screen", + ) + ) + return refuse("unconfirmed_loop_branch", detail, record_field=name) + + # --- the return to the list ------------------------------------------ + return_anchor = None + if tail: + lead = next( + ( + q + for q in body_qs + if q not in optional and rep_step(q).anchor is not None + ), + None, + ) + if lead is None: + return refuse( + "ambiguous_loop", + "the last row skipped the return to the list, and no step of " + "the body has an on-screen target that tells the list apart", + ) + return_anchor = rep_step(lead).anchor + + # --- emit ---------------------------------------------------------------- + taken.update(loop_taken) + for q, name in columns.items(): + example = row_vals[tmpl_k].get(name) or next( + (row_vals[k][name] for k in range(n_iter) if row_vals[k][name]), None + ) + param_specs[name] = ParamSpec( + name=name, + type=ParamKind.STRING, + example=example, + required=q not in optional, + ) + for name in identity_cols: + example = identity_examples.get(name) + param_specs[name] = ParamSpec( + name=name, type=ParamKind.STRING, example=example, required=False + ) + if example is None and identity_only is not None: + identity_only.add(name) - param_name = _param_name_for(field, taken, f"row_{idx}") - taken.add(param_name) - # Representative worklist: the LONGEST demonstrated queue, inlined so the - # bundle is self-contained (a run may override via Replayer worklists=...). - rep = max(loop_toks, key=lambda tk: tk.count) rel_name = f"worklist_{idx}" + rep_rows = [ + {k2: v for k2, v in row_vals[k].items()} + for k in range(n_iter) + if iters[k][0] == rep_t + ] data_sources[rel_name] = Relation( name=rel_name, - rows=rep.rows(param_name), + rows=rep_rows, description=f"inferred worklist for the '{field}' loop", ) - param_specs[param_name] = ParamSpec( - name=param_name, - type=ParamKind.STRING, - example=(rep.rows(param_name)[0][param_name] if rep.count else None), - required=True, - ) - # Body subflow: the repeated step(s), the varying field bound per row. body_id = f"body_{idx}" + order = tail + body_qs + sid_of = {q: f"{body_id}_s{q}" for q in order} + body_end = f"{body_id}_end" body_states: dict[str, State] = {} - prev_ids: list[str] = [] - for bi, bstep in enumerate(body_steps): - sid = f"{body_id}_s{bi}" - step = bstep.model_copy(deep=True) - if step.action in (ActionKind.TYPE, ActionKind.SELECT_OPTION): - step.param = param_name + for i, q in enumerate(order): + step = rep_step(q).model_copy(deep=True) + if q in columns: + step.param = columns[q] step.text = None - st = State( - id=sid, + if q in targets: + step.target_param = targets[q] + tmpl = step.anchor.identity_template if step.anchor else None + if tmpl is not None and q in token_indices: + tmpl.param_token_indices.update(token_indices[q]) + if q in effects_out: + step.effects = effects_out[q] + if q in guards: + step.guard = guards[q] + nxt = sid_of[order[i + 1]] if i + 1 < len(order) else body_end + body_states[sid_of[q]] = State( + id=sid_of[q], kind=StateKind.ACTION, step=step, - transitions=[Transition(target="__PATCH__")], + transitions=[Transition(target=nxt)], ) - body_states[sid] = st - prev_ids.append(sid) - body_end = f"{body_id}_end" + entry = sid_of[order[0]] if order else body_end + if tail and return_anchor is not None: + branch_id = f"{body_id}_return" + body_states[branch_id] = State( + id=branch_id, + kind=StateKind.BRANCH, + transitions=[ + Transition( + guard=Predicate( + kind=PredicateKind.NOT, + operands=[ + Predicate( + kind=PredicateKind.ANCHOR_RESOLVES, + anchor=return_anchor, + intent="the list's first control is on screen", + ) + ], + ), + target=sid_of[tail[0]], + label="not on the list: return to it first", + ), + Transition(target=sid_of[body_qs[0]], label="on the list"), + ], + ) + entry = branch_id body_states[body_end] = State( id=body_end, kind=StateKind.TERMINAL, outcome="success" ) - ordered = list(body_states) - for k, sid in enumerate(ordered): - st = body_states[sid] - if st.kind is StateKind.ACTION: - st.transitions[0].target = ( - ordered[k + 1] if k + 1 < len(ordered) else body_end - ) - body_graph = ProgramGraph(entry=ordered[0], states=body_states) + body_graph = ProgramGraph(entry=entry, states=body_states) - loop_state_id = f"loop_{idx}" + loop_state_id = location node = _Node(loop_state_id) exit_tr = Transition(target="__PATCH__") + spec = LoopSpec(relation=rel_name, body=body_id, var=field) node.states[loop_state_id] = State( id=loop_state_id, kind=StateKind.LOOP, - loop=LoopSpec(relation=rel_name, body=body_id, var=field), + loop=spec, transitions=[exit_tr], ) node.exits = [exit_tr] node._subflow = (body_id, body_graph) # type: ignore[attr-defined] + result.loop_certificates.append( + LoopCertificate( + location=location, + relation=rel_name, + bound=( + f"one pass per row of '{rel_name}' (a finite list), at most " + f"{spec.max_iterations}; more rows halt before the first pass" + ), + max_iterations=spec.max_iterations, + row_columns=sorted(set(columns.values())), + identity_columns=sorted(identity_cols), + selections=selection_notes, + writes=write_notes, + conditional_steps=conditional_notes, + returns_to_list=bool(tail), + ) + ) dec = ColumnDecision( index=idx, kind="loop", align_sig=str(col.align_sig), field=field, - param_name=param_name, + param_name=next(iter(columns.values()), None), present_in=present, counts=counts, note=( - f"loop over '{field}' -- body repeats {counts} across traces " - f"(counts differ => a worklist, not unrolled steps)" + f"loop over '{field}' -- body repeats {counts} across traces" + + (" (returns to the list between rows)" if tail else "") + + (f"; {len(write_notes)} write(s) checked per row" if write_notes else "") ), ) return node, dec +def _loop_branch_text(q, positions, iters, screen_texts) -> Optional[str]: + """Screen text shown on exactly the rows that have body position ``q``. + + Read from the frame before the step (or, for rows without it, before the + row's next step). None when there is no screen reader or no such text. + """ + if screen_texts is None: + return None + with_lines: Optional[set[str]] = None + without: set[str] = set() + for k, (t, _m, steps_k, _sigs, _short) in enumerate(iters): + pos = positions[q] + if k in pos.at: + step = steps_k[pos.at[k]] + else: + later = [ + positions[r].at[k] + for r in range(q + 1, len(positions)) + if k in positions[r].at + ] + if not later: + continue + step = steps_k[later[0]] + lines = screen_texts(t, step.id) + if lines is None: + return None + norm = {" ".join(str(x).split()) for x in lines if str(x).strip()} + if k in pos.at: + with_lines = norm if with_lines is None else with_lines & norm + else: + without |= norm + if not with_lines: + return None + cands = sorted(with_lines - without, key=lambda x: (len(x), x)) + return cands[0] if cands else None + + def _emit_required_single(idx, col, present, taken, param_specs, result, propose=None): toks = [col.tokens[t] for t in present] step0 = toks[0].step diff --git a/packages/flow/openadapt_flow/compiler/loop_shapes.py b/packages/flow/openadapt_flow/compiler/loop_shapes.py index e8d089f65..07bda942f 100644 --- a/packages/flow/openadapt_flow/compiler/loop_shapes.py +++ b/packages/flow/openadapt_flow/compiler/loop_shapes.py @@ -96,31 +96,30 @@ def end(self) -> int: def _match_last( - sigs: Sequence[Any], start: int, positions: list[Position], core: set[int] -) -> tuple[int, int]: + sigs: Sequence[Any], start: int, positions: list[Position] +) -> tuple[int, int, int]: """Greedy match of the last iteration against the body. - Returns ``(end, reached)``: the trace index after the last matched step - and the body index matching stopped at. Skipping an optional position is - allowed; skipping a core position is not (that step then ends the - iteration). + Returns ``(end, reached, matched)``: the trace index after the last + matched step, the body index matching stopped at, and how many steps + matched. Positions the last iteration skips become row-varying steps + (or, at the end of the body, the return it never made). """ ptr = 0 j = start + matched = 0 while j < len(sigs) and ptr < len(positions): - hit = None - for q in range(ptr, len(positions)): - if positions[q].sig == sigs[j]: - hit = q - break - if q in core: - break + hit = next( + (q for q in range(ptr, len(positions)) if positions[q].sig == sigs[j]), + None, + ) if hit is None: break ptr = hit + 1 j += 1 - return j, ptr + matched += 1 + return j, ptr, matched def segment(sigs: Sequence[Any], start: int) -> Iterator[Segmentation]: @@ -129,8 +128,8 @@ def segment(sigs: Sequence[Any], start: int) -> Iterator[Segmentation]: Every occurrence of ``sigs[start]`` after ``start`` may begin a new iteration. A candidate needs two or more iterations, a core of at least two steps shared by every full iteration, few extra steps per iteration, - and a last iteration that reaches the whole core except, possibly, a - trailing suffix. Callers check the steps themselves (that the core varies + and a last iteration that matches at least half the core in order (what + it skips at the end is the return it never made). Callers check the steps themselves (that the core varies by row, that the tail only navigates). """ @@ -147,13 +146,10 @@ def segment(sigs: Sequence[Any], start: int) -> Iterator[Segmentation]: if any((b - a) - len(core) > allowed for a, b in full): continue last_start = cands[k - 1] - end, reached = _match_last(sigs, last_start, positions, set(core)) - tail = [q for q in core if q >= reached] - # The last iteration must reach every core position before the tail. - if any(q not in tail and q >= reached for q in core): - continue - if tail and tail != core[len(core) - len(tail) :]: + end, reached, matched = _match_last(sigs, last_start, positions) + if matched < max(2, (len(core) + 1) // 2): continue + tail = [q for q in core if q >= reached] if len(tail) >= len(core) - 1: continue yield Segmentation( @@ -177,9 +173,7 @@ class BandIdentity: reason: str = "" #: Extra row columns the identity band needs: name -> per-iteration #: value (None where the band was not readable for that iteration). - identity_columns: dict[str, dict[int, Optional[str]]] = field( - default_factory=dict - ) + identity_columns: dict[str, dict[int, Optional[str]]] = field(default_factory=dict) #: Extra columns' demonstrated values (from the template iteration). identity_examples: dict[str, str] = field(default_factory=dict) #: Hashed bands: token indices to index under each column name. @@ -234,8 +228,7 @@ def band_identity( return BandIdentity( ok=False, reason=( - "the row's identity text does not contain the value that " - "selects it" + "the row's identity text does not contain the value that selects it" ), ) return _readable_band_columns( @@ -262,10 +255,7 @@ def band_identity( ok=False, reason="the row's hashed identity does not contain the selecting value", ) - explained = { - tmpl.tokens[j].r - for j in indices - } + explained = {tmpl.tokens[j].r for j in indices} base = {t.r for j, t in enumerate(tmpl.tokens) if j not in indices} for it, other in anchors.items(): if it == template_iter or same_trace.get(it) != same_trace[template_iter]: @@ -335,7 +325,9 @@ def tokens(text: str, values: dict[str, str]) -> list[str]: continue other_toks = tokens(oband, row_values.get(it, {})) ops = SequenceMatcher(a=base, b=other_toks, autojunk=False).get_opcodes() - aligned[it] = [(i1, i2, j1, j2) for tag, i1, i2, j1, j2 in ops if tag != "equal"] + aligned[it] = [ + (i1, i2, j1, j2) for tag, i1, i2, j1, j2 in ops if tag != "equal" + ] for tag, i1, i2, _j1, _j2 in ops: if tag == "insert": return BandIdentity( @@ -391,7 +383,9 @@ def _replace_ci(text: str, old: str, new: str) -> str: # --------------------------------------------------------------------------- -def bind_effect_columns(effects: list[Any], template_values: dict[str, str]) -> tuple[list[Any], set[str]]: +def bind_effect_columns( + effects: list[Any], template_values: dict[str, str] +) -> tuple[list[Any], set[str]]: """Rebind a write's record check to the row's columns. A literal equal to the template row's value of a column becomes a diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 4c5be41ad..363ead556 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -7288,7 +7288,11 @@ def _run_step_timed( # run's value; the identity check verifies the row (param mode). from openadapt_flow.runtime.target_param import bind_target_param + sealed_step = step step, refusal = bind_target_param(step, params, workflow) + # The sealed-leaf check compares the program's step with the + # step the runtime acts on; it must see the sealed one. + self._target_param_binding = (step, sealed_step) if refusal is not None: return StepResult( step_id=step.id, @@ -11301,6 +11305,17 @@ def _revalidate_consequential_actuation( self._active_delivery_region = fresh_region return fresh_resolution, fresh_region, fresh_png, error + def _sealed_step(self, step: Step) -> Step: + """The program's own step for one the runtime rebound to a run value. + + ``runtime.target_param`` rebinds a record-selecting click to the + run's record; integrity checks compare against the sealed step. + """ + binding = getattr(self, "_target_param_binding", None) + if binding is not None and binding[0] is step: + return binding[1] + return step + def _resolve_step_target( self, step: Step, @@ -11737,7 +11752,7 @@ def _active_program_frame_refusal( or leaf.get("state_id") != self._current_state_id or parent_state is None or parent_state.kind is not StateKind.ACTION - or parent_state.step != step + or parent_state.step != self._sealed_step(step) ): return ( "governed program leaf frame no longer matches the exact sealed " diff --git a/packages/flow/tests/test_loop_shapes.py b/packages/flow/tests/test_loop_shapes.py new file mode 100644 index 000000000..7043c8e03 --- /dev/null +++ b/packages/flow/tests/test_loop_shapes.py @@ -0,0 +1,682 @@ +"""Loops that are not back-to-back repeats: induce, then run held-out rows. + +The demonstrations work a list the way people do: search a patient, open the +chart, write a note, save, go back to the list; the last row ends without +going back. Induction must turn that into a loop whose every pass selects the +row's own record (identity-checked) and checks the row's own write. + +The behavioral tests run the induced program through the real interpreter +against a small in-memory chart application, with rows the demonstrations +never showed, and check the application's records afterwards: each held-out +row got its note, nothing else changed, and a row whose identity does not +match stops before any input reaches that record. +""" + +from __future__ import annotations + +import io +from pathlib import Path +from typing import Optional + +import pytest +from PIL import Image + +from openadapt_flow.compiler.induction import induce_program +from openadapt_flow.compiler.loop_shapes import merge, segment +from openadapt_flow.ir import ( + ActionKind, + Anchor, + PredicateKind, + StateKind, + Step, + Workflow, +) +from openadapt_flow.runtime.effects import ( + Effect, + EffectKind, + EffectState, + EffectVerdict, + Verdict, +) +from openadapt_flow.runtime.replayer import Replayer +from tests.test_replayer import FakeBackend, FakeVision, Match, OcrLine + +# -------------------------------------------------------------------------- +# The application: a patient list with search, a chart with a note field +# -------------------------------------------------------------------------- + +PATIENTS = { + "Phil Crawford": "1972-02-09", + "Alice Moreno": "1985-03-04", + "Bob Lindqvist": "1990-07-21", + "Carol Okafor": "1968-11-30", + "Dan Whitaker": "1979-01-15", + "Susan Underwood": "1980-05-05", + "Robert Getting": "1961-09-09", +} + +#: Control -> (template file, region, click point, screen it shows on). +CONTROLS = { + "search": ("templates/search.png", (20, 10, 60, 16), (40, 18), "list"), + "row": ("templates/row.png", (100, 50, 50, 20), (120, 60), "list"), + "note": ("templates/note.png", (40, 90, 80, 16), (60, 98), "chart"), + "save": ("templates/save.png", (200, 140, 40, 16), (210, 148), "chart"), + "back": ("templates/back.png", (10, 175, 40, 16), (20, 183), "chart"), + "signout": ("templates/signout.png", (250, 10, 40, 16), (260, 18), "any"), +} +_COLORS = {name: (20 + 30 * i, 200 - 25 * i, 90) for i, name in enumerate(CONTROLS)} + + +def _png(size, color) -> bytes: + buf = io.BytesIO() + Image.new("RGB", size, color).save(buf, format="PNG") + return buf.getvalue() + + +def _template_bytes(name: str) -> bytes: + _path, region, _point, _screen = CONTROLS[name] + return _png((region[2], region[3]), _COLORS[name]) + + +class ChartApp(FakeBackend): + """Records live in ``notes``; the screen is the list or one chart.""" + + def __init__(self): + super().__init__() + self.notes = {name: "" for name in PATIENTS} + self.screen = "list" + self.focus: Optional[str] = None + self.buffers = {"search": "", "note": ""} + self.results: list[str] = [] + self.open_patient: Optional[str] = None + + def _at(self, x, y) -> Optional[str]: + for name, (_p, (rx, ry, rw, rh), _pt, screen) in CONTROLS.items(): + if ( + screen in (self.screen, "any") + and rx <= x < rx + rw + and ry <= y < ry + rh + ): + return name + return None + + def click(self, x, y, *, double=False): + super().click(x, y, double=double) + hit = self._at(x, y) + if hit in ("search", "note"): + self.focus = hit + self._text_value = self.buffers[hit] + elif hit == "row" and self.results: + self.open_patient = self.results[0] + self.screen = "chart" + self.buffers["note"] = self.notes[self.open_patient] + self.focus = None + elif hit == "save" and self.open_patient is not None: + self.notes[self.open_patient] = self.buffers["note"] + elif hit == "back": + self.screen = "list" + self.open_patient = None + self.focus = None + + def type_text(self, text): + super().type_text(text) + if self.focus is not None: + self.buffers[self.focus] = self._text_value + + def press(self, key): + super().press(key) + if key == "Enter" and self.focus == "search": + q = self.buffers["search"].casefold() + self.results = [n for n in PATIENTS if n.casefold() == q] + self.buffers["search"] = "" + self._text_value = "" + + +class ChartVision(FakeVision): + """Sees what the app shows: its controls, the result row, the row band.""" + + def __init__(self, app: ChartApp): + super().__init__() + self.app = app + self._by_template = {_template_bytes(n): n for n in CONTROLS} + + def find_template(self, screen_png, template_png, **kwargs): + name = self._by_template.get(template_png) + if name is None or CONTROLS[name][3] not in (self.app.screen, "any"): + return None + if name == "row": + # The crop shows the demonstrated row; another row never matches. + return None + _p, region, point, _s = CONTROLS[name] + return Match(point=point, region=region, confidence=0.97) + + def find_text(self, screen_png, text, **kwargs): + if self.app.screen == "list" and self.app.results: + if text.casefold() == self.app.results[0].casefold(): + return Match(point=(120, 60), region=(100, 50, 50, 20), confidence=0.95) + return None + + def ocr(self, screen_png, *, region=None): + if self.app.screen == "list" and self.app.results: + name = self.app.results[0] + return [ + OcrLine( + f"Open chart for {name} {PATIENTS[name]}", region=(160, 52, 240, 16) + ) + ] + return [] + + +class NoteStore: + """Effect verifier: reads the app's records (the system of record).""" + + substrate = "chart-app" + + def __init__(self, app: ChartApp): + self.app = app + + def capture_pre_state(self): + return EffectState(substrate=self.substrate, reachable=True) + + def verify(self, effect, before): + del before + name = str(effect.match["name"]) + ok = self.app.notes.get(name) == str(effect.value) + return EffectVerdict( + verdict=Verdict.CONFIRMED if ok else Verdict.REFUTED, + kind=effect.kind, + substrate=self.substrate, + ) + + +# -------------------------------------------------------------------------- +# Demonstrations (as compiled steps) +# -------------------------------------------------------------------------- + + +def _anchor(control: str, **extra) -> Anchor: + path, region, point, _s = CONTROLS[control] + return Anchor(template=path, region=region, click_point=point, **extra) + + +def _click(sid, control, intent, **kw) -> Step: + anchor_extra = kw.pop("anchor_extra", {}) + return Step( + id=sid, + intent=intent, + action=ActionKind.CLICK, + anchor=_anchor(control, **anchor_extra), + **kw, + ) + + +def _type(sid, label, value) -> Step: + return Step( + id=sid, + intent=f"type '{value}'", + action=ActionKind.TYPE, + text=value, + field_label=label, + ) + + +def _row(i: int, name: str, note: str, *, back: bool, write_check=True) -> list[Step]: + p = f"r{i}" + effects = ( + [ + Effect( + kind=EffectKind.FIELD_EQUALS, + match={"name": name}, + field="note", + value=note, + risk="irreversible", + ) + ] + if write_check + else [] + ) + steps = [ + _click( + f"{p}_search", + "search", + "click 'Search'", + anchor_extra={"ocr_text": "Search"}, + ), + _type(f"{p}_name", "Search", name), + Step(id=f"{p}_enter", intent="press Enter", action=ActionKind.KEY, key="Enter"), + _click( + f"{p}_open", + "row", + f"click '{name}'", + anchor_extra={ + "ocr_text": name, + "context_text": f"Open chart for {name} {PATIENTS[name]}", + }, + ), + _click(f"{p}_note", "note", "click 'Note'", anchor_extra={"ocr_text": "Note"}), + _type(f"{p}_type_note", "Note", note), + _click( + f"{p}_save", + "save", + "click 'Save'", + anchor_extra={"ocr_text": "Save"}, + risk="irreversible", + effects=effects, + ), + ] + if back: + steps.append( + _click( + f"{p}_back", + "back", + "click 'Back to list'", + anchor_extra={"ocr_text": "Back to list"}, + ) + ) + return steps + + +def _demo(rows: list[tuple[str, str]], **kw) -> Workflow: + steps: list[Step] = [] + for i, (name, note) in enumerate(rows): + steps += _row(i, name, note, back=i + 1 < len(rows), **kw) + steps.append( + _click( + "done", "signout", "click 'Sign out'", anchor_extra={"ocr_text": "Sign out"} + ) + ) + return Workflow(name="notes", steps=steps) + + +def _demos(**kw) -> list[Workflow]: + return [ + _demo( + [("Phil Crawford", "Seen, stable"), ("Alice Moreno", "Refill sent")], **kw + ), + _demo( + [ + ("Bob Lindqvist", "Call back"), + ("Carol Okafor", "Labs ordered"), + ("Dan Whitaker", "No change"), + ], + **kw, + ), + ] + + +@pytest.fixture() +def bundle(tmp_path) -> Path: + root = tmp_path / "bundle" + (root / "templates").mkdir(parents=True) + for name, (path, *_rest) in CONTROLS.items(): + (root / path).write_bytes(_template_bytes(name)) + return root + + +# -------------------------------------------------------------------------- +# Shape: segmentation and the induced program +# -------------------------------------------------------------------------- + + +def test_merge_keeps_steps_some_iterations_lack(): + positions = merge([["a", "b", "c"], ["a", "c"], ["a", "x", "c"]]) + assert [p.sig for p in positions] == ["a", "b", "x", "c"] + assert positions[1].at == {0: 1} + assert positions[3].at == {0: 2, 1: 1, 2: 2} + + +def test_segment_finds_a_last_iteration_without_the_return(): + sigs = ["s", "t", "o", "w", "b", "s", "t", "o", "w", "end"] + seg = next(segment(sigs, 0)) + assert seg.chunks == [(0, 5), (5, 9)] + assert [seg.positions[q].sig for q in seg.tail] == ["b"] + + +def test_search_open_act_return_induces_a_checked_loop(): + result = induce_program(_demos()) + assert result.certified, result.render() + [cert] = result.loop_certificates + assert cert.returns_to_list is True + assert set(cert.row_columns) == {"search", "note"} + assert cert.identity_columns == ["search_identity_1"] # the date of birth + assert any("by 'search'" in s for s in cert.selections) + assert any("note" in w and "search" in w for w in cert.writes) + assert "at most 1000" in cert.bound + + wf = result.workflow + body = wf.subflows[wf.program.states[cert.location].loop.body] + entry = body.states[body.entry] + assert entry.kind is StateKind.BRANCH + guard = entry.transitions[0].guard + assert guard.kind is PredicateKind.NOT + assert guard.operands[0].kind is PredicateKind.ANCHOR_RESOLVES + steps = [st.step for st in body.states.values() if st.step is not None] + [opener] = [s for s in steps if s.target_param] + assert opener.target_param == "search" + [save] = [s for s in steps if s.effects] + eff = save.effects[0] + assert eff.match["name"].param == "search" + assert eff.value.param == "note" + rows = wf.data_sources[cert.relation].rows + assert [r["search"] for r in rows] == [ + "Bob Lindqvist", + "Carol Okafor", + "Dan Whitaker", + ] + assert rows[0]["search_identity_1"] == "1990-07-21" + + +def test_a_write_without_a_record_check_is_refused(): + result = induce_program(_demos(write_check=False)) + assert result.certified is False + u = next(u for u in result.uncertainties if u.kind == "ambiguous_loop") + assert "no record check" in u.detail + + +def test_a_record_check_on_the_demonstrated_row_is_refused(): + demos = _demos() + for wf in demos: + for step in wf.steps: + for eff in step.effects: + eff.match = {"name": "Phil Crawford"} # always the first demo's patient + eff.value = "constant" + result = induce_program(demos) + assert result.certified is False + assert any("names no row value" in u.detail for u in result.uncertainties) + + +def test_a_click_no_row_value_names_is_refused(): + demos = _demos() + for wf in demos: + for step in wf.steps: + if step.id.endswith("_open"): + step.anchor.ocr_text = f"Row {step.id}" + step.intent = f"click 'Row {step.id}'" + result = induce_program(demos) + assert result.certified is False + + +def test_a_band_without_the_value_is_refused(): + demos = _demos() + for wf in demos: + for step in wf.steps: + if step.id.endswith("_open"): + step.anchor.context_text = "Open chart" + result = induce_program(demos) + assert result.certified is False + assert any(u.kind == "ambiguous_selection" for u in result.uncertainties) + + +# -------------------------------------------------------------------------- +# Behavior: run rows the demonstrations never showed +# -------------------------------------------------------------------------- + + +def _run(workflow, bundle, run_dir, rows): + app = ChartApp() + vision = ChartVision(app) + relation = next(iter(workflow.data_sources)) + report = Replayer( + app, vision=vision, effect_verifier=NoteStore(app), poll_interval_s=0.01 + ).run(workflow, bundle_dir=bundle, run_dir=run_dir, worklists={relation: rows}) + return app, report + + +HELD_OUT = [ + { + "search": "Susan Underwood", + "search_identity_1": "1980-05-05", + "note": "BP recheck", + }, + { + "search": "Robert Getting", + "search_identity_1": "1961-09-09", + "note": "Flu shot given", + }, +] + + +def test_held_out_rows_each_write_their_own_record(bundle, tmp_path): + result = induce_program(_demos()) + app, report = _run(result.workflow, bundle, tmp_path / "run", HELD_OUT) + errors = [r.error for r in report.results if r.error] + assert report.success is True, errors + assert app.notes["Susan Underwood"] == "BP recheck" + assert app.notes["Robert Getting"] == "Flu shot given" + untouched = { + k: v + for k, v in app.notes.items() + if k not in ("Susan Underwood", "Robert Getting") + } + assert set(untouched.values()) == {""} + opened = [ + r + for r in report.results + if r.identity is not None and r.identity.mode == "param" + ] + assert len(opened) == 2 + assert all(r.identity.status == "verified" for r in opened) + assert report.model_calls == 0 if hasattr(report, "model_calls") else True + + +def test_a_row_whose_identity_differs_stops_before_its_record(bundle, tmp_path): + rows = [dict(HELD_OUT[0], search_identity_1="1999-12-31")] + HELD_OUT[1:] + result = induce_program(_demos()) + app, report = _run(result.workflow, bundle, tmp_path / "run", rows) + assert report.success is False + assert set(app.notes.values()) == {""} # nothing written anywhere + halted = next(r for r in report.results if r.error) + assert halted.identity is not None and halted.identity.status == "mismatch" + + +def test_a_row_missing_its_identity_column_stops(bundle, tmp_path): + rows = [{"search": "Susan Underwood", "note": "BP recheck"}] + result = induce_program(_demos()) + app, report = _run(result.workflow, bundle, tmp_path / "run", rows) + assert report.success is False + assert app.notes["Susan Underwood"] == "" + + +# -------------------------------------------------------------------------- +# Rows that differ: an optional field, a status-dependent write +# -------------------------------------------------------------------------- + + +def _demos_with_optional_note() -> list[Workflow]: + """Some rows get no note at all: the note steps are absent for them.""" + demos = _demos() + for wf in demos: + wf.steps = [ + s + for s in wf.steps + if not (s.id.startswith("r1_") and s.id.endswith(("_note", "_type_note"))) + ] + return demos + + +def test_a_value_some_rows_lack_runs_only_when_the_row_has_it(): + result = induce_program(_demos_with_optional_note()) + assert result.certified, result.render() + [cert] = result.loop_certificates + assert any("note" in c for c in cert.conditional_steps) + body = next(iter(result.workflow.subflows.values())) + [typed] = [ + st.step + for st in body.states.values() + if st.step is not None and st.step.param == "note" + ] + assert typed.guard is not None and typed.guard.on_unmet == "skip" + assert typed.guard.predicate.kind is PredicateKind.NOT + assert typed.guard.predicate.operands[0].kind is PredicateKind.PARAM_EQUALS + + +def _demos_with_status_write() -> list[Workflow]: + """Rows in status 'Pending' also get 'Sign off' (a second write).""" + demos = _demos() + for t, wf in enumerate(demos): + out = [] + for s in wf.steps: + out.append(s) + if s.id.endswith("_save") and s.id.startswith("r0_"): + name = next(x.text for x in wf.steps if x.id == "r0_name") + out.append( + _click( + f"r0_signoff_{t}", + "save", + "click 'Sign off'", + anchor_extra={"ocr_text": "Sign off"}, + risk="irreversible", + effects=[ + Effect( + kind=EffectKind.FIELD_EQUALS, + match={"name": name}, + field="status", + value="signed", + risk="irreversible", + ) + ], + ) + ) + wf.steps = out + return demos + + +def _screens(trace: int, step_id: str): + # Before each row's save, the chart shows its status. + return ["Status: Pending"] if step_id.startswith("r0_") else ["Status: Final"] + + +def test_a_write_some_rows_get_is_refused_with_a_proposed_screen_guard(): + result = induce_program(_demos_with_status_write(), screen_texts=_screens) + assert result.certified is False + u = next(u for u in result.uncertainties if u.kind == "unconfirmed_loop_branch") + assert "Status: Pending" in u.detail + assert any( + p.kind == "guard" and "Status: Pending" in p.content for p in result.proposed + ) + + +def test_an_accepted_screen_guard_certifies_the_status_write(): + demos = _demos_with_status_write() + refused = induce_program(demos, screen_texts=_screens) + u = next(u for u in refused.uncertainties if u.kind == "unconfirmed_loop_branch") + key = u.detail.split("as '")[1].split("'")[0] + result = induce_program( + _demos_with_status_write(), + screen_texts=_screens, + accepted_loop_guards={key: "reviewer"}, + ) + assert result.certified, result.render() + body = next(iter(result.workflow.subflows.values())) + [signoff] = [ + st.step + for st in body.states.values() + if st.step is not None and st.step.intent == "click 'Sign off'" + ] + assert signoff.guard.predicate.kind is PredicateKind.TEXT_PRESENT + assert signoff.guard.predicate.text == "Status: Pending" + + +# -------------------------------------------------------------------------- +# A list inside a form: several columns per row, the save after the loop +# -------------------------------------------------------------------------- + + +def test_list_in_form_binds_every_column(): + def demo(lines): + steps = [ + _click( + "open", + "note", + "click 'New claim'", + anchor_extra={"ocr_text": "New claim"}, + ) + ] + for i, (code, qty) in enumerate(lines): + steps += [ + _click( + f"l{i}_add", + "row", + "click 'Add line'", + anchor_extra={"ocr_text": "Add line"}, + ), + _type(f"l{i}_code", "Code", code), + _type(f"l{i}_qty", "Quantity", qty), + ] + steps.append( + _click("save", "save", "click 'Save'", anchor_extra={"ocr_text": "Save"}) + ) + return Workflow(name="claim", steps=steps) + + result = induce_program( + [ + demo([("A01", "1"), ("B22", "3")]), + demo([("C3", "2"), ("D4", "5"), ("E5", "1")]), + ] + ) + assert result.certified, result.render() + [cert] = result.loop_certificates + assert set(cert.row_columns) == {"code", "quantity"} + rows = result.workflow.data_sources[cert.relation].rows + assert rows == [ + {"code": "C3", "quantity": "2"}, + {"code": "D4", "quantity": "5"}, + {"code": "E5", "quantity": "1"}, + ] + + +def test_one_demonstration_of_several_rows_is_enough(bundle, tmp_path): + demo = _demo( + [ + ("Bob Lindqvist", "Call back"), + ("Carol Okafor", "Labs ordered"), + ("Dan Whitaker", "No change"), + ] + ) + result = induce_program([demo]) + assert result.certified, result.render() + app, report = _run(result.workflow, bundle, tmp_path / "run", HELD_OUT) + assert report.success is True, [r.error for r in report.results if r.error] + assert app.notes["Susan Underwood"] == "BP recheck" + + +def test_hashed_identity_still_verifies_each_row(bundle, tmp_path): + """Induction from recordings hashes the bands after naming the columns.""" + from openadapt_flow.compiler.induction import phi_free_program_identity + + result = induce_program(_demos()) + workflow = result.workflow + phi_free_program_identity(workflow) + body = next(iter(workflow.subflows.values())) + [opener] = [ + st.step for st in body.states.values() if st.step and st.step.target_param + ] + tmpl = opener.anchor.identity_template + assert opener.anchor.context_text is None + assert {"search", "search_identity_1"} <= set(tmpl.param_token_indices) + app, report = _run(workflow, bundle, tmp_path / "run", HELD_OUT) + assert report.success is True, [r.error for r in report.results if r.error] + assert app.notes["Robert Getting"] == "Flu shot given" + + wrong = [dict(HELD_OUT[0], search_identity_1="1999-12-31")] + app, report = _run(workflow, bundle, tmp_path / "run2", wrong) + assert report.success is False + assert set(app.notes.values()) == {""} + + +def test_a_row_without_a_note_skips_the_note(bundle, tmp_path): + result = induce_program(_demos_with_optional_note()) + rows = [ + {"search": "Susan Underwood", "search_identity_1": "1980-05-05", "note": ""}, + { + "search": "Robert Getting", + "search_identity_1": "1961-09-09", + "note": "Flu shot given", + }, + ] + app, report = _run(result.workflow, bundle, tmp_path / "run", rows) + assert report.success is True, [r.error for r in report.results if r.error] + assert app.notes["Robert Getting"] == "Flu shot given" + assert app.notes["Susan Underwood"] == "" + typed = [a for a in app.actions if a[0] == "type"] + assert ("type", "") not in typed diff --git a/packages/flow/tests/test_target_param.py b/packages/flow/tests/test_target_param.py index 2fc284510..3cd6c1694 100644 --- a/packages/flow/tests/test_target_param.py +++ b/packages/flow/tests/test_target_param.py @@ -13,13 +13,13 @@ ActionKind, Anchor, ParamSpec, - StructuralLocator, Step, + StructuralLocator, Workflow, ) +from openadapt_flow.runtime.identity_template import build_identity_template from openadapt_flow.runtime.replayer import Replayer from openadapt_flow.runtime.target_param import NO_TEMPLATE, bind_target_param -from openadapt_flow.runtime.identity_template import build_identity_template from tests.test_replayer import ( FakeBackend, Match, @@ -159,9 +159,7 @@ def test_unsupplied_embedded_identity_parameter_refuses(): param_specs={"dob": ParamSpec(name="dob", required=True)}, steps=[step], ) - _, ok = bind_target_param( - step, {"patient": "Susan", "dob": "1980-01-01"}, workflow - ) + _, ok = bind_target_param(step, {"patient": "Susan", "dob": "1980-01-01"}, workflow) assert ok is None _, refusal = bind_target_param(step, {"patient": "Susan"}, workflow) assert refusal is not None and "'dob'" in refusal From d71cd96893b8cc2023080d781d2bbbe88afbd8df Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 14:29:41 -0400 Subject: [PATCH 4/4] flow: refuse record-selecting clicks for certified and governed runs Qualification's fault driver cannot qualify a target_param click yet (it needs the recorded crop of the row). Until it can, certification under any identity-requiring policy (clinical-write), the run gate's identity gate (Standard, Regulated, governed, whatever --policy) and qualify readiness refuse such a bundle with a plain reason. Demo replay still runs it. Co-Authored-By: Claude Opus 5.5 --- CAPABILITIES.md | 2 +- packages/flow/openadapt_flow/policy.py | 15 +++++++ .../flow/openadapt_flow/qualify/campaign.py | 7 ++++ packages/flow/openadapt_flow/run_gate.py | 12 ++++++ .../openadapt_flow/runtime/target_param.py | 18 ++++++++ .../flow/tests/test_qualify_declarative.py | 25 +++++++++++ packages/flow/tests/test_target_param.py | 41 +++++++++++++++++++ 7 files changed, 119 insertions(+), 1 deletion(-) diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 05d2258be..3d6bbedc8 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -163,7 +163,7 @@ improvising or retrying when something is off. | Learn one program with branches from two or more recordings, or say which recordings are missing | Released | Synthetic | `openadapt-flow induce`, `disambiguate`; [CAPABILITIES.md (flow)](packages/flow/docs/CAPABILITIES.md) | | Learn a guarded branch from several halts fixed the same way | Released | Synthetic | `openadapt-flow teach-branch`, `learning/branch_teach.py` | | Run a program over a worklist | Released | Real app | `openadapt-flow for-each` | -| Learn a worklist loop from recordings that search a record, open it, work it and go back to the list, even when the last row skips the return or rows differ (a field some rows lack, a write some rows get behind a screen check a person accepts). Each pass selects the row's own record by its values, verifies it, and checks that row's write | On main ([#1435](https://github.com/OpenAdaptAI/OpenAdapt/pull/1435)) | Synthetic (held-out rows run through the interpreter against an in-memory app) | `openadapt-flow induce`; `Step.target_param` | +| Learn a worklist loop from recordings that search a record, open it, work it and go back to the list, even when the last row skips the return or rows differ (a field some rows lack, a write some rows get behind a screen check a person accepts). Each pass selects the row's own record by its values, verifies it, and checks that row's write. Certified and governed runs refuse it until qualification covers it | On main ([#1435](https://github.com/OpenAdaptAI/OpenAdapt/pull/1435)) | Synthetic (held-out rows run through the interpreter against an in-memory app) | `openadapt-flow induce`; `Step.target_param` | | Typed business decisions inside a run | Released | Synthetic | `business_decision` states, `openadapt-flow business-decisions` | | Watch many real runs in the background and mine candidate branches (shadow mode) | On main ([#1212](https://github.com/OpenAdaptAI/OpenAdapt/pull/1212)) | Synthetic | [SHADOW_MINING.md](packages/capture/docs/SHADOW_MINING.md), [LEARN_FROM_MANY_RUNS.md](packages/flow/docs/LEARN_FROM_MANY_RUNS.md) | | Fill a workflow's inputs from a referral PDF, fax image or email | Released | Synthetic | `openadapt-flow intake`, [INTAKE.md](packages/flow/docs/INTAKE.md) | diff --git a/packages/flow/openadapt_flow/policy.py b/packages/flow/openadapt_flow/policy.py index d7a548596..eb772b288 100644 --- a/packages/flow/openadapt_flow/policy.py +++ b/packages/flow/openadapt_flow/policy.py @@ -1030,6 +1030,21 @@ def evaluate_policy( ) ) + # A record-selecting click (Step.target_param) cannot be qualified yet: + # any policy that demands identity refuses it, with a plain reason. + if policy.prohibit_unarmed_clicks or policy.require_identity_for: + from openadapt_flow.runtime.target_param import UNQUALIFIED_SELECTION_REASON + + for step in steps: + if step.target_param is not None: + violations.append( + Violation( + rule="unqualified_record_selection", + step_id=step.id, + reason=UNQUALIFIED_SELECTION_REASON, + ) + ) + if policy.max_unverified_steps is not None: vacuous = [s.id for s in steps if is_vacuous(s)] if len(vacuous) > policy.max_unverified_steps: diff --git a/packages/flow/openadapt_flow/qualify/campaign.py b/packages/flow/openadapt_flow/qualify/campaign.py index 2a5fb6bcb..a4077b393 100644 --- a/packages/flow/openadapt_flow/qualify/campaign.py +++ b/packages/flow/openadapt_flow/qualify/campaign.py @@ -286,6 +286,13 @@ def check_ready( problems.append( f"record.fields.{name} uses parameter {expr.param!r}, which has no input" ) + from openadapt_flow.runtime.target_param import ( + UNQUALIFIED_SELECTION_REASON, + record_selecting_steps, + ) + + for step in record_selecting_steps(workflow): + problems.append(f"step {step.id} {UNQUALIFIED_SELECTION_REASON}") if save.anchor is None or not save.identity_armed: problems.append(f"save step {save.id} has no armed identity check") if not save.effects: diff --git a/packages/flow/openadapt_flow/run_gate.py b/packages/flow/openadapt_flow/run_gate.py index 2c1f6bb40..5204852a2 100644 --- a/packages/flow/openadapt_flow/run_gate.py +++ b/packages/flow/openadapt_flow/run_gate.py @@ -656,6 +656,18 @@ def _gate_identity( certifying_policy: Optional[Policy], ) -> GateResult: """Gate 2: every entity-sensitive / consequential action is identity-armed.""" + from openadapt_flow.runtime.target_param import UNQUALIFIED_SELECTION_REASON + + selecting = [step.id for step in steps if step.target_param is not None] + if selecting: + # Whatever the certifying policy: not admitted until qualification + # covers a record-selecting click. + return _result( + GATE_IDENTITY, + False, + f"{len(selecting)} step(s) {UNQUALIFIED_SELECTION_REASON}", + selecting, + ) must_arm = [ step for step in steps diff --git a/packages/flow/openadapt_flow/runtime/target_param.py b/packages/flow/openadapt_flow/runtime/target_param.py index 0981e8005..fd44a9784 100644 --- a/packages/flow/openadapt_flow/runtime/target_param.py +++ b/packages/flow/openadapt_flow/runtime/target_param.py @@ -91,6 +91,24 @@ def has_identity_evidence(anchor: Anchor) -> bool: ) +#: Why a bundle with a record-selecting click is not admitted for +#: certified, production or governed runs yet. +UNQUALIFIED_SELECTION_REASON = ( + "selects a record by a run value (target_param); qualification's fault " + "driver cannot yet qualify such a click (it needs the recorded crop of the " + "row), so this bundle is refused for certified, production and governed " + "runs until it can" +) + + +def record_selecting_steps(workflow: Workflow) -> list[Step]: + """Every step of ``workflow`` that selects a record by a run value.""" + + from openadapt_flow.traversal import iter_workflow_steps + + return [s for s in iter_workflow_steps(workflow) if s.target_param is not None] + + def bind_target_param( step: Step, params: Mapping[str, object], diff --git a/packages/flow/tests/test_qualify_declarative.py b/packages/flow/tests/test_qualify_declarative.py index 1fe3dd4da..eedd6f9d5 100644 --- a/packages/flow/tests/test_qualify_declarative.py +++ b/packages/flow/tests/test_qualify_declarative.py @@ -266,3 +266,28 @@ def test_cli_reports_config_errors(tmp_path): ) # fmt: skip assert result.returncode == 2 assert "verifier" in result.stderr + + +def test_check_ready_refuses_a_record_selecting_click(): + """Qualification's fault driver can't qualify a target_param click yet.""" + workflow = _workflow() + workflow.steps.insert( + 2, + Step( + id="s_pick", + intent="open the patient", + action=ActionKind.CLICK, + anchor=Anchor( + template="templates/pick.png", + region=(10, 40, 80, 20), + click_point=(20, 50), + context_text="Open chart for Doe 1970-01-01", + ), + target_param="lname", + ), + ) + config = _config() + save, _ = apply_verify_config(workflow, config) + problems = " | ".join(check_ready(workflow, config, save, dict(workflow.params))) + assert "step s_pick selects a record by a run value" in problems + assert "fault driver" in problems diff --git a/packages/flow/tests/test_target_param.py b/packages/flow/tests/test_target_param.py index 3cd6c1694..700c603a2 100644 --- a/packages/flow/tests/test_target_param.py +++ b/packages/flow/tests/test_target_param.py @@ -248,3 +248,44 @@ def test_healing_never_adopts_another_record_s_row(bundle, run_dir): ) assert report.success is True assert report.results[0].heal is None + + +# -------------------------------------------------------------------------- +# Not admitted for certified, production or governed runs (yet) +# -------------------------------------------------------------------------- + + +def test_clinical_write_certification_refuses_a_record_selecting_click(): + from openadapt_flow.policy import evaluate_policy, load_policy + + workflow = _wf(_row_click()) + report = evaluate_policy(workflow, load_policy("clinical-write")) + hits = [v for v in report.violations if v.rule == "unqualified_record_selection"] + assert [v.step_id for v in hits] == ["s_open"] + assert "fault driver" in hits[0].reason + + plain = evaluate_policy( + _wf(_row_click(target_param=None)), load_policy("clinical-write") + ) + assert not any(v.rule == "unqualified_record_selection" for v in plain.violations) + + permissive = evaluate_policy(workflow, load_policy("permissive")) + assert not any( + v.rule == "unqualified_record_selection" for v in permissive.violations + ) + + +def test_governed_run_gate_refuses_whatever_the_policy(): + from openadapt_flow.run_gate import GATE_IDENTITY, _gate_identity + + workflow = _wf(_row_click()) + result = _gate_identity( + workflow, + list(workflow.steps), + require_current_risk_certification=False, + certifying_policy=None, + ) + assert result.gate == GATE_IDENTITY + assert result.passed is False + assert result.offenders == ["s_open"] + assert "target_param" in result.detail