diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 5ad09be..0f0a1bb 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -309,6 +309,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 @@ -540,6 +594,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()