From af91caa2b83f8eedc28a6b5f7c8e84cc884d849c Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 15:39:22 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20workflow-program=20IR=20Phase=202?= =?UTF-8?q?=20=E2=80=94=20loops,=20branches,=20subflows,=20exception=20pat?= =?UTF-8?q?hs=20(the=20state=20machine)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Evolve the compiled artifact from a linear action list into a parameterized STATE MACHINE (RFC docs/design/WORKFLOW_PROGRAM_IR.md §2), closing the review's "a workflow is not a list of actions" gap. Phase 1 (typed params, guards, wait_until) added the pieces; Phase 2 adds the control flow a trajectory cannot carry: LOOPS over a worklist, guarded BRANCHES, reusable SUBFLOWS, and EXCEPTION paths — the program the PBD literature (Rousillon, WebRobot, Skill-DisCo, PROLEX) says a demonstration compiler must express. IR (openadapt_flow/ir.py), additive and backward-compatible: - State (action | branch | loop | subflow_call | terminal) + Transition (guarded edge) form a ProgramGraph; an action state's payload IS a Phase-1 Step (the unchanged hardened leaf), a transition's guard IS a Phase-1 Predicate. - Relation (worklist) + LoopSpec (bounded per-row body subflow); Workflow gains optional program / subflows / data_sources. When program is None the linear steps list runs exactly as today. - lift_to_program: mechanical degenerate lift (RFC §2.6) — a linear bundle is the single-path graph. Interpreter (runtime/replayer.py): a deterministic graph interpreter ($0, zero model calls) that REUSES the linear per-action pipeline unchanged — every action state runs through _run_step, so identity / effect / risk / heal gates fire identically inside loop bodies and branches. Adds guarded transition selection (first match wins, no-match HALTs fail-safe), bounded worklist loops, subflow dispatch, and on_exception routing (graph try/except); unhandled failures and halt/escalate terminals stop the run. Bounded against non-terminating graphs (step budget + nesting depth). Linear path is byte-for- byte unchanged (program=None branch). Tests (tests/test_program_ir_phase2.py, 18): loop runs body 3x / 0x / run-time worklist / bound enforced; branch takes each arm (param + screen predicate) and dead-ends HALT; subflow reused as loop body AND direct call; on_exception catches a failed action and continues; identity- and effect-gates fire inside a loop body; the lifted linear graph replays byte-identically to the linear replayer; program round-trips through save/load. Full non-e2e suite green in isolation (859 passed; the concurrent-agent FileNotFoundError errors are environmental). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- openadapt_flow/ir.py | 193 ++++++++ openadapt_flow/runtime/replayer.py | 678 +++++++++++++++++++++++++++-- tests/test_program_ir_phase2.py | 660 ++++++++++++++++++++++++++++ 3 files changed, 1491 insertions(+), 40 deletions(-) create mode 100644 tests/test_program_ir_phase2.py diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index c28a570..b7a7162 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -527,6 +527,165 @@ class Step(BaseModel): ) +# -- Workflow-program IR, Phase 2 (RFC docs/design/WORKFLOW_PROGRAM_IR.md §2) -- +# +# The parameterized STATE MACHINE: the control flow a linear action list cannot +# express -- LOOPS over a worklist, guarded BRANCHES, reusable SUBFLOWS, and +# EXCEPTION paths. Built ADDITIVELY on the Phase-1 pieces: a state's action IS a +# Phase-1 ``Step`` (the unchanged, hardened action leaf -- same anchor/identity/ +# effect/risk machinery), a transition's guard IS a Phase-1 ``Predicate``, and a +# branch reuses the SAME model-free predicate evaluation. BACKWARD-COMPATIBLE: +# ``Workflow.program`` is OPTIONAL -- when it is None the runtime executes +# today's linear ``Workflow.steps`` loop byte-for-byte, and a linear bundle +# lifts mechanically to the degenerate single-path graph (``lift_to_program``, +# RFC §2.6). ZERO model calls at run time -- guards, branches, loops, and +# subflow dispatch are all deterministic ($0 replay). + + +class StateKind(str, Enum): + """The kinds of node in a workflow-program graph (RFC §2.2).""" + + ACTION = "action" # perform a Step (today's hardened action leaf) + BRANCH = "branch" # pick an outgoing transition by guard; performs no action + LOOP = "loop" # iterate a worklist, running a body subflow per row + SUBFLOW_CALL = "subflow_call" # invoke a reusable named subflow + TERMINAL = "terminal" # end this (sub)graph: success | halt | escalate + + +class Transition(BaseModel): + """A guarded edge to a target state (RFC §2.2) -- the thing a linear IR + cannot express. + + ``guard`` is a Phase-1 :class:`Predicate` evaluated (model-free) over the + current frame / run params; ``None`` means UNCONDITIONAL (the RFC's ``TRUE`` + edge -- the default fall-through, and the only edge kind a degenerate linear + program has). A state's ``transitions`` are evaluated IN ORDER; the first + whose guard holds wins. Multiple non-``TRUE`` transitions make a multi-way + branch. + """ + + guard: Optional[Predicate] = None + target: str = Field(description="Id of the state this edge leads to") + label: str = Field(default="", description="Human-readable edge label") + + +class Relation(BaseModel): + """A worklist a ``loop`` state iterates over (RFC §2.3). + + Variable-length ``rows``; each row is a mapping of param name -> value that + is bound into the run params in scope for that loop iteration. Rows may be + INLINED here (deterministic, $0 -- the authored/compiled case) or supplied + at run time (``Replayer.run(worklists=...)``) for a genuinely data-dependent + queue whose length is unknown until run time. Either way iteration stays + BOUNDED (see :class:`LoopSpec.max_iterations`). + """ + + name: str + rows: list[dict[str, str]] = Field( + default_factory=list, + description="Inline worklist rows; each binds params for one iteration", + ) + description: str = "" + + +class LoopSpec(BaseModel): + """The body of a ``loop`` state (RFC §2.3; Rousillon / Helena / WebRobot). + + Binds a ``relation`` (worklist) and a ``body`` subflow that runs ONCE PER + ROW, the row's fields merged into the run params for that iteration (so an + ``entity_ref`` param re-resolves by the identity ladder each pass -- + iteration N acts on the RIGHT row, not a recorded pixel position). A + zero-row worklist runs the body ZERO times. Iteration is BOUNDED by + ``max_iterations`` -- a worklist longer than the bound HALTs (fail-safe), + never runs unbounded. + """ + + relation: str = Field(description="Name of the Relation / worklist to loop") + body: str = Field(description="SubflowId run once per row") + var: str = Field( + default="", + description="Optional human label for the loop variable (for reports)", + ) + max_iterations: int = Field( + default=1000, description="Hard upper bound on iterations (fail-safe)" + ) + + +class State(BaseModel): + """A node in the workflow-program graph (RFC §2.2). + + Its ``kind`` selects the payload: ``action`` carries a hardened Phase-1 + :class:`Step`; ``branch`` picks an edge purely by guard; ``loop`` iterates a + worklist; ``subflow_call`` invokes a reusable subgraph; ``terminal`` ends the + (sub)graph. ``transitions`` are the outgoing edges (empty on a terminal, a + single unconditional edge on a degenerate linear node). ``on_exception`` + routes a FAILED action to a local handler instead of aborting the whole run. + """ + + id: str + kind: StateKind + # kind == ACTION: the hardened Phase-1 Step to perform (unchanged leaf -- + # anchor resolution, identity gate, effects, risk all ride along on it). + step: Optional[Step] = None + # kind == LOOP: the worklist + per-row body subflow. + loop: Optional[LoopSpec] = None + # kind == SUBFLOW_CALL: the reusable subflow to invoke, then continue. + subflow: Optional[str] = None + # Outgoing edges, evaluated IN ORDER (first matching guard wins). Empty on a + # terminal; a single unconditional Transition on a degenerate linear node. + transitions: list[Transition] = Field(default_factory=list) + # Local exception handler (RFC §2.4): when this state's action FAILS (a + # resolution / identity / postcondition / effect HALT), route to THIS state + # instead of aborting the whole run -- the graph analog of try/except. None + # (default) => an unhandled failure HALTs the run, exactly as today. + on_exception: Optional[str] = None + # kind == TERMINAL: how this (sub)graph ends. "success" completes normally + # (returns to the caller for a subflow); "halt" / "escalate" stop the ENTIRE + # run (success=False) -- the safe default for an underdetermined/failed path. + outcome: Optional[Literal["success", "halt", "escalate"]] = None + reason: str = "" + + +class ProgramGraph(BaseModel): + """A directed graph of :class:`State`s with a single ``entry`` (RFC §2.2). + + Used both as the top-level program (``Workflow.program``) and as a reusable + subflow (``Workflow.subflows[name]``, or a ``loop`` body). Walked state by + state from ``entry`` until a terminal (or, for a subflow, until it falls off + / reaches a ``success`` terminal, which RETURNS to the caller). + """ + + entry: str + states: dict[str, State] = Field(default_factory=dict) + + +def lift_to_program(workflow: "Workflow") -> ProgramGraph: + """Mechanically lift a linear ``Workflow`` to the degenerate straight-line + program (RFC §2.6): each ``Step[i]`` becomes an ``action`` State with a + single unconditional ``Transition`` to ``Step[i+1]``, and a final ``success`` + terminal. The graph interpreter over this lift replays byte-for-byte + identically to the linear ``Replayer`` -- the proof that "a linear bundle is + the degenerate single-path graph". + """ + states: dict[str, State] = {} + steps = workflow.steps + end_id = "__end__" + for i, step in enumerate(steps): + sid = f"s::{step.id}" + target = end_id if i + 1 >= len(steps) else f"s::{steps[i + 1].id}" + states[sid] = State( + id=sid, + kind=StateKind.ACTION, + step=step, + transitions=[Transition(target=target, label="")], + ) + states[end_id] = State( + id=end_id, kind=StateKind.TERMINAL, outcome="success" + ) + entry = f"s::{steps[0].id}" if steps else end_id + return ProgramGraph(entry=entry, states=states) + + class Workflow(BaseModel): schema_version: int = 1 name: str @@ -552,6 +711,18 @@ class Workflow(BaseModel): ), ) steps: list[Step] = Field(default_factory=list) + # Workflow-program IR, Phase 2 (RFC §2): the parameterized STATE MACHINE. + # ALL optional and additive -- when ``program`` is None the runtime executes + # the linear ``steps`` loop above byte-for-byte (today's behavior); a linear + # bundle carries none of these. When ``program`` is present the runtime + # interprets the graph (loops / branches / subflows / exception paths), + # reusing the SAME per-action machinery (identity/effect/risk/heal gates) for + # every ``action`` state. ``subflows`` are reusable named subgraphs (a loop + # body or a shared component); ``data_sources`` are the worklists loops + # iterate. See ``lift_to_program`` for the degenerate linear lift (RFC §2.6). + program: Optional["ProgramGraph"] = None + subflows: dict[str, "ProgramGraph"] = Field(default_factory=dict) + data_sources: dict[str, "Relation"] = Field(default_factory=dict) # -- bundle I/O --------------------------------------------------------- @@ -661,6 +832,12 @@ class StepResult(BaseModel): # ``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 + # Workflow-program IR, Phase 2: True when this (failed) action state was + # routed to its ``State.on_exception`` handler instead of aborting the run + # (the graph analog of a caught try/except). The result stays ``ok=False`` + # (the action DID fail) but the run continued via the handler; additive, + # default False for every linear-mode and unhandled result. + exception_handled: bool = False resolution: Optional[Resolution] = None identity: Optional[IdentityCheck] = None # pre-click identity verdict input_verified: Optional[bool] = None # TYPE steps: typed input landed @@ -707,6 +884,14 @@ class RunReport(BaseModel): params: dict[str, str] = Field(default_factory=dict) results: list[StepResult] = Field(default_factory=list) success: bool = False + # Workflow-program IR, Phase 2: the outcome of the terminal state the graph + # interpreter ended on ("success" | "halt" | "escalate"), or None for a + # linear-mode run (no program graph) or a run that fell off the graph. The + # ordered ``visited_states`` trace records the state ids the interpreter + # walked (action/branch/loop/subflow/terminal), for the audit trail. Both + # additive and empty/None on a linear run. + terminal_outcome: Optional[str] = None + visited_states: list[str] = Field(default_factory=list) rung_counts: dict[str, int] = Field(default_factory=dict) heal_count: int = 0 model_calls: int = 0 @@ -744,4 +929,12 @@ def save(self, run_dir: Path | str) -> Path: ApiBinding.model_rebuild() Step.model_rebuild() +# Phase-2 state-machine models: State embeds Step (whose Effect forward ref is +# resolved just above), Transition embeds Predicate, ProgramGraph embeds State, +# and Workflow embeds ProgramGraph/Relation -- rebuild in dependency order so +# every forward reference is resolved before Workflow's schema is completed. +Transition.model_rebuild() +LoopSpec.model_rebuild() +State.model_rebuild() +ProgramGraph.model_rebuild() Workflow.model_rebuild() diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 9990ca9..51ae2bc 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -43,12 +43,16 @@ ActionKind, Anchor, IdentityCheck, + LoopSpec, Point, Predicate, PredicateKind, + ProgramGraph, Region, Resolution, RunReport, + State, + StateKind, Step, StepResult, UnarmedStep, @@ -118,6 +122,70 @@ def secret_env_var(param: str) -> str: return f"{SECRET_ENV_PREFIX}{key}" +# Bounds for the Phase-2 program interpreter -- deterministic termination +# guarantees. A worklist loop is bounded per-loop (LoopSpec.max_iterations); the +# graph walk itself is bounded by a total step budget (so an authored cycle of +# always-true transitions can never spin forever) and subflow/loop nesting is +# depth-bounded. +PROGRAM_MAX_STEPS = 100_000 +PROGRAM_MAX_DEPTH = 64 + + +class _GraphStepContext: + """Per-action-state context the program interpreter feeds ``_run_step``. + + Supplies the two pieces the linear loop derives from list position but a + graph lacks: the previously EXECUTED action (for the click-to-focus TYPE + heuristic) and the SCROLL closed-loop successors (the next anchored action + state and the immediately following state). All optional; a None field means + "no such successor" (the scroll loop then falls back to its open-loop + gesture, exactly as at the end of a linear workflow). + """ + + __slots__ = ("prev_action", "next_anchored", "following") + + def __init__( + self, + prev_action: Optional[ActionKind] = None, + next_anchored: Optional[Step] = None, + following: Optional[Step] = None, + ) -> None: + self.prev_action = prev_action + self.next_anchored = next_anchored + self.following = following + + +class _ProgramHalt(Exception): + """Internal signal: an unrecoverable state was reached (an unhandled action + failure, a dead-end with no matching transition, a bound exceeded, or a + ``halt`` / ``escalate`` terminal). Carries the outcome + reason the run + report records. Caught only by ``_interpret_program`` -- never escapes the + Replayer.""" + + def __init__(self, outcome: str, reason: str) -> None: + super().__init__(reason) + self.outcome = outcome + self.reason = reason + + +def _all_workflow_steps(workflow: Workflow): + """Yield every ``Step`` in a workflow for whole-bundle audits. + + Linear bundle: the ``steps`` list. Program bundle: every ``action`` state's + step across the top-level program graph AND all subflows (a loop body's + steps included). Order is stable but not execution order -- used only for + static coverage counting, not for running. + """ + if workflow.program is None: + yield from workflow.steps + return + graphs = [workflow.program, *workflow.subflows.values()] + for graph in graphs: + for state in graph.states.values(): + if state.kind is StateKind.ACTION and state.step is not None: + yield state.step + + class Replayer: """Replays a Workflow against a Backend using injected vision. @@ -234,18 +302,32 @@ def run( workflow: Workflow, *, params: Optional[dict[str, str]] = None, + worklists: Optional[dict[str, list[dict[str, str]]]] = None, bundle_dir: Path, run_dir: Path, save_healed_to: Optional[Path] = None, ) -> RunReport: """Execute the workflow and write a run directory. + A workflow with no ``program`` graph runs the LINEAR ``steps`` loop + exactly as before (byte-for-byte back-compatible). A workflow carrying a + ``program`` (Workflow-program IR Phase 2) is interpreted as a STATE + MACHINE -- loops, branches, subflows, exception paths -- reusing the + IDENTICAL per-action machinery (resolve / identity gate / effect verify / + risk gate / heal) for every ``action`` state. Both paths make ZERO model + calls on the deterministic path (the $0 runtime guarantee). + Args: workflow: The compiled workflow. Heals are applied to this in-memory object as the run progresses. params: Values for parameterized TYPE steps (``step.param``). Parameters not supplied here fall back to the recorded example/default values in ``workflow.params``. + worklists: Program mode only -- run-time worklist rows for ``loop`` + states, keyed by relation name (each a list of param-dict rows). + Overrides any inline ``Workflow.data_sources`` rows of the same + name; the mechanism that makes a loop VARIABLE-LENGTH at run + time. Ignored for a linear (no-program) workflow. bundle_dir: The workflow bundle directory (source of template crops). run_dir: Output directory for report.json, per-step screenshots @@ -254,9 +336,9 @@ def run( workflow.json + new and unchanged template crops) here. Returns: - The RunReport (also saved as ``run_dir/report.json``). The run - aborts at the first failed step; ``success`` is True only if - every step completed. + The RunReport (also saved as ``run_dir/report.json``). A linear run + aborts at the first failed step; a program run ends at a terminal + state (or an unhandled failure). ``success`` reflects that outcome. """ bundle_dir = Path(bundle_dir) run_dir = Path(run_dir) @@ -310,39 +392,40 @@ def run( report.save(run_dir) return report - for step_index, step in enumerate(workflow.steps): - result = self._run_step( - step, - workflow=workflow, - step_index=step_index, + if workflow.program is not None: + # Workflow-program IR, Phase 2: interpret the STATE MACHINE. The + # interpreter appends StepResults (and accounts each) exactly as the + # linear loop does, but walks the graph -- loops / branches / + # subflows / exception paths -- instead of a straight list. It sets + # report.success / terminal_outcome from the terminal it ends on. + self._interpret_program( + workflow, params=params, + worklists=worklists or {}, bundle_dir=bundle_dir, run_dir=run_dir, + report=report, new_crops=new_crops, ) - report.results.append(result) - if result.ok and result.resolution is not None: - rung = result.resolution.rung - report.rung_counts[rung] = report.rung_counts.get(rung, 0) + 1 - if rung == "grounder": - report.model_calls += 1 - elif result.ok and result.actuation == "api": - # API-tier actuation has no visual resolution rung; count it - # under "api" so the run report shows the deterministic top of - # the ladder in the same place as the visual rungs. Zero model - # calls (the $0 guarantee holds on the API path too). - report.rung_counts["api"] = report.rung_counts.get("api", 0) + 1 - # Drift-oracle state-verifier calls are model calls too (honest - # accounting: a rescued run is not a zero-model run). - report.model_calls += result.drift_oracle_calls - if result.heal is not None: - report.heal_count += 1 - if not result.ok: - break + else: + for step_index, step in enumerate(workflow.steps): + result = self._run_step( + step, + workflow=workflow, + step_index=step_index, + params=params, + bundle_dir=bundle_dir, + run_dir=run_dir, + new_crops=new_crops, + ) + report.results.append(result) + self._account_result(report, result) + if not result.ok: + break - report.success = len(report.results) == len(workflow.steps) and all( - result.ok for result in report.results - ) + report.success = len(report.results) == len(workflow.steps) and all( + result.ok for result in report.results + ) report.total_ms = (time.monotonic() - t_run) * 1000.0 if save_healed_to is not None: @@ -367,7 +450,11 @@ def _record_identity_coverage( identity verification (docs/LIMITS.md), so each one is listed by id with the compile-time reason for the operator. """ - for step in workflow.steps: + # Cover the whole bundle: the linear ``steps`` list, OR (program mode) + # every ``action`` state's step across the program graph and all + # subflows -- so a hand-authored/compiled PROGRAM's identity coverage is + # audited exactly as a linear bundle's is. + for step in _all_workflow_steps(workflow): if step.anchor is None or step.action not in ( ActionKind.CLICK, ActionKind.DOUBLE_CLICK, @@ -391,6 +478,472 @@ def _record_identity_coverage( ) ) + @staticmethod + def _account_result(report: RunReport, result: StepResult) -> None: + """Fold one StepResult's rung / model-call / heal counts into the + report. Shared by the linear loop and the program interpreter so both + account a step identically (a rescued or grounded run is never counted + as a zero-model run).""" + if result.ok and result.resolution is not None: + rung = result.resolution.rung + report.rung_counts[rung] = report.rung_counts.get(rung, 0) + 1 + if rung == "grounder": + report.model_calls += 1 + elif result.ok and result.actuation == "api": + # API-tier actuation has no visual resolution rung; count it under + # "api" so the report shows the deterministic top of the ladder in + # the same place as the visual rungs. Zero model calls. + report.rung_counts["api"] = report.rung_counts.get("api", 0) + 1 + # Drift-oracle state-verifier calls are model calls too (honest + # accounting). + report.model_calls += result.drift_oracle_calls + if result.heal is not None: + report.heal_count += 1 + + # -- Workflow-program IR, Phase 2: the state-machine interpreter ----------- + # + # Deterministic graph interpreter ($0, ZERO model calls on the happy path). + # It REUSES the linear per-action machinery unchanged: every ``action`` + # state is executed by ``_run_step`` -- the SAME settle / resolve / identity + # gate / effect verify / risk gate / heal pipeline the linear replayer runs + # -- so no safety property is weakened by adding control flow AROUND the + # hardened leaf. The interpreter only adds: which state runs next (guarded + # transitions), loops over a worklist, subflow dispatch, and routing a + # failed action to a local exception handler instead of aborting. + + def _interpret_program( + self, + workflow: Workflow, + *, + params: dict[str, str], + worklists: dict[str, list[dict[str, str]]], + bundle_dir: Path, + run_dir: Path, + report: RunReport, + new_crops: dict[str, bytes], + ) -> None: + """Interpret ``workflow.program`` as a state machine, writing results + and the terminal outcome onto ``report``. + + Normal completion (a ``success`` terminal, or falling off a graph with + no further transition) => ``report.success = True``. Any ``_ProgramHalt`` + -- an unhandled action failure, a dead-end branch, a ``halt`` / + ``escalate`` terminal, or a safety bound exceeded -- stops the whole run + with ``success = False`` and records the reason. + """ + assert workflow.program is not None + self._prev_action: Optional[ActionKind] = None + self._program_step_budget = PROGRAM_MAX_STEPS + try: + self._walk_graph( + workflow.program, + workflow=workflow, + params=params, + worklists=worklists, + bundle_dir=bundle_dir, + run_dir=run_dir, + report=report, + new_crops=new_crops, + depth=0, + ) + report.success = True + if report.terminal_outcome is None: + # Fell off the top graph with no explicit terminal: a clean, + # complete run (every executed state succeeded or was handled). + report.terminal_outcome = "success" + except _ProgramHalt as halt: + report.success = False + report.terminal_outcome = halt.outcome + report.results.append( + StepResult( + step_id="", + intent=f"program {halt.outcome}", + ok=False, + error=halt.reason, + ) + ) + + def _walk_graph( + self, + graph: ProgramGraph, + *, + workflow: Workflow, + params: dict[str, str], + worklists: dict[str, list[dict[str, str]]], + bundle_dir: Path, + run_dir: Path, + report: RunReport, + new_crops: dict[str, bytes], + depth: int, + ) -> None: + """Walk one graph from ``entry`` to a terminal / fall-off. + + Returns normally on completion (a ``success`` terminal RETURNS to the + caller -- for a subflow that means "continue after the call"; for the + top program it means the run succeeded). Raises ``_ProgramHalt`` to stop + the ENTIRE run. + """ + if depth > PROGRAM_MAX_DEPTH: + raise _ProgramHalt( + "halt", + f"program nesting exceeded {PROGRAM_MAX_DEPTH} levels " + "(possible unbounded recursion) — run aborted", + ) + state_id: Optional[str] = graph.entry + while state_id is not None: + self._program_step_budget -= 1 + if self._program_step_budget <= 0: + raise _ProgramHalt( + "halt", + f"program exceeded {PROGRAM_MAX_STEPS} state transitions " + "(possible non-terminating graph) — run aborted", + ) + state = graph.states.get(state_id) + if state is None: + raise _ProgramHalt( + "halt", + f"program references undefined state '{state_id}' — " + "run aborted", + ) + report.visited_states.append(state.id) + + if state.kind is StateKind.TERMINAL: + outcome = state.outcome or "success" + report.terminal_outcome = outcome + if outcome == "success": + return # RETURN to caller / top-program success + raise _ProgramHalt( + outcome, + state.reason + or f"reached '{outcome}' terminal state '{state.id}' — " + "run aborted", + ) + + state_id = self._exec_state( + state, + graph, + workflow=workflow, + params=params, + worklists=worklists, + bundle_dir=bundle_dir, + run_dir=run_dir, + report=report, + new_crops=new_crops, + depth=depth, + ) + # Fell off the graph (a state with no outgoing transition): normal + # completion / subflow return. + + def _exec_state( + self, + state: State, + graph: ProgramGraph, + *, + workflow: Workflow, + params: dict[str, str], + worklists: dict[str, list[dict[str, str]]], + bundle_dir: Path, + run_dir: Path, + report: RunReport, + new_crops: dict[str, bytes], + depth: int, + ) -> Optional[str]: + """Execute one non-terminal state; return the next state id (or None to + fall off the current graph). Raises ``_ProgramHalt`` on an unrecoverable + state.""" + if state.kind is StateKind.ACTION: + return self._exec_action_state( + state, + graph, + workflow=workflow, + params=params, + bundle_dir=bundle_dir, + run_dir=run_dir, + report=report, + new_crops=new_crops, + ) + + if state.kind is StateKind.BRANCH: + # A branch performs no action: it picks an outgoing edge purely by + # guard (evaluated on the current frame). No matching edge is a + # fail-safe HALT unless the state routes to an exception handler. + nxt = self._select_transition( + state, params=params, bundle_dir=bundle_dir + ) + if nxt is None: + return self._on_state_failure( + state, + f"branch state '{state.id}' has no transitions to follow " + "— run aborted", + ) + return nxt + + if state.kind is StateKind.LOOP: + return self._exec_loop_state( + state, + workflow=workflow, + params=params, + worklists=worklists, + bundle_dir=bundle_dir, + run_dir=run_dir, + report=report, + new_crops=new_crops, + depth=depth, + ) + + if state.kind is StateKind.SUBFLOW_CALL: + sub = workflow.subflows.get(state.subflow or "") + if sub is None: + return self._on_state_failure( + state, + f"subflow_call state '{state.id}' names undefined subflow " + f"'{state.subflow}' — run aborted", + ) + self._walk_graph( + sub, + workflow=workflow, + params=params, + worklists=worklists, + bundle_dir=bundle_dir, + run_dir=run_dir, + report=report, + new_crops=new_crops, + depth=depth + 1, + ) + return self._select_transition( + state, params=params, bundle_dir=bundle_dir + ) + + raise _ProgramHalt( + "halt", f"state '{state.id}' has unsupported kind {state.kind!r}" + ) + + def _exec_action_state( + self, + state: State, + graph: ProgramGraph, + *, + workflow: Workflow, + params: dict[str, str], + bundle_dir: Path, + run_dir: Path, + report: RunReport, + new_crops: dict[str, bytes], + ) -> Optional[str]: + """Run an ``action`` state's Step through the SHARED per-step pipeline, + then pick the next transition. A genuine failure routes to + ``on_exception`` (recorded as handled) or HALTs the run.""" + if state.step is None: + raise _ProgramHalt( + "halt", f"action state '{state.id}' carries no step" + ) + ctx = self._build_graph_ctx(state, graph) + result = self._run_step( + state.step, + workflow=workflow, + step_index=0, + params=params, + bundle_dir=bundle_dir, + run_dir=run_dir, + new_crops=new_crops, + graph_ctx=ctx, + ) + report.results.append(result) + self._account_result(report, result) + # Track the previously EXECUTED action for the next TYPE step's + # click-to-focus heuristic. A SKIPPED step (guard on_unmet="skip") did + # not act, so it leaves the previous action / click point untouched. + if not result.skipped: + self._prev_action = state.step.action + + if not result.ok: + # A skipped guard is ok=True; only a genuine failure lands here. + if state.on_exception is not None: + result.exception_handled = True + return state.on_exception # graph try/except: route + continue + raise _ProgramHalt( + "halt", + result.error + or f"action state '{state.id}' failed — run aborted", + ) + return self._select_transition( + state, params=params, bundle_dir=bundle_dir + ) + + def _exec_loop_state( + self, + state: State, + *, + workflow: Workflow, + params: dict[str, str], + worklists: dict[str, list[dict[str, str]]], + bundle_dir: Path, + run_dir: Path, + report: RunReport, + new_crops: dict[str, bytes], + depth: int, + ) -> Optional[str]: + """Iterate a worklist, running the body subflow once per row (RFC §2.3). + + Zero rows => the body runs zero times. The row's fields merge into the + params in scope for that iteration (an ``entity_ref`` param then + re-resolves by the identity ladder each pass). Iteration is BOUNDED: + more rows than ``max_iterations`` HALTs (fail-safe).""" + loop = state.loop + if loop is None: + raise _ProgramHalt( + "halt", f"loop state '{state.id}' carries no LoopSpec" + ) + body = workflow.subflows.get(loop.body) + if body is None: + return self._on_state_failure( + state, + f"loop state '{state.id}' body subflow '{loop.body}' is not " + "defined — run aborted", + ) + rows = self._resolve_worklist(state, loop, workflow, worklists) + if len(rows) > loop.max_iterations: + return self._on_state_failure( + state, + f"loop state '{state.id}' worklist has {len(rows)} rows, " + f"exceeding max_iterations={loop.max_iterations} — run aborted", + ) + for row in rows: + iter_params = {**params, **row} + self._walk_graph( + body, + workflow=workflow, + params=iter_params, + worklists=worklists, + bundle_dir=bundle_dir, + run_dir=run_dir, + report=report, + new_crops=new_crops, + depth=depth + 1, + ) + return self._select_transition( + state, params=params, bundle_dir=bundle_dir + ) + + def _on_state_failure(self, state: State, reason: str) -> str: + """A non-action state hit an unrecoverable condition: route to its + ``on_exception`` handler if it has one, else HALT the run.""" + if state.on_exception is not None: + return state.on_exception + raise _ProgramHalt("halt", reason) + + def _resolve_worklist( + self, + state: State, + loop: LoopSpec, + workflow: Workflow, + worklists: dict[str, list[dict[str, str]]], + ) -> list[dict[str, str]]: + """The rows a loop iterates: a run-time ``worklists`` entry (variable + length, highest priority) else the inline ``data_sources`` relation. An + undefined relation is a config error (HALT) -- distinct from a defined + but EMPTY relation, which legitimately runs the body zero times.""" + if loop.relation in worklists: + return list(worklists[loop.relation]) + relation = workflow.data_sources.get(loop.relation) + if relation is not None: + return list(relation.rows) + raise _ProgramHalt( + "halt", + f"loop state '{state.id}' relation '{loop.relation}' is not " + "defined in data_sources and no run-time worklist was supplied — " + "run aborted", + ) + + def _select_transition( + self, + state: State, + *, + params: dict[str, str], + bundle_dir: Path, + ) -> Optional[str]: + """Pick this state's next state id (RFC §2.2): evaluate ``transitions`` + IN ORDER, first whose guard holds wins; ``None`` guard is unconditional. + + Returns the target id, or None when the state has NO transitions (fall + off / subflow return). When transitions exist but NONE match on the + current screen it is a dead end -- a fail-safe HALT (never guess an + edge). Screenshots only when a guard actually needs a frame, so a + degenerate all-unconditional chain replays with no extra settles (the + byte-identical linear lift).""" + transitions = state.transitions + if not transitions: + return None + if all(t.guard is None for t in transitions): + return transitions[0].target + frame = self.vision.wait_settled(self.backend) + for t in transitions: + if t.guard is None or self._predicate_holds( + t.guard, frame, bundle_dir, params + ): + return t.target + raise _ProgramHalt( + "halt", + f"no outgoing transition matched at state '{state.id}' on the " + "current screen (guards: " + + ", ".join( + self._describe_predicate(t.guard) + for t in transitions + if t.guard is not None + ) + + ") — run aborted", + ) + + def _build_graph_ctx( + self, state: State, graph: ProgramGraph + ) -> _GraphStepContext: + """Assemble the ``_GraphStepContext`` for an action state: the previously + executed action (for the TYPE focus heuristic) and, for a SCROLL step, + the closed-loop successors derived from the graph's transitions.""" + next_anchored: Optional[Step] = None + following: Optional[Step] = None + if state.step is not None and state.step.action is ActionKind.SCROLL: + following, next_anchored = self._scroll_successors(state, graph) + return _GraphStepContext( + prev_action=self._prev_action, + next_anchored=next_anchored, + following=following, + ) + + @staticmethod + def _scroll_successors( + state: State, graph: ProgramGraph + ) -> tuple[Optional[Step], Optional[Step]]: + """(immediately-following step, next anchored step) for a SCROLL state's + closed loop, following unconditional transitions within this graph. The + SCROLL-in-a-graph case is rare; this is best-effort and stays inside the + one graph (subflow boundaries end the scan).""" + + def first_target(s: State) -> Optional[str]: + for t in s.transitions: + if t.guard is None: + return t.target + return s.transitions[0].target if s.transitions else None + + following: Optional[Step] = None + next_anchored: Optional[Step] = None + visited: set[str] = set() + cur = first_target(state) + first = True + while cur is not None and cur not in visited: + visited.add(cur) + st = graph.states.get(cur) + if st is None or st.kind is not StateKind.ACTION or st.step is None: + break + if first: + following = st.step + first = False + if st.step.anchor is not None: + next_anchored = st.step + break + cur = first_target(st) + return following, next_anchored + # -- per-step execution --------------------------------------------------- def _run_step( @@ -403,8 +956,16 @@ def _run_step( bundle_dir: Path, run_dir: Path, new_crops: dict[str, bytes], + graph_ctx: Optional["_GraphStepContext"] = None, ) -> StepResult: - """Execute a single step; never raises (failures land in the result).""" + """Execute a single step; never raises (failures land in the result). + + ``graph_ctx`` is set ONLY by the program interpreter (Phase 2): it + supplies the ``prev_action`` / scroll-successor context the linear loop + derives from ``workflow.steps[step_index±1]`` (which does not exist in a + graph). When None (the linear path) every lookup is exactly as before -- + so a linear run is byte-for-byte unchanged. + """ t0 = time.monotonic() result = StepResult(step_id=step.id, intent=step.intent, ok=False) @@ -579,6 +1140,7 @@ def _run_step( bundle_dir=bundle_dir, before_png=before_png, result=result, + graph_ctx=graph_ctx, ) if error is None: @@ -963,6 +1525,7 @@ def _act( bundle_dir: Path, before_png: bytes, result: StepResult, + graph_ctx: Optional["_GraphStepContext"] = None, ) -> Optional[str]: """Perform the step's action through the backend. @@ -1023,10 +1586,7 @@ def _act( # Fresh baseline AFTER the focusing click so its own focus # ring never counts as "input landed". before_png = self.backend.screenshot() - elif step_index > 0 and workflow.steps[step_index - 1].action in ( - ActionKind.CLICK, - ActionKind.DOUBLE_CLICK, - ): + elif self._prev_was_click(workflow, step_index, graph_ctx): field_point = self._last_click_point self.backend.type_text(text) if not text: @@ -1053,10 +1613,35 @@ def _act( bundle_dir=bundle_dir, before_png=before_png, params=params, + graph_ctx=graph_ctx, ) return f"Step '{step.id}' has unsupported action {step.action!r}" + @staticmethod + def _prev_was_click( + workflow: Workflow, + step_index: int, + graph_ctx: Optional["_GraphStepContext"], + ) -> bool: + """Whether the step executed just before this TYPE step was a click. + + Linear mode reads ``workflow.steps[step_index - 1]`` exactly as before; + program mode reads the interpreter-supplied ``graph_ctx.prev_action`` + (the previously EXECUTED action state -- the graph has no positional + predecessor). The click-to-focus-then-type heuristic is thereby + identical on both paths. + """ + if graph_ctx is not None: + return graph_ctx.prev_action in ( + ActionKind.CLICK, + ActionKind.DOUBLE_CLICK, + ) + return step_index > 0 and workflow.steps[step_index - 1].action in ( + ActionKind.CLICK, + ActionKind.DOUBLE_CLICK, + ) + # -- Workflow-program IR gates: guard + wait_until -------------------------- def _apply_step_gates( @@ -1619,6 +2204,7 @@ def _act_scroll( bundle_dir: Path, before_png: bytes, params: dict[str, str], + graph_ctx: Optional["_GraphStepContext"] = None, ) -> Optional[str]: """Execute a SCROLL step as a closed loop on a wait_until predicate. @@ -1648,7 +2234,15 @@ def _act_scroll( """ dx = step.scroll_dx or 0 dy = step.scroll_dy or 0 - next_step = self._next_anchored_step(workflow, step_index) + # The next anchored step (default scroll stop condition): linear mode + # scans forward in ``workflow.steps``; program mode uses the + # interpreter-supplied successor (the target of this state's + # unconditional transition chain). + next_step = ( + graph_ctx.next_anchored + if graph_ctx is not None + else self._next_anchored_step(workflow, step_index) + ) # 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 @@ -1676,9 +2270,13 @@ def _act_scroll( return None following = ( - workflow.steps[step_index + 1] - if step_index + 1 < len(workflow.steps) - else None + graph_ctx.following + if graph_ctx is not None + else ( + workflow.steps[step_index + 1] + if step_index + 1 < len(workflow.steps) + else None + ) ) if following is not None and following.action is ActionKind.SCROLL: # The next SCROLL step continues the loop with its own budget. diff --git a/tests/test_program_ir_phase2.py b/tests/test_program_ir_phase2.py new file mode 100644 index 0000000..e4208ee --- /dev/null +++ b/tests/test_program_ir_phase2.py @@ -0,0 +1,660 @@ +"""Workflow-program IR, Phase 2 (RFC docs/design/WORKFLOW_PROGRAM_IR.md §2). + +The parameterized STATE MACHINE: the control flow a linear action list cannot +express -- LOOPS over a worklist, guarded BRANCHES, reusable SUBFLOWS, and +EXCEPTION paths -- interpreted deterministically (ZERO model calls, $0 replay). + +Built ADDITIVELY on Phase 1: every ``action`` state runs through the SAME +per-step pipeline the linear replayer uses (resolve / identity gate / effect +verify / risk gate / heal), so no safety property is weakened by adding control +flow around the hardened leaf -- proven here by the identity- and effect-gate +tests that fire INSIDE a loop body. A linear (no-``program``) bundle replays +byte-for-byte as today, and its mechanical lift to the degenerate single-path +graph replays identically (``lift_to_program``). + +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, + LoopSpec, + Predicate, + PredicateKind, + ProgramGraph, + Relation, + State, + StateKind, + Step, + Transition, + Workflow, + lift_to_program, +) +from openadapt_flow.runtime.effects import Effect, EffectKind +from openadapt_flow.runtime.replayer import Replayer +from test_replayer import ( + FakeBackend, + FakeVision, + Match, + OcrLine, + click_step, + context_click_step, + make_png, + resolving_vision, +) + + +@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" + + +# -- tiny builders ----------------------------------------------------------- + + +def key_step(step_id, key) -> Step: + return Step(id=step_id, intent=f"press {key}", action=ActionKind.KEY, key=key) + + +def type_step(step_id, param) -> Step: + return Step( + id=step_id, intent=f"type {param}", action=ActionKind.TYPE, param=param + ) + + +def failing_click(step_id) -> Step: + """A click whose target never resolves (no template match scripted). A short + timeout keeps the resolution-retry loop fast in these fail-path tests.""" + step = click_step(step_id) + step.timeout_s = 0.1 + return step + + +def action(step: Step, *, to: str, on_exception: str | None = None) -> State: + """An ACTION state with a single unconditional transition to ``to``.""" + return State( + id=step.id, + kind=StateKind.ACTION, + step=step, + transitions=[Transition(target=to)], + on_exception=on_exception, + ) + + +def terminal(state_id, outcome="success", reason="") -> State: + return State( + id=state_id, kind=StateKind.TERMINAL, outcome=outcome, reason=reason + ) + + +def graph(entry, *states: State) -> ProgramGraph: + return ProgramGraph(entry=entry, states={s.id: s for s in states}) + + +# =========================================================================== +# LOOPS over a worklist (RFC §2.3) +# =========================================================================== + + +def _loop_over_queue_workflow() -> Workflow: + """A loop over the ``queue`` relation whose body types the row's ``patient`` + field once per row (so the recorded actions reveal the per-row binding and + the iteration count).""" + body = graph( + "b_type", + action(type_step("b_type", "patient"), to="b_end"), + terminal("b_end"), + ) + program = graph( + "loop", + State( + id="loop", + kind=StateKind.LOOP, + loop=LoopSpec(relation="queue", body="body", var="patient"), + transitions=[Transition(target="done")], + ), + terminal("done"), + ) + return Workflow( + name="clear-queue", + program=program, + subflows={"body": body}, + ) + + +def test_loop_runs_body_once_per_row_binding_each_row(bundle, run_dir): + """A 3-row worklist runs the body 3x, the loop variable binding each row in + turn (Rousillon/Helena/WebRobot: for every row ...).""" + wf = _loop_over_queue_workflow() + wf.data_sources = { + "queue": Relation( + name="queue", + rows=[{"patient": "Alice"}, {"patient": "Bob"}, {"patient": "Cara"}], + ) + } + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert report.terminal_outcome == "success" + assert backend.actions == [ + ("type", "Alice"), + ("type", "Bob"), + ("type", "Cara"), + ] + assert report.model_calls == 0 # $0 runtime preserved + + +def test_loop_over_zero_rows_runs_body_zero_times(bundle, run_dir): + """A variable-length worklist that happens to be EMPTY runs the body 0x and + still completes successfully -- the data-dependent case a single linear + trace cannot express.""" + wf = _loop_over_queue_workflow() + wf.data_sources = {"queue": Relation(name="queue", rows=[])} + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [] # body never ran + assert report.terminal_outcome == "success" + + +def test_loop_worklist_supplied_at_run_time(bundle, run_dir): + """The worklist can be supplied at RUN time (a genuinely data-dependent + queue whose length is unknown until then), overriding any inline rows.""" + wf = _loop_over_queue_workflow() # no data_sources at all + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, + worklists={"queue": [{"patient": "X"}, {"patient": "Y"}]}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is True + assert backend.actions == [("type", "X"), ("type", "Y")] + + +def test_loop_bound_is_enforced(bundle, run_dir): + """More rows than ``max_iterations`` HALTs (fail-safe) rather than running + unbounded.""" + wf = _loop_over_queue_workflow() + wf.program.states["loop"].loop.max_iterations = 2 + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, + worklists={"queue": [{"patient": str(i)} for i in range(3)]}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is False + assert report.terminal_outcome == "halt" + assert backend.actions == [] # refused before iterating + + +# =========================================================================== +# BRANCHES / conditionals (RFC §2.2) -- guarded transitions pick the arm +# =========================================================================== + + +def _branch_on_param_workflow() -> Workflow: + program = graph( + "pick", + State( + id="pick", + kind=StateKind.BRANCH, + transitions=[ + Transition( + guard=Predicate( + kind=PredicateKind.PARAM_EQUALS, + param="encounter_type", + value="Triage", + ), + target="s_triage", + ), + Transition( + guard=Predicate( + kind=PredicateKind.PARAM_EQUALS, + param="encounter_type", + value="Consult", + ), + target="s_consult", + ), + ], + ), + action(key_step("s_triage", "T"), to="done"), + action(key_step("s_consult", "C"), to="done"), + terminal("done"), + ) + return Workflow(name="pick-encounter", program=program) + + +def test_branch_takes_the_triage_arm(bundle, run_dir): + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + _branch_on_param_workflow(), + params={"encounter_type": "Triage"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is True + assert backend.actions == [("press", "T")] + assert "s_triage" in report.visited_states + assert "s_consult" not in report.visited_states + + +def test_branch_takes_the_consult_arm(bundle, run_dir): + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + _branch_on_param_workflow(), + params={"encounter_type": "Consult"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + assert report.success is True + assert backend.actions == [("press", "C")] + assert "s_consult" in report.visited_states + + +def test_branch_on_screen_predicate_dismisses_optional_modal(bundle, run_dir): + """The MockMed survey-modal case as a guarded BRANCH (RFC §2.2): after Save, + a transition guarded on the modal's presence routes to a dismiss subflow, + else straight to done -- an optional-but-expected popup, not drift/halt.""" + program = graph( + "s_save", + State( + id="s_save", + kind=StateKind.ACTION, + step=key_step("s_save", "S"), + transitions=[ + Transition( + guard=Predicate( + kind=PredicateKind.TEXT_PRESENT, text="Survey" + ), + target="s_dismiss", + ), + Transition(target="done"), # unconditional fall-through + ], + ), + action(key_step("s_dismiss", "Escape"), to="done"), + terminal("done"), + ) + wf = Workflow(name="save-maybe-survey", program=program) + + # Modal PRESENT -> the guarded arm dismisses it, then rejoins. + v_present = FakeVision() + v_present.text_results = {"Survey": Match((10, 10), (0, 0, 5, 5))} + b_present = FakeBackend() + r_present = Replayer(b_present, vision=v_present, poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert r_present.success is True + assert b_present.actions == [("press", "S"), ("press", "Escape")] + + # Modal ABSENT -> the unconditional fall-through goes straight to done. + b_absent = FakeBackend() + r_absent = Replayer(b_absent, vision=FakeVision(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir / "absent" + ) + assert r_absent.success is True + assert b_absent.actions == [("press", "S")] + + +def test_branch_with_no_matching_arm_halts_fail_safe(bundle, run_dir): + """A branch whose guards all fail on the current screen is a dead end: the + run HALTs (never guesses an edge) -- the refuse-rather-than-guess posture.""" + program = graph( + "pick", + State( + id="pick", + kind=StateKind.BRANCH, + transitions=[ + Transition( + guard=Predicate( + kind=PredicateKind.PARAM_EQUALS, param="m", value="a" + ), + target="s_a", + ) + ], + ), + action(key_step("s_a", "A"), to="done"), + terminal("done"), + ) + wf = Workflow(name="dead-end", program=program) + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, params={"m": "z"}, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert report.terminal_outcome == "halt" + assert backend.actions == [] + + +# =========================================================================== +# SUBFLOWS -- reusable named subgraphs (RFC §2.2) +# =========================================================================== + + +def test_subflow_reused_as_loop_body_and_direct_call(bundle, run_dir): + """One subflow ``greet`` (press 'G') is reused BOTH as a loop body (per row) + AND as a direct subflow_call -- the reusable-component construct. 2 rows + + 1 direct call => 3 presses.""" + greet = graph("g", action(key_step("g", "G"), to="g_end"), terminal("g_end")) + program = graph( + "loop", + State( + id="loop", + kind=StateKind.LOOP, + loop=LoopSpec(relation="queue", body="greet"), + transitions=[Transition(target="call")], + ), + State( + id="call", + kind=StateKind.SUBFLOW_CALL, + subflow="greet", + transitions=[Transition(target="done")], + ), + terminal("done"), + ) + wf = Workflow( + name="greet-all", + program=program, + subflows={"greet": greet}, + data_sources={ + "queue": Relation(name="queue", rows=[{"n": "1"}, {"n": "2"}]) + }, + ) + backend = FakeBackend() + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert backend.actions == [("press", "G")] * 3 + + +def test_subflow_call_to_undefined_subflow_halts(bundle, run_dir): + program = graph( + "call", + State( + id="call", + kind=StateKind.SUBFLOW_CALL, + subflow="missing", + transitions=[Transition(target="done")], + ), + terminal("done"), + ) + wf = Workflow(name="bad", program=program) + backend = FakeBackend() + 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 report.terminal_outcome == "halt" + + +# =========================================================================== +# EXCEPTION paths (RFC §2.4) -- on_exception routes a failed action + continues +# =========================================================================== + + +def test_on_exception_catches_failed_action_and_continues(bundle, run_dir): + """An expected-exceptional state (here a click whose target a popup hides, + so the resolution ladder fails) routes to a local handler subflow that + recovers and rejoins -- the run continues instead of aborting (the graph + analog of try/except).""" + recover = graph( + "r", + action(key_step("r", "Escape"), to="r_end"), + terminal("r_end"), + ) + program = graph( + "s_open", + # This click cannot resolve (no template match scripted) -> it FAILS, + # and routes to the recover subflow via on_exception. + State( + id="s_open", + kind=StateKind.ACTION, + step=failing_click("s_open"), + transitions=[Transition(target="done")], + on_exception="s_recover", + ), + State( + id="s_recover", + kind=StateKind.SUBFLOW_CALL, + subflow="recover", + transitions=[Transition(target="done")], + ), + terminal("done"), + ) + wf = Workflow(name="save-or-recover", program=program, subflows={"recover": recover}) + backend = FakeBackend() + # FakeVision with NO template match -> the click's resolution ladder fails. + report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True # the run CONTINUED via the handler + assert report.terminal_outcome == "success" + # The failed action is recorded, marked handled; the handler then ran. + failed = next(r for r in report.results if r.step_id == "s_open") + assert failed.ok is False + assert failed.exception_handled is True + assert backend.actions == [("press", "Escape")] # recovery ran, rejoined + + +def test_unhandled_action_failure_halts_the_whole_run(bundle, run_dir): + """WITHOUT an on_exception handler, a failed action HALTs the whole run -- + no regression vs. today's halt-on-failure.""" + program = graph( + "s_open", + action(failing_click("s_open"), to="done"), # no on_exception + terminal("done"), + ) + wf = Workflow(name="no-handler", program=program) + backend = FakeBackend() + 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 report.terminal_outcome == "halt" + + +def test_halt_terminal_stops_the_run(bundle, run_dir): + """A ``halt`` terminal (e.g. the patient-not-found branch the demo never + showed) stops the run with success=False.""" + program = graph( + "t", + terminal("t", outcome="halt", reason="patient not found"), + ) + wf = Workflow(name="halt-now", program=program) + backend = FakeBackend() + 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 report.terminal_outcome == "halt" + + +# =========================================================================== +# Safety gates STILL FIRE inside loop bodies (RFC §2.1, §2.3) +# =========================================================================== + + +def test_identity_gate_fires_inside_loop_body(bundle, run_dir): + """A loop body's CLICK is identity-gated exactly as a linear click: when the + live band names a DIFFERENT entity than the recorded target, the body action + refuses and the run HALTs -- inside the loop. This is the entity_ref/ + re-resolve-by-identity guarantee (iteration N must click the RIGHT row).""" + body = graph( + "b_click", + action( + context_click_step("Jane Sample Knee pain referral High", step_id="b_click"), + to="b_end", + ), + terminal("b_end"), + ) + program = graph( + "loop", + State( + id="loop", + kind=StateKind.LOOP, + loop=LoopSpec(relation="queue", body="body"), + transitions=[Transition(target="done")], + ), + terminal("done"), + ) + wf = Workflow( + name="clear-queue-identity", + program=program, + subflows={"body": body}, + data_sources={"queue": Relation(name="queue", rows=[{"x": "1"}])}, + ) + vision = resolving_vision() + # Live band names a DIFFERENT entity -> mismatch. + vision.ocr_lines = [OcrLine("Taylor Duplicate Knee pain referral High")] + backend = FakeBackend() + 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 report.terminal_outcome == "halt" + assert backend.actions == [] # never clicked the wrong row + clicked = next(r for r in report.results if r.step_id == "b_click") + assert clicked.identity is not None + assert clicked.identity.status == "mismatch" + + +def test_effect_gate_fires_inside_loop_body(bundle, run_dir): + """A loop body's step that declares system-of-record ``effects`` with NO + verifier configured is a fail-safe HALT -- the effect gate fires for a state + inside a loop body exactly as for a linear step.""" + save = key_step("b_save", "S") + save.effects = [Effect(kind=EffectKind.RECORD_WRITTEN)] + body = graph("b_save", action(save, to="b_end"), terminal("b_end")) + program = graph( + "loop", + State( + id="loop", + kind=StateKind.LOOP, + loop=LoopSpec(relation="queue", body="body"), + transitions=[Transition(target="done")], + ), + terminal("done"), + ) + wf = Workflow( + name="write-queue", + program=program, + subflows={"body": body}, + data_sources={"queue": Relation(name="queue", rows=[{"x": "1"}])}, + ) + backend = FakeBackend() + # No effect_verifier configured on the Replayer -> declared effect HALTs. + 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 report.terminal_outcome == "halt" + saved = next(r for r in report.results if r.step_id == "b_save") + assert saved.effect_verified is False + + +# =========================================================================== +# BACK-COMPAT -- the degenerate linear lift replays identically to today +# =========================================================================== + + +def _linear_workflow() -> Workflow: + return Workflow( + name="wf", + steps=[ + click_step("s1"), + type_step("s2", "note"), + key_step("s3", "Enter"), + ], + params={"note": "hello"}, + ) + + +def _resolving_for_linear() -> FakeVision: + v = FakeVision() + v.template_results = [Match((110, 105), (100, 100, 50, 20), 0.95)] + return v + + +def test_lifted_linear_graph_replays_identically_to_linear(bundle, run_dir): + """A linear workflow lifted to the degenerate single-path graph + (``lift_to_program``, RFC §2.6) performs the IDENTICAL backend actions, in + the identical order, with the identical per-step results as the linear + replayer -- 'a linear bundle is the degenerate single-path graph'.""" + # Linear replay. + lin = _linear_workflow() + b_lin = FakeBackend() + r_lin = Replayer(b_lin, vision=_resolving_for_linear(), poll_interval_s=0.01).run( + lin, bundle_dir=bundle, run_dir=run_dir / "lin" + ) + # Graph replay of the SAME steps, lifted. + grf = _linear_workflow() + grf.program = lift_to_program(grf) + b_grf = FakeBackend() + r_grf = Replayer(b_grf, vision=_resolving_for_linear(), poll_interval_s=0.01).run( + grf, bundle_dir=bundle, run_dir=run_dir / "grf" + ) + + assert r_lin.success is r_grf.success is True + assert b_lin.actions == b_grf.actions # byte-identical actuation + assert [r.step_id for r in r_lin.results] == [ + r.step_id for r in r_grf.results + ] + assert [r.ok for r in r_lin.results] == [r.ok for r in r_grf.results] + assert [ + (r.resolution.rung if r.resolution else None) for r in r_lin.results + ] == [ + (r.resolution.rung if r.resolution else None) for r in r_grf.results + ] + assert r_lin.model_calls == r_grf.model_calls == 0 + + +def test_no_program_field_replays_exactly_as_v0(bundle, run_dir): + """A workflow with ``program=None`` runs the linear ``steps`` loop -- the + additive Phase-2 fields default empty and change nothing.""" + wf = _linear_workflow() + assert wf.program is None and wf.subflows == {} and wf.data_sources == {} + backend = FakeBackend() + report = Replayer(backend, vision=_resolving_for_linear(), poll_interval_s=0.01).run( + wf, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert report.terminal_outcome is None # linear runs set no terminal + assert backend.actions == [ + ("click", 110, 105, False), + ("type", "hello"), + ("press", "Enter"), + ] + + +def test_program_round_trips_through_bundle_save_load(bundle): + """A program graph (states, transitions, loops, subflows, data_sources) + survives ``Workflow.save`` -> ``Workflow.load``.""" + wf = _loop_over_queue_workflow() + wf.data_sources = { + "queue": Relation(name="queue", rows=[{"patient": "Alice"}]) + } + wf.save(bundle) + loaded = Workflow.load(bundle) + assert loaded.program is not None + assert loaded.program.states["loop"].kind is StateKind.LOOP + assert loaded.program.states["loop"].loop.body == "body" + assert loaded.subflows["body"].states["b_type"].step.param == "patient" + assert loaded.data_sources["queue"].rows == [{"patient": "Alice"}] From ab88bddfaef002938b2ced429248dea77323ff39 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 16:14:04 -0400 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20multi-trace=20induction=20=E2=80=94?= =?UTF-8?q?=20infer=20a=20parameterized=20program=20(params/loops/branches?= =?UTF-8?q?)=20from=20multiple=20demos,=20reject-if-underdetermined?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements RFC docs/design/WORKFLOW_PROGRAM_IR.md §3 steps [4]+[5]: the induction loop the PBD lineage (Rousillon, WebRobot, Skill-DisCo, PROLEX) says a demonstration compiler must have. "One demonstration is evidence, not specification." openadapt_flow/compiler/induction.py: - induce_program(traces) aligns multiple demos structurally and infers a Phase-2 ProgramGraph: PARAMS (values that VARY across traces at an aligned position; constant => literal), LOOPS (a repeated body whose count DIFFERS => LoopSpec over an inferred Relation), BRANCHES (a divergent step under a detectable condition => guarded branch, guard proposed/flagged), and OPTIONAL steps (present in some, absent in others, no condition => guarded skip). All deterministic, ZERO model calls. - validate_held_out / reproduction_score: leave-one-out held-out validation (infer from N-1, check reproduction of the held trace). - Reject-rather-than-guess: contradictory / underdetermined traces are QUARANTINED (no program emitted, certified=False) and routed to the disambiguation flow (#74), mirroring the identity gate's posture. - The optional compile-time Proposer (the #78 StepAnnotator fits behind it) only PROPOSES interpretations — flagged, never silently trusted, never flips an underdetermined point to certified. Touch-points kept minimal: reuses the Phase-2 IR + Phase-1 ParamSpec/Guard/ Predicate verbatim (no new IR fields), reuses disambiguation's question model, and the emitted program replays through the EXISTING interpreter unchanged (compile.py untouched; compiler/__init__ re-exports the new API). Tests (tests/test_induction.py, 17 tests): a synthetic MockMed corpus of trace variants covers (a) param, (b) loop, (c) branch/optional, (d) contradiction=> reject; held-out scores a good induction high and an over-specialized one low; underdetermined is flagged not guessed; the induced program round-trips through the real Phase-2 interpreter (faked backend/vision, zero model calls). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- docs/design/WORKFLOW_PROGRAM_IR.md | 10 + openadapt_flow/compiler/__init__.py | 16 + openadapt_flow/compiler/induction.py | 1053 ++++++++++++++++++++++++++ tests/test_induction.py | 511 +++++++++++++ 4 files changed, 1590 insertions(+) create mode 100644 openadapt_flow/compiler/induction.py create mode 100644 tests/test_induction.py diff --git a/docs/design/WORKFLOW_PROGRAM_IR.md b/docs/design/WORKFLOW_PROGRAM_IR.md index a899189..7c7ed81 100644 --- a/docs/design/WORKFLOW_PROGRAM_IR.md +++ b/docs/design/WORKFLOW_PROGRAM_IR.md @@ -652,6 +652,16 @@ Agree the target shape. Reversible by definition. the largest lift and the one most dependent on a real workflow (§7). - Compile-time model use (one-time labeling/risk/param proposal) lands here, guarded to preserve the runtime's $0/0-call property. +- **Status:** §3 step [3] (interactive disambiguation) shipped in + `compiler/disambiguation.py`; §3 steps [4]+[5] (multi-trace induction + + held-out validation / quarantine) shipped in `compiler/induction.py` + (`induce_program` / `validate_held_out` / `reproduction_score`). Structural + alignment infers params (values that VARY across traces), loops (a repeated + body whose count DIFFERS), branches (a divergent step under a detectable + condition — guard proposed/flagged), and optional steps — all deterministic, + ZERO model calls; the optional compile-time `Proposer` only PROPOSES (flagged, + never trusted). Contradictory / underdetermined traces are QUARANTINED (no + program emitted, `certified=False`) and routed to the disambiguation flow. ### Stays as-is (do not rebuild) - The **visual resolution ladder** (`DESIGN.md:152-164`, `resolver.py`) — one diff --git a/openadapt_flow/compiler/__init__.py b/openadapt_flow/compiler/__init__.py index 167d61d..376864f 100644 --- a/openadapt_flow/compiler/__init__.py +++ b/openadapt_flow/compiler/__init__.py @@ -6,6 +6,14 @@ StepEffectMining, mine_step_effects, ) +from openadapt_flow.compiler.induction import ( + HeldOutValidation, + InductionResult, + Proposer, + induce_program, + reproduction_score, + validate_held_out, +) __all__ = [ "compile_recording", @@ -13,4 +21,12 @@ "render_workflow_py", "mine_step_effects", "StepEffectMining", + # Multi-trace induction (RFC §3 [4]+[5]): one demo is the single-trace + # bootstrap; multiple demos induce a parameterized program or refuse. + "induce_program", + "validate_held_out", + "reproduction_score", + "InductionResult", + "HeldOutValidation", + "Proposer", ] diff --git a/openadapt_flow/compiler/induction.py b/openadapt_flow/compiler/induction.py new file mode 100644 index 0000000..1b39134 --- /dev/null +++ b/openadapt_flow/compiler/induction.py @@ -0,0 +1,1053 @@ +"""Multi-trace induction: infer a parameterized PROGRAM from several demos. + +One demonstration is *evidence*, not *specification* (RFC +``docs/design/WORKFLOW_PROGRAM_IR.md`` §1, §3): the same single trace is +consistent with many distinct programs. The whole programming-by-demonstration +lineage a demonstration compiler descends from -- Rousillon/Helena, WebRobot, +Skill-DisCo, PROLEX -- closes that gap with an **induction loop** that +generalizes from *multiple* demonstrations of the same task. This module +implements RFC §3 step [4] (multi-trace induction) + step [5] (held-out +validation / quarantine): it turns ``list[Workflow | recording-dir]`` into a +:class:`~openadapt_flow.ir.ProgramGraph` -- the Phase-2 state machine -- *or an +honest refusal*. + +What it infers, and how (all deterministic + structural -- ZERO model calls): + +* **Parameters.** A value that VARIES across traces at the same aligned + position is a typed :class:`~openadapt_flow.ir.ParamSpec`; a value that is + CONSTANT across traces stays a baked literal (WebRobot-style value + speculation, made *determinate* by cross-trace evidence rather than guessed). +* **Loops.** A repeated aligned sub-sequence whose repetition count DIFFERS + across traces (a worklist of length 2 vs 3) is a + :class:`~openadapt_flow.ir.LoopSpec` over an inferred + :class:`~openadapt_flow.ir.Relation`, its body the repeated subflow + (Rousillon/Helena "for every row ..."). +* **Branches.** A step present/divergent in some traces but not others UNDER A + DETECTABLE CONDITION is a ``branch`` state with guarded transitions -- the + guard is *proposed and flagged for confirmation* (Skill-DisCo: divergences + localize the branch automatically). +* **Optional steps.** Present in some, absent in others, with NO derivable + condition -> an optional/guarded step that SKIPS when its own target is + absent. + +**Refuse rather than guess** (RFC §3 [5]; mirrors ``runtime.identity`` and +``compiler.disambiguation``). When traces CONTRADICT or intent stays +underdetermined -- a divergent branch with no detectable condition, an +irreconcilable ordering -- induction does NOT fabricate a program. It marks the +point ``underdetermined`` with the specific ambiguity, routes it to the +disambiguation flow (:mod:`openadapt_flow.compiler.disambiguation`), leaves +``certified=False``, and does not emit a program. A wrong branch on an +irreversible node is the failure class the whole repo is organized to avoid. + +**The compile-time model only PROPOSES.** An optional :class:`Proposer` (the +compile-time StepAnnotator lives behind this interface) may propose an +interpretation for an ambiguous point, but every proposal is recorded in +``proposed`` / attached to the uncertainty and is NEVER silently trusted: a +proposal cannot flip an ``underdetermined`` point to certified. Deterministic +structural inference is the core; the model is advisory and flagged. + +Touch-points (kept minimal per the stacking constraint on PR #79): + +* Reuses the Phase-2 IR (``ProgramGraph`` / ``State`` / ``Transition`` / + ``LoopSpec`` / ``Relation``) and Phase-1 ``ParamSpec`` / ``Predicate`` / + ``Guard`` VERBATIM -- adds NO new IR fields. +* Reuses ``compiler.disambiguation``'s question model to route uncertainties to + the SAME ask-don't-guess flow (#74). +* ``compiler/compile.py`` is UNCHANGED -- induction runs as a pass OVER + already-compiled workflows (``compile_recording`` is the single-trace + bootstrap, RFC §3 [1]); a recording-dir trace is compiled through it. +* The emitted program replays through the EXISTING Phase-2 interpreter + (``runtime.replayer``) with zero changes -- proven by the round-trip test. +""" + +from __future__ import annotations + +import tempfile +from difflib import SequenceMatcher +from pathlib import Path +from typing import Any, Optional, Protocol, Union, runtime_checkable + +from pydantic import BaseModel, Field + +from openadapt_flow.compiler.disambiguation import ( + AmbiguityKind, + DisambiguationQuestion, + OptionEffect, + QuestionOption, +) +from openadapt_flow.ir import ( + ActionKind, + Guard, + LoopSpec, + ParamKind, + ParamSpec, + Predicate, + PredicateKind, + ProgramGraph, + Relation, + State, + StateKind, + Step, + Transition, + Workflow, +) + +TraceInput = Union[Workflow, str, Path] + +# A typed value shorter than this (stripped) is too weak to speculate on as a +# parameter -- mirrors disambiguation.MIN_PARAM_VALUE_CHARS. +MIN_PARAM_VALUE_CHARS = 1 + + +# =========================================================================== +# Compile-time model, behind an interface: it PROPOSES, it never decides. +# =========================================================================== + + +class Proposal(BaseModel): + """A compile-time-model suggestion for an under-specified point. + + ADVISORY ONLY. A proposal is surfaced in :attr:`InductionResult.proposed` + (and attached to any uncertainty it concerns) so a reviewer sees it, but it + NEVER silently changes the deterministic structural decision and NEVER flips + an ``underdetermined`` point to certified (``trusted`` is always False -- + the field exists to make that contract explicit and auditable). + """ + + target: str = Field(description="Column / state id the proposal concerns") + kind: str = Field(description="'guard' | 'param' | 'label'") + content: str = Field(description="Human-readable proposed interpretation") + source: str = "annotator" + trusted: bool = Field( + default=False, + description="Always False: proposals are flagged, never auto-applied.", + ) + + +@runtime_checkable +class Proposer(Protocol): + """Compile-time interpretation proposer (the #78 StepAnnotator fits here). + + Called at MOST once per ambiguous/branch point at COMPILE time. Any + implementation that reaches a model does so here, once -- never at run time + (the runtime's audited $0 / 0-call property is preserved). Tests pass a + deterministic fake; a real annotator may consult a VLM. Either way the + return value is advisory (see :class:`Proposal`). + """ + + def propose(self, target: str, kind: str, context: dict[str, Any]) -> Optional[str]: + """Return a human-readable proposed interpretation, or None to abstain.""" + ... + + +# =========================================================================== +# Result model +# =========================================================================== + + +class Uncertainty(BaseModel): + """A point where intent stays underdetermined -- the reason induction + REFUSES to emit (RFC §3 [5] quarantine). Routed to the disambiguation flow + via :attr:`question`.""" + + kind: str = Field( + description="'ambiguous_branch' | 'alignment_failure' | 'unobserved'" + ) + location: str = Field(description="Column / position the ambiguity is at") + detail: str + consequential: bool = Field( + default=True, + description="True when a wrong resolution risks an irreversible action.", + ) + question: Optional[DisambiguationQuestion] = Field( + default=None, + description="The grounded question routed to compiler.disambiguation.", + ) + proposal: Optional[Proposal] = Field( + default=None, + description="A compile-time-model suggestion -- flagged, NOT trusted.", + ) + + +class ColumnDecision(BaseModel): + """The induced interpretation of one aligned column, kept for the audit + trail AND for held-out reproduction scoring (:func:`reproduction_score`).""" + + index: int + kind: str # literal | param | loop | branch | optional | divergent + align_sig: str = Field(description="Stringified alignment signature") + field: str = "" + literal_value: Optional[str] = None # kind == literal + param_name: Optional[str] = None # kind == param / loop + present_in: list[int] = Field(default_factory=list) # trace indices + counts: list[int] = Field( + default_factory=list, description="Per-trace loop iteration counts." + ) + note: str = "" + + +class InductionResult(BaseModel): + """The induced program (or an honest refusal) plus the full audit trail.""" + + n_traces: int + program: Optional[ProgramGraph] = None + workflow: Optional[Workflow] = Field( + default=None, + description="A replayable Workflow carrying program/subflows/" + "param_specs/data_sources (None when quarantined).", + ) + param_specs: dict[str, ParamSpec] = Field(default_factory=dict) + column_decisions: list[ColumnDecision] = Field(default_factory=list) + inferred: list[str] = Field( + default_factory=list, + description="What was inferred DETERMINISTICALLY (params/loops/opt).", + ) + proposed: list[Proposal] = Field( + default_factory=list, + description="Compile-time-model suggestions -- flagged, never trusted.", + ) + uncertainties: list[Uncertainty] = Field(default_factory=list) + + @property + def underdetermined(self) -> bool: + return bool(self.uncertainties) + + @property + def certified(self) -> bool: + """False iff any point stays underdetermined (refuse rather than + guess) OR no program was emitted. A flagged proposal does NOT certify.""" + return self.program is not None and not self.uncertainties + + def render(self) -> str: + lines = [ + f"Induction over {self.n_traces} trace(s): " + f"{len(self.inferred)} inferred, {len(self.proposed)} proposed " + f"(flagged), {len(self.uncertainties)} underdetermined." + ] + for line in self.inferred: + lines.append(f" [inferred] {line}") + for p in self.proposed: + lines.append(f" [proposed] {p.target}: {p.content} (NOT trusted)") + for u in self.uncertainties: + lines.append(f" [REFUSED] {u.location}: {u.detail}") + verdict = "CERTIFIED" if self.certified else "NOT CERTIFIED" + lines.append( + f"Certification: {verdict}" + + ("" if self.certified else " -- resolve the point(s) above.") + ) + return "\n".join(lines) + + +class HeldOutValidation(BaseModel): + """Leave-one-out held-out validation (RFC §3 [5]): infer from N-1 traces, + check the induced program reproduces the held-out trace.""" + + per_trace: list[float] = Field(default_factory=list) + mean: float = 0.0 + n_traces: int = 0 + + def render(self) -> str: + scores = ", ".join(f"{s:.2f}" for s in self.per_trace) + return ( + f"Held-out validation ({self.n_traces} folds): " + f"mean={self.mean:.2f} [{scores}]" + ) + + +# =========================================================================== +# Trace normalization + signatures +# =========================================================================== + + +def _as_workflow(trace: TraceInput) -> Workflow: + """Normalize a trace to a linear ``Workflow``. A recording DIRECTORY is + compiled through the single-trace bootstrap ``compile_recording`` (RFC §3 + [1]); a ``Workflow`` is used as-is.""" + if isinstance(trace, Workflow): + return trace + from openadapt_flow.compiler.compile import compile_recording + + recording = Path(trace) + with tempfile.TemporaryDirectory(prefix="induce-boot-") as tmp: + return compile_recording( + recording, Path(tmp) / "bundle", name=recording.name + ) + + +_LEADING_VERBS = frozenset( + {"type", "enter", "input", "click", "press", "select", "set", "fill"} +) + + +def _norm_intent(step: Step) -> str: + """A value-free field label from a step's intent (lowercased, digits + dropped so a per-run value in the intent does not fragment alignment, and a + leading action verb dropped so 'type patient' keys as 'patient').""" + text = "".join(c for c in step.intent.lower() if not c.isdigit()) + words = text.split() + if words and words[0] in _LEADING_VERBS: + words = words[1:] + return " ".join(words) or " ".join(text.split()) + + +def _field_key(step: Step) -> str: + """A stable, VALUE-FREE identity for the field/target a step acts on -- the + dimension alignment matches on (so the SAME field with a DIFFERENT value + aligns, revealing a parameter rather than a structural change).""" + if step.action is ActionKind.KEY: + # Distinct keys are distinct control-flow (Enter vs Escape), so the key + # IS part of the field identity here. + return f"key:{step.key}" + if step.param: + return step.param + if step.anchor is not None and step.anchor.ocr_text: + return step.anchor.ocr_text + return _norm_intent(step) + + +def _sig(step: Step) -> tuple[str, str]: + return (step.action.value, _field_key(step)) + + +def _value(step: Step) -> Optional[str]: + """The per-run VALUE a step carries (the thing that varies for a param).""" + if step.action is ActionKind.TYPE: + return step.text if step.text is not None else step.param + if step.action is ActionKind.KEY: + return step.key + if step.anchor is not None: + return step.anchor.ocr_text + return None + + +# Dialog detection is shared with disambiguation so induction and interactive +# disambiguation agree on what "an optional popup" looks like. +from openadapt_flow.compiler.disambiguation import ( # noqa: E402 + _dialog_text, + _looks_like_dialog, +) + + +# =========================================================================== +# Reduce a trace to tokens (collapse consecutive repeats into loop candidates) +# =========================================================================== + + +class _SingleTok: + kind = "single" + + def __init__(self, step: Step): + self.step = step + self.align_sig: tuple = _sig(step) + + @property + def value(self) -> Optional[str]: + return _value(self.step) + + +class _LoopTok: + kind = "loop" + + def __init__(self, body_steps: list[Step], iterations: list[list[Step]]): + self.body_steps = body_steps + self.iterations = iterations # per-iteration list of steps + body_sig = tuple(_sig(s) for s in body_steps) + self.align_sig = ("__loop__", body_sig) + + @property + def count(self) -> int: + return len(self.iterations) + + def rows(self, field: str) -> list[dict[str, str]]: + """Per-iteration binding of ``field`` -> value (single-column body).""" + out: list[dict[str, str]] = [] + for it in self.iterations: + val = _value(it[0]) if it else None + out.append({field: val if val is not None else ""}) + return out + + +def _reduce_trace(steps: list[Step]) -> list: + """Collapse maximal CONSECUTIVE repeated sub-sequences into ``_LoopTok``s; + everything else stays a ``_SingleTok``. Smallest repeating block wins (the + tightest loop body). Model-free and purely structural.""" + sigs = [_sig(s) for s in steps] + n = len(steps) + toks: list = [] + i = 0 + while i < n: + found: Optional[tuple[int, int]] = None + max_len = (n - i) // 2 + for length in range(1, max_len + 1): + block = sigs[i : i + length] + reps = 1 + while sigs[i + reps * length : i + reps * length + length] == block: + reps += 1 + if reps >= 2: + found = (length, reps) + break + if found is not None: + length, reps = found + body_steps = steps[i : i + length] + iterations = [ + steps[i + r * length : i + r * length + length] + for r in range(reps) + ] + toks.append(_LoopTok(body_steps, iterations)) + i += length * reps + else: + toks.append(_SingleTok(steps[i])) + i += 1 + return toks + + +# =========================================================================== +# Multiple-trace alignment (incremental LCS merge of reduced token sequences) +# =========================================================================== + + +class _Column: + """One aligned position across traces. ``tokens[t]`` is trace ``t``'s token + at this position, or absent when the trace does not have this column.""" + + def __init__(self, align_sig, kind: str): + self.align_sig = align_sig + self.kind = kind # "single" | "loop" + self.tokens: dict[int, Any] = {} + # For a "replace" divergence: the id of the mutually-exclusive column + # this one is paired against (same position, different content). + self.divergent_group: Optional[int] = None + + +def _align(reduced: list[list]) -> list[_Column]: + """Align the reduced token sequences of every trace into ordered columns. + + Incremental merge: start from trace 0's columns, then fold each subsequent + trace in with a ``difflib`` alignment over the token ALIGN-SIGNATURES. + ``equal`` blocks attach to existing columns; ``insert`` blocks add columns + present only where seen; ``delete`` blocks leave columns absent in the new + trace; ``replace`` blocks are recorded as mutually-exclusive DIVERGENCES + (branch-or-contradiction candidates). + """ + if not reduced: + return [] + columns: list[_Column] = [] + for tok in reduced[0]: + col = _Column(tok.align_sig, tok.kind) + col.tokens[0] = tok + columns.append(col) + + div_group = 0 + for t in range(1, len(reduced)): + seq = reduced[t] + a = [c.align_sig for c in columns] + b = [tok.align_sig for tok in seq] + sm = SequenceMatcher(a=a, b=b, autojunk=False) + merged: list[_Column] = [] + for tag, i1, i2, j1, j2 in sm.get_opcodes(): + if tag == "equal": + for k in range(i2 - i1): + col = columns[i1 + k] + col.tokens[t] = seq[j1 + k] + merged.append(col) + elif tag == "delete": + # In consensus, absent from this trace -> optional so far. + merged.extend(columns[i1:i2]) + elif tag == "insert": + # New in this trace only. + for j in range(j1, j2): + tok = seq[j] + col = _Column(tok.align_sig, tok.kind) + col.tokens[t] = tok + merged.append(col) + elif tag == "replace": + # Mutually-exclusive divergence: keep BOTH sides, tagged as one + # group, so interpretation can decide branch vs contradiction. + for col in columns[i1:i2]: + col.divergent_group = div_group + merged.append(col) + for j in range(j1, j2): + tok = seq[j] + col = _Column(tok.align_sig, tok.kind) + col.tokens[t] = tok + col.divergent_group = div_group + merged.append(col) + div_group += 1 + columns = merged + return columns + + +# =========================================================================== +# Interpretation: aligned columns -> ProgramGraph (or refusal) +# =========================================================================== + + +def _is_irreversible(col: _Column) -> bool: + return any( + tok.kind == "single" and tok.step.risk == "irreversible" + for tok in col.tokens.values() + ) + + +def _param_name_for(field: str, taken: set[str], fallback: str) -> str: + base = "".join(c if c.isalnum() else "_" for c in field.lower()).strip("_") + base = "_".join(p for p in base.split("_") if p) or fallback + if base not in taken: + return base + i = 2 + while f"{base}_{i}" in taken: + i += 1 + return f"{base}_{i}" + + +class _Node: + """A contiguous chunk of emitted graph with one ``entry`` state and a set of + ``exits`` (transitions whose target is patched to the following node).""" + + def __init__(self, entry: str): + self.entry = entry + self.states: dict[str, State] = {} + self.exits: list[Transition] = [] + + def set_next(self, next_id: str) -> None: + for tr in self.exits: + tr.target = next_id + + +def induce_program( + traces: list[TraceInput], *, propose: Optional[Proposer] = None +) -> InductionResult: + """Induce a parameterized :class:`ProgramGraph` from multiple demonstrations + of the same task -- the RFC §3 induction loop (step [4] + [5]). + + Deterministic and model-free at its core (align -> params / loops / + branches / optional). An optional :class:`Proposer` may PROPOSE an + interpretation for an ambiguous point; every proposal is flagged and never + silently trusted. When intent stays underdetermined the induced program is + QUARANTINED (not emitted) and ``result.certified is False``. + + Args: + traces: Two or more traces (compiled ``Workflow``s or recording dirs) + of the SAME task. One trace is the degenerate bootstrap. + propose: Optional compile-time interpretation proposer (advisory). + + Returns: + An :class:`InductionResult` with the induced program (or a refusal) and + the full audit trail. + """ + workflows = [_as_workflow(t) for t in traces] + n = len(workflows) + result = InductionResult(n_traces=n) + if n == 0: + return result + + reduced = [_reduce_trace(wf.steps) for wf in workflows] + columns = _align(reduced) + + # Detect an alignment failure: too little shared structure to trust that + # these are traces of the SAME task (refuse rather than induce noise). + shared = sum(1 for c in columns if len(c.tokens) == n) + if n >= 2 and columns and shared == 0: + result.uncertainties.append( + Uncertainty( + kind="alignment_failure", + location="", + detail=( + "traces share no aligned steps -- cannot infer a common " + "program (are these the same task?)." + ), + ) + ) + return result + + decisions: list[ColumnDecision] = [] + nodes: list[_Node] = [] + param_specs: dict[str, ParamSpec] = {} + data_sources: dict[str, Relation] = {} + taken: set[str] = set() + seen_div_groups: set[int] = set() + + for idx, col in enumerate(columns): + present = sorted(col.tokens) + + # --- divergence (replace region): branch or contradiction ---------- + if col.divergent_group is not None: + grp = col.divergent_group + if grp in seen_div_groups: + continue # the whole group is handled once, at its first column + seen_div_groups.add(grp) + group_cols = [c for c in columns if c.divergent_group == grp] + _handle_divergence( + idx, group_cols, workflows, result, propose, decisions + ) + # Underdetermined divergence => refuse to emit (handled below). + continue + + # --- loop ---------------------------------------------------------- + if col.kind == "loop": + node, dec = _emit_loop( + idx, col, present, n, taken, param_specs, data_sources + ) + nodes.append(node) + decisions.append(dec) + result.inferred.append(dec.note) + continue + + # --- single step present in ALL traces: literal vs param ---------- + if len(present) == n: + node, dec = _emit_required_single( + idx, col, present, taken, param_specs, result + ) + nodes.append(node) + decisions.append(dec) + continue + + # --- single step present in SOME traces: branch or optional ------- + node, dec = _emit_optional_single( + idx, col, present, n, result, propose + ) + nodes.append(node) + decisions.append(dec) + + result.column_decisions = decisions + result.param_specs = param_specs + + # Refuse rather than guess: any underdetermined point quarantines the whole + # program (RFC §3 [5]) -- do NOT emit a program that guesses a branch. + if result.uncertainties: + return result + + program, subflows = _wire(nodes) + if program is None: + return result + workflow = Workflow( + name="induced-program", + program=program, + subflows=subflows, + param_specs=param_specs, + params={k: (v.example or "") for k, v in param_specs.items()}, + data_sources=data_sources, + ) + result.program = program + result.workflow = workflow + return result + + +def _emit_loop(idx, col, present, n, taken, param_specs, data_sources): + loop_toks = [col.tokens[t] for t in present] + body_steps = loop_toks[0].body_steps + field = _field_key(body_steps[0]) if body_steps else f"row_{idx}" + counts = [tok.count for tok in loop_toks] + + param_name = _param_name_for(field, taken, f"row_{idx}") + taken.add(param_name) + # Representative worklist: the LONGEST demonstrated queue, inlined so the + # bundle is self-contained (a run may override via Replayer worklists=...). + rep = max(loop_toks, key=lambda tk: tk.count) + rel_name = f"worklist_{idx}" + data_sources[rel_name] = Relation( + name=rel_name, + rows=rep.rows(param_name), + description=f"inferred worklist for the '{field}' loop", + ) + param_specs[param_name] = ParamSpec( + name=param_name, + type=ParamKind.STRING, + example=(rep.rows(param_name)[0][param_name] if rep.count else None), + required=True, + ) + + # Body subflow: the repeated step(s), the varying field bound per row. + body_id = f"body_{idx}" + body_states: dict[str, State] = {} + prev_ids: list[str] = [] + for bi, bstep in enumerate(body_steps): + sid = f"{body_id}_s{bi}" + step = bstep.model_copy(deep=True) + if step.action is ActionKind.TYPE: + step.param = param_name + step.text = None + st = State( + id=sid, + kind=StateKind.ACTION, + step=step, + transitions=[Transition(target="__PATCH__")], + ) + body_states[sid] = st + prev_ids.append(sid) + body_end = f"{body_id}_end" + body_states[body_end] = State( + id=body_end, kind=StateKind.TERMINAL, outcome="success" + ) + ordered = list(body_states) + for k, sid in enumerate(ordered): + st = body_states[sid] + if st.kind is StateKind.ACTION: + st.transitions[0].target = ( + ordered[k + 1] if k + 1 < len(ordered) else body_end + ) + body_graph = ProgramGraph(entry=ordered[0], states=body_states) + + loop_state_id = f"loop_{idx}" + node = _Node(loop_state_id) + exit_tr = Transition(target="__PATCH__") + node.states[loop_state_id] = State( + id=loop_state_id, + kind=StateKind.LOOP, + loop=LoopSpec(relation=rel_name, body=body_id, var=field), + transitions=[exit_tr], + ) + node.exits = [exit_tr] + node._subflow = (body_id, body_graph) # type: ignore[attr-defined] + + dec = ColumnDecision( + index=idx, + kind="loop", + align_sig=str(col.align_sig), + field=field, + param_name=param_name, + present_in=present, + counts=counts, + note=( + f"loop over '{field}' -- body repeats {counts} across traces " + f"(counts differ => a worklist, not unrolled steps)" + ), + ) + return node, dec + + +def _emit_required_single(idx, col, present, taken, param_specs, result): + toks = [col.tokens[t] for t in present] + step0 = toks[0].step + field = _field_key(step0) + values = [tok.value for tok in toks] + sid = f"s{idx}" + step = step0.model_copy(deep=True) + exit_tr = Transition(target="__PATCH__") + + varies = len({v for v in values}) > 1 + if step0.action is ActionKind.TYPE and varies: + # A value that VARIES across traces at the same field is a PARAMETER + # (cross-trace evidence makes WebRobot value-speculation determinate). + name = _param_name_for(field, taken, f"value_{idx}") + taken.add(name) + step.param = name + step.text = None + example = next((v for v in values if v), None) + param_specs[name] = ParamSpec( + name=name, type=ParamKind.STRING, example=example, required=True + ) + result.inferred.append( + f"param '{name}' -- '{field}' varies across traces {values}" + ) + kind, param_name, literal = "param", name, None + else: + literal = values[0] + if step0.action is ActionKind.TYPE: + result.inferred.append( + f"literal '{field}' = {literal!r} -- constant across traces" + ) + kind, param_name = "literal", None + + node = _Node(sid) + node.states[sid] = State( + id=sid, kind=StateKind.ACTION, step=step, transitions=[exit_tr] + ) + node.exits = [exit_tr] + dec = ColumnDecision( + index=idx, + kind=kind, + align_sig=str(col.align_sig), + field=field, + literal_value=literal if kind == "literal" else None, + param_name=param_name, + present_in=present, + ) + return node, dec + + +def _emit_optional_single(idx, col, present, n, result, propose): + tok = col.tokens[present[0]] + step0 = tok.step + field = _field_key(step0) + is_dialog = _looks_like_dialog(step0) + + if is_dialog: + # A DETECTABLE condition (the dialog's own presence) -> a guarded BRANCH + # (RFC §2.2). The guard is PROPOSED and flagged for confirmation. The + # dialog LABEL is extracted the same way disambiguation does (#74), so + # both agree on what text signals the popup. + dialog_text = _dialog_text(step0) + branch_id = f"branch_{idx}" + do_id = f"opt_{idx}" + do_step = step0.model_copy(deep=True) + fall_tr = Transition(target="__PATCH__") + do_tr = Transition(target="__PATCH__") + node = _Node(branch_id) + node.states[branch_id] = State( + id=branch_id, + kind=StateKind.BRANCH, + transitions=[ + Transition( + guard=Predicate( + kind=PredicateKind.TEXT_PRESENT, + text=dialog_text, + intent="optional dialog is present", + ), + target=do_id, + label="dialog present", + ), + fall_tr, + ], + ) + node.states[do_id] = State( + id=do_id, kind=StateKind.ACTION, step=do_step, transitions=[do_tr] + ) + node.exits = [fall_tr, do_tr] + result.inferred.append( + f"branch on optional dialog {dialog_text!r} at column {idx} " + f"(present in {present}/{n} traces) -- guard TEXT_PRESENT" + ) + proposed = _maybe_propose( + propose, branch_id, "guard", {"dialog": dialog_text}, result + ) + result.proposed.append( + Proposal( + target=branch_id, + kind="guard", + content=( + proposed + or f"confirm guard: TEXT_PRESENT({dialog_text!r})" + ), + ) + ) + dec = ColumnDecision( + index=idx, + kind="branch", + align_sig=str(col.align_sig), + field=field, + present_in=present, + note=f"optional dialog branch ({dialog_text!r})", + ) + return node, dec + + # No derivable condition -> an OPTIONAL guarded step that SKIPs when its own + # target is absent (Guard on_unmet='skip', predicate ANCHOR_RESOLVES). + sid = f"opt_{idx}" + step = step0.model_copy(deep=True) + exit_tr = Transition(target="__PATCH__") + guard_pred = ( + Predicate( + kind=PredicateKind.ANCHOR_RESOLVES, + anchor=step.anchor, + intent=f"optional step '{field}' target present", + ) + if step.anchor is not None + else Predicate( + kind=PredicateKind.TEXT_PRESENT, + text=field, + intent=f"optional step '{field}' present", + ) + ) + step.guard = Guard(predicate=guard_pred, on_unmet="skip") + node = _Node(sid) + node.states[sid] = State( + id=sid, kind=StateKind.ACTION, step=step, transitions=[exit_tr] + ) + node.exits = [exit_tr] + result.inferred.append( + f"optional step '{field}' at column {idx} " + f"(present in {present}/{n} traces) -- skip when absent" + ) + dec = ColumnDecision( + index=idx, + kind="optional", + align_sig=str(col.align_sig), + field=field, + present_in=present, + note=f"optional step '{field}' (guarded skip)", + ) + return node, dec + + +def _handle_divergence( + idx, group_cols, workflows, result, propose, decisions +): + """A ``replace`` divergence: mutually-exclusive content at the same aligned + position. If a discriminating condition were DETECTABLE it would be a + branch; here none is derivable structurally, so this is a contradiction the + traces do not resolve -- REFUSE (RFC §3 [5]) and route to disambiguation. + A :class:`Proposer` may suggest a guard, but it is flagged, NEVER trusted. + """ + labels = [] + for col in group_cols: + for tok in col.tokens.values(): + if tok.kind == "single": + labels.append(_value(tok.step) or _field_key(tok.step)) + labels = list(dict.fromkeys(labels)) + consequential = any(_is_irreversible(c) for c in group_cols) + + proposal_content = _maybe_propose( + propose, + f"divergence_{idx}", + "guard", + {"arms": labels}, + result, + ) + proposal = ( + Proposal( + target=f"divergence_{idx}", + kind="guard", + content=proposal_content, + ) + if proposal_content + else None + ) + if proposal is not None: + result.proposed.append(proposal) + + question = DisambiguationQuestion( + id=f"{AmbiguityKind.OPTIONAL_DIALOG.value}:divergence_{idx}", + kind=AmbiguityKind.OPTIONAL_DIALOG, + step_id=f"divergence_{idx}", + prompt=( + f"Traces diverge here between {labels}. Under what condition should " + "the workflow take each branch? (No condition was detectable from " + "the demonstrations.)" + ), + options=[ + QuestionOption( + key="halt", + label="Halt -- the condition is unknown (safe default)", + effect=OptionEffect.NONE, + ), + ], + default_key="halt", + consequential=consequential, + evidence=f"divergent branch between {labels} at column {idx}", + ) + result.uncertainties.append( + Uncertainty( + kind="ambiguous_branch", + location=f"column {idx}", + detail=( + f"traces diverge between {labels} with no detectable " + "condition -- cannot decide the guard; refusing to guess a " + "branch on a" + + (" consequential" if consequential else "") + + " node." + ), + consequential=consequential, + question=question, + proposal=proposal, + ) + ) + decisions.append( + ColumnDecision( + index=idx, + kind="divergent", + align_sig="__divergent__", + note=f"underdetermined divergence between {labels}", + ) + ) + + +def _maybe_propose( + propose: Optional[Proposer], target, kind, context, result +) -> Optional[str]: + if propose is None: + return None + try: + return propose.propose(target, kind, context) + except Exception: # pragma: no cover - advisory path never breaks induction + return None + + +def _wire(nodes: list[_Node]): + """Chain nodes into a single program graph + collect loop-body subflows.""" + if not nodes: + return None, {} + states: dict[str, State] = {} + subflows: dict[str, ProgramGraph] = {} + end_id = "__end__" + for i, node in enumerate(nodes): + states.update(node.states) + sub = getattr(node, "_subflow", None) + if sub is not None: + subflows[sub[0]] = sub[1] + node.set_next(nodes[i + 1].entry if i + 1 < len(nodes) else end_id) + states[end_id] = State(id=end_id, kind=StateKind.TERMINAL, outcome="success") + return ProgramGraph(entry=nodes[0].entry, states=states), subflows + + +# =========================================================================== +# Held-out validation (RFC §3 [5]) + reproduction scoring +# =========================================================================== + + +def reproduction_score(result: InductionResult, trace: TraceInput) -> float: + """Score how well the induced program would REPRODUCE ``trace`` in [0, 1]. + + Deterministic and backend-free: it checks each induced column decision + against the held-out trace's own tokens. A PARAM column reproduces any value + (the run supplies it); a LITERAL column only reproduces the frozen value + (so a param wrongly frozen as a constant scores LOW on a trace with a + different value); LOOP / BRANCH / OPTIONAL columns reproduce whatever count + / presence the trace shows. Extra trace tokens the program cannot account + for are penalized. Returns 0.0 for a quarantined (unemitted) program. + """ + if result.program is None or not result.column_decisions: + return 0.0 + held = _reduce_trace(_as_workflow(trace).steps) + + col_sigs = [ + d.align_sig for d in result.column_decisions if d.kind != "divergent" + ] + decs = [d for d in result.column_decisions if d.kind != "divergent"] + held_sigs = [str(tok.align_sig) for tok in held] + + sm = SequenceMatcher(a=col_sigs, b=held_sigs, autojunk=False) + matched: dict[int, int] = {} # col index -> held index + for tag, i1, i2, j1, j2 in sm.get_opcodes(): + if tag == "equal": + for k in range(i2 - i1): + matched[i1 + k] = j1 + k + + score = 0.0 + for ci, dec in enumerate(decs): + if ci in matched: + tok = held[matched[ci]] + if dec.kind == "literal": + score += 1.0 if tok.value == dec.literal_value else 0.0 + elif dec.kind == "param": + score += 1.0 # a param reproduces any observed value + elif dec.kind == "loop": + score += 1.0 if tok.kind == "loop" else 0.5 + else: # branch / optional + score += 1.0 + else: + # Column not present in the held trace. + score += 1.0 if dec.kind in ("branch", "optional") else 0.0 + + total = len(decs) + extra = len(held) - len(matched) + denom = max(total + max(extra, 0), 1) + return score / denom + + +def validate_held_out( + traces: list[TraceInput], *, propose: Optional[Proposer] = None +) -> HeldOutValidation: + """Leave-one-out held-out validation (RFC §3 [5]): for each trace, induce a + program from the OTHER N-1 traces and score whether it reproduces the held + one. Reports the per-fold scores and their mean. Requires >= 2 traces.""" + n = len(traces) + if n < 2: + return HeldOutValidation(per_trace=[], mean=0.0, n_traces=n) + scores: list[float] = [] + for i in range(n): + train = traces[:i] + traces[i + 1 :] + result = induce_program(train, propose=propose) + scores.append(reproduction_score(result, traces[i])) + mean = sum(scores) / len(scores) if scores else 0.0 + return HeldOutValidation(per_trace=scores, mean=mean, n_traces=n) diff --git a/tests/test_induction.py b/tests/test_induction.py new file mode 100644 index 0000000..123f464 --- /dev/null +++ b/tests/test_induction.py @@ -0,0 +1,511 @@ +"""Multi-trace induction (RFC docs/design/WORKFLOW_PROGRAM_IR.md §3 [4]+[5]). + +Infer a parameterized PROGRAM (the Phase-2 ``ProgramGraph``) from MULTIPLE +demonstrations of the same task -- the induction loop the whole PBD lineage +(Rousillon, WebRobot, Skill-DisCo, PROLEX) says a demonstration compiler must +have. "One demonstration is evidence, not specification." + +Since we lack real multi-worker traces, a small generator builds trace VARIANTS +of a MockMed task: + +* (a) two traces differing only in a typed value -> induce a PARAM +* (b) traces with a worklist of length 2 vs 3 -> induce a LOOP +* (c) traces with an optional dialog present vs absent -> induce a BRANCH +* (d) contradictory traces -> induction REJECTS + +Everything is deterministic and model-free: the compile-time proposer is a +deterministic FAKE (zero model calls), and its suggestions are FLAGGED, never +silently trusted. The induced program is REPLAYED through the real Phase-2 +interpreter (``runtime.replayer``) with the faked backend/vision from +``test_replayer`` -- proving round-trip. +""" + +from __future__ import annotations + + +import pytest + +from openadapt_flow.compiler.induction import ( + induce_program, + reproduction_score, + validate_held_out, +) +from openadapt_flow.ir import ( + ActionKind, + Anchor, + StateKind, + Step, + Workflow, +) +from openadapt_flow.runtime.replayer import Replayer +from test_replayer import FakeBackend, FakeVision, Match, make_png + + +# =========================================================================== +# Synthetic MockMed corpus: trace-variant generators +# =========================================================================== + + +def _type(step_id: str, field: str, value: str, *, risk: str = "reversible") -> Step: + return Step( + id=step_id, + intent=f"type {field}", + action=ActionKind.TYPE, + text=value, + risk=risk, + ) + + +def _key(step_id: str, key: str, *, intent: str | None = None, risk="reversible") -> Step: + return Step( + id=step_id, + intent=intent or f"press {key}", + action=ActionKind.KEY, + key=key, + risk=risk, + ) + + +def mockmed_param_traces() -> list[Workflow]: + """(a) Two+ traces of 'open a chart and record a dose' that differ ONLY in + the patient typed -- the dose is constant. => patient is a PARAM, dose a + literal.""" + + def trace(patient: str) -> Workflow: + return Workflow( + name="record-dose", + steps=[ + _type("s_patient", "patient", patient), + _type("s_dose", "dose", "10mg"), + _key("s_save", "Enter", intent="press Enter to save"), + ], + ) + + return [trace("Alice Alvarez"), trace("Bob Baker"), trace("Cara Chen")] + + +def mockmed_loop_traces() -> list[Workflow]: + """(b) 'Clear the worklist': one trace processes 2 patients, another 3. The + repeated body (type each patient) with a DIFFERING count => a LOOP over a + Relation.""" + + def trace(patients: list[str]) -> Workflow: + steps: list[Step] = [_type("s_login", "clinic", "MockMed General")] + for i, p in enumerate(patients): + steps.append(_type(f"s_row{i}", "patient", p)) + steps.append(_key("s_done", "Enter", intent="press Enter to finish")) + return Workflow(name="clear-worklist", steps=steps) + + return [trace(["Alice", "Bob"]), trace(["Cara", "Dan", "Eve"])] + + +def mockmed_optional_traces() -> list[Workflow]: + """(c) 'Save an encounter': in one trace a Survey popup appears and is + dismissed; in another it does not. => a BRANCH guarded on the popup's + presence (guard proposed / flagged).""" + + def trace(with_survey: bool) -> Workflow: + steps = [ + _type("s_patient", "patient", "Alice"), + _key("s_save", "S", intent="press S to save the encounter"), + ] + if with_survey: + steps.append( + _key( + "s_dismiss", + "Escape", + intent="survey popup appeared - dismiss it", + ) + ) + steps.append(_key("s_close", "Enter", intent="press Enter to close")) + return Workflow(name="save-encounter", steps=steps) + + return [trace(True), trace(False)] + + +def mockmed_contradiction_traces() -> list[Workflow]: + """(d) Two traces that CONTRADICT at the same aligned position with no + detectable condition: one APPROVES the claim, the other REJECTS it -- both + irreversible. => induction REFUSES (underdetermined).""" + + def trace(decision_key: str, label: str) -> Workflow: + return Workflow( + name="adjudicate-claim", + steps=[ + _type("s_patient", "patient", "Alice"), + _key( + "s_decide", + decision_key, + intent=f"{label} the claim", + risk="irreversible", + ), + ], + ) + + return [trace("A", "approve"), trace("R", "reject")] + + +# =========================================================================== +# Deterministic FAKE compile-time proposer (zero model calls) +# =========================================================================== + + +class FakeProposer: + """A deterministic stand-in for the #78 compile-time StepAnnotator. Makes + ZERO model calls; returns a canned suggestion and records that it was + asked. Its output is advisory -- the tests assert it is flagged and NEVER + flips an underdetermined point to certified.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, str, dict]] = [] + + def propose(self, target: str, kind: str, context) -> str: + self.calls.append((target, kind, context)) + return f"[annotator guess for {target}: {kind}]" + + +# =========================================================================== +# Replay harness (round-trip through the real Phase-2 interpreter) +# =========================================================================== + + +@pytest.fixture() +def bundle(tmp_path): + bd = tmp_path / "bundle" + (bd / "templates").mkdir(parents=True) + (bd / "templates" / "btn.png").write_bytes(make_png((50, 20))) + return bd + + +def _run(workflow, bundle, run_dir, *, vision=None, worklists=None, params=None): + backend = FakeBackend() + report = Replayer( + backend, vision=vision or FakeVision(), poll_interval_s=0.01 + ).run( + workflow, + bundle_dir=bundle, + run_dir=run_dir, + worklists=worklists, + params=params or {}, + ) + return backend, report + + +# =========================================================================== +# (a) PARAMETER induction +# =========================================================================== + + +def test_a_varying_value_induces_a_param_constant_stays_literal(): + result = induce_program(mockmed_param_traces()) + assert result.certified is True + assert result.program is not None + + # 'patient' varied -> a param; 'dose' constant -> a baked literal. + kinds = {d.field: d.kind for d in result.column_decisions} + assert kinds["patient"] == "param" + assert kinds["dose"] == "literal" + + pname = next(d.param_name for d in result.column_decisions if d.field == "patient") + assert pname in result.param_specs + assert result.param_specs[pname].example == "Alice Alvarez" + + # The dose literal is frozen in the emitted step; the patient step is a param. + steps = { + s.step.id: s.step + for s in result.program.states.values() + if s.kind is StateKind.ACTION and s.step is not None + } + dose = next(s for s in steps.values() if "dose" in s.intent) + assert dose.text == "10mg" and dose.param is None + patient = next(s for s in steps.values() if "patient" in s.intent) + assert patient.param == pname and patient.text is None + + +def test_a_param_program_round_trips_with_a_new_value(bundle, tmp_path): + result = induce_program(mockmed_param_traces()) + pname = next(d.param_name for d in result.column_decisions if d.field == "patient") + backend, report = _run( + result.workflow, + bundle, + tmp_path / "run", + params={pname: "Zoe Zhang"}, # a value never demonstrated + ) + assert report.success is True + assert backend.actions == [ + ("type", "Zoe Zhang"), + ("type", "10mg"), + ("press", "Enter"), + ] + assert report.model_calls == 0 # $0 runtime preserved + + +# =========================================================================== +# (b) LOOP induction +# =========================================================================== + + +def test_b_differing_repetition_count_induces_a_loop_over_a_relation(): + result = induce_program(mockmed_loop_traces()) + assert result.certified is True + + loops = [s for s in result.program.states.values() if s.kind is StateKind.LOOP] + assert len(loops) == 1 + loop_state = loops[0] + rel = loop_state.loop.relation + assert rel in result.workflow.data_sources + # The body subflow exists and is bound to the loop. + assert loop_state.loop.body in result.workflow.subflows + + dec = next(d for d in result.column_decisions if d.kind == "loop") + assert sorted(dec.counts) == [2, 3] # counts DIFFER across traces + + +def test_b_loop_program_replays_body_once_per_row(bundle, tmp_path): + result = induce_program(mockmed_loop_traces()) + loop_state = next( + s for s in result.program.states.values() if s.kind is StateKind.LOOP + ) + rel = loop_state.loop.relation + + # Run-time worklist of a length neither demo used (data-dependent queue). + backend, report = _run( + result.workflow, + bundle, + tmp_path / "run", + worklists={rel: [{"patient": "Q1"}, {"patient": "Q2"}, {"patient": "Q3"}, + {"patient": "Q4"}]}, + ) + assert report.success is True + assert backend.actions == [ + ("type", "MockMed General"), + ("type", "Q1"), + ("type", "Q2"), + ("type", "Q3"), + ("type", "Q4"), + ("press", "Enter"), + ] + + +def test_b_loop_inline_rows_replay_without_a_supplied_worklist(bundle, tmp_path): + """The bundle is self-contained: the representative (longest) demonstrated + worklist is inlined, so a run with no worklist still iterates it.""" + result = induce_program(mockmed_loop_traces()) + backend, report = _run(result.workflow, bundle, tmp_path / "run") + assert report.success is True + assert backend.actions == [ + ("type", "MockMed General"), + ("type", "Cara"), + ("type", "Dan"), + ("type", "Eve"), + ("press", "Enter"), + ] + + +# =========================================================================== +# (c) BRANCH / optional-step induction +# =========================================================================== + + +def test_c_optional_dialog_induces_a_guarded_branch(): + result = induce_program(mockmed_optional_traces()) + assert result.certified is True + + branches = [ + s for s in result.program.states.values() if s.kind is StateKind.BRANCH + ] + assert len(branches) == 1 + branch = branches[0] + # Two arms: guarded (dialog present) + unconditional fall-through. + guarded = [t for t in branch.transitions if t.guard is not None] + fall = [t for t in branch.transitions if t.guard is None] + assert len(guarded) == 1 and len(fall) == 1 + + # The guard is PROPOSED / flagged for confirmation, never silently trusted. + assert any(p.kind == "guard" and p.trusted is False for p in result.proposed) + + +def test_c_branch_program_dismisses_when_present_skips_when_absent(bundle, tmp_path): + result = induce_program(mockmed_optional_traces()) + branch = next( + s for s in result.program.states.values() if s.kind is StateKind.BRANCH + ) + guard_text = next(t.guard.text for t in branch.transitions if t.guard) + + # Popup PRESENT -> the guarded arm dismisses it. + v_present = FakeVision() + v_present.text_results = {guard_text: Match((10, 10), (0, 0, 5, 5))} + backend, report = _run( + result.workflow, bundle, tmp_path / "present", vision=v_present + ) + assert report.success is True + assert backend.actions == [("type", "Alice"), ("press", "S"), + ("press", "Escape"), ("press", "Enter")] + + # Popup ABSENT -> fall-through skips the dismiss. + backend2, report2 = _run( + result.workflow, bundle, tmp_path / "absent", vision=FakeVision() + ) + assert report2.success is True + assert backend2.actions == [("type", "Alice"), ("press", "S"), + ("press", "Enter")] + + +def test_c_optional_non_dialog_step_becomes_a_guarded_skip(bundle, tmp_path): + """An optional step with NO derivable condition (not a dialog) becomes a + guarded step that SKIPs when its own target is absent -- not a branch.""" + + def trace(with_extra: bool) -> Workflow: + steps = [_type("s_patient", "patient", "Alice")] + if with_extra: + # A non-dialog optional click with a resolvable anchor. + steps.append( + Step( + id="s_flag", + intent="mark the chart reviewed", + action=ActionKind.CLICK, + anchor=Anchor( + template="templates/btn.png", + region=(100, 100, 50, 20), + click_point=(110, 105), + ocr_text="Reviewed", + ), + ) + ) + steps.append(_key("s_done", "Enter")) + return Workflow(name="review", steps=steps) + + result = induce_program([trace(True), trace(False)]) + assert result.certified is True + assert any(d.kind == "optional" for d in result.column_decisions) + step = next( + s.step + for s in result.program.states.values() + if s.kind is StateKind.ACTION and s.step and s.step.id == "s_flag" + ) + assert step.guard is not None and step.guard.on_unmet == "skip" + + # The anchor never resolves (empty vision) -> the guard is unmet -> the + # step is SKIPPED and the run still succeeds. + backend, report = _run(result.workflow, bundle, tmp_path / "run") + assert report.success is True + assert backend.actions == [("type", "Alice"), ("press", "Enter")] + + +# =========================================================================== +# (d) CONTRADICTION -> REJECT rather than guess +# =========================================================================== + + +def test_d_contradictory_traces_are_rejected_not_guessed(): + result = induce_program(mockmed_contradiction_traces()) + assert result.underdetermined is True + assert result.certified is False + assert result.program is None # quarantined -- NOT emitted + assert result.workflow is None + + u = result.uncertainties[0] + assert u.kind == "ambiguous_branch" + assert u.consequential is True # both arms are irreversible + # Routed to the disambiguation flow (#74). + assert u.question is not None + + +def test_d_proposal_is_flagged_but_never_flips_underdetermined_to_certified(): + proposer = FakeProposer() + result = induce_program(mockmed_contradiction_traces(), propose=proposer) + # The proposer WAS consulted... + assert proposer.calls, "proposer should be asked about the divergence" + # ...its guess is surfaced (flagged, not trusted)... + assert result.proposed and all(p.trusted is False for p in result.proposed) + assert result.uncertainties[0].proposal is not None + # ...but the point stays underdetermined and the program is NOT emitted. + assert result.certified is False + assert result.program is None + + +# =========================================================================== +# Held-out validation (RFC §3 [5]) +# =========================================================================== + + +def test_held_out_scores_a_good_param_induction_high(): + hv = validate_held_out(mockmed_param_traces()) + assert hv.n_traces == 3 + assert hv.mean == pytest.approx(1.0) # reproduces every held-out trace + + +def test_held_out_scores_an_over_specialized_induction_low(): + """Two IDENTICAL demos cannot reveal that the value varies, so the compiler + freezes it as a literal (over-specialization). Held against a trace with a + DIFFERENT value, that literal does not reproduce -> a LOW score. Contrast + with the param induction, which reproduces it exactly.""" + traces = mockmed_param_traces() + + identical = [traces[0], traces[0].model_copy(deep=True)] + frozen = induce_program(identical) + # No param was inferred -- the varying field is frozen as a literal. + assert not frozen.param_specs + bad = reproduction_score(frozen, traces[1]) # a different patient + + good = induce_program([traces[0], traces[1]]) + good_score = reproduction_score(good, traces[2]) + + assert good_score == pytest.approx(1.0) + assert bad < good_score + assert bad < 1.0 + + +def test_held_out_reproduces_the_loop_corpus_exactly(): + # A loop is inducible from ANY single held-out fold's training trace (each + # remaining trace still shows the repeated body), so every fold reproduces. + assert validate_held_out(mockmed_loop_traces()).mean == pytest.approx(1.0) + + +def test_held_out_partially_reproduces_a_two_trace_branch_corpus(): + """A branch needs BOTH the present and absent variant to be inducible, so + leave-one-out over just two traces (training on one) cannot reproduce the + branch fully -- an HONEST partial score, not a fabricated 1.0. It is still + high (the linear skeleton reproduces); a third variant would raise it.""" + mean = validate_held_out(mockmed_optional_traces()).mean + assert 0.7 <= mean < 1.0 + + +# =========================================================================== +# Bootstrap + invariants +# =========================================================================== + + +def test_single_trace_induces_the_degenerate_linear_bootstrap(): + """One demo is the bootstrap seed (RFC §3 [1]): a linear program, every + typed value a literal (a single demo cannot reveal a parameter).""" + [one] = [mockmed_param_traces()[0]] + result = induce_program([one]) + assert result.certified is True + assert result.program is not None + assert not result.param_specs # nothing varies -> nothing is a param + assert all(d.kind in ("literal",) for d in result.column_decisions + if d.field in ("patient", "dose")) + + +def test_induction_makes_zero_model_calls_and_replays_at_zero_cost(bundle, tmp_path): + proposer = FakeProposer() + result = induce_program(mockmed_param_traces(), propose=proposer) + pname = next(d.param_name for d in result.column_decisions if d.field == "patient") + _, report = _run( + result.workflow, bundle, tmp_path / "run", params={pname: "New Name"} + ) + # The proposer is deterministic (no model); the runtime is $0 / 0-call. + assert report.model_calls == 0 + assert report.est_model_cost_usd == 0.0 + + +def test_alignment_failure_on_unrelated_traces_is_refused(): + """Traces that share no aligned steps are not the same task -- refuse rather + than induce noise.""" + a = Workflow(name="a", steps=[_type("x", "alpha", "1"), _key("y", "F1")]) + b = Workflow(name="b", steps=[_type("z", "omega", "2"), _key("w", "F9")]) + result = induce_program([a, b]) + assert result.certified is False + assert result.program is None + assert result.uncertainties[0].kind == "alignment_failure"