From dfb3fb1ff49db199aae5d55aa22ca434c5a49afa Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 15:26:47 -0400 Subject: [PATCH] feat: opt-in compile-time model annotation (label/risk/param proposals, confirm-don't-trust; runtime stays $0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reviews' 'use the model at compile time, not just repair time' cheap win. A StepAnnotator Protocol proposes step labels, richer risk classifications, and parameter inferences from a demonstration; the model runs ONCE at compile, OFF by default, behind an interface (fake for tests, lazy Anthropic impl). A proposed risk UPGRADE applies (safe direction); a downgrade or consequential param is FLAGGED needs_operator_confirmation, never silently trusted. The runtime/replayer is untouched — zero model calls at replay. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- openadapt_flow/compiler/__init__.py | 14 + openadapt_flow/compiler/annotate.py | 528 ++++++++++++++++++++++++++++ openadapt_flow/compiler/compile.py | 54 ++- tests/test_annotate.py | 380 ++++++++++++++++++++ 4 files changed, 975 insertions(+), 1 deletion(-) create mode 100644 openadapt_flow/compiler/annotate.py create mode 100644 tests/test_annotate.py diff --git a/openadapt_flow/compiler/__init__.py b/openadapt_flow/compiler/__init__.py index 167d61d..a244fa0 100644 --- a/openadapt_flow/compiler/__init__.py +++ b/openadapt_flow/compiler/__init__.py @@ -1,5 +1,13 @@ """Compiler: recording directory -> workflow bundle.""" +from openadapt_flow.compiler.annotate import ( + AnnotationResult, + AnthropicStepAnnotator, + FakeStepAnnotator, + StepAnnotator, + WorkflowProposals, + apply_annotations, +) from openadapt_flow.compiler.codegen import render_workflow_py from openadapt_flow.compiler.compile import compile_recording, lint_param_leakage from openadapt_flow.compiler.effect_mining import ( @@ -13,4 +21,10 @@ "render_workflow_py", "mine_step_effects", "StepEffectMining", + "apply_annotations", + "AnnotationResult", + "AnthropicStepAnnotator", + "FakeStepAnnotator", + "StepAnnotator", + "WorkflowProposals", ] diff --git a/openadapt_flow/compiler/annotate.py b/openadapt_flow/compiler/annotate.py new file mode 100644 index 0000000..41cc56f --- /dev/null +++ b/openadapt_flow/compiler/annotate.py @@ -0,0 +1,528 @@ +"""Opt-in COMPILE-TIME model annotation of a compiled workflow. + +The compiler already labels steps, classifies risk, and infers typed parameters +HEURISTICALLY: intent strings are templated from the click's OCR label +(``compile._`` intent building), risk is keyword-matched +(:mod:`openadapt_flow.risk`, issue #65), and every recorded value is typed as a +bare ``string`` (``ParamSpec`` in ``compile_recording``, Phase 1). Those +heuristics are cheap and model-free, but they miss things a language model would +catch at a glance: a write-shaped control the keyword list doesn't cover +(``"Commit charges"``), a demonstrated value that is really a DATE or an +ENTITY_REF rather than a constant string, a terse ``click at (640, 424)`` intent +that a model could render as ``"submit the triage encounter"``. + +This module adds a model-backed pass that runs at COMPILE time and is OFF by +default. Compiling is a one-time step, so a single model call there buys runtime +robustness at ZERO per-run cost -- the replayer (``runtime.replayer``) never +calls a model on this path, and nothing here is read at replay (the annotations +live in a bundle sidecar, ``annotations.json``, never in ``workflow.json``'s hot +fields beyond the risk upgrades described below). The runtime ``$0`` guarantee is +therefore preserved unchanged. + +Confirm-don't-trust (mirrors #74 disambiguation and #75 effect-mining) +--------------------------------------------------------------------- +A model PROPOSES; the compiler never blindly trusts it. Proposals are applied +only in the SAFE direction and are otherwise FLAGGED for an operator, exactly as +the identity ladder refuses rather than guesses: + +- A **risk UPGRADE** (``reversible`` -> ``irreversible``) is APPLIED. It only ever + ARMS a safeguard (the irreversible-step refusals in ``runtime.Replayer``); the + cost of a false upgrade is availability, never a silent wrong write. +- A **risk DOWNGRADE** (``irreversible`` -> ``reversible``) is NEVER applied -- it + would DISARM a safeguard. It is flagged ``needs_operator_confirmation`` and the + heuristic risk stands. +- A **label** proposal is advisory (a human-readable intent for review); it never + changes replay behaviour, so it is always attached, never gated. +- A **parameter** proposal that only ENRICHES the TYPE of an already-declared + parameter (``string`` -> ``date`` / ``enum`` / ``entity_ref``) is metadata-only + and APPLIED. A CONSEQUENTIAL parameter inference -- one that would turn a + demonstrated constant into a new run-varying parameter, changing what the + workflow does -- is FLAGGED, never applied. + +The model call sits behind the :class:`StepAnnotator` ``Protocol``. Tests use +:class:`FakeStepAnnotator` (deterministic, no network). The real +:class:`AnthropicStepAnnotator` calls Anthropic ONLY at compile and ONLY when the +caller opts in (``compile_recording(..., annotate=True)`` with no annotator, or +by passing an ``AnthropicStepAnnotator`` explicitly); ``anthropic`` and the API +key are imported/resolved LAZILY, so nothing here needs a key unless the real +annotator actually runs. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any, Optional, Protocol, runtime_checkable + +from pydantic import BaseModel, Field + +from openadapt_flow.ir import ParamKind, ParamSpec, Step, Workflow + +logger = logging.getLogger(__name__) + +Risk = str # "reversible" | "irreversible" (mirrors ir.Step.risk) + + +# -- proposals (what a StepAnnotator returns) -------------------------------- + + +class LabelProposal(BaseModel): + """A richer human-readable intent for a step (advisory only).""" + + label: str = Field(description="Proposed human-readable intent") + rationale: str = "" + + +class RiskProposal(BaseModel): + """A proposed refinement to a step's heuristic risk classification.""" + + proposed_risk: Risk = Field(description="'reversible' or 'irreversible'") + rationale: str = "" + + +class ParamProposal(BaseModel): + """A proposed typed parameter inference for a demonstrated value. + + ``consequential`` is True when applying the proposal would change what the + workflow DOES -- i.e. it would turn a demonstrated CONSTANT into a new + run-varying parameter. A non-consequential proposal only ENRICHES the TYPE of + a parameter the demonstration already declared (Phase-1 ``string`` -> + ``date`` / ``enum`` / ``entity_ref``), which is metadata and safe to apply. + """ + + name: str = Field(description="Parameter name this value maps to") + type: ParamKind = ParamKind.STRING + example: Optional[str] = None + choices: list[str] = Field(default_factory=list) + consequential: bool = Field( + default=False, + description=( + "True when applying would make a demonstrated constant vary per run" + " (a behaviour change -> FLAG, never apply). False for a pure" + " type-enrichment of an existing parameter (safe -> apply)." + ), + ) + rationale: str = "" + + +class StepAnnotation(BaseModel): + """A model's proposals for ONE compiled step (all fields optional).""" + + step_id: str + label: Optional[LabelProposal] = None + risk: Optional[RiskProposal] = None + params: list[ParamProposal] = Field(default_factory=list) + + +class WorkflowProposals(BaseModel): + """The raw proposals a :class:`StepAnnotator` returns -- nothing applied.""" + + steps: list[StepAnnotation] = Field(default_factory=list) + + def for_step(self, step_id: str) -> Optional[StepAnnotation]: + for sa in self.steps: + if sa.step_id == step_id: + return sa + return None + + +# -- the Protocol + a fake + the real (Anthropic) annotator ------------------ + + +@runtime_checkable +class StepAnnotator(Protocol): + """Given a compiled workflow (+ optional captured state), propose label, + risk, and parameter refinements. Implementations MUST NOT mutate the + workflow -- they only PROPOSE; :func:`apply_annotations` decides what is + safe to apply. The real implementation calls a model; the fake does not.""" + + def annotate( + self, workflow: Workflow, *, captured_state: Optional[dict] = None + ) -> WorkflowProposals: ... + + +class FakeStepAnnotator: + """A deterministic, network-free :class:`StepAnnotator` for tests. + + It returns exactly the ``WorkflowProposals`` it was constructed with (keyed + per step id), so a test scripts the "model's" output and asserts how + :func:`apply_annotations` applies vs flags it. No model, no key, no network. + """ + + def __init__(self, proposals: Optional[WorkflowProposals] = None) -> None: + self._proposals = proposals or WorkflowProposals() + + def annotate( + self, workflow: Workflow, *, captured_state: Optional[dict] = None + ) -> WorkflowProposals: + # Return only proposals whose step id actually exists in the workflow, + # so a scripted proposal for a removed step is silently ignored (the + # real model is prompted from the same workflow and cannot invent ids). + ids = {s.id for s in workflow.steps} + return WorkflowProposals( + steps=[sa for sa in self._proposals.steps if sa.step_id in ids] + ) + + +class AnthropicStepAnnotator: + """The real, COMPILE-TIME-ONLY, opt-in :class:`StepAnnotator`. + + Calls Anthropic once per compile to propose richer labels, risk refinements, + and typed-parameter inferences over the compiled workflow's step evidence + (intent, action, OCR label, current risk, declared params). ``anthropic`` is + imported LAZILY and the API key is resolved LAZILY (env + ``ANTHROPIC_API_KEY`` or ``~/.anthropic/api_key`` via the existing + ``benchmark.agent_baseline.load_api_key``), so importing this module needs no + key and no network -- only :meth:`annotate` does. NEVER called at replay. + """ + + def __init__( + self, + *, + model: Optional[str] = None, + client: Any = None, + max_tokens: int = 4096, + ) -> None: + self._model = model + self._client = client + self._max_tokens = max_tokens + + def annotate( + self, workflow: Workflow, *, captured_state: Optional[dict] = None + ) -> WorkflowProposals: + client = self._client + model = self._model + if client is None or model is None: + # Lazy: resolve the SDK, key, and default model only when actually + # invoked (keeps the module import-light and key-free). The default + # model reuses the repo's single Anthropic-model constant so a bump + # there follows here too. + from openadapt_flow.benchmark.agent_baseline import ( + MODEL as DEFAULT_MODEL, + load_api_key, + ) + + if model is None: + model = DEFAULT_MODEL + if client is None: + import anthropic + + client = anthropic.Anthropic(api_key=load_api_key()) + + prompt = _build_prompt(workflow, captured_state) + response = client.messages.create( + model=model, + max_tokens=self._max_tokens, + system=_SYSTEM_PROMPT, + messages=[{"role": "user", "content": prompt}], + ) + text = "".join( + block.text for block in response.content if block.type == "text" + ) + return _parse_proposals(text, workflow) + + +# -- prompt + parsing for the real annotator --------------------------------- + +_SYSTEM_PROMPT = ( + "You label steps of a compiled desktop/web automation. You PROPOSE " + "refinements only; a downstream safety layer decides what to apply. Reply " + "with a single JSON object matching the requested schema and nothing else. " + "Bias risk toward 'irreversible' for any step that writes, submits, sends, " + "pays, deletes, or otherwise commits a consequential change." +) + + +def _step_evidence(step: Step) -> dict: + label = step.anchor.ocr_text if step.anchor is not None else None + return { + "id": step.id, + "action": step.action.value, + "intent": step.intent, + "ocr_label": label, + "current_risk": step.risk, + "param": step.param, + "text": None if step.secret else step.text, + } + + +def _build_prompt( + workflow: Workflow, captured_state: Optional[dict] +) -> str: + payload = { + "workflow_name": workflow.name, + "declared_params": { + name: spec.type.value for name, spec in workflow.param_specs.items() + }, + "steps": [_step_evidence(s) for s in workflow.steps], + "captured_state": captured_state or {}, + } + schema_hint = { + "steps": [ + { + "step_id": "step_000", + "label": {"label": "human-readable intent", "rationale": "..."}, + "risk": { + "proposed_risk": "reversible|irreversible", + "rationale": "...", + }, + "params": [ + { + "name": "param_name", + "type": "string|date|enum|number|entity_ref", + "example": "value", + "choices": [], + "consequential": False, + "rationale": "...", + } + ], + } + ] + } + return ( + "Compiled workflow:\n" + + json.dumps(payload, indent=2) + + "\n\nReturn a JSON object with this shape (omit fields you have no " + "proposal for; only include steps you want to annotate):\n" + + json.dumps(schema_hint, indent=2) + ) + + +def _parse_proposals(text: str, workflow: Workflow) -> WorkflowProposals: + """Parse the model's JSON reply into ``WorkflowProposals``, defensively. + + The reply may be wrapped in prose or a ``json`` code fence; extract the first + balanced JSON object. Unknown step ids and malformed entries are dropped + (fail-safe: a bad model reply yields NO proposals, never a crash or a + fabricated annotation). + """ + ids = {s.id for s in workflow.steps} + obj = _extract_json_object(text) + if not isinstance(obj, dict): + logger.warning("annotator: no JSON object in model reply; ignoring") + return WorkflowProposals() + steps: list[StepAnnotation] = [] + for raw in obj.get("steps", []) or []: + if not isinstance(raw, dict) or raw.get("step_id") not in ids: + continue + try: + steps.append(StepAnnotation.model_validate(raw)) + except Exception as exc: # noqa: BLE001 - one bad step must not crash all + logger.warning("annotator: dropping malformed step proposal: %s", exc) + return WorkflowProposals(steps=steps) + + +def _extract_json_object(text: str) -> Any: + text = text.strip() + start = text.find("{") + if start == -1: + return None + depth = 0 + for i in range(start, len(text)): + c = text[i] + if c == "{": + depth += 1 + elif c == "}": + depth -= 1 + if depth == 0: + try: + return json.loads(text[start : i + 1]) + except json.JSONDecodeError: + return None + return None + + +# -- application: confirm-don't-trust ---------------------------------------- + + +class AppliedAnnotation(BaseModel): + """A proposal that was APPLIED (safe direction).""" + + kind: str # "label" | "risk_upgrade" | "param_type" + step_id: str + detail: str + + +class FlaggedAnnotation(BaseModel): + """A proposal that was NOT applied and needs operator confirmation. + + Mirrors ``effect.needs_operator_confirmation`` (#75) and the disambiguation + ``unresolved_consequential`` posture (#74): a model-proposed weakening of a + safeguard, or a consequential behaviour change, is surfaced for a human -- + never silently trusted. + """ + + kind: str # "risk_downgrade" | "consequential_param" + step_id: str + detail: str + needs_operator_confirmation: bool = True + + +class AnnotationResult(BaseModel): + """Outcome of applying a :class:`StepAnnotator`'s proposals to a workflow. + + ``workflow`` is the (copied) workflow with SAFE proposals applied (risk + upgrades in ``step.risk``, parameter type enrichments in ``param_specs``). + ``labels`` maps step id -> proposed human-readable intent (advisory). + ``applied`` / ``flagged`` are the audit trail. ``proposals`` is the raw model + output for the record. + """ + + workflow: Workflow + proposals: WorkflowProposals = Field(default_factory=WorkflowProposals) + labels: dict[str, str] = Field(default_factory=dict) + applied: list[AppliedAnnotation] = Field(default_factory=list) + flagged: list[FlaggedAnnotation] = Field(default_factory=list) + + @property + def clean(self) -> bool: + """True iff nothing needs operator confirmation (mirrors #74 + ``certified``): no safeguard was proposed-weakened and no consequential + parameter inference is outstanding.""" + return not self.flagged + + def render(self) -> str: + lines = [ + f"Annotation: {len(self.labels)} label(s), " + f"{len(self.applied)} applied, {len(self.flagged)} flagged " + "(need operator confirmation)." + ] + for a in self.applied: + lines.append(f" [applied] {a.step_id}: {a.kind} -- {a.detail}") + for f in self.flagged: + lines.append(f" [FLAGGED] {f.step_id}: {f.kind} -- {f.detail}") + verdict = "CLEAN" if self.clean else "NEEDS OPERATOR CONFIRMATION" + lines.append(f"Verdict: {verdict}") + return "\n".join(lines) + + +def apply_annotations( + workflow: Workflow, + annotator: StepAnnotator, + *, + captured_state: Optional[dict] = None, +) -> AnnotationResult: + """Run ``annotator`` over a COPY of ``workflow`` and apply its proposals + with the confirm-don't-trust asymmetry (see the module docstring). + + Pure with respect to ``workflow`` (it is deep-copied first). The model call, + if any, happens inside ``annotator.annotate`` -- this function itself makes no + network calls and is safe to unit-test with :class:`FakeStepAnnotator`. + + Returns: + An :class:`AnnotationResult` carrying the annotated (copied) workflow and + the full applied/flagged audit trail. + """ + resolved = workflow.model_copy(deep=True) + proposals = annotator.annotate(resolved, captured_state=captured_state) + result = AnnotationResult(workflow=resolved, proposals=proposals) + by_id = {s.id: s for s in resolved.steps} + + for sa in proposals.steps: + step = by_id.get(sa.step_id) + if step is None: + continue # unknown step id: ignore (fail-safe) + + # LABEL -- advisory, never changes replay behaviour: always attach. + if sa.label is not None and sa.label.label.strip(): + result.labels[step.id] = sa.label.label + result.applied.append( + AppliedAnnotation( + kind="label", step_id=step.id, detail=sa.label.label + ) + ) + + # RISK -- upgrade applied (arms a safeguard), downgrade flagged. + if sa.risk is not None: + _apply_risk(step, sa.risk, result) + + # PARAMS -- type-enrichment applied, consequential flagged. + for p in sa.params: + _apply_param(resolved, step, p, result) + + return result + + +def _apply_risk( + step: Step, proposal: RiskProposal, result: AnnotationResult +) -> None: + proposed = proposal.proposed_risk + if proposed not in ("reversible", "irreversible"): + logger.warning( + "annotator: ignoring invalid risk %r for %s", proposed, step.id + ) + return + current = step.risk + if proposed == current: + return # agrees with the heuristic: nothing to do + if proposed == "irreversible": + # UPGRADE: safe direction -- only ever arms a safeguard. Apply. + step.risk = "irreversible" + detail = "reversible -> irreversible" + if proposal.rationale: + detail += f" ({proposal.rationale})" + result.applied.append( + AppliedAnnotation( + kind="risk_upgrade", step_id=step.id, detail=detail + ) + ) + else: + # DOWNGRADE: would DISARM a safeguard. Never apply silently -- flag. + detail = ( + "model proposed irreversible -> reversible; heuristic risk " + "'irreversible' KEPT" + ) + if proposal.rationale: + detail += f" (model: {proposal.rationale})" + result.flagged.append( + FlaggedAnnotation( + kind="risk_downgrade", step_id=step.id, detail=detail + ) + ) + + +def _apply_param( + resolved: Workflow, + step: Step, + proposal: ParamProposal, + result: AnnotationResult, +) -> None: + name = proposal.name + existing = resolved.param_specs.get(name) + # A proposal is safe to apply ONLY when it is a pure TYPE enrichment of a + # parameter the demonstration already declared. Anything else -- explicitly + # flagged consequential, or naming a value that is not yet a parameter -- + # would change what the workflow does, so it is flagged, never applied. + if proposal.consequential or existing is None: + detail = ( + f"model proposed making {name!r} a {proposal.type.value} parameter" + ) + if existing is None: + detail += " (would make a demonstrated constant vary per run)" + if proposal.rationale: + detail += f"; {proposal.rationale}" + result.flagged.append( + FlaggedAnnotation( + kind="consequential_param", step_id=step.id, detail=detail + ) + ) + return + if existing.type == proposal.type and not proposal.choices: + return # no change + # Type-only enrichment: keep the recorded name/example/required, refine type + # (and enum choices when given). Metadata only -- the value still comes from + # params[name] at replay exactly as before. + resolved.param_specs[name] = ParamSpec( + name=existing.name, + type=proposal.type, + example=existing.example, + required=existing.required, + choices=proposal.choices or existing.choices, + ) + result.applied.append( + AppliedAnnotation( + kind="param_type", + step_id=step.id, + detail=f"{name}: {existing.type.value} -> {proposal.type.value}", + ) + ) diff --git a/openadapt_flow/compiler/compile.py b/openadapt_flow/compiler/compile.py index fb3b1ca..fa689d0 100644 --- a/openadapt_flow/compiler/compile.py +++ b/openadapt_flow/compiler/compile.py @@ -14,7 +14,10 @@ import math from datetime import date, datetime from pathlib import Path -from typing import Optional +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from openadapt_flow.compiler.annotate import StepAnnotator import cv2 import numpy as np @@ -727,6 +730,8 @@ def compile_recording( name: str, risk_overrides: Optional[dict[str, str]] = None, mine_effects: bool = False, + annotate: bool = False, + annotator: Optional["StepAnnotator"] = None, ) -> Workflow: """Compile a recording directory into a workflow bundle. @@ -777,6 +782,24 @@ def compile_recording( or nothing (with an honest "no verifiable effect derivable" log). Default False keeps the bundle byte-identical to before; even when True a bundle is unchanged wherever mining derives nothing. + annotate: Opt-in COMPILE-TIME model annotation + (``compiler.annotate``). When True, a :class:`StepAnnotator` proposes + richer step LABELS, RISK refinements, and typed PARAMETER inferences + over the compiled workflow, and its proposals are applied with the + confirm-don't-trust asymmetry: a risk UPGRADE (reversible -> + irreversible) and a pure parameter TYPE enrichment are applied; a + risk DOWNGRADE or a consequential parameter inference is FLAGGED for + an operator, never applied. Applied risk upgrades land in + ``step.risk``; the full proposal/flag audit trail is written to a + bundle sidecar ``annotations.json``. This is COMPILE-TIME ONLY -- the + replayer never reads it and makes ZERO model calls. Default False + keeps the bundle byte-identical to today (heuristic only, no model, + no key needed). + annotator: The :class:`StepAnnotator` to use when ``annotate`` is True. + None defaults to :class:`~openadapt_flow.compiler.annotate. + AnthropicStepAnnotator` (the real model, resolved lazily -- it needs + an API key only when it actually runs). Tests pass a network-free + fake. Ignored when ``annotate`` is False. Returns: The compiled :class:`Workflow` (also saved to the bundle). @@ -1159,6 +1182,35 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: + "\n ".join(violations) ) + # Opt-in COMPILE-TIME model annotation (off by default). Runs LAST, over the + # fully-built workflow. Applies model proposals with the confirm-don't-trust + # asymmetry (risk upgrade / type-enrichment applied; downgrade / + # consequential param FLAGGED), and writes the audit trail to a bundle + # sidecar. COMPILE-TIME ONLY: the replayer never reads this and makes ZERO + # model calls. Off -> `workflow` and the bundle are byte-identical to today. + if annotate: + from openadapt_flow.compiler.annotate import ( + AnthropicStepAnnotator, + apply_annotations, + ) + + ann = annotator if annotator is not None else AnthropicStepAnnotator() + result = apply_annotations(workflow, ann) + workflow = result.workflow + (bundle / "annotations.json").write_text( + result.model_dump_json(indent=2, exclude={"workflow"}) + ) + for flag in result.flagged: + logger.warning( + "annotation FLAGGED (needs operator confirmation) %s: %s", + flag.step_id, + flag.detail, + ) + for applied in result.applied: + logger.info( + "annotation applied %s: %s", applied.step_id, applied.detail + ) + workflow.save(bundle) (bundle / "workflow.py").write_text(render_workflow_py(workflow)) return workflow diff --git a/tests/test_annotate.py b/tests/test_annotate.py new file mode 100644 index 0000000..46c421f --- /dev/null +++ b/tests/test_annotate.py @@ -0,0 +1,380 @@ +"""Opt-in COMPILE-TIME model annotation (``compiler.annotate``). + +A model PROPOSES richer labels, risk refinements, and typed-parameter +inferences at COMPILE time; the compiler applies them with a confirm-don't-trust +asymmetry (safe upgrades applied, weakenings / consequential changes FLAGGED). +The model call lives behind the ``StepAnnotator`` Protocol; every test here uses +the deterministic ``FakeStepAnnotator`` (or a stub client) -- ZERO network, ZERO +model calls, no API key. The runtime (replayer) must make ZERO model calls too; +the final test asserts exactly that against an annotated bundle. +""" + +from __future__ import annotations + +import json + +import pytest + +from openadapt_flow.compiler import compile_recording +from openadapt_flow.compiler.annotate import ( + AnthropicStepAnnotator, + FakeStepAnnotator, + LabelProposal, + ParamProposal, + RiskProposal, + StepAnnotation, + WorkflowProposals, + apply_annotations, +) +from openadapt_flow.ir import ( + ActionKind, + Anchor, + ParamKind, + ParamSpec, + Postcondition, + PostconditionKind, + Step, + Workflow, +) +from openadapt_flow.runtime.replayer import Replayer + +# Reuse the compile-recording synthetic builder + drawing helpers. +from test_compiler import ( + _write_recording, + blank, + draw_button, + draw_text, +) + +# Reuse the faked backend/vision + step builders (ZERO model, no Playwright). +from test_replayer import FakeBackend, FakeVision, Match, click_step, make_png + + +# -- helpers ----------------------------------------------------------------- + + +def _wf(*steps: Step, params=None, param_specs=None) -> Workflow: + return Workflow( + name="wf", + params=params or {}, + param_specs=param_specs or {}, + steps=list(steps), + ) + + +def _click(step_id="step_000", *, risk="reversible", ocr="Open") -> Step: + return Step( + id=step_id, + intent=f"click '{ocr}'", + action=ActionKind.CLICK, + anchor=Anchor( + template=f"templates/{step_id}.png", + region=(10, 10, 50, 20), + click_point=(30, 20), + ocr_text=ocr, + ), + risk=risk, + ) + + +# -- proposals attach: richer labels, param types, risk proposals ------------ + + +def test_label_proposal_attaches_as_advisory_intent(): + wf = _wf(_click("step_000", ocr="Open")) + ann = FakeStepAnnotator( + WorkflowProposals( + steps=[ + StepAnnotation( + step_id="step_000", + label=LabelProposal(label="open the triage encounter"), + ) + ] + ) + ) + result = apply_annotations(wf, ann) + assert result.labels["step_000"] == "open the triage encounter" + assert any(a.kind == "label" for a in result.applied) + # A label never changes replay behaviour, so nothing is flagged. + assert result.clean is True + # The step's own intent is untouched (label is advisory, in the sidecar). + assert result.workflow.steps[0].intent == "click 'Open'" + + +@pytest.mark.parametrize( + "kind", [ParamKind.DATE, ParamKind.ENUM, ParamKind.ENTITY_REF] +) +def test_param_type_enrichment_applies_richer_than_phase1_string(kind): + # Phase 1 types every recorded value as a bare string; the model enriches it. + wf = _wf( + Step(id="step_000", intent="type ", action=ActionKind.TYPE, + param="dob", text="1980-02-03"), + params={"dob": "1980-02-03"}, + param_specs={"dob": ParamSpec(name="dob", type=ParamKind.STRING, + example="1980-02-03")}, + ) + choices = ["Triage", "Consult"] if kind is ParamKind.ENUM else [] + ann = FakeStepAnnotator( + WorkflowProposals(steps=[StepAnnotation( + step_id="step_000", + params=[ParamProposal(name="dob", type=kind, choices=choices, + consequential=False)], + )]) + ) + result = apply_annotations(wf, ann) + assert result.workflow.param_specs["dob"].type is kind + if kind is ParamKind.ENUM: + assert result.workflow.param_specs["dob"].choices == choices + # example/required carried over unchanged; only the TYPE was enriched. + assert result.workflow.param_specs["dob"].example == "1980-02-03" + assert any(a.kind == "param_type" for a in result.applied) + assert result.clean is True + + +def test_risk_upgrade_the_keyword_heuristic_missed_applies(): + # "Commit charges" is write-shaped but not in the keyword list -> heuristic + # leaves it reversible; the model upgrades it (safe direction) and it sticks. + wf = _wf(_click("step_000", risk="reversible", ocr="Commit charges")) + ann = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="step_000", + risk=RiskProposal(proposed_risk="irreversible", + rationale="commits a billing write"), + )])) + result = apply_annotations(wf, ann) + assert result.workflow.steps[0].risk == "irreversible" + assert any(a.kind == "risk_upgrade" for a in result.applied) + assert result.clean is True # arming a safeguard needs no confirmation + + +# -- confirm-don't-trust: downgrade / consequential are FLAGGED, not applied -- + + +def test_risk_downgrade_is_flagged_not_applied(): + wf = _wf(_click("step_000", risk="irreversible", ocr="Delete")) + ann = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="step_000", + risk=RiskProposal(proposed_risk="reversible", + rationale="thinks it is a soft delete"), + )])) + result = apply_annotations(wf, ann) + # The safeguard is NEVER silently weakened. + assert result.workflow.steps[0].risk == "irreversible" + assert not any(a.kind == "risk_upgrade" for a in result.applied) + flags = [f for f in result.flagged if f.kind == "risk_downgrade"] + assert len(flags) == 1 + assert flags[0].needs_operator_confirmation is True + assert result.clean is False + + +def test_consequential_param_is_flagged_not_applied(): + # Turning a demonstrated constant into a run-varying parameter changes what + # the workflow DOES -> flagged, never applied. + wf = _wf(Step(id="step_000", intent="type 'Acme Corp'", + action=ActionKind.TYPE, text="Acme Corp")) + ann = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="step_000", + params=[ParamProposal(name="company", type=ParamKind.ENTITY_REF, + example="Acme Corp", consequential=True)], + )])) + result = apply_annotations(wf, ann) + assert "company" not in result.workflow.param_specs + assert result.workflow.params == {} + flags = [f for f in result.flagged if f.kind == "consequential_param"] + assert len(flags) == 1 + assert flags[0].needs_operator_confirmation is True + assert result.clean is False + + +def test_param_naming_a_non_parameter_value_is_flagged_not_applied(): + # Even non-`consequential` flagged, a proposal for a value that is NOT an + # already-declared parameter cannot be a safe type-enrichment -> flag. + wf = _wf(Step(id="step_000", intent="type 'x'", action=ActionKind.TYPE, + text="x")) + ann = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="step_000", + params=[ParamProposal(name="new_param", type=ParamKind.DATE, + consequential=False)], + )])) + result = apply_annotations(wf, ann) + assert "new_param" not in result.workflow.param_specs + assert any(f.kind == "consequential_param" for f in result.flagged) + + +# -- purity / robustness ----------------------------------------------------- + + +def test_original_workflow_is_never_mutated(): + wf = _wf(_click("step_000", risk="reversible", ocr="Save")) + ann = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="step_000", + risk=RiskProposal(proposed_risk="irreversible"), + )])) + apply_annotations(wf, ann) + # apply_annotations deep-copies; the caller's workflow is untouched. + assert wf.steps[0].risk == "reversible" + + +def test_unknown_step_id_is_ignored(): + wf = _wf(_click("step_000")) + ann = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="does_not_exist", + risk=RiskProposal(proposed_risk="irreversible"), + )])) + result = apply_annotations(wf, ann) + assert result.applied == [] + assert result.flagged == [] + + +def test_empty_annotator_leaves_workflow_unchanged(): + wf = _wf(_click("step_000", risk="reversible")) + result = apply_annotations(wf, FakeStepAnnotator()) + assert result.workflow.model_dump() == wf.model_dump() + assert result.clean is True + + +# -- compile.py opt-in hook -------------------------------------------------- + + +def _annotatable_recording(tmp_path): + """A 2-step recording: a benign nav click + a write-shaped 'Save' click.""" + before = blank() + draw_button(before, 40, 40, 120, 40, "Home") + draw_button(before, 560, 400, 160, 48, "Save") + mid = before.copy() + draw_text(mid, 120, 244, "Patient opened") + after = mid.copy() + draw_text(after, 380, 560, "Saved") + events = [ + {"i": 0, "kind": "click", "x": 100, "y": 60, "t": 1.0}, + {"i": 1, "kind": "click", "x": 640, "y": 424, "t": 2.0}, + ] + recording = tmp_path / "rec" + _write_recording(recording, events, {0: (before, mid), 1: (mid, after)}) + return recording + + +def test_default_off_is_byte_identical_and_writes_no_sidecar(tmp_path): + recording = _annotatable_recording(tmp_path) + b_off = tmp_path / "b_off" + b_default = tmp_path / "b_default" + # annotate defaults to False; passing it explicitly must be identical. + off = compile_recording(recording, b_off, name="wf", annotate=False) + default = compile_recording(recording, b_default, name="wf") + assert not (b_off / "annotations.json").exists() + assert not (b_default / "annotations.json").exists() + # Byte-identical modulo the always-varying created_at timestamp: annotate + # off changes nothing about compilation (heuristic-only, no model). + a = off.model_dump() + b = default.model_dump() + a.pop("created_at"), b.pop("created_at") + assert a == b + + +def test_annotate_on_attaches_proposals_and_applies_safe_upgrade(tmp_path): + recording = _annotatable_recording(tmp_path) + bundle = tmp_path / "bundle" + # A fake that (a) proposes a richer label for step_000, (b) upgrades the + # nav click to irreversible (safe), (c) proposes a downgrade for the Save + # click (flagged, not applied). + fake = FakeStepAnnotator(WorkflowProposals(steps=[ + StepAnnotation(step_id="step_000", + label=LabelProposal(label="open the patient chart"), + risk=RiskProposal(proposed_risk="irreversible")), + StepAnnotation(step_id="step_001", + risk=RiskProposal(proposed_risk="reversible")), + ])) + wf = compile_recording(recording, bundle, name="wf", annotate=True, + annotator=fake) + # Safe upgrade landed in the saved workflow. + assert wf.steps[0].risk == "irreversible" + # The Save click's heuristic risk (irreversible) was NOT weakened. + assert wf.steps[1].risk == "irreversible" + # Sidecar carries the full audit trail. + sidecar = json.loads((bundle / "annotations.json").read_text()) + assert sidecar["labels"]["step_000"] == "open the patient chart" + assert any(a["kind"] == "risk_upgrade" for a in sidecar["applied"]) + downgrades = [f for f in sidecar["flagged"] if f["kind"] == "risk_downgrade"] + assert len(downgrades) == 1 + assert downgrades[0]["needs_operator_confirmation"] is True + # The saved workflow.json reflects the applied upgrade on disk. + saved = json.loads((bundle / "workflow.json").read_text()) + assert saved["steps"][0]["risk"] == "irreversible" + + +# -- the real annotator parses without ANY network --------------------------- + + +class _StubMessages: + def __init__(self, text): + self._text = text + + def create(self, **kwargs): # mimics anthropic client.messages.create + block = type("B", (), {"type": "text", "text": self._text})() + return type("R", (), {"content": [block]})() + + +class _StubClient: + def __init__(self, text): + self.messages = _StubMessages(text) + + +def test_anthropic_annotator_parses_json_with_a_stub_client(): + # Exercises the REAL annotator's prompt + parse path with a stub client -- + # NO network, NO key, NO anthropic import. Proves the wiring, not the model. + wf = _wf(_click("step_000", ocr="Pay invoice")) + reply = json.dumps({"steps": [{ + "step_id": "step_000", + "label": {"label": "pay the outstanding invoice"}, + "risk": {"proposed_risk": "irreversible", "rationale": "money moves"}, + }]}) + annotator = AnthropicStepAnnotator(model="stub-model", + client=_StubClient(reply)) + proposals = annotator.annotate(wf) + sa = proposals.for_step("step_000") + assert sa is not None and sa.risk.proposed_risk == "irreversible" + # And it flows through apply with the same confirm-don't-trust rules. + result = apply_annotations(wf, annotator) + assert result.workflow.steps[0].risk == "irreversible" + + +def test_anthropic_annotator_ignores_a_junk_reply_fail_safe(): + wf = _wf(_click("step_000")) + annotator = AnthropicStepAnnotator(model="stub", + client=_StubClient("sorry, no JSON here")) + proposals = annotator.annotate(wf) + assert proposals.steps == [] + + +# -- the RUNTIME stays $0: an annotated bundle replays with ZERO model calls -- + + +def test_runtime_replay_of_annotated_bundle_makes_zero_model_calls(tmp_path): + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "btn.png").write_bytes(make_png((50, 20))) + run_dir = tmp_path / "run" + + wf = Workflow(name="wf", steps=[ + click_step(expect=[Postcondition( + kind=PostconditionKind.TEXT_PRESENT, text="Saved", timeout_s=0.2)]), + Step(id="s2", intent="type note", action=ActionKind.TYPE, param="note"), + ]) + # Annotate the compiled workflow (label-only: behaviour-preserving) and + # replay the ANNOTATED workflow -- the replayer must never touch a model. + fake = FakeStepAnnotator(WorkflowProposals(steps=[StepAnnotation( + step_id="s1", label=LabelProposal(label="save the record"))])) + annotated = apply_annotations(wf, fake).workflow + + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95)] + vision.text_results = { + "Saved": Match(point=(50, 10), region=(30, 5, 40, 10), confidence=0.9)} + backend = FakeBackend() + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + annotated, params={"note": "hello"}, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + assert report.model_calls == 0 + assert report.est_model_cost_usd == 0.0 + assert report.rung_counts.get("grounder", 0) == 0 # no VLM rung either