diff --git a/packages/flow/docs/CAPABILITIES.md b/packages/flow/docs/CAPABILITIES.md index f66fe931e..81f24c983 100644 --- a/packages/flow/docs/CAPABILITIES.md +++ b/packages/flow/docs/CAPABILITIES.md @@ -65,6 +65,10 @@ at replaying one. These capabilities layer onto the same $0, model-free runtime. underdetermined. The healthy replay path still makes no model call. [`LEARN_FROM_MANY_RUNS.md`](LEARN_FROM_MANY_RUNS.md) covers `induce`, `teach-branch` and shadow mining together, with each one's status. + With `--probe-on test-copy.yaml`, a refused optional step is instead + replayed both ways on a declared test copy and scored by reading the record + back; it is certified only when exactly one reading works for every + recording ([`PROBE_INDUCTION.md`](PROBE_INDUCTION.md)). - **Effect verification against the system of record.** The read-back checks our own work: a wrong field, a wrong record, a duplicate submit, or a partial save can all look like success on screen. A step diff --git a/packages/flow/docs/LEARN_FROM_MANY_RUNS.md b/packages/flow/docs/LEARN_FROM_MANY_RUNS.md index f28324a7e..a4eec454c 100644 --- a/packages/flow/docs/LEARN_FROM_MANY_RUNS.md +++ b/packages/flow/docs/LEARN_FROM_MANY_RUNS.md @@ -39,6 +39,12 @@ When the recordings don't settle a question, it refuses instead of guessing: each of these two paths under a condition you can see" or "record the path where this dialog doesn't appear". Record those and run it again. +Or, when a test copy of the app is available, let it answer: +`induce --probe-on test-copy.yaml` replays each recording with the disputed +step dropped and with it kept, scores each run by reading the record back, +and certifies the step only when exactly one reading works for every +recording. See [PROBE_INDUCTION.md](PROBE_INDUCTION.md). + `admit rec-1 rec-2` runs the same induction as its first step, then proposes the checks and stops once for a decision (see [SHADOW_CANDIDATES.md](SHADOW_CANDIDATES.md#how-a-candidate-is-admitted)). diff --git a/packages/flow/docs/PROBE_INDUCTION.md b/packages/flow/docs/PROBE_INDUCTION.md new file mode 100644 index 000000000..1ad8d1c74 --- /dev/null +++ b/packages/flow/docs/PROBE_INDUCTION.md @@ -0,0 +1,167 @@ +# Settle a disagreement between recordings on a test copy + +`induce` learns one program from several recordings of a task. When one +recording has a consequential step that another lacks, the recordings alone +can't say whether the step was noise or a branch the task needs. Plain +`induce` refuses and names the recording that would settle it, for example +"Record the path where Sign chart is absent." + +Recording more clean, agreeing demonstrations is slow. `induce --probe-on` +lets a test copy of the app answer the question instead. It's opt-in, and it +runs only on a target the deployment file declares a test copy. + +Status: in review. Tested in CI on MockMed, OpenAdapt's synthetic test app, +with its transactional server as the system of record. It has not run +against a third-party application yet. + +## How it works + +For each refused point, probing tries two readings: + +| Reading | What every recording replays | +|---|---| +| `incidental` | The recording without the step. | +| `conditional` | The recording with the step, set to run only when its target (or its dialog) is on screen and to skip otherwise. This is the guard plain `induce` would put on the step. | + +Each reading replays in every recording's context: that recording's own +steps, values and subject, with only the point in question changed. Before +each run the test copy is reset to its baseline, and each run starts a new +session. + +Each run is scored by reading the system of record back, with the same judge +`serve-reward` uses (`openadapt_flow.reward.worker.judge_episode`) and the +oracle the reward contract names. A run counts only when the judge says +`verified` for that recording's subject: every required effect is present +and no forbidden effect is. + +A point is certified only when exactly one reading produced the recorded +effect in every context and the other reading conclusively failed in at least +one. In every other case the point stays refused, with its record-next item, +exactly as without probing: + +- Both readings produced the effect. The probe can't tell them apart. +- Neither did. +- A read-back couldn't decide (store unreachable, effect uncertain, replay + crashed). Such a run is logged as inconclusive and never counts as a miss. + +Probing covers the points plain `induce` reports as `unconfirmed_optional` +and `unconfirmed_branch`. Other refusals (contradicting paths, step order, +loops with a consequential body) are logged as not probed and stay refused. + +## Examples + +These are the three MockMed cases in `tests/test_probe_induction.py`. + +- **Incidental click.** Recording A clicked Sign chart on a chart that was + already signed, so the button did nothing. Recording B didn't click it. On + the test copy, the `conditional` reading signs B's chart, which writes a + signature record the reward contract forbids. Only `incidental` holds, so + the program drops the step. +- **Real branch.** Recording A's patient opened behind an allergy alert, and + the person pressed Accept. Recording B's patient had no alert. The + `incidental` reading leaves A's Save button behind the alert, and the run + halts before saving. Only `conditional` holds, so the program gets a branch + guarded on the alert. +- **Can't decide.** Recording A clicked Print summary. Printing writes + nothing to the record, so both readings produce the recorded effect. The + point stays refused. + +## Declare the test copy + +Probing reads one deployment file. It refuses, before it resets or replays +anything, unless that file has a `test_copy` section like this one: + +```yaml +name: mockmed-test-copy +backend: + url: http://127.0.0.1:8765/ +test_copy: + declared: true + reset_url: http://127.0.0.1:8765/api/reset # or reset_command: [argv...] + reward_contract: contracts/chart-note # serve-reward bundle +``` + +| Field | Meaning | +|---|---| +| `declared` | The operator's statement that every system the file names is a test copy. Must be `true`. | +| `reset_url` | Probing sends an HTTP POST here before every run to restore the baseline. | +| `reset_command` | Or a command, as an argv list run without a shell. | +| `reward_contract` | A reward contract bundle in the `serve-reward` format. Its required and forbidden effects define the recorded effect; its oracle recipe is the read-back. A relative path resolves against the deployment file. | + +Probing also refuses when: + +- there's no reset or no reward contract, +- `runtime.profile` is `standard` or `regulated` (production profiles), or +- the backend attaches to an already-open browser + (`browser_cdp_endpoint`, `browser_page_url`), since each run needs a fresh + session. + +The reward contract's identity keys (for example `patient_id`) must be in +each recording's `params`, so each run is judged against that recording's own +subject. A recording without them isn't probed. + +## Run it + +```bash +openadapt-flow induce rec-a rec-b --out program \ + --probe-on test-copy.yaml --report induce-report.json +``` + +The output lists every probe run and one verdict per point: + +```text +probes on test copy 'mockmed-test-copy': 4 run(s) + [probe] opt_2 incidental in rec-a: verified (completed) -- every required effect is present exactly as declared + [probe] opt_2 incidental in rec-b: verified (completed) -- every required effect is present exactly as declared + [probe] opt_2 conditional in rec-a: verified (completed) -- every required effect is present exactly as declared + [probe] opt_2 conditional in rec-b: wrong_effect (completed) -- a forbidden effect is present in the system of record + [verdict] opt_2: certified incidental [incidental=passed, conditional=failed] -- only incidental produced the recorded effect in all 2 recording contexts +Certification: CERTIFIED +``` + +Once a reading conclusively fails in one context, probing doesn't run it in +the remaining ones. + +`--report` writes the same result as JSON, including every run's reading, +context, runtime signal, judge outcome, reason and model-call count. +`--probe-work-dir` keeps the staged bundles and run directories (screenshots +and replay reports); by default they go to a temporary directory that is +removed afterwards. + +Exit codes: 0 when the program is certified and written, 2 when a point stays +refused, 3 when the target is refused. + +## What probing will not do + +- **Probe production.** The `test_copy` declaration is the only thing that + opens the gate, and a production profile closes it again. +- **Let a model decide.** A `Proposer` stays advisory: its proposals are + recorded as untrusted, and probing never reads them. The test + `test_proposer_stays_advisory` gives a proposer that argues for the wrong + reading; the read-back decides. +- **Call a model by default.** Probe replays run with model grounding and + egress off. Each run's model-call count is in the report. +- **Hide a run.** Every run is in the report, including inconclusive ones and + runs that didn't change the outcome. + +## Limits + +- One point at a time. While a point is probed, every other point keeps what + each recording did. +- Only two readings per point. A step that should run under a condition other + than its own target being on screen (say, a field value) needs a recording, + as before. +- The check is only as good as the reward contract. A contract with no + forbidden effects can't tell a harmless extra write from a harmful one. +- The CI tests drive the real replayer against a scripted MockMed screen. The + browser path (`DeploymentProbeTarget` for `web` backends) isn't covered + end to end yet. + +## Code + +- `openadapt_flow/compiler/probing.py`: `probe_and_induce`, + `probe_target_refusals`, `DeploymentProbeTarget` +- `openadapt_flow/compiler/induction.py`: `optional_points`, + `optional_guard`, `ProbeRun`, `ProbeVerdict` +- `openadapt_flow/deployment.py`: `ProbeCopySection` (`test_copy`) +- `tests/test_probe_induction.py` diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 7b61d75b7..6af77a20b 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -1847,8 +1847,19 @@ def _cmd_induce(args: argparse.Namespace) -> int: else: traces.append(path) - result = induce_program(traces) + if args.probe_on: + probed = _induce_with_probes(args, traces) + if isinstance(probed, int): + return probed + result = probed + else: + result = induce_program(traces) print(result.render()) + if args.report: + report_path = Path(args.report) + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text(result.model_dump_json(indent=2) + "\n") + print(f"Induction report written to {report_path}") if args.held_out and len(traces) >= 2: print(validate_held_out(traces).render()) @@ -1876,6 +1887,64 @@ def _cmd_induce(args: argparse.Namespace) -> int: return 0 +def _induce_with_probes(args: argparse.Namespace, traces: list) -> Any: + """``induce --probe-on``: refuse a non-test target, else probe and induce. + + Returns the InductionResult, or an exit code when the target is refused. + """ + import tempfile + + from openadapt_flow.compiler import probing + from openadapt_flow.deployment import load_deployment + + deployment = Path(args.probe_on) + try: + cfg = load_deployment(deployment) + except (FileNotFoundError, ValueError) as exc: + print(f"PROBE REFUSED: {exc}") + return 3 + reasons = probing.probe_target_refusals(cfg) + if reasons: + print("PROBE REFUSED: nothing was reset or replayed.") + for reason in reasons: + print(f" - {reason}") + return 3 + contract = Path(str(cfg.test_copy.reward_contract)).expanduser() + if not contract.is_absolute(): + contract = deployment.parent / contract + + def _run(work: Path) -> Any: + from openadapt_flow.compiler.compile import compile_recording + from openadapt_flow.ir import Workflow + + probe_traces = [] + seen: set[str] = set() + for i, (arg, trace) in enumerate(zip(args.recording, traces)): + path = Path(arg) + label = path.name if path.name not in seen else f"{path.name}#{i}" + seen.add(label) + if isinstance(trace, Workflow): + bundle = path + else: + bundle = work / "compiled" / f"{i}-{path.name}" + trace = compile_recording(path, bundle, name=path.name) + probe_traces.append( + probing.ProbeTrace(workflow=trace, bundle_dir=bundle, label=label) + ) + target = probing.DeploymentProbeTarget(cfg) + print(f"Probing on declared test copy {cfg.name!r} (read-back: {contract})") + return probing.probe_and_induce( + probe_traces, target=target, reward_contract=contract, work_dir=work + ) + + if args.probe_work_dir: + work = Path(args.probe_work_dir) + work.mkdir(parents=True, exist_ok=True) + return _run(work) + with tempfile.TemporaryDirectory(prefix="induce-probe-") as tmp: + return _run(Path(tmp)) + + def _cmd_for_each(args: argparse.Namespace) -> int: import shutil @@ -7430,6 +7499,31 @@ def build_parser() -> argparse.ArgumentParser: "reproduction scores (needs >= 2 traces)" ), ) + p.add_argument( + "--probe-on", + default=None, + metavar="DEPLOYMENT", + help=( + "Settle refused optional steps by replaying each reading (step " + "dropped / step kept when present) in a fresh session on the test " + "copy this deployment file declares (test_copy.declared: true), " + "scored by its reward contract's read-back. Refuses any target " + "not declared a test copy. See docs/PROBE_INDUCTION.md" + ), + ) + p.add_argument( + "--probe-work-dir", + default=None, + help=( + "Keep probe bundles and run directories here (default: a " + "temporary directory, removed afterwards)" + ), + ) + p.add_argument( + "--report", + default=None, + help="Write the induction report (JSON, every probe run included) here", + ) p.set_defaults(func=_cmd_induce) p = sub.add_parser( diff --git a/packages/flow/openadapt_flow/compiler/induction.py b/packages/flow/openadapt_flow/compiler/induction.py index a357cc5fb..dc6f30065 100644 --- a/packages/flow/openadapt_flow/compiler/induction.py +++ b/packages/flow/openadapt_flow/compiler/induction.py @@ -252,6 +252,53 @@ class ColumnDecision(BaseModel): note: str = "" +class ProbeRun(BaseModel): + """One replay of one interpretation in one recording's context. + + Written for every run ``induce --probe-on`` makes on the test copy, + whether or not it decided anything (:mod:`openadapt_flow.compiler.probing`). + """ + + location: str = Field(description="The underdetermined point probed") + interpretation: str = Field(description="'incidental' | 'conditional'") + context: str = Field(description="Label of the recording whose context ran") + runtime_signal: str = Field( + description="'completed' | 'halted_before_effect' | 'refused' | " + "'failed_platform', from the replay" + ) + outcome: str = Field( + description="The serve-reward judge's outcome for the read-back" + ) + produced_recorded_effect: bool = Field( + description="True only when the judge says VERIFIED." + ) + conclusive: bool = Field( + description="False when the read-back could not decide (store " + "unreachable, effect uncertain, platform failure)." + ) + reason: str = "" + model_calls: int = 0 + + +class ProbeVerdict(BaseModel): + """What probing decided about one underdetermined point.""" + + location: str + kind: str = Field(description="The uncertainty kind that was probed") + interpretations: dict[str, str] = Field( + default_factory=dict, + description="interpretation -> 'passed' | 'failed' | 'inconclusive' " + "| 'not_run'", + ) + certified: Optional[str] = Field( + default=None, + description="The one interpretation that produced the recorded effect " + "in every context while every other one conclusively failed; None " + "when the point stays refused.", + ) + detail: str = "" + + class InductionResult(BaseModel): """The induced program (or an honest refusal) plus the full audit trail.""" @@ -280,6 +327,18 @@ class InductionResult(BaseModel): "uncovered-branch notes on an emitted program (non-blocking)." ), ) + probe_target: Optional[str] = Field( + default=None, + description="Name of the declared test copy probed (None: no probing).", + ) + probe_runs: list[ProbeRun] = Field( + default_factory=list, + description="Every probe run, in the order it ran.", + ) + probe_verdicts: list[ProbeVerdict] = Field( + default_factory=list, + description="One verdict per underdetermined point probing looked at.", + ) @property def underdetermined(self) -> bool: @@ -303,6 +362,24 @@ def render(self) -> str: lines.append(f" [proposed] {p.target}: {p.content} (NOT trusted)") for u in self.uncertainties: lines.append(f" [REFUSED] {u.location}: {u.detail}") + if self.probe_target is not None: + lines.append( + f"probes on test copy {self.probe_target!r}: " + f"{len(self.probe_runs)} run(s)" + ) + for run in self.probe_runs: + lines.append( + f" [probe] {run.location} {run.interpretation} in " + f"{run.context}: {run.outcome} ({run.runtime_signal})" + + (f" -- {run.reason}" if run.reason else "") + ) + for pv in self.probe_verdicts: + shown = ", ".join(f"{k}={v}" for k, v in pv.interpretations.items()) + head = f"certified {pv.certified}" if pv.certified else "still refused" + lines.append( + f" [verdict] {pv.location}: {head} [{shown}]" + + (f" -- {pv.detail}" if pv.detail else "") + ) verdict = "CERTIFIED" if self.certified else "NOT CERTIFIED" lines.append( f"Certification: {verdict}" @@ -894,7 +971,10 @@ def set_next(self, next_id: str) -> None: def induce_program( - traces: list[TraceInput], *, propose: Optional[Proposer] = None + traces: list[TraceInput], + *, + propose: Optional[Proposer] = None, + probe_resolutions: Optional[dict[str, str]] = None, ) -> InductionResult: """Induce a parameterized :class:`ProgramGraph` from multiple demonstrations of the same task -- the RFC ยง3 induction loop (step [4] + [5]). @@ -909,6 +989,12 @@ def induce_program( 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). + probe_resolutions: Optional ``{location: interpretation}`` for + consequential optional points that probing on a declared test copy + settled (:func:`openadapt_flow.compiler.probing.probe_and_induce` + is the only caller that sets it). ``"incidental"`` drops the step; + ``"conditional"`` keeps it guarded on its own presence. A point + not listed here is refused exactly as without probing. Returns: An :class:`InductionResult` with the induced program (or a refusal) and @@ -981,8 +1067,11 @@ def induce_program( continue # --- single step present in SOME traces: branch or optional ------- - node, dec = _emit_optional_single(idx, col, present, n, result, propose) - nodes.append(node) + node, dec = _emit_optional_single( + idx, col, present, n, result, propose, probe_resolutions or {} + ) + if node is not None: + nodes.append(node) decisions.append(dec) result.column_decisions = decisions @@ -1270,11 +1359,132 @@ def _emit_required_single(idx, col, present, taken, param_specs, result, propose return node, dec -def _emit_optional_single(idx, col, present, n, result, propose): +def optional_guard(step: Step) -> Guard: + """The skip-when-absent guard induction puts on an optional step. + + A dialog step runs when its dialog text is on screen; any other step runs + when its own target resolves (or, with no anchor, its field text shows). + Probing replays exactly this guard for the ``conditional`` interpretation. + """ + field = _field_key(step) + if _looks_like_dialog(step): + predicate = Predicate( + kind=PredicateKind.TEXT_PRESENT, + text=_dialog_text(step), + intent="optional dialog is present", + ) + elif step.anchor is not None: + predicate = Predicate( + kind=PredicateKind.ANCHOR_RESOLVES, + anchor=step.anchor, + intent=f"optional step '{field}' target present", + ) + else: + predicate = Predicate( + kind=PredicateKind.TEXT_PRESENT, + text=field, + intent=f"optional step '{field}' present", + ) + return Guard(predicate=predicate, on_unmet="skip") + + +class OptionalPoint(BaseModel): + """A consequential step some traces have and others lack. + + These are the points induction refuses as ``unconfirmed_optional`` or + ``unconfirmed_branch``, located well enough to replay each recording with + the step dropped or kept (see :mod:`openadapt_flow.compiler.probing`). + """ + + model_config = {"arbitrary_types_allowed": True} + + location: str + kind: str + column: int + source_trace: int = Field(description="First trace that has the step") + step: Step + #: trace -> index of the step in that trace's ``steps`` (present traces). + index_in_trace: dict[int, int] = Field(default_factory=dict) + #: trace -> index the step would be inserted at (absent traces). + insert_at: dict[int, int] = Field(default_factory=dict) + + +def optional_points(traces: list[TraceInput]) -> dict[str, OptionalPoint]: + """Locate every consequential optional point, keyed by its location id. + + Uses the same reduction, alignment and location ids as + :func:`induce_program`, so a key here matches an + :attr:`Uncertainty.location` there. + """ + workflows = [_as_workflow(t) for t in traces] + n = len(workflows) + columns = _align([_reduce_trace(wf.steps) for wf in workflows]) + positions = [{id(step): k for k, step in enumerate(wf.steps)} for wf in workflows] + + def last_index(tok: Any, t: int) -> int: + if tok.kind == "loop": + return positions[t][id(tok.iterations[-1][-1])] + return positions[t][id(tok.step)] + + points: dict[str, OptionalPoint] = {} + for idx, col in enumerate(columns): + present = sorted(col.tokens) + if ( + col.divergent_group is not None + or col.kind != "single" + or not 0 < len(present) < n + ): + continue + step0 = col.tokens[present[0]].step + if not _step_consequential(step0): + continue + is_dialog = _looks_like_dialog(step0) + point = OptionalPoint( + location=f"branch_{idx}" if is_dialog else f"opt_{idx}", + kind="unconfirmed_branch" if is_dialog else "unconfirmed_optional", + column=idx, + source_trace=present[0], + step=step0, + ) + for t in range(n): + if t in col.tokens: + point.index_in_trace[t] = positions[t][id(col.tokens[t].step)] + continue + before = 0 + for prev in reversed(columns[:idx]): + if t in prev.tokens: + before = last_index(prev.tokens[t], t) + 1 + break + point.insert_at[t] = before + points[point.location] = point + return points + + +def _emit_optional_single(idx, col, present, n, result, propose, resolutions=None): tok = col.tokens[present[0]] step0 = tok.step field = _field_key(step0) is_dialog = _looks_like_dialog(step0) + location = f"branch_{idx}" if is_dialog else f"opt_{idx}" + resolved = (resolutions or {}).get(location) + if resolved is not None and not _step_consequential(step0): + # Probing only ever settles a point induction refused; a step that + # was never refused keeps its ordinary interpretation. + resolved = None + if resolved == "incidental": + result.inferred.append( + f"step '{field}' at column {idx} (present in {present}/{n} " + "traces) is incidental -- dropped; certified by probing on a " + "declared test copy" + ) + return None, ColumnDecision( + index=idx, + kind="incidental", + align_sig=str(col.align_sig), + field=field, + present_in=present, + note=f"incidental step '{field}' dropped (probe-certified)", + ) if is_dialog: # A DETECTABLE condition (the dialog's own presence) -> a guarded BRANCH @@ -1326,7 +1536,12 @@ def _emit_optional_single(idx, col, present, n, result, propose): # evidence -- running (or skipping) an irreversible action on a GUESSED # condition is exactly the failure class to avoid -- so ALSO emit an # Uncertainty requiring operator confirmation (certified stays False). - if _step_consequential(do_step): + if resolved == "conditional": + result.inferred.append( + f"guard TEXT_PRESENT({dialog_text!r}) at {branch_id} certified " + "by probing on a declared test copy" + ) + elif _step_consequential(do_step): _add_uncertainty( result, kind="unconfirmed_branch", @@ -1372,20 +1587,7 @@ def _emit_optional_single(idx, col, present, n, result, propose): 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") + step.guard = optional_guard(step0) node = _Node(sid) node.states[sid] = State( id=sid, kind=StateKind.ACTION, step=step, transitions=[exit_tr] @@ -1398,7 +1600,13 @@ def _emit_optional_single(idx, col, present, n, result, propose): # not cover the case is UNDERDETERMINED, and auto-skipping an irreversible # action on an unconfirmed guard is not a safe default. So it becomes a # QUESTION (Uncertainty), not a silent skip -- certified stays False. - if _step_consequential(step0): + if resolved == "conditional": + result.inferred.append( + f"optional step '{field}' at column {idx} (present in " + f"{present}/{n} traces) -- run when present, skip when absent; " + "certified by probing on a declared test copy" + ) + elif _step_consequential(step0): _add_uncertainty( result, kind="unconfirmed_optional", diff --git a/packages/flow/openadapt_flow/compiler/probing.py b/packages/flow/openadapt_flow/compiler/probing.py new file mode 100644 index 000000000..bcef2d9fd --- /dev/null +++ b/packages/flow/openadapt_flow/compiler/probing.py @@ -0,0 +1,628 @@ +"""Settle a recording disagreement by replaying it on a declared test copy. + +:func:`~openadapt_flow.compiler.induction.induce_program` refuses when one +recording has a consequential step another lacks: from the recordings alone it +cannot tell an incidental extra click from a required conditional branch. It +names the recording that would settle the question (a record-next item). + +This module tries the other route, opt-in and only on a resettable test copy. +For each such point it replays each recording twice in a fresh session, once +per interpretation: + +* ``incidental``: the step is noise. Every recording runs without it. +* ``conditional``: the step is a required branch. Every recording runs it, + guarded to skip when its target (or its dialog) is not on screen. + +Each run is scored by reading the system of record back through the reward +contract's oracle and judging it with +:func:`openadapt_flow.reward.worker.judge_episode`, the function +``serve-reward`` scores with. A run "produces the recorded effect" only when +that judge says ``verified`` for the recording's own subject. A point is +certified only when exactly one interpretation produces the recorded effect in +every recording's context and every other interpretation conclusively fails +in at least one. Anything else (both pass, neither passes, a read-back that +cannot decide) leaves the point refused, as without probing. + +What this module will not do: + +* Probe a target whose deployment file does not declare it a test copy + (:func:`probe_target_refusals`), or whose runtime profile is a production one. +* Let a :class:`~openadapt_flow.compiler.induction.Proposer` choose. Proposals + stay advisory in the result; probing never reads them. +* Call a model. Probe replays run with model grounding off, and each run's + model-call count is written to the report. + +Every run is written to :attr:`InductionResult.probe_runs`, decisive or not. +""" + +from __future__ import annotations + +import shutil +import subprocess +import urllib.request +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional, Protocol, Union, runtime_checkable + +from openadapt_flow.compiler.induction import ( + InductionResult, + OptionalPoint, + ProbeRun, + ProbeVerdict, + Proposer, + TraceInput, + induce_program, + optional_guard, + optional_points, +) +from openadapt_flow.ir import Step, Workflow + +INTERPRETATIONS: tuple[str, ...] = ("incidental", "conditional") + +#: Runtime signals the serve-reward judge accepts. +RUNTIME_SIGNALS = frozenset( + { + "completed", + "halted_before_effect", + "refused", + "rejected_policy", + "failed_platform", + } +) + +#: Judge outcomes that do not decide anything about the interpretation. +_INCONCLUSIVE_OUTCOMES = frozenset({"failed_platform", "reconciliation_required"}) + +#: Runtime profiles a production deployment selects (deployment.RuntimeSection). +_PRODUCTION_PROFILES = frozenset({"standard", "regulated"}) + + +class ProbeRefused(Exception): + """The target may not be probed. Nothing ran.""" + + +@dataclass +class ProbeTrace: + """One recording, as probing needs it. + + ``bundle_dir`` holds the files the workflow's anchors reference + (template crops). ``identity`` is the recording's subject, keyed by the + reward contract's identity keys; when None it is read from the + workflow's ``params``. + """ + + workflow: Workflow + bundle_dir: Path + label: str + identity: Optional[dict[str, str]] = None + + +@dataclass +class RunSignal: + """What a probe replay reports back, before the read-back judges it.""" + + signal: str + model_calls: int = 0 + detail: str = "" + + +@runtime_checkable +class ProbeTarget(Protocol): + """A resettable test copy that can replay a workflow in a fresh session.""" + + name: str + + def reset(self) -> None: + """Restore the copy to its baseline. Raise if that fails.""" + ... + + def run( + self, + workflow: Workflow, + *, + bundle_dir: Path, + run_dir: Path, + params: dict[str, str], + ) -> RunSignal: + """Replay ``workflow`` in a new session; never reuse one.""" + ... + + +# --------------------------------------------------------------------------- +# The test-copy gate +# --------------------------------------------------------------------------- + + +def probe_target_refusals(cfg: Any) -> list[str]: + """Why ``cfg`` (a DeploymentConfig) may not be probed; empty when it may.""" + + reasons: list[str] = [] + section = getattr(cfg, "test_copy", None) + if section is None or section.declared is not True: + reasons.append( + "the deployment file does not declare its target a test copy " + "(test_copy.declared: true)" + ) + return reasons + if not section.reset_url and not section.reset_command: + reasons.append( + "test_copy names no reset (reset_url or reset_command); a probe " + "needs a copy it can restore before every run" + ) + if not section.reward_contract: + reasons.append( + "test_copy names no reward_contract; a probe is scored by its " + "read-back and nothing else" + ) + profile = getattr(cfg.runtime, "profile", None) + if profile in _PRODUCTION_PROFILES: + reasons.append( + f"runtime.profile is {profile!r}, a production profile; a test " + "copy runs without one or with 'demo'" + ) + backend = cfg.backend + if backend.browser_cdp_endpoint or getattr(backend, "browser_page_url", None): + reasons.append( + "backend attaches to an already-open browser; each probe run needs " + "a fresh session" + ) + return reasons + + +def require_test_copy(cfg: Any) -> None: + reasons = probe_target_refusals(cfg) + if reasons: + raise ProbeRefused("; ".join(reasons)) + + +# --------------------------------------------------------------------------- +# Interpretations -> concrete per-recording workflows +# --------------------------------------------------------------------------- + + +def _guarded_copy(step: Step) -> Step: + copy = step.model_copy(deep=True) + copy.guard = optional_guard(step) + return copy + + +def interpretation_workflow( + traces: list[ProbeTrace], point: OptionalPoint, t: int, interpretation: str +) -> tuple[Workflow, Optional[int]]: + """Recording ``t`` replayed under ``interpretation`` of ``point``. + + Every other step is the recording's own, so every other point keeps what + that recording did. Returns the workflow and, when the step was copied in + from another recording, its index (its anchor's files live in that + recording's bundle). + """ + + wf = traces[t].workflow + steps = [s.model_copy(deep=True) for s in wf.steps] + inserted: Optional[int] = None + if interpretation == "incidental": + if t in point.index_in_trace: + del steps[point.index_in_trace[t]] + elif interpretation == "conditional": + if t in point.index_in_trace: + k = point.index_in_trace[t] + steps[k] = _guarded_copy(wf.steps[k]) + else: + inserted = point.insert_at[t] + steps.insert(inserted, _guarded_copy(point.step)) + else: + raise ValueError(f"unknown interpretation {interpretation!r}") + out = wf.model_copy(update={"steps": steps, "program": None}) + return out, inserted + + +def _stage_bundle( + traces: list[ProbeTrace], + point: OptionalPoint, + t: int, + workflow: Workflow, + inserted: Optional[int], + dest: Path, +) -> Workflow: + """Copy recording ``t``'s bundle to ``dest`` and bring in the files an + inserted step's anchor needs from the recording it came from.""" + + if dest.exists(): + shutil.rmtree(dest) + src = traces[t].bundle_dir + if src.is_dir(): + shutil.copytree(src, dest) + else: + dest.mkdir(parents=True) + if inserted is None: + return workflow + origin = traces[point.source_trace].bundle_dir + step = workflow.steps[inserted] + prefix = f"probe-from-{point.source_trace}" + for anchor in (step.anchor, getattr(step, "drag_end_anchor", None)): + if anchor is None or not anchor.template: + continue + source = origin / anchor.template + if not source.is_file(): + continue + target = dest / prefix / anchor.template + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, target) + anchor.template = f"{prefix}/{anchor.template}" + return workflow + + +# --------------------------------------------------------------------------- +# Scoring: the serve-reward judge over the contract's own oracle +# --------------------------------------------------------------------------- + + +@dataclass +class _Scorer: + bundle: Any # openadapt_flow.reward.models.RewardBundle + oracle: Any = field(init=False) + + def __post_init__(self) -> None: + from openadapt_flow.reward.oracles import build_oracle + + self.oracle = build_oracle(self.bundle.oracle, self.bundle.directory) + + @property + def substrate(self) -> str: + return self.bundle.oracle.channel.value + + def baseline(self, identity: dict[str, str]) -> Any: + from openadapt_flow.reward.oracles import effect_state_of + + return effect_state_of(self.oracle.read(identity), self.substrate) + + def judge(self, identity: dict[str, str], signal: str, before: Any) -> Any: + from openadapt_flow.reward.worker import judge_episode + + observed = self.oracle.read(identity) + return judge_episode(self.bundle, identity, signal, before, observed) + + +def _identity(trace: ProbeTrace, keys: tuple[str, ...]) -> dict[str, str]: + if trace.identity is not None: + return {k: str(v) for k, v in trace.identity.items()} + params = trace.workflow.params or {} + missing = [k for k in keys if params.get(k) in (None, "")] + if missing: + raise ValueError( + f"recording {trace.label} has no value for the reward contract's " + f"identity key(s) {', '.join(missing)} in its params" + ) + return {k: str(params[k]) for k in keys} + + +def _params(trace: ProbeTrace) -> dict[str, str]: + return {k: str(v) for k, v in (trace.workflow.params or {}).items()} + + +def _probe_once( + *, + target: ProbeTarget, + scorer: _Scorer, + traces: list[ProbeTrace], + point: OptionalPoint, + t: int, + interpretation: str, + identity: dict[str, str], + work_dir: Path, +) -> ProbeRun: + slug = f"{point.location}-{interpretation}-{t}" + workflow, inserted = interpretation_workflow(traces, point, t, interpretation) + base: dict[str, Any] = { + "location": point.location, + "interpretation": interpretation, + "context": traces[t].label, + } + try: + workflow = _stage_bundle( + traces, point, t, workflow, inserted, work_dir / "bundles" / slug + ) + target.reset() + before = scorer.baseline(identity) + except Exception as exc: # noqa: BLE001 - logged, and the run decides nothing + return ProbeRun( + **base, + runtime_signal="failed_platform", + outcome="failed_platform", + produced_recorded_effect=False, + conclusive=False, + reason=f"could not prepare the run: {exc}", + ) + try: + signal = target.run( + workflow, + bundle_dir=work_dir / "bundles" / slug, + run_dir=work_dir / "runs" / slug, + params=_params(traces[t]), + ) + except Exception as exc: # noqa: BLE001 - a crashed replay is a platform failure + signal = RunSignal("failed_platform", detail=f"replay raised: {exc}") + if signal.signal not in RUNTIME_SIGNALS: + signal = RunSignal( + "failed_platform", + signal.model_calls, + f"unknown runtime signal {signal.signal!r}", + ) + judged = scorer.judge(identity, signal.signal, before) + outcome = judged.outcome.value + reason = judged.reason + if signal.detail: + reason = f"{reason}; {signal.detail}" if reason else signal.detail + return ProbeRun( + **base, + runtime_signal=signal.signal, + outcome=outcome, + produced_recorded_effect=outcome == "verified", + conclusive=outcome not in _INCONCLUSIVE_OUTCOMES, + reason=reason, + model_calls=signal.model_calls, + ) + + +def _probe_point( + *, + target: ProbeTarget, + scorer: _Scorer, + traces: list[ProbeTrace], + identities: list[dict[str, str]], + point: OptionalPoint, + work_dir: Path, + runs: list[ProbeRun], +) -> ProbeVerdict: + status: dict[str, str] = {} + for interpretation in INTERPRETATIONS: + state = "passed" + for t in range(len(traces)): + run = _probe_once( + target=target, + scorer=scorer, + traces=traces, + point=point, + t=t, + interpretation=interpretation, + identity=identities[t], + work_dir=work_dir, + ) + runs.append(run) + if run.produced_recorded_effect: + continue + if run.conclusive: + state = "failed" + break # one conclusive miss rules this interpretation out + state = "inconclusive" # keep looking for a conclusive miss + status[interpretation] = state + passed = [k for k, v in status.items() if v == "passed"] + others_failed = all(v == "failed" for k, v in status.items() if k not in passed) + if len(passed) == 1 and others_failed: + return ProbeVerdict( + location=point.location, + kind=point.kind, + interpretations=status, + certified=passed[0], + detail=( + f"only {passed[0]} produced the recorded effect in all " + f"{len(traces)} recording contexts" + ), + ) + if len(passed) > 1: + detail = "more than one interpretation produced the recorded effect" + elif not passed and all(v == "failed" for v in status.values()): + detail = "no interpretation produced the recorded effect" + else: + detail = "a read-back could not decide" + return ProbeVerdict( + location=point.location, + kind=point.kind, + interpretations=status, + detail=detail + "; the point stays refused", + ) + + +def probe_and_induce( + traces: list[ProbeTrace], + *, + target: ProbeTarget, + reward_contract: Union[Any, str, Path], + work_dir: Path, + propose: Optional[Proposer] = None, +) -> InductionResult: + """Induce, then probe each refused optional point on ``target``. + + ``target`` must already have passed :func:`require_test_copy` (the CLI + does this from the deployment file). ``reward_contract`` is a + ``serve-reward`` contract bundle (a directory or a loaded + :class:`~openadapt_flow.reward.models.RewardBundle`). + + The returned result carries every probe run and one verdict per refused + point. Points probing did not settle stay refused, with their record-next + items, exactly as plain induction leaves them. + """ + + from openadapt_flow.reward.models import RewardBundle + + bundle = ( + reward_contract + if isinstance(reward_contract, RewardBundle) + else RewardBundle.load(reward_contract) + ) + workflows: list[TraceInput] = [tr.workflow for tr in traces] + # The proposer is advisory and never decides a probe, so the first pass + # runs without it; it is consulted once, in the pass whose result returns. + first = induce_program(workflows) + name = str(getattr(target, "name", "test copy")) + if not first.uncertainties: + result = induce_program(workflows, propose=propose) if propose else first + result.probe_target = name + return result + + runs: list[ProbeRun] = [] + verdicts: list[ProbeVerdict] = [] + resolutions: dict[str, str] = {} + try: + identities = [_identity(tr, bundle.identity_keys) for tr in traces] + for identity in identities: + bundle.check_identity(identity) + except ValueError as exc: + verdicts = [ + ProbeVerdict( + location=u.location, + kind=u.kind, + detail=f"not probed: {exc}; the point stays refused", + ) + for u in first.uncertainties + ] + identities = [] + + if identities: + scorer = _Scorer(bundle) + points = optional_points(workflows) + for u in first.uncertainties: + point = points.get(u.location) + if point is None or point.kind != u.kind: + verdicts.append( + ProbeVerdict( + location=u.location, + kind=u.kind, + detail=( + f"not probed: {u.kind} has no replayable " + "interpretations; the point stays refused" + ), + ) + ) + continue + verdict = _probe_point( + target=target, + scorer=scorer, + traces=traces, + identities=identities, + point=point, + work_dir=Path(work_dir), + runs=runs, + ) + verdicts.append(verdict) + if verdict.certified: + resolutions[point.location] = verdict.certified + try: + target.reset() # leave the copy at its baseline + except Exception: # noqa: BLE001 - every run already reset first + pass + + result = induce_program( + workflows, propose=propose, probe_resolutions=resolutions or None + ) + result.probe_target = name + result.probe_runs = runs + result.probe_verdicts = verdicts + return result + + +# --------------------------------------------------------------------------- +# The deployment-file target +# --------------------------------------------------------------------------- + + +def runtime_signal_of(report: Any) -> str: + """Map a replay report onto the serve-reward runtime signal.""" + + if getattr(report, "success", False): + return "completed" + return "halted_before_effect" + + +class DeploymentProbeTarget: + """The test copy a deployment file describes. + + Reset: ``test_copy.reset_url`` (HTTP POST) or ``test_copy.reset_command`` + (argv, no shell). Run: a new browser (``web``) or a new backend for every + run, wired by :func:`openadapt_flow.deployment.build_replayer` with model + grounding and egress off and no effect verifier (the probe's read-back + scores the run). + """ + + def __init__(self, cfg: Any, *, reset_timeout_s: float = 30.0) -> None: + require_test_copy(cfg) + self.cfg = cfg + self.name = str(cfg.name) + self.reset_timeout_s = reset_timeout_s + + def reset(self) -> None: + section = self.cfg.test_copy + if section.reset_url: + request = urllib.request.Request( + section.reset_url, + data=b"{}", + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen( # noqa: S310 - operator-declared URL + request, timeout=self.reset_timeout_s + ) as response: + if not 200 <= response.status < 300: + raise RuntimeError(f"reset returned HTTP {response.status}") + if section.reset_command: + subprocess.run( # noqa: S603 - argv list, no shell + list(section.reset_command), + check=True, + timeout=self.reset_timeout_s, + capture_output=True, + ) + + def _replay(self, backend: Any, workflow, bundle_dir, run_dir, params) -> RunSignal: + from openadapt_flow.deployment import build_replayer + + replayer = build_replayer( + backend, + allow_egress=False, + effect_verifier=None, + api_actuator=None, + durable=False, + use_structural=True, + ) + report = replayer.run( + workflow, bundle_dir=bundle_dir, run_dir=run_dir, params=params + ) + return RunSignal( + runtime_signal_of(report), + model_calls=int(getattr(report, "model_calls", 0) or 0), + ) + + def run( + self, + workflow: Workflow, + *, + bundle_dir: Path, + run_dir: Path, + params: dict[str, str], + ) -> RunSignal: + from openadapt_flow.backends.factory import build_backend + + backend_cfg = self.cfg.backend + if backend_cfg.kind != "web": + backend = build_backend(backend_cfg) + try: + return self._replay(backend, workflow, bundle_dir, run_dir, params) + finally: + close = getattr(backend, "close", None) + if callable(close): + close() + if not backend_cfg.url: + raise ValueError("a web test copy needs backend.url") + from playwright.sync_api import sync_playwright + + from openadapt_flow._browser_setup import launch_chromium + + with sync_playwright() as p: + browser = launch_chromium(p.chromium, headless=not backend_cfg.headed) + try: + context = browser.new_context(viewport={"width": 1280, "height": 800}) + page = context.new_page() + page.goto(backend_cfg.url) + backend = build_backend(backend_cfg, page=page) + return self._replay(backend, workflow, bundle_dir, run_dir, params) + finally: + browser.close() diff --git a/packages/flow/openadapt_flow/deployment.py b/packages/flow/openadapt_flow/deployment.py index 002780421..42c42a484 100644 --- a/packages/flow/openadapt_flow/deployment.py +++ b/packages/flow/openadapt_flow/deployment.py @@ -29,6 +29,7 @@ from pydantic import ( BaseModel, + ConfigDict, Field, field_validator, model_serializer, @@ -831,6 +832,34 @@ class IdempotencySection(BaseModel): key_env: Optional[str] = None +class ProbeCopySection(BaseModel): + """Declares that this deployment drives a resettable test copy. + + Only ``induce --probe-on`` reads it. Probing replays guessed + interpretations of a recording disagreement, some of which are wrong on + purpose, so it runs only where the operator has said in writing that the + target is a disposable copy. See docs/PROBE_INDUCTION.md. + """ + + model_config = ConfigDict(extra="forbid") + + #: The operator's declaration that every system this file names (the GUI, + #: the read-back store, the reset hook) is a test copy, not production. + #: Probing refuses unless this is exactly ``true``. + declared: bool = False + #: Restores the copy to its baseline before each probe run: an HTTP POST + #: to this URL. One of ``reset_url`` / ``reset_command`` is required. + reset_url: Optional[str] = None + #: Restores the copy to its baseline before each probe run: a command run + #: without a shell, as an argv list. + reset_command: list[str] = Field(default_factory=list) + #: Reward contract bundle directory, in the ``serve-reward`` format. Its + #: required and forbidden effects say what "the recorded effect" is, and + #: its oracle recipe is the read-back. A relative path resolves against + #: the deployment file's directory. + reward_contract: Optional[str] = None + + #: Sentinel written by ``config init`` for fields the operator must fill. UNRESOLVED = "UNRESOLVED" @@ -859,6 +888,9 @@ class DeploymentConfig(BaseModel): #: declares. ``--surface NAME=...`` flags apply over these entries. Empty #: (and omitted when serialized) otherwise. surfaces: dict[str, BackendConfig] = Field(default_factory=dict) + #: Declares the target a resettable test copy (``induce --probe-on`` + #: only). None (and omitted when serialized) otherwise. + test_copy: Optional[ProbeCopySection] = None @model_serializer(mode="wrap") def _serialize_compatible(self, handler: Any) -> dict[str, Any]: @@ -867,6 +899,8 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: data: dict[str, Any] = handler(self) if not self.surfaces: data.pop("surfaces", None) + if self.test_copy is None: + data.pop("test_copy", None) return data diff --git a/packages/flow/tests/test_probe_induction.py b/packages/flow/tests/test_probe_induction.py new file mode 100644 index 000000000..40939eb2f --- /dev/null +++ b/packages/flow/tests/test_probe_induction.py @@ -0,0 +1,603 @@ +"""``induce --probe-on``: settle a recording disagreement on a test copy. + +The test copy is MockMed's transactional server +(:mod:`openadapt_flow.mockmed.fault_server`): a real HTTP store that +``POST /api/reset`` restores and ``GET /api/db`` reads. Probe runs are scored +by a ``serve-reward`` contract whose oracle is the ``rest`` recipe over that +store, so a run counts only when the judge ``serve-reward`` uses reads the +expected record back. + +The screen side is a scripted MockMed session (a fresh one per run) driven by +the real :class:`~openadapt_flow.runtime.replayer.Replayer`. Its buttons write +through the same HTTP API the MockMed app uses under ``?fault=ok``: + +* ``p1`` opens behind an allergy alert. Until someone presses Accept, nothing + under it can be clicked. +* ``p2``'s chart is already signed, so its Sign chart button does nothing. +* ``p3`` is a plain chart. Sign chart writes a Signature record. + +All data is synthetic. +""" + +from __future__ import annotations + +import json +import urllib.request +from pathlib import Path +from typing import Iterator + +import pytest + +from openadapt_flow.compiler.induction import induce_program +from openadapt_flow.compiler.probing import ( + ProbeRefused, + ProbeTrace, + RunSignal, + probe_and_induce, + probe_target_refusals, + require_test_copy, + runtime_signal_of, +) +from openadapt_flow.deployment import DeploymentConfig +from openadapt_flow.ir import ActionKind, Anchor, StateKind, Step, Workflow +from openadapt_flow.mockmed.fault_server import serve as serve_fault +from openadapt_flow.reward.seed import write_bundle +from openadapt_flow.runtime.replayer import Replayer +from tests.test_replayer import FakeBackend, FakeVision, Match, make_png + +NOTE = "Synthetic triage note for probe tests" + +#: Button label -> (template colour, click point). Each template is a distinct +#: PNG, which is how the scripted vision tells the buttons apart. +BUTTONS: dict[str, tuple[tuple[int, int, int], tuple[int, int]]] = { + "Accept": ((200, 40, 40), (150, 60)), + "Sign chart": ((40, 200, 40), (60, 150)), + "Save Encounter": ((40, 40, 200), (150, 150)), + "Print summary": ((200, 200, 40), (240, 150)), +} +ALLERGY_ALERT = {"p1"} +ALREADY_SIGNED = {"p2"} + + +def _template(label: str) -> bytes: + return make_png((40, 16), BUTTONS[label][0]) + + +# --------------------------------------------------------------------------- +# The scripted MockMed session (screen side of the test copy) +# --------------------------------------------------------------------------- + + +class MockMedSession: + """One fresh MockMed session. Writes go to the test copy over HTTP.""" + + def __init__(self, base_url: str) -> None: + self.base_url = base_url + self.patient: str | None = None + self.alert = False + self.note = "" + + def visible(self) -> set[str]: + if self.patient is None: + return set() + if self.alert: + return {"Accept", "Allergy alert"} + return {"Sign chart", "Save Encounter", "Print summary"} + + def type(self, text: str) -> bool: + if self.patient is None: + self.patient = text + self.alert = text in ALLERGY_ALERT + return True + if self.alert: + return False # the modal swallows the keystrokes + self.note += text + return True + + def click(self, label: str) -> None: + if label not in self.visible(): + return + if label == "Accept": + self.alert = False + elif label == "Save Encounter": + self._write("Triage", self.note) + elif label == "Sign chart" and self.patient not in ALREADY_SIGNED: + self._write("Signature", "") + + def _write(self, kind: str, note: str) -> None: + body = json.dumps( + {"patient_id": self.patient, "type": kind, "note": note} + ).encode() + request = urllib.request.Request( + self.base_url + "api/encounter?fault=ok", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + urllib.request.urlopen(request, timeout=5).close() + + +class MockMedBackend(FakeBackend): + def __init__(self, session: MockMedSession) -> None: + super().__init__() + self.session = session + + def type_text(self, text): + landed = self.session.type(text) + self._type_accept_results = [landed] + super().type_text(text) + + def click(self, x, y, *, double=False): + super().click(x, y, double=double) + for label, (_colour, point) in BUTTONS.items(): + if abs(point[0] - x) <= 12 and abs(point[1] - y) <= 12: + self.session.click(label) + return + + +class MockMedVision(FakeVision): + def __init__(self, session: MockMedSession) -> None: + super().__init__() + self.session = session + self._by_png = {_template(label): label for label in BUTTONS} + + def _match(self, label: str): + if label not in self.session.visible(): + return None + point = BUTTONS[label][1] if label in BUTTONS else (10, 10) + return Match(point, (point[0] - 20, point[1] - 8, 40, 16)) + + def find_template(self, screen_png, template_png, **_kw): + label = self._by_png.get(template_png) + return self._match(label) if label else None + + def find_text(self, screen_png, text, **_kw): + return self._match(text) + + def text_present(self, screen_png, text, **_kw): + return text in self.session.visible() + + +class MockMedTestCopy: + """A ``ProbeTarget`` over the MockMed transactional server.""" + + def __init__(self, base_url: str, *, reset: bool = True) -> None: + self.name = "mockmed-test-copy" + self.base_url = base_url + self.resets = 0 + self.sessions = 0 + self._reset = reset + + def reset(self) -> None: + self.resets += 1 + if not self._reset: + return + request = urllib.request.Request( + self.base_url + "api/reset", data=b"{}", method="POST" + ) + urllib.request.urlopen(request, timeout=5).close() + + def run(self, workflow, *, bundle_dir, run_dir, params) -> RunSignal: + self.sessions += 1 + session = MockMedSession(self.base_url) # a fresh session every run + report = Replayer( + MockMedBackend(session), + vision=MockMedVision(session), + poll_interval_s=0.01, + ).run(workflow, bundle_dir=bundle_dir, run_dir=run_dir, params=params) + return RunSignal(runtime_signal_of(report), model_calls=report.model_calls) + + +# --------------------------------------------------------------------------- +# Recordings +# --------------------------------------------------------------------------- + + +def _type(step_id: str, field: str, value: str) -> Step: + return Step(id=step_id, intent=f"type {field}", action=ActionKind.TYPE, text=value) + + +def _click(step_id: str, label: str, *, intent: str | None = None) -> Step: + point = BUTTONS[label][1] + return Step( + id=step_id, + intent=intent or f"click {label}", + action=ActionKind.CLICK, + risk="irreversible", + anchor=Anchor( + template=f"templates/{label.replace(' ', '_').lower()}.png", + region=(point[0] - 20, point[1] - 8, 40, 16), + click_point=point, + ocr_text=label, + ), + ) + + +def _recording(tmp: Path, label: str, patient: str, extra: list[Step], at: int): + steps = [ + _type("s_patient", "patient", patient), + _type("s_note", "note", NOTE), + _click("s_save", "Save Encounter"), + ] + steps[at:at] = extra + workflow = Workflow(name="chart-note", steps=steps, params={"patient_id": patient}) + bundle = tmp / label + (bundle / "templates").mkdir(parents=True) + for step in steps: + if step.anchor is not None: + (bundle / step.anchor.template).write_bytes(_template(step.anchor.ocr_text)) + return ProbeTrace(workflow=workflow, bundle_dir=bundle, label=label) + + +def incidental_traces(tmp: Path) -> list[ProbeTrace]: + """A clicked Sign chart on p2, where it was already signed (a no-op).""" + return [ + _recording(tmp, "rec-a", "p2", [_click("s_sign", "Sign chart")], at=2), + _recording(tmp, "rec-b", "p3", [], at=2), + ] + + +def branch_traces(tmp: Path) -> list[ProbeTrace]: + """A accepted p1's allergy alert; B's patient had none.""" + accept = _click("s_accept", "Accept", intent="accept the allergy alert") + return [ + _recording(tmp, "rec-a", "p1", [accept], at=1), + _recording(tmp, "rec-b", "p3", [], at=1), + ] + + +def undecidable_traces(tmp: Path) -> list[ProbeTrace]: + """A printed a summary. Printing writes nothing, so both readings hold.""" + return [ + _recording(tmp, "rec-a", "p2", [_click("s_print", "Print summary")], at=2), + _recording(tmp, "rec-b", "p3", [], at=2), + ] + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def mockmed() -> Iterator[tuple[str, object]]: + url, db, stop = serve_fault(port=0) + yield url, db + stop() + + +def _contract(directory: Path, base_url: str) -> Path: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + write_bundle( + directory, + contract_id="reward_contract_mockmed_probe", + oracle={"kind": "rest", "base_url": base_url.rstrip("/")}, + channel="api", + key=Ed25519PrivateKey.generate(), + issuer_key_id="self_signed:test", + certify=False, + required_effects=[ + { + "kind": "record_written", + "match": { + "patient_id": {"param": "patient_id"}, + "type": "Triage", + "note": NOTE, + }, + "expected_count": 1, + "count_new_only": True, + } + ], + forbidden_effects=[ + { + "kind": "record_written", + "match": {"patient_id": {"param": "patient_id"}, "type": "Signature"}, + "expected_count": 1, + } + ], + ) + return directory + + +@pytest.fixture() +def contract(tmp_path: Path, mockmed) -> Path: + return _contract(tmp_path / "contract", mockmed[0]) + + +def _probe(traces, mockmed, contract, tmp_path, **kw): + target = MockMedTestCopy(mockmed[0]) + result = probe_and_induce( + traces, + target=target, + reward_contract=contract, + work_dir=tmp_path / "probe", + **kw, + ) + return result, target + + +# --------------------------------------------------------------------------- +# Without probing, each disagreement is refused (the baseline) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "make, kind", + [ + (incidental_traces, "unconfirmed_optional"), + (branch_traces, "unconfirmed_branch"), + (undecidable_traces, "unconfirmed_optional"), + ], +) +def test_plain_induction_refuses_each_disagreement(tmp_path, make, kind): + result = induce_program([t.workflow for t in make(tmp_path)]) + assert result.certified is False + assert [u.kind for u in result.uncertainties] == [kind] + assert result.probe_target is None and result.probe_runs == [] + + +# --------------------------------------------------------------------------- +# Done criteria +# --------------------------------------------------------------------------- + + +def test_incidental_click_is_resolved(tmp_path, mockmed, contract): + result, target = _probe(incidental_traces(tmp_path), mockmed, contract, tmp_path) + + assert result.certified is True + [verdict] = result.probe_verdicts + assert verdict.certified == "incidental" + assert verdict.interpretations == {"incidental": "passed", "conditional": "failed"} + # The program has no Sign chart step at all. + steps = [ + s.step.id for s in result.program.states.values() if s.kind is StateKind.ACTION + ] + assert "s_sign" not in steps + assert any(d.kind == "incidental" for d in result.column_decisions) + # The conditional reading signed p3's chart: a forbidden effect. + signed = [ + r + for r in result.probe_runs + if r.interpretation == "conditional" and r.context == "rec-b" + ] + assert [r.outcome for r in signed] == ["wrong_effect"] + + +def test_real_branch_is_resolved(tmp_path, mockmed, contract): + result, _target = _probe(branch_traces(tmp_path), mockmed, contract, tmp_path) + + assert result.certified is True + [verdict] = result.probe_verdicts + assert verdict.certified == "conditional" + assert verdict.interpretations == {"incidental": "failed", "conditional": "passed"} + branches = [s for s in result.program.states.values() if s.kind is StateKind.BRANCH] + assert len(branches) == 1 + [guard] = [t.guard for t in branches[0].transitions if t.guard is not None] + assert guard.text == "Accept" + # Dropping Accept left p1's save behind the alert: the run halted. + missed = [ + r + for r in result.probe_runs + if r.interpretation == "incidental" and r.context == "rec-a" + ] + assert [r.outcome for r in missed] == ["halted_before_effect"] + + +def test_probe_that_cannot_decide_keeps_the_point_refused(tmp_path, mockmed, contract): + result, _target = _probe(undecidable_traces(tmp_path), mockmed, contract, tmp_path) + + assert result.certified is False + assert result.program is None + [verdict] = result.probe_verdicts + assert verdict.certified is None + assert verdict.interpretations == {"incidental": "passed", "conditional": "passed"} + assert "stays refused" in verdict.detail + # The record-next item is still there, as without probing. + assert [i.code for i in result.record_next] == ["unconfirmed_optional"] + + +def test_inconclusive_read_back_keeps_the_point_refused(tmp_path, mockmed): + """A store the probe cannot read decides nothing, even when one reading + would otherwise have won.""" + contract = _contract(tmp_path / "contract", "http://127.0.0.1:9") + result, _target = _probe(incidental_traces(tmp_path), mockmed, contract, tmp_path) + assert result.certified is False + [verdict] = result.probe_verdicts + assert verdict.certified is None + assert set(verdict.interpretations.values()) == {"inconclusive"} + assert all(not r.conclusive for r in result.probe_runs) + + +def test_probing_a_non_test_target_is_refused(tmp_path, mockmed, contract): + url, db = mockmed + # A record the probe's reset would wipe: proof nothing was touched. + db.add("p3", "Triage", "pre-existing") + traces = incidental_traces(tmp_path) + deployment = tmp_path / "production.yaml" + deployment.write_text(f"name: clinic\nbackend:\n url: {url}\n") + + from openadapt_flow.__main__ import main + + rc = main( + [ + "induce", + *[str(_save(t)) for t in traces], + "--out", + str(tmp_path / "out"), + "--probe-on", + str(deployment), + ] + ) + assert rc == 3 + assert len(db.snapshot()["records"]) == 1 + assert not (tmp_path / "out").exists() + + +@pytest.mark.parametrize( + "section, profile, needle", + [ + (None, None, "does not declare"), + ( + {"declared": False, "reset_url": "x", "reward_contract": "c"}, + None, + "does not declare", + ), + ({"declared": True, "reward_contract": "c"}, None, "no reset"), + ({"declared": True, "reset_url": "x"}, None, "no reward_contract"), + ( + {"declared": True, "reset_url": "x", "reward_contract": "c"}, + "regulated", + "production profile", + ), + ( + {"declared": True, "reset_url": "x", "reward_contract": "c"}, + "standard", + "production profile", + ), + ], +) +def test_test_copy_gate(section, profile, needle): + data = {"runtime": {"profile": profile}} + if section is not None: + data["test_copy"] = section + cfg = DeploymentConfig.model_validate(data) + assert any(needle in r for r in probe_target_refusals(cfg)) + with pytest.raises(ProbeRefused): + require_test_copy(cfg) + + +def test_declared_test_copy_passes_the_gate(): + cfg = DeploymentConfig.model_validate( + { + "runtime": {"profile": "demo"}, + "test_copy": {"declared": True, "reset_url": "x", "reward_contract": "c"}, + } + ) + assert probe_target_refusals(cfg) == [] + + +def test_attached_browser_is_refused(): + cfg = DeploymentConfig.model_validate( + { + "backend": {"browser_cdp_endpoint": "http://127.0.0.1:9222"}, + "test_copy": {"declared": True, "reset_url": "x", "reward_contract": "c"}, + } + ) + assert any("fresh session" in r for r in probe_target_refusals(cfg)) + + +# --------------------------------------------------------------------------- +# Logging, fresh sessions, no model calls, advisory proposer +# --------------------------------------------------------------------------- + + +def test_every_run_is_logged_reset_first_and_in_a_fresh_session( + tmp_path, mockmed, contract +): + result, target = _probe(branch_traces(tmp_path), mockmed, contract, tmp_path) + # incidental fails on the first context and stops; conditional runs both. + assert len(result.probe_runs) == target.sessions == 3 + assert target.resets == target.sessions + 1 # one before each run, one after + rendered = result.render() + assert rendered.count("[probe]") == len(result.probe_runs) + assert "[verdict] branch_" in rendered + dumped = json.loads(result.model_dump_json()) + assert len(dumped["probe_runs"]) == 3 + + +def test_probing_makes_no_model_calls(tmp_path, mockmed, contract): + result, _target = _probe(incidental_traces(tmp_path), mockmed, contract, tmp_path) + assert result.probe_runs + assert sum(r.model_calls for r in result.probe_runs) == 0 + + +class _WrongProposer: + """Always argues that the step is noise. Zero model calls.""" + + def __init__(self) -> None: + self.calls = 0 + + def propose(self, target, kind, context): + self.calls += 1 + return "drop this step: the demonstrator clicked it by habit" + + +def test_proposer_stays_advisory(tmp_path, mockmed, contract): + proposer = _WrongProposer() + result, _target = _probe( + branch_traces(tmp_path), mockmed, contract, tmp_path, propose=proposer + ) + # The read-back decides, not the proposal. + assert result.probe_verdicts[0].certified == "conditional" + assert result.certified is True + assert proposer.calls >= 1 + assert result.proposed and all(p.trusted is False for p in result.proposed) + + +def test_missing_identity_is_not_probed(tmp_path, mockmed, contract): + traces = incidental_traces(tmp_path) + traces[1].workflow = traces[1].workflow.model_copy(update={"params": {}}) + result, target = _probe(traces, mockmed, contract, tmp_path) + assert target.sessions == 0 + assert result.certified is False + assert "not probed" in result.probe_verdicts[0].detail + + +# --------------------------------------------------------------------------- +# CLI wiring +# --------------------------------------------------------------------------- + + +def _save(trace: ProbeTrace) -> Path: + trace.workflow.save(trace.bundle_dir) + return trace.bundle_dir + + +def test_cli_probe_on_declared_test_copy_certifies_and_logs( + tmp_path, mockmed, contract, monkeypatch, capsys +): + url, _db = mockmed + import openadapt_flow.compiler.probing as probing + + made: list[MockMedTestCopy] = [] + + def fake_target(cfg): + probing.require_test_copy(cfg) + made.append(MockMedTestCopy(url)) + return made[-1] + + monkeypatch.setattr(probing, "DeploymentProbeTarget", fake_target) + deployment = tmp_path / "test-copy.yaml" + deployment.write_text( + "name: mockmed-copy\n" + f"backend:\n url: {url}\n" + "test_copy:\n" + " declared: true\n" + f" reset_url: {url}api/reset\n" + " reward_contract: contract\n" + ) + traces = incidental_traces(tmp_path) + out = tmp_path / "out" + report = tmp_path / "report.json" + + from openadapt_flow.__main__ import main + + rc = main( + [ + "induce", + *[str(_save(t)) for t in traces], + "--out", + str(out), + "--probe-on", + str(deployment), + "--report", + str(report), + ] + ) + printed = capsys.readouterr().out + assert rc == 0, printed + assert (out / "workflow.json").is_file() + assert "[probe]" in printed and "certified incidental" in printed + saved = json.loads(report.read_text()) + assert len(saved["probe_runs"]) == made[0].sessions > 0 diff --git a/packages/flow/tests/test_probe_induction_surfaces.py b/packages/flow/tests/test_probe_induction_surfaces.py new file mode 100644 index 000000000..c3b332205 --- /dev/null +++ b/packages/flow/tests/test_probe_induction_surfaces.py @@ -0,0 +1,115 @@ +"""``induce --probe-on`` probes a test copy on a desktop or remote surface. + +The deployment file declares a Windows, macOS, Linux, RDP or Citrix test copy. +:class:`~openadapt_flow.compiler.probing.DeploymentProbeTarget` resets it with +``reset_command`` and opens a fresh backend from the file's ``backend:`` +section for every probe run, through the factory. It never opens a browser. +The screen side is the scripted MockMed session from test_probe_induction.py, +standing in for the desktop app; writes land in the same HTTP test copy, and +the read-back decides. See docs/BACKEND_PARITY.md. +""" + +# ruff: noqa: F811 - pytest fixtures imported from test_probe_induction + +from __future__ import annotations + +import sys + +import pytest + +from openadapt_flow.compiler import probing +from openadapt_flow.compiler.probing import ( + DeploymentProbeTarget, + RunSignal, + probe_and_induce, + runtime_signal_of, +) +from openadapt_flow.deployment import DeploymentConfig +from openadapt_flow.runtime.replayer import Replayer +from tests.test_probe_induction import ( # noqa: F401 - pytest fixtures + MockMedBackend, + MockMedSession, + MockMedVision, + branch_traces, + contract, + mockmed, +) + +SECTIONS = { + "windows": {"kind": "windows", "agent_url": "http://127.0.0.1:5001"}, + "macos": {"kind": "macos", "macos_app": "TextEdit"}, + "linux": {"kind": "linux", "linux_app": "notes", "linux_window_title": "Notes"}, + "rdp": {"kind": "rdp", "rdp_host": "10.0.0.5"}, + "citrix": { + "kind": "citrix", + "rdp_window": "wfica32", + "rdp_readiness_text": "Patient Notes", + }, +} + + +@pytest.mark.parametrize("kind", sorted(SECTIONS)) +def test_a_desktop_test_copy_is_probed_through_its_own_backend( + tmp_path, + mockmed, + contract, + monkeypatch, + kind, +): + import openadapt_flow.backends.factory as factory + + url, _db = mockmed + built: list = [] + closed: list = [] + + def build_backend(cfg, **kwargs): + assert kwargs.get("page") is None + built.append(cfg.kind) + backend = MockMedBackend(MockMedSession(url)) + backend.close = lambda: closed.append(cfg.kind) # type: ignore[attr-defined] + return backend + + def replay(self, backend, workflow, bundle_dir, run_dir, params) -> RunSignal: + # The scripted session's templates are told apart by its own vision; + # everything else is the target's real code path. + report = Replayer( + backend, vision=MockMedVision(backend.session), poll_interval_s=0.01 + ).run(workflow, bundle_dir=bundle_dir, run_dir=run_dir, params=params) + return RunSignal(runtime_signal_of(report), model_calls=report.model_calls) + + monkeypatch.setattr(factory, "build_backend", build_backend) + monkeypatch.setattr(DeploymentProbeTarget, "_replay", replay) + sync_api = pytest.importorskip("playwright.sync_api") + monkeypatch.setattr( + sync_api, "sync_playwright", lambda: pytest.fail("no browser for a desktop") + ) + + reset = ( + "import urllib.request; urllib.request.urlopen(urllib.request.Request(" + f"{url + 'api/reset'!r}, data=b'{{}}', method='POST'), timeout=5).close()" + ) + cfg = DeploymentConfig.model_validate( + { + "name": f"{kind}-test-copy", + "backend": SECTIONS[kind], + "test_copy": { + "declared": True, + "reset_command": [sys.executable, "-c", reset], + "reward_contract": str(contract), + }, + } + ) + assert probing.probe_target_refusals(cfg) == [] + target = DeploymentProbeTarget(cfg) + result = probe_and_induce( + branch_traces(tmp_path), + target=target, + reward_contract=contract, + work_dir=tmp_path / "probe", + ) + + assert result.certified is True + assert result.probe_verdicts[0].certified == "conditional" + assert result.probe_runs and set(built) == {kind} + assert len(built) == len(result.probe_runs) == len(closed) + assert sum(r.model_calls for r in result.probe_runs) == 0