From 1f8674f78e0d424cc058bf30beddf122c58bff4d Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 8 Aug 2026 17:56:07 +0200 Subject: [PATCH 1/3] fix: bind Teach corrections to halted runs --- engine/controller.py | 16 ++- engine/dispatch.py | 194 ++++++++++++++++++++++++----- src/screens/Teach.tsx | 61 ++++++++- tests/test_engine/test_dispatch.py | 135 +++++++++++++++++++- 4 files changed, 364 insertions(+), 42 deletions(-) diff --git a/engine/controller.py b/engine/controller.py index 3d07433..b758478 100644 --- a/engine/controller.py +++ b/engine/controller.py @@ -123,12 +123,20 @@ def current_capture_id(self) -> str | None: """ID of the current capture session, or None if idle.""" return self._current_capture_id - def start(self, quality: str | None = None, task_description: str = "") -> str: + def start( + self, + quality: str | None = None, + task_description: str = "", + *, + metadata: dict[str, object] | None = None, + ) -> str: """Start a new recording session. Args: quality: Override the default quality preset. task_description: Optional description of the recording task. + metadata: Optional Desktop-owned metadata that binds this capture + to a governed local operation. The recorder does not consume it. Returns: The capture ID for the new session. @@ -155,6 +163,12 @@ def start(self, quality: str | None = None, task_description: str = "") -> str: "task_description": task_description, "quality": quality or self.quality, } + if metadata: + reserved = set(meta) & set(metadata) + if reserved: + names = ", ".join(sorted(reserved)) + raise ValueError(f"Recording metadata cannot replace reserved fields: {names}") + meta.update(metadata) (capture_dir / "meta.json").write_text(json.dumps(meta, indent=2)) state_path = capture_dir / "state.json" diff --git a/engine/dispatch.py b/engine/dispatch.py index d1f79a7..9bcc8cf 100644 --- a/engine/dispatch.py +++ b/engine/dispatch.py @@ -59,6 +59,8 @@ } ) _TEACHABLE_RUN_OUTCOMES = frozenset({"HALTED", "halt"}) +_TEACH_CORRECTION_SCHEMA = "openadapt.desktop-teach-correction/v1" +_TEACH_CORRECTION_META_KEY = "teach_correction" def _noop_emit(event: str, data: dict) -> None: @@ -351,6 +353,22 @@ def start_recording(self, **params: Any) -> dict: self.emit("recording_error", {"error": message}) raise ValueError(message) from None + task = str(params.get("purpose") or params.get("task") or params.get("name") or "") + recording_metadata: dict[str, object] | None = None + if task == "teach_fix": + workflow_id = str(params.get("workflow_id") or "") + run_id = str(params.get("run_id") or "") + run, error = self._teachable_run(workflow_id, expected_run_id=run_id) + if run is None: + raise ValueError(error or "The halted run is not available for Teach") + recording_metadata = { + _TEACH_CORRECTION_META_KEY: { + "schema": _TEACH_CORRECTION_SCHEMA, + "workflow_id": workflow_id, + "run_id": run_id, + } + } + needs_native_input = target is None or target.backend != "web" if ( sys.platform == "darwin" @@ -367,7 +385,6 @@ def start_recording(self, **params: Any) -> dict: ) self.emit("recording_error", {"error": message}) raise PermissionError(message) - task = str(params.get("purpose") or params.get("task") or params.get("name") or "") if target is not None: if target.backend == "web": ensure_browser = getattr( @@ -463,7 +480,10 @@ def start_recording(self, **params: Any) -> dict: self.emit("status_update", self._status_dict(controller)) return {"capture_id": capture_id, "recording": True} - capture_id = controller.start(task_description=str(task)) + capture_id = controller.start( + task_description=str(task), + metadata=recording_metadata, + ) self.services.audit.log("recording_started", capture_id=capture_id) self.emit("recording_started", {"capture_id": capture_id}) self.emit("status_update", self._status_dict(controller)) @@ -486,6 +506,10 @@ def stop_recording(self, **params: Any) -> dict: self.emit("recording_stopped", metadata) self.emit("status_update", self._status_dict(controller)) stopped = {"capture_id": metadata.get("id"), **metadata} + correction = self._stopped_teach_correction(stopped) + if correction is not None: + stopped["correction"] = correction + return stopped stopped["compile"] = self._compile_registered_capture( str(stopped["capture_id"]), automatic=True, @@ -1477,52 +1501,160 @@ def _workflow_step_count(self, workflow_id: str | None) -> int | None: def teach_fix(self, **params: Any) -> dict: """Teach a fix for a halted workflow via ``openadapt-flow teach``.""" - workflow_id = params.get("workflow_id") + workflow_id = str(params.get("workflow_id") or "") bundle = self._bundle_dir(workflow_id) if bundle is None: return {"promoted": False, "message": f"Unknown workflow {workflow_id}"} - # Teach is a response to the latest execution state, not a search for - # any historical halt. Selecting an older halt after a newer VERIFIED, - # FAILED, or otherwise terminal run would promote evidence against a - # stale application state. + run, error = self._teachable_run( + workflow_id, + expected_run_id=str(params.get("run_id") or ""), + ) + if run is None: + return {"promoted": False, "message": error or "No halted run to teach against"} + try: + fix = self._teach_correction_path(params, workflow_id=workflow_id, run=run) + except ValueError as exc: + return {"promoted": False, "message": str(exc)} + out_dir = self.config.data_dir / "bundles" / f"{workflow_id}_taught_{uuid.uuid4().hex[:6]}" + try: + result = self.services.flow_bridge.teach( + Path(run["run_path"]), + bundle, + out_dir, + fix=fix, + ) + except Exception as exc: + return {"promoted": False, "message": str(exc)} + message = "Fix promoted." if result.ok else (result.stderr or "Teach did not promote.") + return {"promoted": result.ok, "message": message} + + def _teachable_run( + self, + workflow_id: str, + *, + expected_run_id: str, + ) -> tuple[dict | None, str | None]: + """Return the exact latest halted run or a safe refusal reason.""" + + if not workflow_id: + return None, "A workflow is required for Teach." + if not expected_run_id: + return None, "The halted run is required for Teach." + # Teach responds to the latest execution state. It must not search for + # an older halt after a newer VERIFIED, FAILED, or other terminal run. run = next( - (r for r in self.services.db.list_runs(limit=100) if r.get("bundle_id") == workflow_id), + ( + item + for item in self.services.db.list_runs(limit=100) + if item.get("bundle_id") == workflow_id + ), None, ) - pending = self._pending_run(str(workflow_id)) + pending = self._pending_run(workflow_id) pending_matches_saved_run = bool( pending is not None and run is not None and pending[1].get("run_id") == run.get("run_id") ) if pending_matches_saved_run or self._pending_run_is_newer(pending, run): - return { - "promoted": False, - "message": ( - "The latest run is not saved in local history. Retry the " - "local history save before teaching a fix." - ), - } + return ( + None, + "The latest run is not saved in local history. Retry the " + "local history save before teaching a fix.", + ) if run is None: - return {"promoted": False, "message": "No halted run to teach against"} + return None, "No halted run to teach against" + if str(run.get("run_id") or "") != expected_run_id: + return None, "The selected halt is stale. Open the latest run before teaching." if run.get("status") not in _TEACHABLE_RUN_OUTCOMES: status = str(run.get("status") or "unknown") - return { - "promoted": False, - "message": f"The latest run ended as {status}, so it is not teachable.", - } + return None, f"The latest run ended as {status}, so it is not teachable." if not run.get("run_path"): - return { - "promoted": False, - "message": "The latest halted run has no evidence path", - } - out_dir = self.config.data_dir / "bundles" / f"{workflow_id}_taught_{uuid.uuid4().hex[:6]}" + return None, "The latest halted run has no evidence path" + return run, None + + def _stopped_teach_correction(self, stopped: dict) -> dict | None: + """Expose the exact local correction artifact when Teach recorded it.""" + + capture_id = str(stopped.get("capture_id") or "") + if not capture_id: + return None + capture = self.services.db.get_capture(capture_id) + if not capture or capture.get("task_description") != "teach_fix": + return None + path, binding = self._validated_recorded_correction(capture_id) + return { + "schema": _TEACH_CORRECTION_SCHEMA, + "capture_id": capture_id, + "fix_path": str(path), + "workflow_id": binding["workflow_id"], + "run_id": binding["run_id"], + } + + def _validated_recorded_correction(self, capture_id: str) -> tuple[Path, dict]: + """Resolve one completed correction capture and validate its run binding.""" + + capture = self.services.db.get_capture(capture_id) + if not capture or capture.get("task_description") != "teach_fix": + raise ValueError("The selected artifact is not a Teach correction recording.") + if not capture.get("stopped_at"): + raise ValueError("Stop the correction recording before you submit it.") try: - result = self.services.flow_bridge.teach(Path(run["run_path"]), bundle, out_dir) - except Exception as exc: - return {"promoted": False, "message": str(exc)} - message = "Fix promoted." if result.ok else (result.stderr or "Teach did not promote.") - return {"promoted": result.ok, "message": message} + path = self._presentation_capture_dir({"capture_id": capture_id}) + except (OSError, ValueError) as exc: + raise ValueError("The correction recording is not available.") from exc + meta_path = path / "meta.json" + if not meta_path.is_file() or meta_path.is_symlink(): + raise ValueError("The correction recording has no trusted metadata.") + try: + meta = json.loads(meta_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError("The correction recording metadata is invalid.") from exc + binding = meta.get(_TEACH_CORRECTION_META_KEY) if isinstance(meta, dict) else None + if ( + not isinstance(binding, dict) + or set(binding) != {"schema", "workflow_id", "run_id"} + or binding.get("schema") != _TEACH_CORRECTION_SCHEMA + or not isinstance(binding.get("workflow_id"), str) + or not isinstance(binding.get("run_id"), str) + ): + raise ValueError("The correction recording is not bound to a halted run.") + return path, binding + + def _teach_correction_path(self, params: dict, *, workflow_id: str, run: dict) -> Path: + """Return the explicit fix artifact for the exact current halted run.""" + + mode = str(params.get("mode") or "record") + if mode == "describe": + note = str(params.get("note") or "").strip() + if not note: + raise ValueError("Describe the corrective action before you submit it.") + binding_key = hashlib.sha256( + f"{workflow_id}\0{run['run_id']}".encode("utf-8") + ).hexdigest()[:16] + correction_dir = self.config.data_dir / "corrections" / binding_key + correction_dir.mkdir(parents=True, exist_ok=True, mode=0o700) + fix = correction_dir / f"{uuid.uuid4().hex}.json" + fix.write_text( + json.dumps({"resolution_steps": [{"intent": note}]}, indent=2), + encoding="utf-8", + ) + fix.chmod(0o600) + return fix + if mode != "record": + raise ValueError("The Teach correction mode is not supported.") + capture_id = str(params.get("fix_capture_id") or "") + supplied_path = str(params.get("fix_path") or "") + if not capture_id or not supplied_path: + raise ValueError("Record and stop the corrective action before you submit it.") + fix, binding = self._validated_recorded_correction(capture_id) + if Path(supplied_path).resolve() != fix: + raise ValueError("The correction artifact path does not match the recorded capture.") + if binding["workflow_id"] != workflow_id or binding["run_id"] != str( + run.get("run_id") or "" + ): + raise ValueError("The correction recording belongs to a different halted run.") + return fix def _bundle_dir(self, bundle_id: str | None) -> Path | None: if not bundle_id: diff --git a/src/screens/Teach.tsx b/src/screens/Teach.tsx index de18e5b..3582ede 100644 --- a/src/screens/Teach.tsx +++ b/src/screens/Teach.tsx @@ -7,6 +7,19 @@ import { CMD, engineInvoke, engineTry } from "../lib/engine"; import type { RunPersistenceRetryResponse, RunReport } from "../lib/types"; import { Button, Card, CardHead, Callout, Field } from "../ui/primitives"; +interface TeachCorrection { + schema: "openadapt.desktop-teach-correction/v1"; + capture_id: string; + fix_path: string; + workflow_id: string; + run_id: string; +} + +interface TeachRecordingResult { + capture_id?: string | null; + correction?: TeachCorrection; +} + export function Teach({ workflowId, onDone, @@ -20,6 +33,7 @@ export function Teach({ const [phase, setPhase] = useState<"idle" | "recording" | "teaching">("idle"); const [result, setResult] = useState(null); const [retryingPersistence, setRetryingPersistence] = useState(false); + const [correction, setCorrection] = useState(null); useEffect(() => { engineTry( @@ -30,28 +44,61 @@ export function Teach({ }, [workflowId]); async function recordFix() { + if (!report?.run_id) { + setResult("Open the halted run before you record a correction."); + return; + } setPhase("recording"); + setResult(null); + setCorrection(null); try { - await engineInvoke(CMD.START_RECORDING, { purpose: "teach_fix" }); - } catch { + await engineInvoke(CMD.START_RECORDING, { + purpose: "teach_fix", + workflow_id: workflowId, + run_id: report.run_id, + }); + } catch (error) { + setResult(String(error)); setPhase("idle"); } } async function stopFix() { try { - await engineInvoke(CMD.STOP_RECORDING, {}); + const stopped = await engineInvoke( + CMD.STOP_RECORDING, + {}, + ); + if (stopped.correction) { + setCorrection(stopped.correction); + setResult("Correction recorded. Submit it when you are ready."); + } else { + setResult("Desktop could not bind this recording to the halted run."); + } + } catch (error) { + setResult(String(error)); } finally { setPhase("idle"); } } async function teach() { + if (!report?.run_id) { + setResult("Open the halted run before you submit a correction."); + return; + } setPhase("teaching"); setResult(null); try { const r = await engineInvoke<{ promoted?: boolean; message?: string }>( CMD.TEACH_FIX, - { workflow_id: workflowId, note, mode }, + { + workflow_id: workflowId, + run_id: report.run_id, + note, + mode, + fix_capture_id: correction?.capture_id, + fix_path: correction?.fix_path, + }, ); setResult( r?.promoted @@ -176,7 +223,11 @@ export function Teach({