From 1820e5bfed0f2594d1c320a54f8c54dbc81fd97f Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 18 Oct 2026 16:01:44 -0400 Subject: [PATCH 1/3] feat(flow): search repair bridges on a staging copy, scored by read-back Self-repair's probe stops before the save, so it cannot tell two bridges apart that reach the save the same way but write different records. openadapt_flow.repair.staging_search runs each candidate through self-repair on a resettable staging copy, then runs the patched bundle through the save there and reads the copy's store with the reward worker. The best verified candidate is the one proposal production self-repair receives; its probe, campaigns, approval, staging and canary are unchanged. The search never opens a production session. - StagingCopy, StagingBridgeSearch, StagingSearchAgent (a RepairAgent). - ScriptedBridges for tests; ModelBridges reads the halted trace and the last verified traces and answers with control labels mapped to points. - MockMed ?drift=menu2: a second save path in the menu (Submit as Consult). Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/HARNESS2.md | 35 ++ .../flow/openadapt_flow/mockmed/static/app.js | 29 +- .../openadapt_flow/repair/staging_search.py | 437 ++++++++++++++++++ packages/flow/tests/test_staging_search.py | 227 +++++++++ 4 files changed, 724 insertions(+), 4 deletions(-) create mode 100644 packages/flow/openadapt_flow/repair/staging_search.py create mode 100644 packages/flow/tests/test_staging_search.py diff --git a/packages/flow/docs/HARNESS2.md b/packages/flow/docs/HARNESS2.md index 63dace023..509560046 100644 --- a/packages/flow/docs/HARNESS2.md +++ b/packages/flow/docs/HARNESS2.md @@ -129,6 +129,41 @@ export still says which write hit the wrong patient. In `readback` mode the same identity verdict also appears in the trajectory Harness² proposers read, as an observation after each write. +## Repair search on a staging copy + +Self-repair ([SELF_REPAIR.md](SELF_REPAIR.md)) probes a proposed fix up to +the last point before the save and stops there. That shows the fix reaches +the save on the right record. It cannot show what the save writes, and two +bridges can look the same up to the save and still write different records. + +`openadapt_flow.repair.staging_search` closes that gap without touching +production. For each candidate bridge, on a resettable staging copy, it: + +1. runs self-repair with that one bridge (halt, patch, probe, campaigns) and + stops at the approval request, which nobody answers; +2. runs the patched bundle end to end on a fresh copy, through the save; +3. reads the copy's system of record with the reward worker and scores it. + +`StagingSearchAgent` hands production self-repair one proposal: the best +candidate the read-back verified, or none. From there the pipeline is +unchanged. Production probes it before the save and runs the campaigns, and +one person approves it before anything is saved (candidate, replay_passed, +fault_passed, approved, staged, canary, active). The search never opens a +production session. + +Candidates come from `ScriptedBridges` (tests) or `ModelBridges`. A model +reads the halted step, the controls OCR finds on the live screen, and the +full traces of the halted run and the last verified runs, and answers with +control labels. The labels are mapped to points on the screen, so the model +never invents coordinates. + +MockMed `?drift=menu2` is the test case. The save moved into a menu with two +save paths: Actions > Billing > Submit encounter (a Triage encounter, as +recorded) and Actions > Clinical > Submit as Consult. Both pass self-repair's +probe. On staging the Consult path writes a Consult record (wrong value, +-0.5) and the Billing path verifies (1.0), so production sees only the +Billing bridge. + ## Install Harness² needs Python 3.12 and is not on PyPI. Install it from its diff --git a/packages/flow/openadapt_flow/mockmed/static/app.js b/packages/flow/openadapt_flow/mockmed/static/app.js index 8ca36578e..2eab95f84 100644 --- a/packages/flow/openadapt_flow/mockmed/static/app.js +++ b/packages/flow/openadapt_flow/mockmed/static/app.js @@ -413,15 +413,25 @@ function renderEncounter() { // "menu" drift: the save moved behind a menu, Actions > Billing > Submit // encounter. The menu opens in place; nothing is saved until Submit. - var saveHtml = DRIFT.has('menu') + // "menu2" drift: the same menu with a second save path, Actions > Clinical + // > Submit as Consult, which saves the encounter as a Consult. Both paths + // look the same up to the save; only the saved record tells them apart. + var MENU = DRIFT.has('menu') || DRIFT.has('menu2'); + var saveHtml = MENU ? '
' + '' + '
' + '' + + (DRIFT.has('menu2') + ? '' + + '' + : '') + + '' : ''; - var saveTarget = DRIFT.has('menu') ? 'menu-submit' : SAVE_ID; + var saveTarget = MENU ? 'menu-submit' : SAVE_ID; var triageBtn = ''; @@ -475,7 +485,7 @@ function renderEncounter() { }); }); - if (DRIFT.has('menu')) { + if (MENU) { document.getElementById('actions-menu') .addEventListener('click', function () { var list = document.getElementById('actions-list'); @@ -485,6 +495,17 @@ function renderEncounter() { .addEventListener('click', function () { document.getElementById('billing-list').hidden = false; }); + if (DRIFT.has('menu2')) { + document.getElementById('menu-clinical') + .addEventListener('click', function () { + document.getElementById('clinical-list').hidden = false; + }); + document.getElementById('menu-consult') + .addEventListener('click', function () { + state.encounterType = 'Consult'; + document.getElementById(saveTarget).click(); + }); + } } document.getElementById(saveTarget) diff --git a/packages/flow/openadapt_flow/repair/staging_search.py b/packages/flow/openadapt_flow/repair/staging_search.py new file mode 100644 index 000000000..252106ac7 --- /dev/null +++ b/packages/flow/openadapt_flow/repair/staging_search.py @@ -0,0 +1,437 @@ +"""Search for a multi-step bridge on a staging copy, scored by read-back. + +Self-repair (:mod:`openadapt_flow.repair.self_repair`) probes each proposed +fix up to the last point before the save and stops there. That proves the +bridge reaches the save on the right record. It cannot show what the save +writes. Two bridges can look the same up to the save and still write +different records: one menu path files the encounter as Triage, another as +Consult. + +:class:`StagingBridgeSearch` closes that gap without touching production. +For each candidate bridge, on a resettable staging copy of the app, it: + +1. runs self-repair itself with that one bridge (halt, patch, probe, + campaigns), stopping at the approval request, which nobody answers; +2. runs the patched bundle end to end on a fresh copy, through the save; +3. reads the copy's system of record with the reward worker and scores it. + +The best candidate that the read-back verifies becomes the one proposal +:class:`StagingSearchAgent` hands to the production self-repair. From there +nothing changes: production probes it before the save, runs the campaigns, +and a person approves it before anything is saved +(candidate -> replay_passed -> fault_passed -> approved -> staged -> canary +-> active). The search never opens a production session. + +Candidates come from a :class:`BridgeProposer`: :class:`ScriptedBridges` for +tests, :class:`ModelBridges` for a model that reads the halted trace and the +last verified traces (:mod:`openadapt_flow.integrations.harness2.traces`). +""" + +from __future__ import annotations + +import json +import re +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any, Callable, Mapping, Optional, Protocol, Sequence + +from openadapt_flow.integrations.harness2 import traces +from openadapt_flow.integrations.harness2.scorer import ( + EpisodeScore, + Penalties, + ReadBackScorer, +) +from openadapt_flow.integrations.harness2.tasks import GuiTask +from openadapt_flow.ir import RunReport, Workflow +from openadapt_flow.repair.self_repair import ( + RepairProposal, + RepairRequest, + ScreenAccess, + SelfRepair, + SelfRepairTarget, +) + +#: Runtime outcome -> the runtime signal the reward worker takes. +_SIGNAL = { + "VERIFIED": "completed", + "COMPLETED_UNVERIFIED": "completed", + "HALTED": "halted_before_effect", + "ROLLED_BACK": "halted_before_effect", + "FAILED": "failed_platform", +} + + +@dataclass(frozen=True) +class StagingCopy: + """A resettable copy of the app that the search may write to. + + ``target`` opens sessions on the copy only. ``reset`` empties the copy's + store. ``oracle_recipe`` is how the reward worker reads that store. + ``identity`` is the item's subject (for MockMed, ``{"patient_id": "p1"}``) + and ``record`` the values the item's save must write. + """ + + target: SelfRepairTarget + reset: Callable[[], None] + oracle_recipe: Mapping[str, Any] + identity: Mapping[str, str] + record: Mapping[str, str] + fault_probes: Sequence[str] = () + + +class BridgeProposer(Protocol): + """Proposes candidate bridges for one halted step, best guess first.""" + + dialect: str + model_calls: int + cost_usd: float + + def bridges(self, request: RepairRequest, history: str) -> list[RepairProposal]: ... + + +class ScriptedBridges: + """Fixed candidates, for tests. No model calls.""" + + model_calls = 0 + cost_usd = 0.0 + + def __init__(self, *bridges: Sequence[Mapping[str, Any]], dialect: str = "generic"): + self._bridges = [list(b) for b in bridges] + self.dialect = dialect + self.seen_history = "" + + def bridges(self, request: RepairRequest, history: str) -> list[RepairProposal]: + self.seen_history = history + return [RepairProposal(actions=b) for b in self._bridges] + + +_BRIDGES_PROMPT = """\ +A compiled GUI program halted at one step because the screen changed. Propose +up to {n} different bridges: each is 1 to {max_actions} clicks, by the visible +label of the control, that would reach the step's goal on the current screen. +The last click must do what the halted step did. + +Halted step: {intent} +Recorded control label: {label} +Why it halted: {reason} +Mode: {mode} ({mode_help}) + +Visible text on the current screen, one control per line: +{screen} + +Earlier runs of the same item, the halted one first: +{history} + +Reply with JSON only: {{"bridges": [["label", "label", ...], ...]}} +""" + + +class ModelBridges: + """A model proposes bridges as control labels; OCR maps labels to points. + + The model reads the halted step, the controls OCR finds on the live + screen, and the full traces of the halted run and the last verified runs. + It answers with labels, which keeps it from inventing coordinates. + """ + + dialect = "generic" + + def __init__(self, model: Any, *, n: int = 3, max_actions: int = 3) -> None: + self.model = model + self.n = n + self.max_actions = max_actions + self.model_calls = 0 + self.cost_usd = 0.0 + + def bridges(self, request: RepairRequest, history: str) -> list[RepairProposal]: + from openadapt_flow.vision.ocr import ocr + + lines = ocr(request.live_png) + points = { + _squash(ln.text): ( + int(ln.region[0] + ln.region[2] / 2), + int(ln.region[1] + ln.region[3] / 2), + ) + for ln in lines + if ln.text.strip() + } + prompt = _BRIDGES_PROMPT.format( + n=self.n, + max_actions=min(self.max_actions, request.max_actions), + intent=request.intent, + label=request.recorded_label or "(none)", + reason=request.halt_reason[:500], + mode=request.mode, + mode_help=( + "the control is gone; the last click replaces it" + if request.mode == "replace" + else "the click acted but a new screen appeared; the bridge continues after it" + ), + screen="\n".join(ln.text for ln in lines if ln.text.strip()), + history=history[:20_000] or "(none)", + ) + reply = self.model.complete( + "You repair GUI automations.", prompt, max_tokens=600 + ) + self.model_calls += 1 + self.cost_usd += reply.usd + try: + raw = json.loads( + reply.text[reply.text.find("{") : reply.text.rfind("}") + 1] + ) + labels = [list(map(str, b)) for b in raw.get("bridges", [])][: self.n] + except (ValueError, TypeError, AttributeError): + return [] + out = [] + for bridge in labels: + actions = [] + for label in bridge[: request.max_actions]: + point = points.get(_squash(label)) + if point is None: + break + actions.append({"kind": "click", "x": point[0], "y": point[1]}) + else: + if actions: + out.append(RepairProposal(actions=actions, note=" > ".join(bridge))) + return out + + +@dataclass +class CandidateOutcome: + """What one candidate bridge did on the staging copy.""" + + index: int + actions: list[dict[str, Any]] + note: str = "" + #: Self-repair's status on staging (``awaiting_approval`` = probe and + #: campaigns passed and it reached the approval request). + repair_status: str = "" + reason: str = "" + runtime_outcome: str = "" + outcome: str = "" + reward: Optional[float] = None + wrong_record: bool = False + duplicate: bool = False + wrong_value: bool = False + receipt_id: str = "" + + +@dataclass +class SearchResult: + winner: Optional[RepairProposal] + winner_index: Optional[int] + candidates: list[CandidateOutcome] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "schema": "openadapt.staging-bridge-search/v1", + "winner_index": self.winner_index, + "candidates": [asdict(c) for c in self.candidates], + } + + +class StagingBridgeSearch: + """Try each candidate bridge on the staging copy; keep the verified best.""" + + def __init__( + self, + proposer: BridgeProposer, + staging: StagingCopy, + bundle_dir: Path, + work_dir: Path, + *, + halted_run_dir: Optional[Path] = None, + verified_run_dirs: Sequence[Path] = (), + last_verified: int = 3, + max_candidates: int = 4, + penalties: Penalties = Penalties(), + ) -> None: + self.proposer = proposer + self.staging = staging + self.bundle_dir = Path(bundle_dir) + self.work_dir = Path(work_dir) + self.halted_run_dir = halted_run_dir + self.verified_run_dirs = ( + list(verified_run_dirs)[-last_verified:] if last_verified else [] + ) + self.max_candidates = max_candidates + self.penalties = penalties + + def history(self) -> str: + """The halted trace, then the last verified ones, rendered in full.""" + + parts = [] + dirs = ( + [self.halted_run_dir] if self.halted_run_dir else [] + ) + self.verified_run_dirs + for d in dirs: + if d is not None and (Path(d) / "report.json").is_file(): + parts.append(traces.render(traces.from_flow_run(Path(d)))) + return "\n\n".join(parts) + + def search(self, request: RepairRequest) -> SearchResult: + proposals = self.proposer.bridges(request, self.history())[ + : self.max_candidates + ] + result = SearchResult(winner=None, winner_index=None) + for i, proposal in enumerate(proposals): + result.candidates.append(self._try(i, proposal)) + verified = [ + c + for c in result.candidates + if c.outcome == "verified" and c.reward is not None + ] + if verified: + best = max(verified, key=lambda c: (c.reward, -c.index)) + result.winner_index = best.index + result.winner = proposals[best.index] + self.work_dir.mkdir(parents=True, exist_ok=True) + (self.work_dir / "search.json").write_text( + json.dumps(result.to_dict(), indent=2, default=str), encoding="utf-8" + ) + return result + + def _try(self, index: int, proposal: RepairProposal) -> CandidateOutcome: + out = CandidateOutcome( + index=index, + actions=[dict(a) for a in proposal.bridge()], + note=proposal.note, + ) + cdir = self.work_dir / f"candidate-{index}" + self.staging.reset() + repair = SelfRepair( + _OneProposal(proposal, self.proposer.dialect), + self.staging.target, + store=cdir / "store", + screen_access=ScreenAccess( + enabled_by="staging-search", mode="allow_unscrubbed" + ), + max_attempts=1, + fault_probes=self.staging.fault_probes, + approval_wait_s=0, + ) + repaired = repair.run(self.bundle_dir, cdir / "halt") + out.repair_status = repaired.status + out.reason = repaired.evidence.reason or "" + if repaired.status != "awaiting_approval" or repaired.patched_bundle is None: + return out + # The probe stopped before the save. Run the patched bundle through + # the save on a fresh copy, and read what it wrote. + self.staging.reset() + task = GuiTask( + task_id=f"repair/{request_step(repaired)}", + family="staging-repair", + app="staging", + instruction="", + params={**dict(self.staging.record), **dict(self.staging.identity)}, + identity=dict(self.staging.identity), + ) + scorer = ReadBackScorer( + task, + dict(self.staging.oracle_recipe), + cdir / "reward", + penalties=self.penalties, + ) + episode = f"episode_staging_{index:04d}_{cdir.parent.name[-16:] or 'search'}" + episode = re.sub(r"[^A-Za-z0-9._:-]", "_", episode) + scorer.begin(episode) + patched = Path(repaired.patched_bundle) + session = self.staging.target.open() + report: Optional[RunReport] = None + try: + report = self.staging.target.execute( + Workflow.load(patched), patched, session, cdir / "full" + ) + except Exception as exc: # noqa: BLE001 - recorded, scored as a platform failure + out.reason = f"full run failed: {type(exc).__name__}: {exc}"[:500] + finally: + session.close() + out.runtime_outcome = ( + str(report.execution_outcome or "") if report else "FAILED" + ) + signal = _SIGNAL.get(out.runtime_outcome, "halted_before_effect") + if report is not None and report.success and not report.execution_outcome: + signal = "completed" + score: EpisodeScore = scorer.score(episode, signal) + out.outcome = score.outcome + out.reward = score.reward + out.wrong_record = score.wrong_record + out.duplicate = score.duplicate + out.wrong_value = score.wrong_value + out.receipt_id = score.receipt_id + return out + + +def request_step(repaired: Any) -> str: + return str(getattr(repaired.evidence, "halted_step_id", "") or "step") + + +class _OneProposal: + """Hands self-repair exactly one fixed proposal, once.""" + + agent_id = "agent:staging-search-candidate" + requires_model_key = False + + def __init__(self, proposal: RepairProposal, dialect: str) -> None: + self.proposal: Optional[RepairProposal] = proposal + self.dialect = dialect + + def propose(self, request: RepairRequest) -> Optional[RepairProposal]: + proposal, self.proposal = self.proposal, None + return proposal + + +class StagingSearchAgent: + """A self-repair agent whose one proposal is the staging search's winner. + + Give it to the production :class:`~openadapt_flow.repair.self_repair.SelfRepair`. + On the first request it runs the search on staging and returns the best + verified bridge, or ``None`` (give up) when no candidate verified. It + proposes once; a refused winner is not retried with a worse candidate. + """ + + def __init__( + self, + search: StagingBridgeSearch, + *, + agent_id: str = "agent:staging-search", + ) -> None: + self.search = search + self.agent_id = agent_id + self.dialect = search.proposer.dialect + # A model-backed proposer runs on the customer's key, like any agent. + self.requires_model_key = not isinstance(search.proposer, ScriptedBridges) + self.result: Optional[SearchResult] = None + + def propose(self, request: RepairRequest) -> Optional[RepairProposal]: + if self.result is not None: + return None + self.result = self.search.search(request) + winner = self.result.winner + if winner is None: + return None + return RepairProposal( + actions=list(winner.bridge()), + model_calls=int(getattr(self.search.proposer, "model_calls", 0)), + cost_usd=float(getattr(self.search.proposer, "cost_usd", 0.0)), + note=( + f"verified on a staging copy as candidate {self.result.winner_index} " + f"of {len(self.result.candidates)}" + ), + ) + + +def _squash(text: str) -> str: + return re.sub(r"\s+", "", text).lower() + + +__all__ = [ + "BridgeProposer", + "CandidateOutcome", + "ModelBridges", + "ScriptedBridges", + "SearchResult", + "StagingBridgeSearch", + "StagingCopy", + "StagingSearchAgent", +] diff --git a/packages/flow/tests/test_staging_search.py b/packages/flow/tests/test_staging_search.py new file mode 100644 index 000000000..0f6b12dfa --- /dev/null +++ b/packages/flow/tests/test_staging_search.py @@ -0,0 +1,227 @@ +"""Bridge search on a staging copy, scored by read-back, then one approval. + +MockMed ``?drift=menu2`` moves the save behind a menu with two save paths: +Actions > Billing > Submit encounter (a Triage encounter, as recorded) and +Actions > Clinical > Submit as Consult. Self-repair's probe stops before the +save, so both bridges pass it. Only the saved record tells them apart. + +The search runs both on a separate staging server, writes through the save +there, and reads what landed. The Consult path is refuted; the Billing path +is verified and becomes the one proposal production self-repair sees. A +person approves it on the phone and the resumed production run verifies. +Production gets one record; staging writes never reach it. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from openadapt_flow.repair.lifecycle import RepairStore +from openadapt_flow.repair.staging_search import ( + ScriptedBridges, + StagingBridgeSearch, + StagingCopy, + StagingSearchAgent, +) +from tests.test_self_repair import ( # noqa: F401 - the shared MockMed fixture + _browser, + _phone, + _records, + _reset, + _self_repair, + _target, + mockmed, +) + +pytest.importorskip("playwright.sync_api") + +CONSULT = [ + {"kind": "click", "target": "#actions-menu"}, + {"kind": "click", "target": "#menu-clinical"}, + {"kind": "click", "target": "#menu-consult"}, +] +BILLING = [ + {"kind": "click", "target": "#actions-menu"}, + {"kind": "click", "target": "#menu-billing"}, + {"kind": "click", "target": "#menu-submit"}, +] + + +@pytest.fixture +def staging(): + from openadapt_flow.mockmed.fault_server import serve + from openadapt_flow.tutorial import TUTORIAL_NOTE + + url, _db, stop = serve() + base = url.rstrip("/") + try: + yield ( + base, + StagingCopy( + target=_target(base, "menu2"), + reset=lambda: _reset(base), + oracle_recipe={ + "kind": "rest", + "base_url": base, + "records_path": "/api/db", + "records_key": "records", + }, + identity={"patient_id": "p1"}, + record={"type": "Triage", "note": TUTORIAL_NOTE}, + fault_probes=("missing",), + ), + ) + finally: + stop() + + +def test_staging_search_picks_the_bridge_whose_record_verifies( + mockmed, # noqa: F811 - the shared fixture + staging, + tmp_path: Path, +) -> None: + base_url, bundle = mockmed + staging_url, copy = staging + _reset(base_url) + proposer = ScriptedBridges(CONSULT, BILLING) + search = StagingBridgeSearch(proposer, copy, bundle, tmp_path / "search") + agent = StagingSearchAgent(search) + run_dir = tmp_path / "run" + seen: list[int] = [] + with _browser(): + result = _self_repair( + base_url, + tmp_path, + agent, + drift="menu2", + approval_wait_s=120, + sleep=_phone(run_dir, approve=True, seen=seen, base_url=base_url), + ).run(bundle, run_dir) + + found = agent.result + assert found is not None + consult, billing = found.candidates + # Both reach the save on the right record; self-repair cannot tell them + # apart before the save. + assert consult.repair_status == billing.repair_status == "awaiting_approval" + # The read-back can. + # The runtime's record check halts the Consult run after its save, and + # the read-back shows a Consult record on the right patient. + assert consult.wrong_value and consult.reward == -0.5 + assert billing.outcome == "verified" and billing.reward == 1.0 + assert found.winner_index == 1 + assert (tmp_path / "search" / "search.json").is_file() + + # Production saw one proposal, the winner, and a person approved it. + assert result.status == "resumed", result.evidence.reason + assert result.report.execution_outcome == "VERIFIED" + assert seen == [0] + records = _records(base_url) + assert len(records) == 1 and records[0]["type"] == "Triage" + candidate = RepairStore(tmp_path / "repair-store").load_candidate( + result.evidence.candidate_id + ) + assert candidate.state == "canary" + # The staging copy is a different server: its writes never reached + # production (the last staging run left its own record there). + assert staging_url != base_url + assert len(_records(staging_url)) == 1 + + +def test_no_verified_candidate_means_no_proposal( + mockmed, # noqa: F811 - the shared fixture + staging, + tmp_path: Path, +) -> None: + from openadapt_flow.repair.self_repair import RepairRequest + + _base_url, bundle = mockmed + _staging_url, copy = staging + search = StagingBridgeSearch(ScriptedBridges(CONSULT), copy, bundle, tmp_path / "s") + agent = StagingSearchAgent(search) + request = RepairRequest( + step_id="step_005", + intent="Save the encounter", + action="click", + halt_reason="Could not resolve target", + live_png=b"", + viewport=(1280, 800), + recorded_label="Save Encounter", + recorded_region=(0, 0, 10, 10), + recorded_click_point=(5, 5), + recorded_crop_png=None, + dialect="generic", + attempt=1, + max_actions=3, + ) + with _browser(): + assert agent.propose(request) is None + assert agent.result is not None + [only] = agent.result.candidates + assert only.wrong_value and only.reward == -0.5 + # It proposes once; a second request gets nothing. + assert agent.propose(request) is None + + +def test_model_bridges_map_labels_to_screen_points(monkeypatch) -> None: + from dataclasses import dataclass + + from openadapt_flow.integrations.harness2.llm import Completion + from openadapt_flow.repair.self_repair import RepairRequest + from openadapt_flow.repair.staging_search import ModelBridges + + @dataclass + class Line: + text: str + region: tuple[int, int, int, int] + + screen = [ + Line("Actions", (100, 500, 60, 20)), + Line("Billing", (100, 530, 60, 20)), + Line("Submit encounter", (100, 560, 120, 20)), + ] + import sys + + import openadapt_flow.vision.ocr # noqa: F401 - the package re-exports a function by that name + + monkeypatch.setattr( + sys.modules["openadapt_flow.vision.ocr"], "ocr", lambda png, **kw: screen + ) + + class Model: + prompt = "" + + def complete(self, system, user, *, max_tokens=800): + Model.prompt = user + return Completion( + '{"bridges": [["Actions", "Billing", "Submit encounter"],' + ' ["Actions", "Nowhere"]]}', + 0.002, + ) + + request = RepairRequest( + step_id="step_005", + intent="Save the encounter", + action="click", + halt_reason="Could not resolve target", + live_png=b"png", + viewport=(1280, 800), + recorded_label="Save Encounter", + recorded_region=(0, 0, 10, 10), + recorded_click_point=(5, 5), + recorded_crop_png=None, + dialect="generic", + attempt=1, + max_actions=3, + ) + proposer = ModelBridges(Model()) + [bridge] = proposer.bridges(request, "# Trace halted-run\n") + assert [dict(a) for a in bridge.bridge()] == [ + {"kind": "click", "x": 130, "y": 510}, + {"kind": "click", "x": 130, "y": 540}, + {"kind": "click", "x": 160, "y": 570}, + ] + assert "# Trace halted-run" in Model.prompt and "Submit encounter" in Model.prompt + assert proposer.model_calls == 1 and proposer.cost_usd == 0.002 From 404a18d070e703514863c77cf769f57f33b0a912 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 18 Oct 2026 16:15:41 -0400 Subject: [PATCH 2/3] wip: review fixes --- .../openadapt_flow/repair/staging_search.py | 48 +++++++++++++++++-- packages/flow/tests/test_staging_search.py | 11 +++++ 2 files changed, 56 insertions(+), 3 deletions(-) diff --git a/packages/flow/openadapt_flow/repair/staging_search.py b/packages/flow/openadapt_flow/repair/staging_search.py index 252106ac7..c2e125d62 100644 --- a/packages/flow/openadapt_flow/repair/staging_search.py +++ b/packages/flow/openadapt_flow/repair/staging_search.py @@ -31,6 +31,7 @@ import json import re +import shutil from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Callable, Mapping, Optional, Protocol, Sequence @@ -41,6 +42,7 @@ Penalties, ReadBackScorer, ) +from openadapt_flow.integrations.harness2.rollout import drop_worker_secrets from openadapt_flow.integrations.harness2.tasks import GuiTask from openadapt_flow.ir import RunReport, Workflow from openadapt_flow.repair.self_repair import ( @@ -298,6 +300,9 @@ def _try(self, index: int, proposal: RepairProposal) -> CandidateOutcome: note=proposal.note, ) cdir = self.work_dir / f"candidate-{index}" + # Each search starts clean: a store or episode left by an earlier + # search would refuse this one. + shutil.rmtree(cdir, ignore_errors=True) self.staging.reset() repair = SelfRepair( _OneProposal(proposal, self.proposer.dialect), @@ -313,8 +318,17 @@ def _try(self, index: int, proposal: RepairProposal) -> CandidateOutcome: repaired = repair.run(self.bundle_dir, cdir / "halt") out.repair_status = repaired.status out.reason = repaired.evidence.reason or "" - if repaired.status != "awaiting_approval" or repaired.patched_bundle is None: + try: + if repaired.status != "awaiting_approval" or repaired.patched_bundle is None: + return out + self._full_run(out, repaired, cdir, index) return out + finally: + _withdraw(repaired, cdir) + + def _full_run( + self, out: CandidateOutcome, repaired: Any, cdir: Path, index: int + ) -> None: # The probe stopped before the save. Run the patched bundle through # the save on a fresh copy, and read what it wrote. self.staging.reset() @@ -325,6 +339,7 @@ def _try(self, index: int, proposal: RepairProposal) -> CandidateOutcome: instruction="", params={**dict(self.staging.record), **dict(self.staging.identity)}, identity=dict(self.staging.identity), + record_fields=tuple(self.staging.record), ) scorer = ReadBackScorer( task, @@ -352,14 +367,41 @@ def _try(self, index: int, proposal: RepairProposal) -> CandidateOutcome: signal = _SIGNAL.get(out.runtime_outcome, "halted_before_effect") if report is not None and report.success and not report.execution_outcome: signal = "completed" - score: EpisodeScore = scorer.score(episode, signal) + try: + score: EpisodeScore = scorer.score(episode, signal) + finally: + drop_worker_secrets(cdir / "reward" / "worker") out.outcome = score.outcome out.reward = score.reward out.wrong_record = score.wrong_record out.duplicate = score.duplicate out.wrong_value = score.wrong_value out.receipt_id = score.receipt_id - return out + + +def _withdraw(repaired: Any, cdir: Path) -> None: + """Close what the staging self-repair left open for a person. + + Each staging candidate stops at an approval request. Nobody may answer + it: the request is removed and the staging candidate is rejected, so a + portal pointed at the search's directory has nothing to approve. + """ + + from openadapt_flow.repair.candidate import Actor + from openadapt_flow.repair.lifecycle import RepairLifecycleError, RepairStore + + shutil.rmtree(cdir / "halt" / "item-approval", ignore_errors=True) + candidate_id = getattr(repaired.evidence, "candidate_id", None) + if not candidate_id: + return + try: + RepairStore(cdir / "store").reject( + candidate_id, + Actor(kind="automation", id="openadapt-staging-search"), + "staging-only candidate: scored on a staging copy, never for approval", + ) + except (RepairLifecycleError, OSError, ValueError): + pass def request_step(repaired: Any) -> str: diff --git a/packages/flow/tests/test_staging_search.py b/packages/flow/tests/test_staging_search.py index 0f6b12dfa..ffbe72c64 100644 --- a/packages/flow/tests/test_staging_search.py +++ b/packages/flow/tests/test_staging_search.py @@ -113,6 +113,13 @@ def test_staging_search_picks_the_bridge_whose_record_verifies( assert billing.outcome == "verified" and billing.reward == 1.0 assert found.winner_index == 1 assert (tmp_path / "search" / "search.json").is_file() + # Nothing on staging is left for a person to approve, and no secrets stay. + for i in (0, 1): + cdir = tmp_path / "search" / f"candidate-{i}" + assert not (cdir / "halt" / "item-approval").exists() + [staged] = RepairStore(cdir / "store").list_candidates() + assert staged.state == "rejected" + assert not (cdir / "reward" / "worker" / "keys" / "ed25519.pem").exists() # Production saw one proposal, the winner, and a person approved it. assert result.status == "resumed", result.evidence.reason @@ -163,6 +170,10 @@ def test_no_verified_candidate_means_no_proposal( assert only.wrong_value and only.reward == -0.5 # It proposes once; a second request gets nothing. assert agent.propose(request) is None + # A second search in the same directory starts clean. + with _browser(): + again = search.search(request) + assert again.candidates[0].reward == -0.5 def test_model_bridges_map_labels_to_screen_points(monkeypatch) -> None: From 133feb6c5a1efc179c981795d743efc119808ce6 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 18 Oct 2026 16:24:23 -0400 Subject: [PATCH 3/3] fix(flow): staging search review fixes - Each search runs in its own directory; run paths, stores and episodes are single-use, so a second search no longer fails. - Every staging candidate's approval request is withdrawn and the staging candidate rejected, so nothing on staging can be approved. - The rollout's reward worker key and token are deleted after scoring. - The reward contract checks every field of StagingCopy.record. Co-Authored-By: Claude Opus 5.5 --- .../openadapt_flow/repair/staging_search.py | 23 ++++++++++++------- packages/flow/tests/test_staging_search.py | 8 ++++--- 2 files changed, 20 insertions(+), 11 deletions(-) diff --git a/packages/flow/openadapt_flow/repair/staging_search.py b/packages/flow/openadapt_flow/repair/staging_search.py index c2e125d62..3a4ea4e91 100644 --- a/packages/flow/openadapt_flow/repair/staging_search.py +++ b/packages/flow/openadapt_flow/repair/staging_search.py @@ -32,17 +32,18 @@ import json import re import shutil +import time from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Callable, Mapping, Optional, Protocol, Sequence from openadapt_flow.integrations.harness2 import traces +from openadapt_flow.integrations.harness2.rollout import drop_worker_secrets from openadapt_flow.integrations.harness2.scorer import ( EpisodeScore, Penalties, ReadBackScorer, ) -from openadapt_flow.integrations.harness2.rollout import drop_worker_secrets from openadapt_flow.integrations.harness2.tasks import GuiTask from openadapt_flow.ir import RunReport, Workflow from openadapt_flow.repair.self_repair import ( @@ -258,6 +259,8 @@ def __init__( ) self.max_candidates = max_candidates self.penalties = penalties + #: The directory of the latest search (``/search-