feat: guide qualification and recover local run history

engine/db.py

@@ -332,13 +332,26 @@ def update_bundle(self, bundle_id: str, **fields: object) -> None:
 
     # --- Run operations (local replay/run executions) ---
 
-    def insert_run(self, run_id: str, run_path: str, *, bundle_id: str | None = None) -> None:
-        """Record a local replay/run execution."""
-        self.conn.execute(
-            "INSERT INTO runs (run_id, bundle_id, run_path, created_at) VALUES (?, ?, ?, ?)",
-            (run_id, bundle_id, run_path, _now()),
-        )
-        self.conn.commit()
+    def insert_run(
+        self,
+        run_id: str,
+        run_path: str,
+        *,
+        bundle_id: str | None = None,
+        status: str = "pending",
+    ) -> None:
+        """Record a local execution and its known outcome atomically."""
+        try:
+            self.conn.execute(
+                "INSERT INTO runs "
+                "(run_id, bundle_id, run_path, status, created_at) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (run_id, bundle_id, run_path, status, _now()),
+            )
+            self.conn.commit()
+        except Exception:
+            self.conn.rollback()
+            raise
 
     def get_run(self, run_id: str) -> dict | None:
         """Get a single run by ID."""

engine/dispatch.py

@@ -45,6 +45,20 @@
 
 EmitFn = Callable[[str, dict], None]
 
+_RUN_PERSISTENCE_MARKER = ".desktop-run-persistence.json"
+_KNOWN_RUN_OUTCOMES = frozenset(
+    {
+        "VERIFIED",
+        "COMPLETED_UNVERIFIED",
+        "HALTED",
+        "FAILED",
+        "ROLLED_BACK",
+        "success",
+        "halt",
+        "unknown",
+    }
+)
+
 
 def _noop_emit(event: str, data: dict) -> None:
     """Default event sink -- drops events when no emitter is wired."""
@@ -211,6 +225,7 @@ def _register(self) -> None:
             "replay_workflow": self.replay_workflow,
             "run_workflow": self.run_workflow,
             "get_run_report": self.get_run_report,
+            "retry_run_persistence": self.retry_run_persistence,
             "teach_fix": self.teach_fix,
             # qualification cockpit (canonical Flow graph/policy/manifests)
             "get_qualification": self.get_qualification,
@@ -923,18 +938,20 @@ def _replay_or_run(self, params: dict, *, run: bool) -> dict:
                     "an action."
                 )
 
-        try:
-            self.services.db.insert_run(run_id, str(run_dir), bundle_id=workflow_id)
-            self.services.db.update_run(run_id, status=outcome)
-        except Exception:
-            pass
+        persistence = self._persist_local_run(
+            run_id=run_id,
+            run_dir=run_dir,
+            workflow_id=str(workflow_id),
+            outcome=outcome,
+        )
         report = self._run_report(
             run_dir,
             workflow_id,
             run_id,
             outcome=outcome,
             error=error,
         )
+        report["persistence"] = persistence
         progress_state = {
             "VERIFIED": "done",
             "COMPLETED_UNVERIFIED": "completed_unverified",
@@ -982,6 +999,190 @@ def _replay_or_run(self, params: dict, *, run: bool) -> dict:
         )
         return report
 
+    @staticmethod
+    def _persistence_marker_path(run_dir: Path) -> Path:
+        return run_dir / _RUN_PERSISTENCE_MARKER
+
+    def _persist_local_run(
+        self,
+        *,
+        run_id: str,
+        run_dir: Path,
+        workflow_id: str,
+        outcome: str,
+    ) -> dict:
+        """Persist one local run or leave a bounded recovery marker."""
+
+        marker = self._persistence_marker_path(run_dir)
+        payload = {
+            "schema": "openadapt.desktop-run-persistence/v1",
+            "run_id": run_id,
+            "workflow_id": workflow_id,
+            "outcome": outcome,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+        }
+        staging = marker.with_name(f"{marker.name}.{uuid.uuid4().hex}.tmp")
+        try:
+            staging.write_text(json.dumps(payload, sort_keys=True))
+            staging.replace(marker)
+        except Exception:
+            staging.unlink(missing_ok=True)
+            return {
+                "state": "failed",
+                "retryable": False,
+                "message": (
+                    "The run report is available for this session, but Desktop "
+                    "could not create a local history recovery record. Preserve "
+                    "the run evidence before closing Desktop."
+                ),
+            }
+
+        try:
+            self.services.db.insert_run(
+                run_id,
+                str(run_dir),
+                bundle_id=workflow_id,
+                status=outcome,
+            )
+        except Exception:
+            return self._degraded_run_persistence()
+
+        try:
+            marker.unlink(missing_ok=True)
+        except OSError:
+            logger.warning("Could not remove reconciled run persistence marker")
+        return {
+            "state": "persisted",
+            "retryable": False,
+            "message": "The report is saved in local history.",
+        }
+
+    @staticmethod
+    def _degraded_run_persistence() -> dict:
+        return {
+            "state": "degraded",
+            "retryable": True,
+            "message": (
+                "The report remains available, but Desktop could not add this run "
+                "to local history. History and Teach will not use it until the "
+                "local history save succeeds."
+            ),
+        }
+
+    def _pending_run(
+        self,
+        workflow_id: str,
+        *,
+        run_id: str | None = None,
+    ) -> tuple[Path, dict] | None:
+        """Find the newest valid local-history recovery marker."""
+
+        runs_root = self.config.data_dir / "runs"
+        if not runs_root.is_dir():
+            return None
+        matches: list[tuple[float, Path, dict]] = []
+        for marker in runs_root.glob(f"*/{_RUN_PERSISTENCE_MARKER}"):
+            if marker.parent.is_symlink():
+                continue
+            try:
+                payload = json.loads(marker.read_text())
+                valid = (
+                    payload.get("schema") == "openadapt.desktop-run-persistence/v1"
+                    and payload.get("workflow_id") == workflow_id
+                    and payload.get("outcome") in _KNOWN_RUN_OUTCOMES
+                    and isinstance(payload.get("run_id"), str)
+                    and isinstance(payload.get("created_at"), str)
+                    and (run_id is None or payload.get("run_id") == run_id)
+                )
+                if valid:
+                    matches.append((marker.stat().st_mtime, marker.parent, payload))
+            except (OSError, ValueError, TypeError):
+                continue
+        if not matches:
+            return None
+        _mtime, run_dir, payload = max(matches, key=lambda item: item[0])
+        return run_dir, payload
+
+    @staticmethod
+    def _pending_run_is_newer(pending: tuple[Path, dict] | None, run: dict | None) -> bool:
+        if pending is None:
+            return False
+        if run is None:
+            return True
+        return str(pending[1].get("created_at") or "") >= str(
+            run.get("created_at") or ""
+        )
+
+    def retry_run_persistence(self, **params: Any) -> dict:
+        """Retry a failed local-history save from its bounded recovery marker."""
+
+        workflow_id = str(params.get("workflow_id") or "")
+        run_id = str(params.get("run_id") or "")
+        if not workflow_id or not run_id:
+            return {"ok": False, "error": "workflow_id and run_id are required"}
+
+        existing = self.services.db.get_run(run_id)
+        if existing is not None:
+            if existing.get("bundle_id") != workflow_id:
+                return {"ok": False, "error": "The saved run belongs to another workflow"}
+            if existing.get("status") in _KNOWN_RUN_OUTCOMES:
+                report = self._run_report(
+                    Path(str(existing["run_path"])),
+                    workflow_id,
+                    run_id,
+                    outcome=str(existing["status"]),
+                )
+                report["persistence"] = {
+                    "state": "persisted",
+                    "retryable": False,
+                    "message": "The report is saved in local history.",
+                }
+                return {"ok": True, "report": report}
+
+        pending = self._pending_run(workflow_id, run_id=run_id)
+        if pending is None:
+            return {
+                "ok": False,
+                "error": "No retryable local history record was found for this run",
+            }
+        run_dir, payload = pending
+        try:
+            if existing is None:
+                self.services.db.insert_run(
+                    run_id,
+                    str(run_dir),
+                    bundle_id=workflow_id,
+                    status=str(payload["outcome"]),
+                )
+            else:
+                if Path(str(existing.get("run_path") or "")) != run_dir:
+                    return {
+                        "ok": False,
+                        "error": "The recovery record does not match the saved run path",
+                    }
+                self.services.db.update_run(run_id, status=str(payload["outcome"]))
+        except Exception:
+            return {
+                "ok": False,
+                "error": "Local history is still unavailable. Fix local storage and retry.",
+            }
+        try:
+            self._persistence_marker_path(run_dir).unlink(missing_ok=True)
+        except OSError:
+            logger.warning("Could not remove reconciled run persistence marker")
+        report = self._run_report(
+            run_dir,
+            workflow_id,
+            run_id,
+            outcome=str(payload["outcome"]),
+        )
+        report["persistence"] = {
+            "state": "persisted",
+            "retryable": False,
+            "message": "The report is saved in local history.",
+        }
+        return {"ok": True, "report": report}
+
     @staticmethod
     def _pre_action_refusal(error: str) -> dict:
         """Return the only response that proves Flow was never invoked."""
@@ -1038,34 +1239,69 @@ def _execution_target(self, params: dict) -> tuple[Any | None, Path | None]:
 
     def get_run_report(self, **params: Any) -> dict | None:
         """Return the latest ``RunReport`` for a workflow, or None if none."""
-        workflow_id = params.get("workflow_id")
+        workflow_id = str(params.get("workflow_id") or "")
         runs = [
             r for r in self.services.db.list_runs(limit=100) if r.get("bundle_id") == workflow_id
         ]
+        pending = self._pending_run(workflow_id)
+        pending_is_newest = self._pending_run_is_newer(
+            pending,
+            runs[0] if runs else None,
+        )
+        if pending_is_newest:
+            assert pending is not None
+            run_dir, payload = pending
+            report = self._run_report(
+                run_dir,
+                workflow_id,
+                str(payload["run_id"]),
+                outcome=str(payload["outcome"]),
+            )
+            report["persistence"] = self._degraded_run_persistence()
+            return report
         if not runs:
             return None
         run = runs[0]
         run_dir = run.get("run_path")
         if not run_dir:
             return None
         stored_outcome = run.get("status")
-        known_outcomes = {
-            "VERIFIED",
-            "COMPLETED_UNVERIFIED",
-            "HALTED",
-            "FAILED",
-            "ROLLED_BACK",
-            "success",
-            "halt",
-            "unknown",
-        }
-        outcome = stored_outcome if stored_outcome in known_outcomes else None
-        return self._run_report(
+        if stored_outcome not in _KNOWN_RUN_OUTCOMES:
+            pending = self._pending_run(workflow_id, run_id=str(run.get("run_id") or ""))
+            if pending is not None:
+                pending_dir, payload = pending
+                report = self._run_report(
+                    pending_dir,
+                    workflow_id,
+                    str(payload["run_id"]),
+                    outcome=str(payload["outcome"]),
+                )
+                report["persistence"] = self._degraded_run_persistence()
+                return report
+        outcome = stored_outcome if stored_outcome in _KNOWN_RUN_OUTCOMES else None
+        report = self._run_report(
             Path(run_dir),
             workflow_id,
             run.get("run_id", ""),
             outcome=outcome,
         )
+        report["persistence"] = (
+            {
+                "state": "persisted",
+                "retryable": False,
+                "message": "The report is saved in local history.",
+            }
+            if outcome is not None
+            else {
+                "state": "failed",
+                "retryable": False,
+                "message": (
+                    "Desktop found the run evidence, but its local history outcome "
+                    "is incomplete and no recovery record is available."
+                ),
+            }
+        )
+        return report
 
     def _run_report(
         self,
@@ -1230,9 +1466,22 @@ def teach_fix(self, **params: Any) -> dict:
         if bundle is None:
             return {"promoted": False, "message": f"Unknown workflow {workflow_id}"}
         run = next(
-            (r for r in self.services.db.list_runs(limit=100) if r.get("bundle_id") == workflow_id),
+            (
+                r
+                for r in self.services.db.list_runs(limit=100)
+                if r.get("bundle_id") == workflow_id
+                and r.get("status") in _KNOWN_RUN_OUTCOMES
+            ),
             None,
         )
+        if self._pending_run_is_newer(self._pending_run(str(workflow_id)), run):
+            return {
+                "promoted": False,
+                "message": (
+                    "The latest run is not saved in local history. Retry the "
+                    "local history save before teaching a fix."
+                ),
+            }
         if run is None or not run.get("run_path"):
             return {"promoted": False, "message": "No halted run to teach against"}
         out_dir = self.config.data_dir / "bundles" / f"{workflow_id}_taught_{uuid.uuid4().hex[:6]}"

src/lib/engine.ts

@@ -32,6 +32,7 @@ export const CMD = {
   REPLAY_WORKFLOW: "replay_workflow",
   RUN_WORKFLOW: "run_workflow",
   GET_RUN_REPORT: "get_run_report",
+  RETRY_RUN_PERSISTENCE: "retry_run_persistence",
   TEACH_FIX: "teach_fix",
   GET_QUALIFICATION: "get_qualification",
   INITIALIZE_QUALIFICATION: "initialize_qualification",