From 093749ac3364b23203decf1d3e5ffd1af1a5b1ac Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 11:39:18 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20workflow-program=20IR=20Phase=201?= =?UTF-8?q?=20=E2=80=94=20typed=20params,=20guards,=20wait=5Funtil=20(addi?= =?UTF-8?q?tive,=20back-compatible)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the RFC's Phase 1 (docs/design/WORKFLOW_PROGRAM_IR.md): the first additive, backward-compatible step from a linear macro IR toward a parameterized program. Typed parameters on Workflow (substituted at replay), an optional per-step guard (deterministic precondition; fail-safe), and wait_until (bounded readiness predicate that subsumes the SCROLL closed-loop). A bundle with none of these replays byte-identically to today. $0 / zero model calls. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- openadapt_flow/compiler/compile.py | 14 ++ openadapt_flow/ir.py | 148 +++++++++++++ openadapt_flow/runtime/replayer.py | 284 ++++++++++++++++++++---- tests/test_program_ir_phase1.py | 336 +++++++++++++++++++++++++++++ 4 files changed, 738 insertions(+), 44 deletions(-) create mode 100644 tests/test_program_ir_phase1.py diff --git a/openadapt_flow/compiler/compile.py b/openadapt_flow/compiler/compile.py index ef59ae5..7daa2b7 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, @@ -1063,6 +1065,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() + }, steps=steps, ) diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index e09a44b..7acf66b 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") @@ -194,6 +320,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 @@ -232,6 +371,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) steps: list[Step] = Field(default_factory=list) # -- bundle I/O --------------------------------------------------------- @@ -321,6 +465,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 4b3c535..9c3fed5 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -43,6 +43,8 @@ Anchor, IdentityCheck, Point, + Predicate, + PredicateKind, Region, Resolution, RunReport, @@ -211,9 +213,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, @@ -225,6 +235,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, @@ -330,6 +367,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 ) @@ -735,10 +796,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( @@ -1160,15 +1362,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 @@ -1176,9 +1382,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. @@ -1186,11 +1393,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) @@ -1200,7 +1416,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 = ( @@ -1211,12 +1427,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 @@ -1227,30 +1447,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 From 6cd7c18439dbc752f0cd6f68ddad712d199a7f86 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 12:25:47 -0400 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20interactive=20disambiguation=20?= =?UTF-8?q?=E2=80=94=20Socrates-style=20compile-time=20questions=20?= =?UTF-8?q?=E2=86=92=20guards/params=20(ask,=20don't=20guess)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the RFC (docs/design/WORKFLOW_PROGRAM_IR.md §3 [3]) induction stage: where a single demonstration under-specifies intent, surface CONCRETE multiple-choice questions and apply the answer deterministically as a Phase-1 guard/param — instead of silently freezing an accidental interpretation. New module openadapt_flow/compiler/disambiguation.py detects three ambiguity kinds structurally (ZERO model calls): - parameter candidate — an untagged typed value → ParamSpec + param binding - absent-result handling — an identity-armed entity selection after a search with no 0/>1-match branch → Guard(ANCHOR_RESOLVES, on_unmet="halt") - optional dialog — a once-handled popup → Guard(TEXT_PRESENT, on_unmet="skip") Answers map to #71's Guard/Predicate/ParamSpec types verbatim (no new IR fields). Refuse-rather-than-guess (mirrors runtime.identity): an UNANSWERED consequential ambiguity (one gating an irreversible write) is flagged and the resolved skill is marked NOT certified until answered — never silently defaulted. Non-consequential unanswered ambiguities fall back to a safe no-op default. Core is a pure, testable API — detect_ambiguities(workflow) and apply_answers(workflow, answers) — plus a thin `disambiguate` CLI subcommand (interactive prompt or --answers JSON). compile.py is UNCHANGED; disambiguation is an opt-in pass over a compiled bundle. Stacks on #71 (feat/workflow-program-ir-phase1); retarget base to main after #71 merges. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- openadapt_flow/__main__.py | 78 +++ openadapt_flow/compiler/disambiguation.py | 583 ++++++++++++++++++++++ tests/test_disambiguation.py | 334 +++++++++++++ 3 files changed, 995 insertions(+) create mode 100644 openadapt_flow/compiler/disambiguation.py create mode 100644 tests/test_disambiguation.py diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index db03b46..09d61f5 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -279,6 +279,60 @@ def _cmd_certify(args: argparse.Namespace) -> int: return 0 if report.passed else 2 +def _cmd_disambiguate(args: argparse.Namespace) -> int: + import json + + from openadapt_flow.compiler.disambiguation import ( + apply_answers, + detect_ambiguities, + ) + from openadapt_flow.ir import Workflow + + bundle = Path(args.bundle) + workflow = Workflow.load(bundle) + questions = detect_ambiguities(workflow) + + if not questions: + print("No ambiguities detected; the demo is fully specified.") + return 0 + + answers: dict[str, str] = {} + if args.answers: + answers = json.loads(Path(args.answers).read_text()) + elif args.interactive: + # Thin interactive wrapper -- prompts a human, then calls the same API + # the tests drive directly. The core stays non-interactive. + for q in questions: + print(f"\n{q.prompt}") + for opt in q.options: + print(f" ({opt.key}) {opt.label}") + tag = "" if q.consequential else f" [default: {q.default_key}]" + reply = input(f"Answer for {q.id}{tag}: ").strip() + if reply: + answers[q.id] = reply + else: + # Non-interactive listing: surface the questions and exit nonzero if + # any is a consequential (must-answer) ambiguity. + for q in questions: + flag = " (CONSEQUENTIAL)" if q.consequential else "" + print(f"\n[{q.kind.value}] {q.id}{flag}\n {q.prompt}") + for opt in q.options: + print(f" ({opt.key}) {opt.label}") + consequential = any(q.consequential for q in questions) + print( + f"\n{len(questions)} question(s) detected. Re-run with " + "--interactive or --answers to resolve." + ) + return 2 if consequential else 0 + + result = apply_answers(workflow, answers) + print(result.render()) + if args.write: + result.workflow.save(bundle) + print(f"Resolved workflow written to {bundle}") + return 0 if result.certified else 2 + + def _cmd_emit_skill(args: argparse.Namespace) -> int: from openadapt_flow.emit.skill import emit_skill @@ -472,6 +526,30 @@ def build_parser() -> argparse.ArgumentParser: ) p.set_defaults(func=_cmd_certify) + p = sub.add_parser( + "disambiguate", + help=( + "Surface compile-time multiple-choice questions for an ambiguous " + "demo and apply the answers as guards/params (ask, don't guess)" + ), + ) + p.add_argument("bundle", help="Workflow bundle directory") + p.add_argument( + "--interactive", + action="store_true", + help="Prompt for each question on the terminal", + ) + p.add_argument( + "--answers", + help="JSON file mapping question id -> chosen option key", + ) + p.add_argument( + "--write", + action="store_true", + help="Save the resolved workflow back into the bundle", + ) + p.set_defaults(func=_cmd_disambiguate) + p = sub.add_parser( "emit-skill", help="Emit an Agent Skills folder for a bundle" ) diff --git a/openadapt_flow/compiler/disambiguation.py b/openadapt_flow/compiler/disambiguation.py new file mode 100644 index 0000000..8e527e0 --- /dev/null +++ b/openadapt_flow/compiler/disambiguation.py @@ -0,0 +1,583 @@ +"""Interactive disambiguation: compile-time Socrates-style questions. + +A single demonstration under-determines the operator's intent (RFC +``docs/design/WORKFLOW_PROGRAM_IR.md`` §1, §3): the same trace is consistent +with several distinct programs. The compiler already *guesses* at some of +these (volatility mining, the ``exclude_texts`` param machinery) -- but where +the guess is CONSEQUENTIAL, guessing silently freezes an accidental +interpretation. This module implements the RFC §3 step [3] induction stage: +where candidate interpretations disagree, **do not guess -- ask the operator a +concrete, grounded, multiple-choice question**, and apply the chosen answer +DETERMINISTICALLY as a Phase-1 guard/param (``ir.Guard`` / ``ir.ParamSpec``, +from the workflow-program IR). + +Detected ambiguity kinds (all detected structurally, ZERO model calls): + +- **parameter candidate** -- a value typed in the demo that was NOT tagged as + a parameter is plausibly a parameter the author forgot to mark, not a fixed + constant (RFC §1 "which literal values are parameters"). Answer maps to a + ``ParamSpec`` + param binding (make it vary per run) or a no-op (keep the + literal). +- **absent-result handling** -- a step that selects a SEARCHED entity has no + recorded branch for the 0-results / >1-match case (RFC §1 + "absent-result handling", ``docs/LIMITS.md`` "no conditionals"). Answer maps + to a ``Guard`` on the selection step (the searched entity must resolve on the + results screen, else HALT) -- so a 0-result run halts instead of clicking the + recorded position blindly. +- **optional dialog** -- a popup dismissed once in the demo whose + expected-vs-exceptional status is unknown (RFC §1 "branches and + expected-vs-exceptional popups"). Answer maps to a guarded/optional step + (``Guard`` with ``on_unmet="skip"`` gated on the dialog's presence -- the RFC + "guarded branch WITHOUT the Phase-2 state machine") or a no-op (always + expected). + +**Refuse rather than guess** (mirrors ``runtime.identity`` and the RFC §3 step +[5] quarantine): an UNANSWERED ambiguity on a CONSEQUENTIAL step (one that, or +whose downstream, performs an irreversible write) is NOT silently defaulted -- +it is flagged and the resolved skill is marked NOT certified until the operator +answers. Non-consequential unanswered ambiguities fall back to the conservative +no-op default (leave the demo interpretation as recorded). + +The CORE is non-interactive and testable: :func:`detect_ambiguities` and +:func:`apply_answers` are pure functions over a :class:`~openadapt_flow.ir. +Workflow`. A thin CLI wrapper (``openadapt-flow disambiguate``) prompts a human +and calls the same API. + +Touch-points (kept minimal, per the stacking constraint on PR #71): +- Reuses #71's ``ir.Guard`` / ``ir.Predicate`` / ``ir.ParamSpec`` verbatim; adds + NO new IR fields. +- ``compiler/compile.py`` is UNCHANGED -- disambiguation runs as an opt-in pass + over an already-compiled bundle (``detect_ambiguities`` / ``apply_answers``), + not inside ``compile_recording``. +- ``__main__.py`` gains one thin ``disambiguate`` subcommand. +""" + +from __future__ import annotations + +import re +from enum import Enum +from typing import Optional + +from pydantic import BaseModel, Field + +from openadapt_flow.ir import ( + ActionKind, + Guard, + ParamKind, + ParamSpec, + Predicate, + PredicateKind, + Step, + Workflow, +) + +# A typed value shorter than this (stripped) is too weak to be a meaningful +# parameter candidate (single keystrokes, "OK") -- skip it. +MIN_PARAM_VALUE_CHARS = 2 + +# Dialog-dismissal labels: a click whose target label / intent matches one of +# these is plausibly dismissing an optional popup rather than driving the main +# task. Matched case-insensitively on WORD boundaries so "ok" does not fire on +# "lookup" and "close" does not fire on "closed". Conservative on purpose -- +# an over-broad list would flag ordinary navigation as an optional dialog. +_DIALOG_STEMS: tuple[str, ...] = ( + r"dismiss", + r"close", + r"okay", + r"ok", + r"accept", + r"got\s*it", + r"no\s*thanks", + r"not\s*now", + r"maybe\s*later", + r"survey", +) +_DIALOG_RE = re.compile( + r"\b(?:" + "|".join(_DIALOG_STEMS) + r")\b", re.IGNORECASE +) + + +class AmbiguityKind(str, Enum): + """The three deterministically-detected under-specification kinds.""" + + PARAMETER_CANDIDATE = "parameter_candidate" + ABSENT_RESULT = "absent_result" + OPTIONAL_DIALOG = "optional_dialog" + + +class OptionEffect(str, Enum): + """What applying an option does to the workflow. Data-driven so + :func:`apply_answers` needs no per-kind branching on option identity.""" + + #: No change -- keep the recorded interpretation (the conservative default). + NONE = "none" + #: Bind the step's typed value to a new ``ParamSpec`` (make it vary/run). + MAKE_PARAM = "make_param" + #: Add a ``Guard`` that HALTs when its predicate is unmet. + GUARD_HALT = "guard_halt" + #: Add a ``Guard`` that SKIPs the step when its predicate is unmet. + GUARD_SKIP = "guard_skip" + + +class QuestionOption(BaseModel): + """One multiple-choice answer, carrying the deterministic effect it + applies. ``policy`` records a run-time strategy the operator intends that + Phase 1 cannot yet express (e.g. "pick the newest match") -- it is stored + on the applied guard's predicate intent as an audit note; the Phase-1 + realization is always the safe HALT.""" + + key: str + label: str + effect: OptionEffect + on_unmet: Optional[str] = None # "halt" | "skip" for GUARD_* effects + policy: Optional[str] = None # deferred run-time strategy (audit note) + + +class DisambiguationQuestion(BaseModel): + """A concrete, screenshot-grounded multiple-choice question about one + ambiguity, plus the data :func:`apply_answers` needs to apply any answer + deterministically.""" + + id: str = Field(description="Stable ':' identifier") + kind: AmbiguityKind + step_id: str + prompt: str + options: list[QuestionOption] + default_key: str = Field( + description="The conservative no-op-or-safe option, used only for " + "NON-consequential unanswered ambiguities." + ) + consequential: bool = Field( + description="True when a wrong answer could cause a wrong irreversible " + "action (this or a downstream step is irreversible). Unanswered " + "consequential questions are refused, never defaulted." + ) + evidence: str = Field( + description="The concrete demonstrated fact the question is about " + "(the typed value, the selected entity, the dialog label)." + ) + # Apply-time payloads (kind-specific; unused fields stay None). + param_name: Optional[str] = None # PARAMETER_CANDIDATE + example_value: Optional[str] = None # PARAMETER_CANDIDATE + dialog_text: Optional[str] = None # OPTIONAL_DIALOG + + def option(self, key: str) -> QuestionOption: + for opt in self.options: + if opt.key == key: + return opt + raise ValueError( + f"{self.id}: unknown answer {key!r} " + f"(choices: {', '.join(o.key for o in self.options)})" + ) + + +class DisambiguationResult(BaseModel): + """Outcome of applying answers: the resolved workflow, plus the audit + trail (which questions were applied, which consequential ones remain + unresolved, and the certification verdict).""" + + workflow: Workflow + questions: list[DisambiguationQuestion] + applied: dict[str, str] = Field( + default_factory=dict, description="question id -> chosen option key" + ) + defaulted: list[str] = Field( + default_factory=list, + description="Non-consequential questions left at their safe default.", + ) + unresolved_consequential: list[str] = Field( + default_factory=list, + description="Consequential questions with no answer -- the reason the " + "skill is not certified.", + ) + + @property + def certified(self) -> bool: + """False iff a consequential ambiguity is unresolved (refuse rather + than guess -- mirrors the identity gate).""" + return not self.unresolved_consequential + + def render(self) -> str: + lines = [ + f"Disambiguation: {len(self.questions)} question(s), " + f"{len(self.applied)} answered, {len(self.defaulted)} defaulted, " + f"{len(self.unresolved_consequential)} unresolved (consequential)." + ] + for q in self.questions: + if q.id in self.applied: + mark, note = "answered", self.applied[q.id] + elif q.id in self.defaulted: + mark, note = "defaulted", q.default_key + else: + mark, note = "UNRESOLVED", "(consequential -- must answer)" + lines.append(f" [{mark}] {q.id}: {note}") + verdict = "CERTIFIED" if self.certified else "NOT CERTIFIED" + lines.append( + f"Certification: {verdict}" + + ( + "" + if self.certified + else " -- resolve the consequential question(s) above." + ) + ) + return "\n".join(lines) + + +# -- consequentiality --------------------------------------------------------- + + +def _is_consequential(workflow: Workflow, step_index: int) -> bool: + """True when the ambiguity at ``step_index`` gates an irreversible action: + the step itself or any LATER step is ``risk="irreversible"``. A wrong + answer then risks a wrong write, so the question must be answered, never + silently defaulted (RFC §3 [5] quarantine; ``docs/LIMITS.md`` posture).""" + return any(s.risk == "irreversible" for s in workflow.steps[step_index:]) + + +# -- parameter-name synthesis ------------------------------------------------- + + +def _slug(text: str) -> str: + """Deterministic short identifier from a typed value: lowercase, first + three word-tokens joined by underscore, non-alphanumerics dropped.""" + words = re.findall(r"[A-Za-z0-9]+", text.lower()) + return "_".join(words[:3]) + + +def _unique_param_name(base: str, taken: set[str], fallback: str) -> str: + name = base or fallback + if name not in taken: + return name + i = 2 + while f"{name}_{i}" in taken: + i += 1 + return f"{name}_{i}" + + +# -- question synthesis (templated, NOT model-generated) ---------------------- + + +def _parameter_question( + step: Step, consequential: bool, param_name: str +) -> DisambiguationQuestion: + value = step.text or "" + preview = value if len(value) <= 40 else value[:39] + "…" + return DisambiguationQuestion( + id=f"{AmbiguityKind.PARAMETER_CANDIDATE.value}:{step.id}", + kind=AmbiguityKind.PARAMETER_CANDIDATE, + step_id=step.id, + prompt=( + f"The demo typed {preview!r} but did not mark it as a parameter. " + "Is this a FIXED value or a PARAMETER that varies per run?" + ), + options=[ + QuestionOption( + key="fixed", + label="Fixed value -- keep the literal as demonstrated", + effect=OptionEffect.NONE, + ), + QuestionOption( + key="param", + label=( + f"Parameter -- vary per run (bind as '{param_name}', " + "demo value becomes the default)" + ), + effect=OptionEffect.MAKE_PARAM, + ), + ], + default_key="fixed", + consequential=consequential, + evidence=f"typed value {preview!r} at {step.id}", + param_name=param_name, + example_value=value, + ) + + +def _absent_result_question( + step: Step, consequential: bool +) -> DisambiguationQuestion: + entity = None + if step.anchor is not None: + entity = step.anchor.context_text or step.anchor.ocr_text + entity_desc = f"'{entity}'" if entity else "the searched entity" + return DisambiguationQuestion( + id=f"{AmbiguityKind.ABSENT_RESULT.value}:{step.id}", + kind=AmbiguityKind.ABSENT_RESULT, + step_id=step.id, + prompt=( + f"The demo selected {entity_desc} from a search, but showed only " + "the one-match case. When the search returns 0 matches or >1 " + "matches at run time, what should the workflow do?" + ), + options=[ + QuestionOption( + key="halt", + label="Halt the run", + effect=OptionEffect.GUARD_HALT, + on_unmet="halt", + policy="halt", + ), + QuestionOption( + key="newest", + label="Pick the newest match (deferred; halts until unique)", + effect=OptionEffect.GUARD_HALT, + on_unmet="halt", + policy="select_newest", + ), + QuestionOption( + key="compare", + label=( + "Compare a second field (e.g. DOB) and pick the exact " + "match (deferred; halts until unique)" + ), + effect=OptionEffect.GUARD_HALT, + on_unmet="halt", + policy="compare_second_field", + ), + QuestionOption( + key="ask", + label="Ask the operator (durable pause; halts in Phase 1)", + effect=OptionEffect.GUARD_HALT, + on_unmet="halt", + policy="escalate", + ), + ], + default_key="halt", + consequential=consequential, + evidence=f"entity selection at {step.id} ({entity_desc})", + ) + + +def _optional_dialog_question( + step: Step, consequential: bool, dialog_text: str +) -> DisambiguationQuestion: + return DisambiguationQuestion( + id=f"{AmbiguityKind.OPTIONAL_DIALOG.value}:{step.id}", + kind=AmbiguityKind.OPTIONAL_DIALOG, + step_id=step.id, + prompt=( + f"The demo handled a dialog ({dialog_text!r}) once. Is this dialog " + "ALWAYS expected, or does it only appear SOMETIMES?" + ), + options=[ + QuestionOption( + key="always", + label="Always expected -- keep it as a required step", + effect=OptionEffect.NONE, + ), + QuestionOption( + key="sometimes", + label=( + "Only sometimes -- dismiss it when present, skip when " + "absent (guarded optional step)" + ), + effect=OptionEffect.GUARD_SKIP, + on_unmet="skip", + ), + ], + default_key="always", + consequential=consequential, + evidence=f"dialog {dialog_text!r} handled at {step.id}", + dialog_text=dialog_text, + ) + + +# -- detection ---------------------------------------------------------------- + + +def detect_ambiguities(workflow: Workflow) -> list[DisambiguationQuestion]: + """Deterministically detect under-specified points in a compiled workflow + and synthesize one grounded multiple-choice question per point. + + ZERO model calls: the detection is structural (typed-but-untagged values, + search-then-select without a branch, dialog-dismissal labels) and the + question text is templated. Ordering follows step order, so the returned + list is stable. + """ + questions: list[DisambiguationQuestion] = [] + taken: set[str] = set(workflow.params) | set(workflow.param_specs) + seen_type_before = False + + for idx, step in enumerate(workflow.steps): + consequential = _is_consequential(workflow, idx) + + # (1) parameter candidate: an untagged, non-trivial typed value. + if ( + step.action is ActionKind.TYPE + and not step.param + and step.text + and len(step.text.strip()) >= MIN_PARAM_VALUE_CHARS + ): + base = _slug(step.text) + name = _unique_param_name(base, taken, f"value_{idx}") + taken.add(name) + questions.append( + _parameter_question(step, consequential, name) + ) + + # (2) absent-result: an identity-armed entity selection that follows a + # typed search query, with no branch for 0/>1 matches yet. + if ( + step.action in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK) + and step.guard is None + and step.identity_armed + and seen_type_before + and not _looks_like_dialog(step) + ): + questions.append(_absent_result_question(step, consequential)) + + # (3) optional dialog: a click that dismisses a popup, no guard yet. + if ( + step.action in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK) + and step.guard is None + and _looks_like_dialog(step) + ): + dialog_text = _dialog_text(step) + questions.append( + _optional_dialog_question(step, consequential, dialog_text) + ) + + if step.action is ActionKind.TYPE: + seen_type_before = True + + return questions + + +def _looks_like_dialog(step: Step) -> bool: + label = step.anchor.ocr_text if step.anchor else None + return bool(_DIALOG_RE.search(f"{step.intent} {label or ''}")) + + +def _dialog_text(step: Step) -> str: + """A distinctive label for the dialog, preferring the target's own OCR + text, else the matched keyword from the intent.""" + if step.anchor and step.anchor.ocr_text: + return step.anchor.ocr_text + m = _DIALOG_RE.search(step.intent) + return m.group(0) if m else step.intent + + +# -- application -------------------------------------------------------------- + + +def _apply_option( + workflow: Workflow, q: DisambiguationQuestion, opt: QuestionOption +) -> None: + """Mutate ``workflow`` in place to realize ``opt`` using #71's Phase-1 + IR types. ``workflow`` is expected to be a caller-owned copy.""" + if opt.effect is OptionEffect.NONE: + return + + step = _find_step(workflow, q.step_id) + + if opt.effect is OptionEffect.MAKE_PARAM: + name = q.param_name or _slug(q.example_value or "") or q.step_id + step.param = name + step.text = q.example_value + workflow.params[name] = q.example_value or "" + workflow.param_specs[name] = ParamSpec( + name=name, + type=ParamKind.STRING, + example=q.example_value, + required=True, + ) + return + + if opt.effect in (OptionEffect.GUARD_HALT, OptionEffect.GUARD_SKIP): + predicate = _guard_predicate(q, opt) + step.guard = Guard( + predicate=predicate, + on_unmet=opt.on_unmet or "halt", + ) + return + + raise ValueError(f"unhandled option effect {opt.effect!r}") + + +def _guard_predicate( + q: DisambiguationQuestion, opt: QuestionOption +) -> Predicate: + if q.kind is AmbiguityKind.ABSENT_RESULT: + # The searched entity must RESOLVE on the results screen; on 0 results + # (or a screen where it does not resolve) the guard is unmet and the + # run halts instead of clicking the recorded position blindly. The + # richer run-time strategy the operator chose (newest / DOB compare / + # escalate) is recorded as the predicate intent for a later phase. + note = f" [policy={opt.policy}]" if opt.policy else "" + return Predicate( + kind=PredicateKind.ANCHOR_RESOLVES, + intent=f"searched entity resolves before selection{note}", + ) + if q.kind is AmbiguityKind.OPTIONAL_DIALOG: + # The step runs only when the dialog is actually present; otherwise it + # is a no-op success (on_unmet="skip") -- a guarded branch without the + # Phase-2 state machine (RFC §2.2). + return Predicate( + kind=PredicateKind.TEXT_PRESENT, + text=q.dialog_text, + intent="optional dialog is present", + ) + raise ValueError(f"no guard predicate for kind {q.kind!r}") + + +def _find_step(workflow: Workflow, step_id: str) -> Step: + for step in workflow.steps: + if step.id == step_id: + return step + raise ValueError(f"no step {step_id!r} in workflow {workflow.name!r}") + + +def apply_answers( + workflow: Workflow, answers: Optional[dict[str, str]] = None +) -> DisambiguationResult: + """Resolve a workflow's ambiguities from an answers map. + + Pure and deterministic -- no model calls, no I/O. ``answers`` maps a + question id (see :func:`detect_ambiguities`) to a chosen option key. + + For each detected ambiguity: + * answered -> apply the chosen option deterministically (ParamSpec / + Guard, using #71's Phase-1 IR types); + * unanswered + NON-consequential -> apply the conservative default + (typically a no-op that keeps the demonstrated interpretation); + * unanswered + CONSEQUENTIAL -> refuse: leave the step untouched and flag + it. The result is then NOT certified (``result.certified is False``), + mirroring the identity ladder's refuse-rather-than-guess posture -- the + skill must not silently freeze an accidental interpretation on a step + that gates an irreversible write. + + An unknown option key for a question raises ``ValueError`` (an invalid + answer is not silently ignored). + + Returns: + A :class:`DisambiguationResult` with the resolved (copied) workflow and + the full audit trail. + """ + answers = answers or {} + resolved = workflow.model_copy(deep=True) + questions = detect_ambiguities(resolved) + valid_ids = {q.id for q in questions} + for qid in answers: + if qid not in valid_ids: + raise ValueError( + f"answer for unknown question {qid!r} " + f"(questions: {', '.join(sorted(valid_ids)) or 'none'})" + ) + + result = DisambiguationResult(workflow=resolved, questions=questions) + for q in questions: + if q.id in answers: + opt = q.option(answers[q.id]) + _apply_option(resolved, q, opt) + result.applied[q.id] = opt.key + elif q.consequential: + # Refuse rather than guess: do not default a consequential edge. + result.unresolved_consequential.append(q.id) + else: + opt = q.option(q.default_key) + _apply_option(resolved, q, opt) + result.defaulted.append(q.id) + + # Re-attach the (possibly mutated) workflow. + result.workflow = resolved + return result diff --git a/tests/test_disambiguation.py b/tests/test_disambiguation.py new file mode 100644 index 0000000..9dfd5d5 --- /dev/null +++ b/tests/test_disambiguation.py @@ -0,0 +1,334 @@ +"""Interactive disambiguation: compile-time Socrates-style questions. + +Covers the RFC ``docs/design/WORKFLOW_PROGRAM_IR.md`` §3 [3] induction stage: +an ambiguous demo yields the expected grounded multiple-choice questions; +applying answers produces the right Phase-1 IR (``ParamSpec`` / ``Guard``); an +UNANSWERED consequential ambiguity marks the skill NOT certified (refuse rather +than guess); a fully-answered workflow resolves clean. + +Pure API tests -- no backend, no vision, no Playwright, ZERO model calls. +""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.compiler.disambiguation import ( + AmbiguityKind, + OptionEffect, + apply_answers, + detect_ambiguities, +) +from openadapt_flow.ir import ( + ActionKind, + Anchor, + ParamKind, + PredicateKind, + Step, + Workflow, +) + + +def _anchor(label: str, context: str | None) -> Anchor: + return Anchor( + template="templates/x.png", + region=(0, 0, 10, 10), + click_point=(5, 5), + ocr_text=label, + context_text=context, + ) + + +def _select_step(step_id: str, label: str, context: str) -> Step: + """An identity-armed entity selection (a searched patient row).""" + return Step( + id=step_id, + intent=f"click '{label}'", + action=ActionKind.CLICK, + anchor=_anchor(label, context), + identity_armed=True, + ) + + +def ambiguous_workflow(*, with_irreversible: bool = True) -> Workflow: + """A demo exercising all three ambiguity kinds. + + search (type) -> select patient (identity-armed click) -> type an untagged + note -> dismiss a survey dialog -> Save (irreversible). + """ + steps = [ + Step( + id="step_000", + intent="type 'Belford'", + action=ActionKind.TYPE, + text="Belford", # a search query (untagged typed value too) + ), + _select_step("step_001", "Belford, Phil", "Belford, Phil 1980-02-03"), + Step( + id="step_002", + intent="type 'Follow-up in 2 weeks'", + action=ActionKind.TYPE, + text="Follow-up in 2 weeks", # untagged -> parameter candidate + ), + Step( + id="step_003", + intent="click 'Dismiss survey'", + action=ActionKind.CLICK, + anchor=_anchor("Dismiss survey", "We value your feedback"), + identity_armed=False, + ), + ] + if with_irreversible: + steps.append( + Step( + id="step_004", + intent="click 'Save Encounter'", + action=ActionKind.CLICK, + anchor=_anchor("Save Encounter", None), + risk="irreversible", + ) + ) + return Workflow(name="add-patient-note", steps=steps) + + +# -- detection ---------------------------------------------------------------- + + +def test_detects_expected_question_set(): + questions = detect_ambiguities(ambiguous_workflow()) + kinds = {(q.kind, q.step_id) for q in questions} + # Two untagged typed values -> two parameter candidates (search + note), + # one absent-result on the identity-armed selection, one optional dialog. + assert (AmbiguityKind.PARAMETER_CANDIDATE, "step_000") in kinds + assert (AmbiguityKind.PARAMETER_CANDIDATE, "step_002") in kinds + assert (AmbiguityKind.ABSENT_RESULT, "step_001") in kinds + assert (AmbiguityKind.OPTIONAL_DIALOG, "step_003") in kinds + assert len(questions) == 4 + + +def test_all_questions_consequential_when_downstream_irreversible(): + for q in detect_ambiguities(ambiguous_workflow()): + assert q.consequential, q.id + + +def test_questions_not_consequential_without_irreversible_step(): + wf = ambiguous_workflow(with_irreversible=False) + for q in detect_ambiguities(wf): + assert not q.consequential, q.id + + +def test_dialog_click_is_not_also_an_absent_result(): + # The survey-dismiss click is identity_armed=False, so it must NOT be + # mistaken for an entity selection. + questions = detect_ambiguities(ambiguous_workflow()) + absent = [q for q in questions if q.kind is AmbiguityKind.ABSENT_RESULT] + assert [q.step_id for q in absent] == ["step_001"] + + +def test_no_questions_for_fully_specified_workflow(): + wf = Workflow( + name="clean", + params={"note": "hi"}, + steps=[ + Step( + id="s0", + intent="type ", + action=ActionKind.TYPE, + text="hi", + param="note", + ), + ], + ) + assert detect_ambiguities(wf) == [] + + +# -- applying answers --------------------------------------------------------- + + +def test_parameter_answer_creates_paramspec_and_binding(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + note_q = next(q for q in qs if q.step_id == "step_002") + result = apply_answers(wf, {note_q.id: "param"}) + + name = note_q.param_name + assert name in result.workflow.param_specs + spec = result.workflow.param_specs[name] + assert spec.type is ParamKind.STRING + assert spec.example == "Follow-up in 2 weeks" + assert result.workflow.params[name] == "Follow-up in 2 weeks" + step = next(s for s in result.workflow.steps if s.id == "step_002") + assert step.param == name + assert step.text == "Follow-up in 2 weeks" # demo value stays as default + + +def test_parameter_fixed_answer_is_a_noop(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + note_q = next(q for q in qs if q.step_id == "step_002") + result = apply_answers(wf, {note_q.id: "fixed"}) + step = next(s for s in result.workflow.steps if s.id == "step_002") + assert step.param is None + assert result.workflow.param_specs == {} + + +def test_absent_result_answer_installs_halt_guard(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + q = next(q for q in qs if q.kind is AmbiguityKind.ABSENT_RESULT) + result = apply_answers(wf, {q.id: "halt"}) + step = next(s for s in result.workflow.steps if s.id == "step_001") + assert step.guard is not None + assert step.guard.on_unmet == "halt" + assert step.guard.predicate.kind is PredicateKind.ANCHOR_RESOLVES + + +def test_absent_result_strategy_recorded_as_policy_note(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + q = next(q for q in qs if q.kind is AmbiguityKind.ABSENT_RESULT) + result = apply_answers(wf, {q.id: "compare"}) + step = next(s for s in result.workflow.steps if s.id == "step_001") + # Phase-1 realization is the safe halt; the chosen run-time strategy is + # recorded on the predicate intent for a later phase. + assert step.guard.on_unmet == "halt" + assert "compare_second_field" in step.guard.predicate.intent + + +def test_optional_dialog_answer_installs_skip_guard(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + q = next(q for q in qs if q.kind is AmbiguityKind.OPTIONAL_DIALOG) + result = apply_answers(wf, {q.id: "sometimes"}) + step = next(s for s in result.workflow.steps if s.id == "step_003") + assert step.guard is not None + assert step.guard.on_unmet == "skip" + assert step.guard.predicate.kind is PredicateKind.TEXT_PRESENT + assert step.guard.predicate.text == "Dismiss survey" + + +def test_optional_dialog_always_answer_is_a_noop(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + q = next(q for q in qs if q.kind is AmbiguityKind.OPTIONAL_DIALOG) + result = apply_answers(wf, {q.id: "always"}) + step = next(s for s in result.workflow.steps if s.id == "step_003") + assert step.guard is None + + +# -- refuse rather than guess ------------------------------------------------- + + +def test_unanswered_consequential_ambiguity_is_not_certified(): + wf = ambiguous_workflow() # every question is consequential + result = apply_answers(wf, {}) # answer nothing + assert not result.certified + assert len(result.unresolved_consequential) == 4 + # Nothing was silently defaulted onto a consequential step. + assert result.defaulted == [] + # The original steps are untouched (no guessed guard/param). + assert all(s.guard is None for s in result.workflow.steps) + assert result.workflow.param_specs == {} + + +def test_partially_answered_still_not_certified(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + absent = next(q for q in qs if q.kind is AmbiguityKind.ABSENT_RESULT) + result = apply_answers(wf, {absent.id: "halt"}) + assert not result.certified + assert absent.id in result.applied + assert len(result.unresolved_consequential) == 3 + + +def test_fully_answered_workflow_certifies_clean(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + answers = {} + for q in qs: + if q.kind is AmbiguityKind.PARAMETER_CANDIDATE: + answers[q.id] = "param" + elif q.kind is AmbiguityKind.ABSENT_RESULT: + answers[q.id] = "halt" + else: + answers[q.id] = "sometimes" + result = apply_answers(wf, answers) + assert result.certified + assert result.unresolved_consequential == [] + assert len(result.applied) == 4 + + +def test_non_consequential_unanswered_falls_back_to_default(): + wf = ambiguous_workflow(with_irreversible=False) + result = apply_answers(wf, {}) + assert result.certified # no consequential ambiguity to block + assert set(result.defaulted) == {q.id for q in result.questions} + # Param/dialog defaults are no-ops (keep the demo interpretation); the + # absent-result default is the SAFE halt-guard, not a no-op. + assert result.workflow.param_specs == {} + select = next(s for s in result.workflow.steps if s.id == "step_001") + assert select.guard is not None and select.guard.on_unmet == "halt" + dialog = next(s for s in result.workflow.steps if s.id == "step_003") + assert dialog.guard is None + + +# -- input validation + round-trip ------------------------------------------- + + +def test_unknown_answer_key_raises(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + note_q = next(q for q in qs if q.step_id == "step_002") + with pytest.raises(ValueError, match="unknown answer"): + apply_answers(wf, {note_q.id: "bogus"}) + + +def test_answer_for_unknown_question_raises(): + wf = ambiguous_workflow() + with pytest.raises(ValueError, match="unknown question"): + apply_answers(wf, {"parameter_candidate:does_not_exist": "param"}) + + +def test_apply_does_not_mutate_the_input_workflow(): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + note_q = next(q for q in qs if q.step_id == "step_002") + apply_answers(wf, {note_q.id: "param"}) + # The caller's workflow is untouched (apply works on a deep copy). + assert wf.param_specs == {} + assert next(s for s in wf.steps if s.id == "step_002").param is None + + +def test_resolved_workflow_roundtrips_through_bundle(tmp_path): + wf = ambiguous_workflow() + qs = detect_ambiguities(wf) + answers = {} + for q in qs: + answers[q.id] = { + AmbiguityKind.PARAMETER_CANDIDATE: "param", + AmbiguityKind.ABSENT_RESULT: "halt", + AmbiguityKind.OPTIONAL_DIALOG: "sometimes", + }[q.kind] + result = apply_answers(wf, answers) + bundle = tmp_path / "bundle" + result.workflow.save(bundle) + reloaded = Workflow.load(bundle) + + step1 = next(s for s in reloaded.steps if s.id == "step_001") + assert step1.guard.predicate.kind is PredicateKind.ANCHOR_RESOLVES + step3 = next(s for s in reloaded.steps if s.id == "step_003") + assert step3.guard.on_unmet == "skip" + assert reloaded.param_specs # note + search bound as params + + +def test_option_effects_are_data_driven(): + # Every option's effect is a known enum member (guards apply generically). + wf = ambiguous_workflow() + for q in detect_ambiguities(wf): + for opt in q.options: + assert isinstance(opt.effect, OptionEffect) + + +def test_render_reports_certification_verdict(): + wf = ambiguous_workflow() + assert "NOT CERTIFIED" in apply_answers(wf, {}).render()