diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 6b511d68b..e541d413f 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -166,6 +166,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. 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/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 new file mode 100644 index 000000000..07bda942f --- /dev/null +++ b/packages/flow/openadapt_flow/compiler/loop_shapes.py @@ -0,0 +1,430 @@ +"""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] +) -> tuple[int, int, int]: + """Greedy match of the last iteration against the body. + + 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 = 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 + matched += 1 + return j, ptr, matched + + +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 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). + """ + + 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, 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( + 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/ir.py b/packages/flow/openadapt_flow/ir.py index d79a7734d..6c88d7982 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/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 d89f1a8c7..d07913cad 100644 --- a/packages/flow/openadapt_flow/qualify/campaign.py +++ b/packages/flow/openadapt_flow/qualify/campaign.py @@ -321,6 +321,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 e8c869dcb..22f86aa97 100644 --- a/packages/flow/openadapt_flow/run_gate.py +++ b/packages/flow/openadapt_flow/run_gate.py @@ -667,6 +667,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/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 38756fb7c..e5df26bc3 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -7598,6 +7598,25 @@ 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 + + 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, + 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: @@ -8963,6 +8982,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 @@ -10537,9 +10560,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 ( @@ -10685,7 +10710,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 " @@ -10698,6 +10723,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 " @@ -11602,6 +11629,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, @@ -12038,7 +12076,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 " @@ -18108,6 +18146,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..fd44a9784 --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/target_param.py @@ -0,0 +1,185 @@ +"""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)) + ) + + +#: 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], + 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 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_qualify_declarative.py b/packages/flow/tests/test_qualify_declarative.py index 63ff2faea..3400e219d 100644 --- a/packages/flow/tests/test_qualify_declarative.py +++ b/packages/flow/tests/test_qualify_declarative.py @@ -268,6 +268,31 @@ def test_cli_reports_config_errors(tmp_path): 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 + + def test_a_qualified_bundle_can_be_qualified_again(): # A sealed or repaired copy of a qualified bundle carries its project. # Qualifying it again replaces the project with its next revision. diff --git a/packages/flow/tests/test_target_param.py b/packages/flow/tests/test_target_param.py new file mode 100644 index 000000000..700c603a2 --- /dev/null +++ b/packages/flow/tests/test_target_param.py @@ -0,0 +1,291 @@ +"""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, + 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 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 + + +# -------------------------------------------------------------------------- +# 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