diff --git a/openadapt_flow/compiler/compile.py b/openadapt_flow/compiler/compile.py index ae6ec15..863d8f2 100644 --- a/openadapt_flow/compiler/compile.py +++ b/openadapt_flow/compiler/compile.py @@ -23,6 +23,8 @@ ActionKind, Anchor, Landmark, + ParamKind, + ParamSpec, Point, Postcondition, PostconditionKind, @@ -1089,6 +1091,18 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: recording_id=meta.get("id"), viewport=tuple(viewport) if viewport else None, params=params, + # Workflow-program IR, Phase 1: emit a TYPED spec for each recorded + # parameter alongside the frozen ``params`` dict -- generalizing the + # recorder's single "note value at replay" into a first-class, + # typed+required param. Phase 1 types every recorded value as a + # string with its demo value as the example/default; richer types + # (entity_ref/enum/date) come from disambiguation in a later phase. + param_specs={ + pname: ParamSpec( + name=pname, type=ParamKind.STRING, example=value + ) + for pname, value in params.items() + }, secret_params=list(meta.get("secret_params") or []), steps=steps, ) diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index 520976a..4ec6ae8 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -170,6 +170,132 @@ class Postcondition(BaseModel): ) +# -- Workflow-program IR, Phase 1 (RFC docs/design/WORKFLOW_PROGRAM_IR.md §6) -- +# +# Additive, backward-compatible first step toward the parameterized workflow +# program: typed parameters, a per-step `wait_until` readiness predicate, and a +# per-step `guard` precondition. ALL optional -- a bundle that declares none of +# them loads and replays EXACTLY as a v0 linear bundle does. These fields are +# deliberately ORTHOGONAL to Step.effects / Step.risk / the Anchor identity +# rungs: they add control-flow *around* the existing hardened action leaf, they +# do not restructure it (RFC §2.1). Full branches/loops/subflows are Phase 2 -- +# NOT built here. + + +class ParamKind(str, Enum): + """Typed-parameter kinds (RFC §2.2 ``ParamSpec.type``). + + ``entity_ref`` names an ENTITY to be re-resolved by the identity ladder at + run time (the "which patient" fix, docs/LIMITS.md), not a literal to blindly + substitute; the other kinds are literal values. Phase 1 stores the type for + typing/validation/emit -- kind-specific run-time resolution (entity_ref + re-resolution) is Phase 2+. + """ + + STRING = "string" + DATE = "date" + ENUM = "enum" + NUMBER = "number" + ENTITY_REF = "entity_ref" + + +class ParamSpec(BaseModel): + """A TYPED workflow parameter (RFC §2.2). Supersedes a bare + ``params: dict[str, str]`` entry by carrying a type, the recorded demo + value (``example``, which doubles as the replay default), whether it is + required, and enum choices. Additive: ``Workflow.param_specs`` lives + ALONGSIDE the frozen ``Workflow.params`` dict; a bundle with an empty + ``param_specs`` behaves exactly as before. + """ + + name: str + type: ParamKind = ParamKind.STRING + example: Optional[str] = Field( + default=None, + description="Recorded demo value; also the replay default when the " + "caller supplies no value for this parameter.", + ) + required: bool = True + choices: list[str] = Field( + default_factory=list, description="Allowed values for an enum param." + ) + + +class PredicateKind(str, Enum): + """Deterministic, model-free predicate kinds (RFC §2.2 ``Predicate``). + + A predicate is evaluated over the CURRENT observed frame / run parameters + with ZERO model calls -- it is the thing a linear IR cannot express. Phase 1 + ships the concrete kinds needed to (a) subsume today's SCROLL closed loop + (``anchor_resolves``), (b) turn the optional-modal case into a guarded + branch (``text_present``), and (c) branch on a parameter (``param_equals``), + plus boolean composition. ``worklist_nonempty`` (loops) is Phase 2. + """ + + #: The embedded ``anchor`` resolves on the current frame via the (model-free) + #: resolution ladder -- today's closed-loop scroll stop condition, now a + #: first-class predicate. + ANCHOR_RESOLVES = "anchor_resolves" + #: ``text`` is present on the current frame (tolerant OCR presence check). + TEXT_PRESENT = "text_present" + #: ``text`` is NOT present on the current frame. + TEXT_ABSENT = "text_absent" + #: The run's value for parameter ``param`` equals ``value`` (string compare). + PARAM_EQUALS = "param_equals" + AND = "and" + OR = "or" + NOT = "not" + + +class Predicate(BaseModel): + """A deterministic condition over observed state (RFC §2.2 ``Predicate``). + + Used two ways in Phase 1: as a ``Step.wait_until`` readiness predicate (the + replayer polls it, BOUNDED by ``timeout_s``, and HALTS on timeout -- never + proceeds-anyway) and as the condition inside a ``Guard``. Model-free by + construction (see ``runtime.replayer._predicate_holds``); an unknown kind + fails safe (does not hold). + """ + + kind: PredicateKind + anchor: Optional[Anchor] = None # ANCHOR_RESOLVES + text: Optional[str] = None # TEXT_PRESENT / TEXT_ABSENT + param: Optional[str] = None # PARAM_EQUALS + value: Optional[str] = None # PARAM_EQUALS + intent: Optional[str] = Field( + default=None, + description="Human-readable label (also the resolution-ladder intent " + "for an ANCHOR_RESOLVES predicate).", + ) + operands: list["Predicate"] = Field( + default_factory=list, description="Sub-predicates for AND / OR / NOT." + ) + timeout_s: float = Field( + default=5.0, + description="wait_until bound: how long the replayer polls this " + "predicate before HALTing (fail-safe; never proceed-anyway).", + ) + + +class Guard(BaseModel): + """A deterministic precondition on a step (RFC §2.2, Phase 1 scope). + + ``predicate`` is evaluated over the step's entry frame. When it does NOT + hold, ``on_unmet`` decides: ``"halt"`` (the DEFAULT -- the safe direction + for an unmet precondition, per the RFC's refuse-rather-than-guess posture) + stops the run naming the step; ``"skip"`` makes the step a no-op success + (the expected-but-optional case, e.g. dismissing a survey modal only when it + appeared -- a guarded branch WITHOUT the Phase-2 state machine). Full + multi-way branching is Phase 2. + """ + + predicate: Predicate + on_unmet: Literal["halt", "skip"] = "halt" + + +Predicate.model_rebuild() # resolve the self-referential `operands` + + class Step(BaseModel): id: str intent: str = Field(description="Human-readable purpose of the step") @@ -206,6 +332,19 @@ class Step(BaseModel): # deployment error that HALTS (never a silent unverifiable write). effects: list["Effect"] = Field(default_factory=list) risk: Literal["reversible", "irreversible"] = "reversible" + # Workflow-program IR, Phase 1 (RFC §6) -- both OPTIONAL and additive; a + # step with neither replays EXACTLY as a v0 step. Orthogonal to effects / + # risk / identity above. + # + # ``wait_until``: a BOUNDED readiness predicate the replayer polls BEFORE + # acting; timeout => HALT (fail-safe, never proceed-anyway). This subsumes + # today's SCROLL closed loop as its first concrete predicate -- a SCROLL + # step's default readiness is "the next anchored step's anchor resolves", + # now expressed as an ANCHOR_RESOLVES predicate (see runtime.replayer). + wait_until: Optional[Predicate] = None + # ``guard``: a deterministic precondition evaluated on the entry frame. + # Unmet => HALT (default) or SKIP the step (see Guard.on_unmet). + guard: Optional[Guard] = None timeout_s: float = 10.0 # Identity-protection audit trail (clicks and anchored TYPE steps): # whether this step's click is guarded by the pre-click identity check @@ -244,6 +383,11 @@ class Workflow(BaseModel): params: dict[str, str] = Field( default_factory=dict, description="param name -> example/default value" ) + # Workflow-program IR, Phase 1 (RFC §2.2, §6): TYPED parameter specs, ADDITIVE + # alongside the frozen ``params`` dict above. Keyed by param name. Empty by + # default, so a v0 bundle is unaffected; when present, the replayer folds each + # spec's ``example`` in as a default and fails fast on a missing required one. + param_specs: dict[str, "ParamSpec"] = Field(default_factory=dict) secret_params: list[str] = Field( default_factory=list, description=( @@ -341,6 +485,10 @@ class StepResult(BaseModel): step_id: str intent: str ok: bool + # Workflow-program IR, Phase 1: True when a step was SKIPPED because its + # ``guard`` was unmet with ``on_unmet="skip"`` (a no-op success -- the + # step did not act). False for every executed step; additive. + skipped: bool = False resolution: Optional[Resolution] = None identity: Optional[IdentityCheck] = None # pre-click identity verdict input_verified: Optional[bool] = None # TYPE steps: typed input landed diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 5bfb8dc..d7b2a66 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -44,6 +44,8 @@ Anchor, IdentityCheck, Point, + Predicate, + PredicateKind, Region, Resolution, RunReport, @@ -224,9 +226,17 @@ def run( bundle_dir = Path(bundle_dir) run_dir = Path(run_dir) (run_dir / "steps").mkdir(parents=True, exist_ok=True) - # Caller-supplied params override the recorded defaults; a bundle - # with recorded example values replays without any explicit params. - params = {**workflow.params, **(params or {})} + # Parameter resolution (Workflow-program IR, Phase 1): recorded defaults + # (``params`` dict) plus each TYPED ``param_specs`` example as a default, + # with caller-supplied values overriding both. A v0 bundle (empty + # ``param_specs``) collapses to exactly the old ``{**workflow.params, + # **caller}`` merge. + merged: dict[str, str] = {**workflow.params} + for pname, spec in workflow.param_specs.items(): + if spec.example is not None: + merged.setdefault(pname, spec.example) + merged.update(params or {}) + params = merged report = RunReport( workflow_name=workflow.name, @@ -238,6 +248,33 @@ def run( self._last_click_point: Optional[Point] = None t_run = time.monotonic() + # Fail fast, naming them, when a REQUIRED typed parameter has no value + # (neither a caller override nor a recorded example) -- never start a + # run that will substitute a hole into a step. + missing = sorted( + pname + for pname, spec in workflow.param_specs.items() + if spec.required and not params.get(pname) + ) + if missing: + report.results.append( + StepResult( + step_id="", + intent="validate required workflow parameters", + ok=False, + error=( + "Required workflow parameter(s) not supplied and no " + "recorded example is available: " + + ", ".join(missing) + + " — refusing to run; run aborted" + ), + ) + ) + report.success = False + report.total_ms = (time.monotonic() - t_run) * 1000.0 + report.save(run_dir) + return report + for step_index, step in enumerate(workflow.steps): result = self._run_step( step, @@ -343,6 +380,30 @@ def _run_step( effect_pre_state: Optional[EffectState] = None try: + # Workflow-program IR, Phase 1: evaluate the step's guard + # (precondition) then its wait_until (readiness) BEFORE resolving / + # acting. Both are model-free. A SCROLL step's wait_until is + # consumed by its own closed loop (see _act_scroll), not here. + proceed, gate_error, before_png = self._apply_step_gates( + step, before_png, bundle_dir, params + ) + if before_png is not last_frame: + result.before_png = self._save_step_png( + run_dir, step.id, "before", before_png + ) + last_frame = before_png + if not proceed: + # gate_error None => guard skipped the step (no-op success); + # gate_error set => guard/wait_until HALTed the run. + result.skipped = gate_error is None + result.ok = gate_error is None + result.error = gate_error + result.after_png = self._save_step_png( + run_dir, step.id, "after", last_frame + ) + result.elapsed_ms = (time.monotonic() - t0) * 1000.0 + return result + resolution, matched_region, error = self._resolve_step( step, before_png, bundle_dir ) @@ -774,10 +835,151 @@ def _act( step_index=step_index, bundle_dir=bundle_dir, before_png=before_png, + params=params, ) return f"Step '{step.id}' has unsupported action {step.action!r}" + # -- Workflow-program IR gates: guard + wait_until -------------------------- + + def _apply_step_gates( + self, + step: Step, + before_png: bytes, + bundle_dir: Path, + params: dict[str, str], + ) -> tuple[bool, Optional[str], bytes]: + """Evaluate the step's guard then its wait_until before acting. + + Returns ``(proceed, error, before_png)``: + + - ``(True, None, frame)`` — proceed to resolve/act (frame may have been + re-settled while polling a wait_until predicate). + - ``(False, None, frame)`` — the guard was unmet with ``on_unmet="skip"``: + the step is a no-op success (caller marks it ``skipped``). + - ``(False, error, frame)`` — HALT: an unmet ``halt`` guard, or a + ``wait_until`` predicate that never held within its bound (fail-safe; + the run NEVER proceeds-anyway on an unmet readiness predicate). + + Model-free: predicates are evaluated by :meth:`_predicate_holds`. + """ + # Guard (precondition) on the entry frame. + if step.guard is not None and not self._predicate_holds( + step.guard.predicate, before_png, bundle_dir, params + ): + if step.guard.on_unmet == "skip": + return False, None, before_png + return False, ( + f"Guard precondition for step '{step.id}' ({step.intent}) is " + f"unmet on the current screen " + f"({self._describe_predicate(step.guard.predicate)}) — " + "refusing to act; run aborted" + ), before_png + + # wait_until (readiness). SCROLL consumes its own predicate as the stop + # condition of its closed loop (_act_scroll), so it is skipped here. + if step.wait_until is not None and step.action is not ActionKind.SCROLL: + pred = step.wait_until + deadline = time.monotonic() + pred.timeout_s + while True: + if self._predicate_holds(pred, before_png, bundle_dir, params): + return True, None, before_png + if time.monotonic() >= deadline: + return False, ( + f"wait_until predicate for step '{step.id}' " + f"({step.intent}) did not hold within " + f"{pred.timeout_s:.1f}s " + f"({self._describe_predicate(pred)}) — readiness never " + "reached; refusing to proceed-anyway; run aborted" + ), before_png + time.sleep(self.poll_interval_s) + before_png = self.vision.wait_settled(self.backend) + + return True, None, before_png + + def _predicate_holds( + self, + pred: Predicate, + frame_png: bytes, + bundle_dir: Path, + params: dict[str, str], + ) -> bool: + """Evaluate a Predicate against the current frame / run params. + + Deterministic and ZERO model calls (the $0 runtime guarantee): + ``anchor_resolves`` runs the resolution ladder with NO grounder, + ``text_present`` / ``text_absent`` use the tolerant OCR presence check, + ``param_equals`` is a string compare, and ``and`` / ``or`` / ``not`` + compose. An unknown kind fails safe (does not hold). + """ + kind = pred.kind + if kind is PredicateKind.ANCHOR_RESOLVES: + if pred.anchor is None: + return False + template_png: Optional[bytes] = None + template_path = Path(bundle_dir) / pred.anchor.template + if template_path.is_file(): + template_png = template_path.read_bytes() + return resolve( + pred.anchor, + frame_png, + self.vision, + None, # NEVER ground inside a predicate probe: stay model-free + pred.intent or pred.anchor.ocr_text or "", + template_png=template_png, + viewport=self.backend.viewport, + ) is not None + if kind is PredicateKind.TEXT_PRESENT: + return bool(pred.text) and self.vision.text_present( + frame_png, pred.text + ) + if kind is PredicateKind.TEXT_ABSENT: + return not ( + pred.text and self.vision.text_present(frame_png, pred.text) + ) + if kind is PredicateKind.PARAM_EQUALS: + return ( + pred.param is not None + and str(params.get(pred.param)) == str(pred.value) + ) + if kind is PredicateKind.AND: + return all( + self._predicate_holds(op, frame_png, bundle_dir, params) + for op in pred.operands + ) + if kind is PredicateKind.OR: + return any( + self._predicate_holds(op, frame_png, bundle_dir, params) + for op in pred.operands + ) + if kind is PredicateKind.NOT: + return bool(pred.operands) and not self._predicate_holds( + pred.operands[0], frame_png, bundle_dir, params + ) + return False + + @staticmethod + def _describe_predicate(pred: Predicate) -> str: + """Human-readable one-liner for a predicate (for HALT messages).""" + kind = pred.kind.value if hasattr(pred.kind, "value") else pred.kind + if kind == "anchor_resolves": + label = ( + pred.intent + or (pred.anchor.ocr_text if pred.anchor else None) + or "target" + ) + return f"anchor_resolves({label!r})" + if kind in ("text_present", "text_absent"): + return f"{kind}({pred.text!r})" + if kind == "param_equals": + return f"param_equals({pred.param!r}=={pred.value!r})" + if kind in ("and", "or", "not"): + inner = ", ".join( + Replayer._describe_predicate(op) for op in pred.operands + ) + return f"{kind}({inner})" + return str(kind) + # -- identity verification (pre-click) -------------------------------------- def _verify_identity( @@ -1199,15 +1401,19 @@ def _act_scroll( step_index: int, bundle_dir: Path, before_png: bytes, + params: dict[str, str], ) -> Optional[str]: - """Execute a SCROLL step as a closed loop on the next anchor. - - A recorded scroll's purpose is to bring the next target into view, - so the step scrolls by its recorded delta until the NEXT anchored - step's anchor resolves on a settled frame — not a fixed number of - times. The step probes BEFORE scrolling (a preceding SCROLL step may - already have brought the target into view, making this one a no-op) - and stops as soon as a probe resolves. + """Execute a SCROLL step as a closed loop on a wait_until predicate. + + A recorded scroll's purpose is to bring the next target into view, so + the step scrolls by its recorded delta until a READINESS predicate holds + on a settled frame — not a fixed number of times. That predicate is the + step's own ``wait_until`` when set; otherwise it defaults to + ``ANCHOR_RESOLVES`` on the NEXT anchored step's anchor — today's closed + loop, now expressed as the first concrete ``wait_until`` predicate + rather than a special case (RFC §6). The step probes BEFORE scrolling + (a preceding SCROLL step may already have brought the target into view, + making this one a no-op) and stops as soon as the predicate holds. The loop is bounded: this step may scroll at most ``SCROLL_BUDGET_FACTOR`` times its own recorded distance. On budget @@ -1215,9 +1421,10 @@ def _act_scroll( step is another SCROLL step, which inherits the loop (so a recorded run of N scrolls shares a combined ~2.5x budget). - Falls back to the fixed recorded delta (open-loop, one gesture) when - no later step has an anchor or the recorded delta is zero. Probes - never call the grounder: closed-loop scrolling must stay model-free. + Falls back to the fixed recorded delta (open-loop, one gesture) when no + readiness predicate exists (no ``wait_until`` and no later anchor) or + the recorded delta is zero. Predicate probes never call the grounder: + closed-loop scrolling must stay model-free. Returns: An error string on budget exhaustion (see above) or None. @@ -1225,11 +1432,20 @@ def _act_scroll( dx = step.scroll_dx or 0 dy = step.scroll_dy or 0 next_step = self._next_anchored_step(workflow, step_index) - if next_step is None or (dx == 0 and dy == 0): + # The scroll's readiness is a wait_until predicate: an explicit + # step.wait_until wins; otherwise the next anchor's ANCHOR_RESOLVES. + stop_pred = step.wait_until + if stop_pred is None and next_step is not None: + stop_pred = Predicate( + kind=PredicateKind.ANCHOR_RESOLVES, + anchor=next_step.anchor, + intent=next_step.intent, + ) + if stop_pred is None or (dx == 0 and dy == 0): self.backend.scroll(dx, dy) return None - if self._probe_anchor(next_step, before_png, bundle_dir): + if self._predicate_holds(stop_pred, before_png, bundle_dir, params): return None # target already in view; nothing to scroll increment = math.hypot(dx, dy) @@ -1239,7 +1455,7 @@ def _act_scroll( self.backend.scroll(dx, dy) scrolled += increment frame = self.vision.wait_settled(self.backend) - if self._probe_anchor(next_step, frame, bundle_dir): + if self._predicate_holds(stop_pred, frame, bundle_dir, params): return None following = ( @@ -1250,12 +1466,16 @@ def _act_scroll( if following is not None and following.action is ActionKind.SCROLL: # The next SCROLL step continues the loop with its own budget. return None + target_desc = ( + f"the anchor of step '{next_step.id}' ({next_step.intent})" + if next_step is not None + else self._describe_predicate(stop_pred) + ) return ( f"Step '{step.id}' ({step.intent}): closed-loop scroll exhausted " f"its budget ({scrolled:.0f}px of {budget:.0f}px allowed, " - f"{SCROLL_BUDGET_FACTOR}x the recorded distance) without the " - f"anchor of step '{next_step.id}' ({next_step.intent}) resolving " - "— target never came into view; run aborted" + f"{SCROLL_BUDGET_FACTOR}x the recorded distance) without " + f"{target_desc} resolving — target never came into view; run aborted" ) @staticmethod @@ -1266,30 +1486,6 @@ def _next_anchored_step(workflow: Workflow, step_index: int) -> Optional[Step]: return candidate return None - def _probe_anchor( - self, step: Step, frame_png: bytes, bundle_dir: Path - ) -> bool: - """Single ladder pass for ``step``'s anchor against ``frame_png``. - - Used by the closed-loop scroll to test whether the scroll target is - in view. No timeout retries and no grounder (a probe per scroll - gesture must stay fast and model-free). - """ - assert step.anchor is not None # guaranteed by _next_anchored_step - template_png: Optional[bytes] = None - template_path = Path(bundle_dir) / step.anchor.template - if template_path.is_file(): - template_png = template_path.read_bytes() - return resolve( - step.anchor, - frame_png, - self.vision, - None, # never ground during a scroll probe - step.intent, - template_png=template_png, - viewport=self.backend.viewport, - ) is not None - # -- postconditions -------------------------------------------------------- def _structural_state(self) -> dict[str, Any]: diff --git a/tests/test_program_ir_phase1.py b/tests/test_program_ir_phase1.py new file mode 100644 index 0000000..c900d84 --- /dev/null +++ b/tests/test_program_ir_phase1.py @@ -0,0 +1,336 @@ +"""Workflow-program IR, Phase 1 (RFC docs/design/WORKFLOW_PROGRAM_IR.md §6). + +Additive, backward-compatible: typed parameters (``Workflow.param_specs``), a +per-step ``wait_until`` readiness predicate, and a per-step ``guard`` +precondition. A bundle that declares none of them replays EXACTLY as a v0 +linear bundle (see also tests/test_replayer.py, all of whose cases keep +passing unchanged). + +Backend and vision are faked (reused from test_replayer) -- no Playwright, no +OCR stack, ZERO model calls. +""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.ir import ( + ActionKind, + Guard, + ParamKind, + ParamSpec, + Predicate, + PredicateKind, + Step, + Workflow, +) +from openadapt_flow.runtime.replayer import Replayer +from test_replayer import FakeBackend, FakeVision, Match, click_step, make_png + + +@pytest.fixture() +def bundle(tmp_path): + bundle_dir = tmp_path / "bundle" + (bundle_dir / "templates").mkdir(parents=True) + (bundle_dir / "templates" / "btn.png").write_bytes(make_png((50, 20))) + return bundle_dir + + +@pytest.fixture() +def run_dir(tmp_path): + return tmp_path / "run" + + +def key_step(step_id="k1", key="Enter", *, guard=None, wait_until=None) -> Step: + return Step( + id=step_id, + intent=f"press {key}", + action=ActionKind.KEY, + key=key, + guard=guard, + wait_until=wait_until, + ) + + +# -- worked example: typed param + wait_until + guard, round-tripped ---------- + + +def test_worked_example_roundtrips_through_bundle_save_load(bundle): + """The RFC add-patient-note-style skill expressed with a typed param, a + wait_until, and a guard survives ``Workflow.save`` -> ``Workflow.load`` + byte-for-byte in its Phase-1 fields.""" + wf = Workflow( + name="add-patient-note", + params={"note": "Follow-up in 2 weeks", "encounter_type": "Triage"}, + param_specs={ + "note": ParamSpec( + name="note", type=ParamKind.STRING, + example="Follow-up in 2 weeks", + ), + "encounter_type": ParamSpec( + name="encounter_type", type=ParamKind.ENUM, + example="Triage", choices=["Triage", "Consult"], + ), + }, + steps=[ + # optional-branch step: only click "Triage" when the param says so + Step( + id="s_triage", + intent="click Triage", + action=ActionKind.CLICK, + anchor=click_step().anchor, + guard=Guard( + predicate=Predicate( + kind=PredicateKind.PARAM_EQUALS, + param="encounter_type", value="Triage", + ), + on_unmet="skip", + ), + ), + # readiness-gated typed step + Step( + id="s_note", + intent="type the note", + action=ActionKind.TYPE, + param="note", + wait_until=Predicate( + kind=PredicateKind.TEXT_PRESENT, + text="Save Encounter", timeout_s=2.0, + ), + ), + ], + ) + wf.save(bundle) + loaded = Workflow.load(bundle) + + assert loaded.param_specs["encounter_type"].type is ParamKind.ENUM + assert loaded.param_specs["encounter_type"].choices == ["Triage", "Consult"] + assert loaded.param_specs["note"].example == "Follow-up in 2 weeks" + g = loaded.steps[0].guard + assert g.on_unmet == "skip" + assert g.predicate.kind is PredicateKind.PARAM_EQUALS + assert g.predicate.param == "encounter_type" + w = loaded.steps[1].wait_until + assert w.kind is PredicateKind.TEXT_PRESENT and w.text == "Save Encounter" + assert w.timeout_s == 2.0 + + +# -- typed params supplied / defaulted at replay ----------------------------- + + +def _type_param_workflow() -> Workflow: + return Workflow( + name="wf", + param_specs={ + "note": ParamSpec( + name="note", type=ParamKind.STRING, example="recorded default" + ) + }, + steps=[Step(id="t1", intent="type note", action=ActionKind.TYPE, + param="note")], + ) + + +def test_typed_param_example_is_the_replay_default(bundle, run_dir): + """No caller value -> the ParamSpec.example is substituted (the generalized + 'note value at replay').""" + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + _type_param_workflow(), bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [("type", "recorded default")] + assert report.params["note"] == "recorded default" + + +def test_caller_param_overrides_typed_example(bundle, run_dir): + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + _type_param_workflow(), params={"note": "run value"}, + bundle_dir=bundle, run_dir=run_dir, + ) + assert report.success is True + assert backend.actions == [("type", "run value")] + + +def test_missing_required_param_fails_fast_naming_it(bundle, run_dir): + """A required typed param with no caller value AND no example halts the run + BEFORE any step executes, naming the parameter.""" + wf = Workflow( + name="wf", + param_specs={ + "patient": ParamSpec( + name="patient", type=ParamKind.ENTITY_REF, required=True + ) + }, + steps=[key_step()], + ) + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision()).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert len(report.results) == 1 + assert report.results[0].step_id == "" + assert "patient" in report.results[0].error + assert backend.actions == [] # nothing ran + + +# -- wait_until: bounded readiness, fail-safe HALT on timeout ----------------- + + +def test_wait_until_holds_then_step_proceeds(bundle, run_dir): + vision = FakeVision() + # "Ready" absent on the first probe, present on the second. + vision.text_results = {"Ready": [None, Match((10, 10), (0, 0, 5, 5))]} + backend = FakeBackend() + wf = Workflow(name="wf", steps=[ + key_step(wait_until=Predicate( + kind=PredicateKind.TEXT_PRESENT, text="Ready", timeout_s=1.0)) + ]) + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [("press", "Enter")] + + +def test_wait_until_timeout_halts_and_never_proceeds(bundle, run_dir): + vision = FakeVision() # "Ready" is never present + backend = FakeBackend() + wf = Workflow(name="wf", steps=[ + key_step(wait_until=Predicate( + kind=PredicateKind.TEXT_PRESENT, text="Ready", timeout_s=0.05)) + ]) + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert len(report.results) == 1 + err = report.results[0].error + assert "wait_until" in err and "k1" in err + assert backend.actions == [] # HALT: never proceeded-anyway + assert report.results[0].skipped is False + + +# -- guards: HALT-on-unmet (default) vs SKIP --------------------------------- + + +def test_guard_unmet_halts_by_default(bundle, run_dir): + backend = FakeBackend() + wf = Workflow(name="wf", params={"mode": "user"}, steps=[ + key_step(guard=Guard(predicate=Predicate( + kind=PredicateKind.PARAM_EQUALS, param="mode", value="admin"))) + ]) + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert "Guard precondition" in report.results[0].error + assert backend.actions == [] + + +def test_guard_unmet_skip_makes_step_a_noop_success(bundle, run_dir): + """on_unmet='skip' turns an expected-but-optional step (e.g. dismiss a + survey modal only when present) into a no-op success -- the next step + still runs. This is a guarded branch WITHOUT the Phase-2 state machine.""" + vision = FakeVision() # "Survey" never present + backend = FakeBackend() + wf = Workflow(name="wf", steps=[ + key_step("dismiss", key="Escape", guard=Guard( + predicate=Predicate(kind=PredicateKind.TEXT_PRESENT, text="Survey"), + on_unmet="skip")), + key_step("next", key="Tab"), + ]) + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert report.results[0].skipped is True + assert report.results[0].ok is True + assert report.results[1].skipped is False + assert backend.actions == [("press", "Tab")] # dismiss skipped, next ran + + +def test_guard_met_executes_step_normally(bundle, run_dir): + vision = FakeVision() + vision.text_results = {"Survey": Match((10, 10), (0, 0, 5, 5))} + backend = FakeBackend() + wf = Workflow(name="wf", steps=[ + key_step("dismiss", key="Escape", guard=Guard( + predicate=Predicate(kind=PredicateKind.TEXT_PRESENT, text="Survey"), + on_unmet="skip")), + ]) + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert report.results[0].skipped is False + assert backend.actions == [("press", "Escape")] + + +# -- wait_until subsumes the SCROLL closed loop ------------------------------ + + +def test_scroll_wait_until_predicate_is_the_stop_condition(bundle, run_dir): + """A SCROLL step's readiness is a wait_until predicate: with an explicit + text_present predicate the scroll loops until that text appears -- the same + machinery that, by default, waits on the next anchor's ANCHOR_RESOLVES.""" + vision = FakeVision() + # "Bottom" absent on the pre-scroll probe, present after the first scroll. + vision.text_results = {"Bottom": [None, Match((10, 10), (0, 0, 5, 5))]} + backend = FakeBackend() + scroll = Step( + id="sc1", intent="scroll to bottom", action=ActionKind.SCROLL, + scroll_dx=0, scroll_dy=400, + wait_until=Predicate(kind=PredicateKind.TEXT_PRESENT, text="Bottom"), + ) + wf = Workflow(name="wf", steps=[scroll, key_step("k1", "Enter")]) + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [("scroll", 0, 400), ("press", "Enter")] + + +def test_default_scroll_still_waits_on_next_anchor(bundle, run_dir): + """No explicit wait_until: the SCROLL default readiness is ANCHOR_RESOLVES + on the next anchored step -- today's closed loop, now a predicate. Probe + misses pre-scroll (local+global), resolves after the first scroll.""" + vision = FakeVision() + target = Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95) + vision.template_results = [None, None, target, target] + backend = FakeBackend() + scroll = Step(id="sc1", intent="scroll", action=ActionKind.SCROLL, + scroll_dx=0, scroll_dy=400) + wf = Workflow(name="wf", steps=[scroll, click_step()]) + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [("scroll", 0, 400), ("click", 110, 105, False)] + + +# -- back-compat: a bundle with none of the Phase-1 fields is unchanged ------- + + +def test_no_new_fields_replays_exactly_as_before(bundle, run_dir): + """A workflow declaring no param_specs / guard / wait_until behaves + identically to a v0 bundle: same actions, same success, skipped=False.""" + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95) + ] + backend = FakeBackend() + wf = Workflow(name="wf", steps=[click_step(), + key_step("k1", "Enter")]) + # sanity: the additive fields default to empty/None + assert wf.param_specs == {} + assert wf.steps[0].guard is None and wf.steps[0].wait_until is None + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [("click", 110, 105, False), ("press", "Enter")] + assert all(r.skipped is False for r in report.results) + assert report.model_calls == 0 # $0 runtime preserved