feat(imports): recover an import whose background job was killed

wikify/api/imports.py

@@ -4,7 +4,9 @@
 
 import frappe
 
+from wikify import tasks
 from wikify.engine import preview_wiki as _preview_wiki
+from wikify.jobs._util import publish_progress
 from wikify.seed import seed_uncategorized_project
 
 #: Guard against a runaway drag-and-drop — one worker chews through these serially.
@@ -89,6 +91,43 @@ def trigger_remediation(import_name: str, scope: str = "flagged") -> str:
 	return import_name
 
 
+@frappe.whitelist()
+def retry_import(import_name: str) -> str:
+	"""Restart an Import that failed, or whose worker died mid-job.
+
+	Runs from `Failed`, or from a running status that has gone stale (no progress for
+	`tasks.STALE_MINUTES`) — a live job is never interrupted. The stale traceback is
+	cleared; an import with nothing parsed yet gets its parse job re-enqueued, and one
+	that already has a document is handed back to the stage before the one that died,
+	so remediation or wiki generation can be re-run.
+	"""
+	imp = frappe.get_doc("Wikify Import", import_name)
+	if imp.status not in ("Failed", *tasks.RUNNING_STATUSES):
+		frappe.throw(f"Nothing to retry — the import is in {imp.status}.")
+	if imp.status != "Failed" and not tasks.is_stale(imp):
+		frappe.throw(f"This import is still running ({imp.stage_label or imp.status}).")
+
+	# The transition is persisted before it is broadcast — a realtime hiccup must not
+	# leave the import stuck in the status it is being rescued from.
+	if not imp.source_document:
+		imp.db_set({"status": "Queued", "error": None})
+		publish_progress(import_name, 0, "Queued for retry", status="Queued")
+		frappe.enqueue(
+			"wikify.jobs.parse.run",
+			queue="long",
+			timeout=3600,
+			import_name=import_name,
+		)
+		return import_name
+
+	# The parse result (and any approved tree) is intact — hand the import back to the
+	# stage it was in before the lost job so the user can re-run it from the UI.
+	resume_status = "Graphed" if imp.status == "Generating Wiki" else "Review"
+	imp.db_set({"status": resume_status, "error": None})
+	publish_progress(import_name, 100, f"Ready to retry from {resume_status}", status=resume_status)
+	return import_name
+
+
 @frappe.whitelist()
 def reclassify(import_name: str) -> str:
 	"""Re-tag the doc's Source Sections after manual tree edits.

wikify/hooks.py

@@ -157,6 +157,13 @@
 # Scheduled Tasks
 # ---------------
 
+scheduler_events = {
+	"cron": {
+		# Frequent enough that a lost worker surfaces while the user is still watching.
+		"*/15 * * * *": ["wikify.tasks.fail_stuck_imports"],
+	},
+}
+
 # scheduler_events = {
 # 	"all": [
 # 		"wikify.tasks.all"

wikify/jobs/generate.py

@@ -23,6 +23,7 @@ def run(
 	imp = frappe.get_doc("Wikify Import", import_name)
 	try:
 		imp.db_set("status", "Generating Wiki")
+		imp.db_set("error", None)
 		publish_progress(import_name, 0, "Generating wiki", status="Generating Wiki")
 		log(import_name, "info", "generate", f"Generating wiki for {imp.import_title}")
 

wikify/jobs/parse.py

@@ -19,6 +19,7 @@ def run(import_name: str) -> None:
 	imp = frappe.get_doc("Wikify Import", import_name)
 	try:
 		imp.db_set("status", "Parsing")
+		imp.db_set("error", None)
 		imp.db_set("started_at", now_datetime())
 		publish_progress(import_name, 0, "Starting parse", status="Parsing")
 		log(import_name, "info", "parse", f"Starting parse of {imp.import_title}")

wikify/jobs/remediate.py

@@ -17,6 +17,7 @@ def run(import_name: str, scope: str = "flagged", instruction: str = "") -> None
 	imp = frappe.get_doc("Wikify Import", import_name)
 	try:
 		imp.db_set("status", "Remediating")
+		imp.db_set("error", None)
 		publish_progress(import_name, 0, f"Starting remediation ({scope})", status="Remediating")
 		log(import_name, "info", "remediate", f"Remediating {scope} pages of {imp.import_title}")
 

wikify/tasks.py

@@ -0,0 +1,49 @@
+"""Scheduled maintenance for the Imports flow."""
+
+from __future__ import annotations
+
+import frappe
+from frappe.utils import add_to_date, get_datetime, now_datetime
+
+from wikify.jobs._util import log, publish_progress
+
+#: Statuses that mean a background job owns the import right now.
+RUNNING_STATUSES = ("Queued", "Parsing", "Remediating", "Generating Wiki")
+
+#: A running import whose progress hasn't moved for this long has lost its worker.
+#: Jobs write progress every few seconds (a 400-page remediation runs for the better
+#: part of an hour but publishes on every page), so this keys off time-since-last-
+#: progress and never touches a legitimately slow pass.
+STALE_MINUTES = 30
+
+
+def stale_cutoff():
+	"""Progress older than this means the job is gone, not slow."""
+	return add_to_date(now_datetime(), minutes=-STALE_MINUTES)
+
+
+def is_stale(import_doc) -> bool:
+	"""Is this import sitting in a running status with no progress written since the cutoff?"""
+	return import_doc.status in RUNNING_STATUSES and get_datetime(import_doc.modified) < stale_cutoff()
+
+
+def fail_stuck_imports() -> None:
+	"""Fail imports whose worker died mid-job.
+
+	Nothing else moves an import out of a running status, so a killed worker (OOM,
+	deploy, hard timeout) leaves the UI showing an in-flight job forever with no way
+	back except editing `status` by hand.
+	"""
+	stuck = frappe.get_all(
+		"Wikify Import",
+		filters={"status": ("in", RUNNING_STATUSES), "modified": ("<", stale_cutoff())},
+		fields=["name", "status"],
+	)
+	for imp in stuck:
+		message = (
+			f"No progress for {STALE_MINUTES} minutes while {imp.status} — the worker running "
+			f"this import was lost. Retry the import to restart it."
+		)
+		frappe.db.set_value("Wikify Import", imp.name, {"status": "Failed", "error": message})
+		publish_progress(imp.name, 100, "Worker lost", status="Failed")
+		log(imp.name, "error", "scheduler", message)

wikify/tests/test_job_recovery.py

@@ -0,0 +1,175 @@
+# Copyright (c) 2026, BWH and contributors
+# For license information, please see license.txt
+
+"""Recovering an Import whose background job died: the stale error is cleared when a
+new run starts, `retry_import` restarts a failed or stuck import, and the scheduled
+`fail_stuck_imports` fails one whose progress stopped.
+
+The progress/log helpers are patched out throughout — they `frappe.db.commit()`, which
+would break the test rollback and leak rows into the site.
+"""
+
+import tempfile
+from pathlib import Path
+from unittest.mock import DEFAULT, patch
+
+import frappe
+from frappe.tests.utils import FrappeTestCase
+from frappe.utils import add_to_date, now_datetime
+from frappe.utils.file_manager import save_file
+
+from wikify import tasks
+from wikify.api import imports as imports_api
+from wikify.engine import store
+from wikify.jobs import generate as generate_job
+from wikify.jobs import remediate as remediate_job
+from wikify.tests.test_parse_pipeline import _make_sample_pdf
+
+OLD_ERROR = "Traceback (most recent call last):\n  RuntimeError: from a run in August"
+
+
+def _without_realtime(module):
+	helpers = {name: DEFAULT for name in ("publish_progress", "log") if hasattr(module, name)}
+	return patch.multiple(module, **helpers)
+
+
+class TestJobRecovery(FrappeTestCase):
+	def setUp(self):
+		self.project = frappe.get_doc(
+			{
+				"doctype": "Wikify Project",
+				"project_name": f"Job Recovery {frappe.generate_hash(length=6)}",
+			}
+		).insert(ignore_permissions=True)
+
+	def _import(self, status: str, error: str | None = OLD_ERROR, with_document: bool = False) -> str:
+		imp = frappe.get_doc(
+			{
+				"doctype": "Wikify Import",
+				"import_title": "Handbook",
+				"project": self.project.name,
+				"pdf": "/private/files/handbook.pdf",
+				"status": status,
+				"error": error,
+				"source_document": store.create_document("Handbook") if with_document else None,
+			}
+		).insert(ignore_permissions=True)
+		return imp.name
+
+	def _age(self, import_name: str, minutes: int) -> None:
+		frappe.db.set_value(
+			"Wikify Import",
+			import_name,
+			"modified",
+			add_to_date(now_datetime(), minutes=-minutes),
+			update_modified=False,
+		)
+
+	def _read(self, import_name: str) -> dict:
+		return frappe.db.get_value("Wikify Import", import_name, ["status", "error"], as_dict=True)
+
+	def test_remediate_job_clears_the_error_of_an_earlier_run(self):
+		import_name = self._import("Review", with_document=True)
+		path = Path(tempfile.mkdtemp()) / "handbook.pdf"
+		_make_sample_pdf(str(path))
+		pdf = save_file("handbook.pdf", path.read_bytes(), "Wikify Import", import_name, is_private=1)
+		frappe.db.set_value("Wikify Import", import_name, "pdf", pdf.file_url)
+		summary = {"targets": 1, "adopted": 1, "canonical_mean": 0.9, "sections": 2, "cost": 0}
+
+		with (
+			_without_realtime(remediate_job),
+			patch.object(remediate_job, "remediate_pdf", return_value=summary),
+		):
+			remediate_job.run(import_name, scope="flagged")
+
+		self.assertIsNone(self._read(import_name).error)
+
+	def test_generate_job_clears_the_error_of_an_earlier_run(self):
+		import_name = self._import("Graphed", with_document=True)
+		result = {"space": "Handbook", "space_route": "handbook", "pages": 2, "groups": 1}
+		result.update({"deleted": 0, "links": 0})
+
+		with (
+			_without_realtime(generate_job),
+			patch.object(generate_job, "generate_wiki", return_value=result),
+		):
+			generate_job.run(import_name, wiki_space="Handbook")
+
+		row = self._read(import_name)
+		self.assertIsNone(row.error)
+		self.assertEqual(row.status, "Completed")
+
+	def test_retry_hands_a_failed_import_back_to_review(self):
+		import_name = self._import("Failed", with_document=True)
+
+		with _without_realtime(imports_api), patch.object(frappe, "enqueue") as enqueue:
+			self.assertEqual(imports_api.retry_import(import_name), import_name)
+			enqueue.assert_not_called()
+
+		row = self._read(import_name)
+		self.assertEqual(row.status, "Review")
+		self.assertIsNone(row.error)
+
+	def test_retry_re_enqueues_the_parse_when_nothing_was_parsed(self):
+		import_name = self._import("Failed")
+
+		with _without_realtime(imports_api), patch.object(frappe, "enqueue") as enqueue:
+			imports_api.retry_import(import_name)
+
+		self.assertEqual(enqueue.call_args.args[0], "wikify.jobs.parse.run")
+		self.assertEqual(enqueue.call_args.kwargs["import_name"], import_name)
+		self.assertIsNone(self._read(import_name).error)
+
+	def test_retry_recovers_an_import_whose_worker_was_lost(self):
+		import_name = self._import("Remediating", with_document=True)
+		self._age(import_name, tasks.STALE_MINUTES + 5)
+
+		with _without_realtime(imports_api), patch.object(frappe, "enqueue"):
+			imports_api.retry_import(import_name)
+
+		self.assertEqual(self._read(import_name).status, "Review")
+
+	def test_retry_refuses_a_job_that_is_still_running(self):
+		import_name = self._import("Remediating", with_document=True)
+
+		with _without_realtime(imports_api), patch.object(frappe, "enqueue") as enqueue:
+			with self.assertRaises(frappe.ValidationError):
+				imports_api.retry_import(import_name)
+			enqueue.assert_not_called()
+
+		self.assertEqual(self._read(import_name).status, "Remediating")
+
+	def test_retry_refuses_a_completed_import(self):
+		import_name = self._import("Completed", error=None, with_document=True)
+
+		with _without_realtime(imports_api), self.assertRaises(frappe.ValidationError):
+			imports_api.retry_import(import_name)
+
+	def test_a_running_import_with_no_progress_is_failed(self):
+		import_name = self._import("Remediating", error=None, with_document=True)
+		self._age(import_name, tasks.STALE_MINUTES + 5)
+
+		with _without_realtime(tasks):
+			tasks.fail_stuck_imports()
+
+		row = self._read(import_name)
+		self.assertEqual(row.status, "Failed")
+		self.assertIn("worker running this import was lost", row.error)
+
+	def test_a_slow_job_still_reporting_progress_is_left_alone(self):
+		import_name = self._import("Remediating", error=None, with_document=True)
+		self._age(import_name, tasks.STALE_MINUTES - 5)
+
+		with _without_realtime(tasks):
+			tasks.fail_stuck_imports()
+
+		self.assertEqual(self._read(import_name).status, "Remediating")
+
+	def test_an_import_at_rest_is_left_alone(self):
+		import_name = self._import("Review", error=None, with_document=True)
+		self._age(import_name, tasks.STALE_MINUTES * 4)
+
+		with _without_realtime(tasks):
+			tasks.fail_stuck_imports()
+
+		self.assertEqual(self._read(import_name).status, "Review")